From ffc8a7e44e5dfd27abff444ac87f618983592f53 Mon Sep 17 00:00:00 2001 From: JanHuang <384099566@qq.com> Date: Thu, 29 Oct 2026 16:22:14 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20=E6=96=B0=E5=A2=9E=20FastD=20?= =?UTF-8?q?=E5=BE=AE=E6=9C=8D=E5=8A=A1=E9=9B=86=E6=88=90=E5=8C=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - etcd v3 HTTP gateway 客户端 - 服务注册、解析与负载均衡 - etcd 配置源 - 按服务名解析的 PSR-18 客户端 - W3C traceparent 传播 - 重试、熔断、进程内限流 - 单元测试与 CI --- .github/workflows/ci.yml | 31 +++ .gitignore | 3 + LICENSE | 21 +++ README.md | 168 ++++++++++++++++- composer.json | 50 +++++ phpunit.xml.dist | 12 ++ src/Config/ConfigSourceInterface.php | 13 ++ src/Config/EtcdConfigSource.php | 72 +++++++ src/Discovery/EtcdResolver.php | 77 ++++++++ src/Discovery/EtcdServiceRegistry.php | 70 +++++++ .../LoadBalancer/LoadBalancerInterface.php | 15 ++ src/Discovery/LoadBalancer/RandomBalancer.php | 19 ++ .../LoadBalancer/RoundRobinBalancer.php | 24 +++ .../LoadBalancer/WeightedRandomBalancer.php | 35 ++++ src/Discovery/ResolverInterface.php | 13 ++ src/Discovery/ServiceInstance.php | 78 ++++++++ src/Discovery/ServiceLease.php | 15 ++ src/Discovery/ServiceRegistryInterface.php | 14 ++ src/Etcd/EtcdClient.php | 178 ++++++++++++++++++ src/Etcd/EtcdKeyValue.php | 44 +++++ src/Exception/CircuitOpenException.php | 9 + src/Exception/DiscoveryException.php | 9 + src/Exception/EtcdException.php | 9 + src/Exception/IntegrationException.php | 11 ++ src/Exception/ServiceUnavailableException.php | 9 + src/Governance/CircuitBreaker.php | 69 +++++++ src/Governance/CircuitBreakerClient.php | 43 +++++ src/Governance/InMemoryRateLimiter.php | 54 ++++++ src/Governance/RateLimitMiddleware.php | 41 ++++ src/Governance/RateLimiterInterface.php | 10 + src/Governance/RetryClient.php | 53 ++++++ src/Governance/RetryPolicy.php | 53 ++++++ src/HttpClient/ServiceClient.php | 55 ++++++ src/Tracing/TraceClient.php | 30 +++ src/Tracing/TraceContext.php | 73 +++++++ src/Tracing/TracingMiddleware.php | 38 ++++ tests/Config/EtcdConfigSourceTest.php | 51 +++++ tests/Discovery/DiscoveryTest.php | 111 +++++++++++ tests/Etcd/EtcdClientTest.php | 64 +++++++ tests/Governance/GovernanceTest.php | 113 +++++++++++ tests/HttpClient/ServiceClientTest.php | 49 +++++ tests/Support/FakeClient.php | 40 ++++ tests/Support/FakeHandler.php | 26 +++ tests/Support/FakeResolver.php | 27 +++ tests/Tracing/TracingTest.php | 68 +++++++ 45 files changed, 2065 insertions(+), 2 deletions(-) create mode 100644 .github/workflows/ci.yml create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 composer.json create mode 100644 phpunit.xml.dist create mode 100644 src/Config/ConfigSourceInterface.php create mode 100644 src/Config/EtcdConfigSource.php create mode 100644 src/Discovery/EtcdResolver.php create mode 100644 src/Discovery/EtcdServiceRegistry.php create mode 100644 src/Discovery/LoadBalancer/LoadBalancerInterface.php create mode 100644 src/Discovery/LoadBalancer/RandomBalancer.php create mode 100644 src/Discovery/LoadBalancer/RoundRobinBalancer.php create mode 100644 src/Discovery/LoadBalancer/WeightedRandomBalancer.php create mode 100644 src/Discovery/ResolverInterface.php create mode 100644 src/Discovery/ServiceInstance.php create mode 100644 src/Discovery/ServiceLease.php create mode 100644 src/Discovery/ServiceRegistryInterface.php create mode 100644 src/Etcd/EtcdClient.php create mode 100644 src/Etcd/EtcdKeyValue.php create mode 100644 src/Exception/CircuitOpenException.php create mode 100644 src/Exception/DiscoveryException.php create mode 100644 src/Exception/EtcdException.php create mode 100644 src/Exception/IntegrationException.php create mode 100644 src/Exception/ServiceUnavailableException.php create mode 100644 src/Governance/CircuitBreaker.php create mode 100644 src/Governance/CircuitBreakerClient.php create mode 100644 src/Governance/InMemoryRateLimiter.php create mode 100644 src/Governance/RateLimitMiddleware.php create mode 100644 src/Governance/RateLimiterInterface.php create mode 100644 src/Governance/RetryClient.php create mode 100644 src/Governance/RetryPolicy.php create mode 100644 src/HttpClient/ServiceClient.php create mode 100644 src/Tracing/TraceClient.php create mode 100644 src/Tracing/TraceContext.php create mode 100644 src/Tracing/TracingMiddleware.php create mode 100644 tests/Config/EtcdConfigSourceTest.php create mode 100644 tests/Discovery/DiscoveryTest.php create mode 100644 tests/Etcd/EtcdClientTest.php create mode 100644 tests/Governance/GovernanceTest.php create mode 100644 tests/HttpClient/ServiceClientTest.php create mode 100644 tests/Support/FakeClient.php create mode 100644 tests/Support/FakeHandler.php create mode 100644 tests/Support/FakeResolver.php create mode 100644 tests/Tracing/TracingTest.php diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..9371961 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,31 @@ +name: CI + +on: + push: + branches: [master] + pull_request: + +permissions: + contents: read + +jobs: + test: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + php: ['8.4', '8.5'] + steps: + - uses: actions/checkout@v4 + - name: Setup PHP + uses: shivammathur/setup-php@v2 + with: + php-version: ${{ matrix.php }} + coverage: none + tools: composer:v2 + - name: Validate composer.json + run: composer validate --no-check-publish --no-check-lock + - name: Install dependencies + run: composer update --prefer-dist --no-progress --no-interaction + - name: Run tests + run: vendor/bin/phpunit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..6ede8ef --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +/vendor/ +/.phpunit.cache/ +/composer.lock diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..940595e --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) FastD + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index ff700ed..c16d349 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,166 @@ -# integrations -Optional FastD microservice integrations: etcd discovery/config, service clients, tracing and governance. +# fastd/integrations + +FastD 的可选微服务集成包。 + +定位是 **集成层**，不是微服务治理平台： + +- FastD 负责适配、装配、中间件和客户端包装 +- etcd、OpenTelemetry、限流/熔断算法仍由外部实现 +- 不把注册中心、RPC、网关或分布式事务塞进 FastD 核心 + +## 模块 + +| 模块 | 作用 | +|---|---| +| `Etcd` | etcd v3 HTTP gateway 客户端 | +| `Discovery` | 服务注册、服务解析、实例负载均衡 | +| `Config` | 从 etcd 读取配置中心键值 | +| `HttpClient` | 按服务名解析实例的 PSR-18 客户端 | +| `Tracing` | W3C `traceparent` 提取、生成与下游传播 | +| `Governance` | 重试、熔断、令牌桶限流 | + +## 安装 + +```bash +composer require fastd/integrations +``` + +本包只依赖 PSR 接口。`fastd/http` 仅作为开发测试依赖，用于构造 PSR-7 请求/响应。 + +## 配置示例 + +```php +use FastD\Http\Factory; +use FastD\Integrations\Discovery\EtcdResolver; +use FastD\Integrations\Discovery\LoadBalancer\RoundRobinBalancer; +use FastD\Integrations\Etcd\EtcdClient; +use FastD\Integrations\HttpClient\ServiceClient; + +$factory = new Factory(); + +$etcd = new EtcdClient( + httpClient: $psr18Client, + requestFactory: $factory, + streamFactory: $factory, + baseUri: 'http://127.0.0.1:2379', +); + +$resolver = new EtcdResolver($etcd, prefix: '/fastd/services'); +$client = new ServiceClient($psr18Client, $resolver, new RoundRobinBalancer()); + +// 请求 URI 的 host 作为服务名 +$request = $factory->createRequest('GET', 'http://user-service/users/1'); +$response = $client->sendRequest($request); +``` + +## 服务注册 + +```php +use FastD\Integrations\Discovery\EtcdServiceRegistry; +use FastD\Integrations\Discovery\ServiceInstance; + +$registry = new EtcdServiceRegistry($etcd, prefix: '/fastd/services', ttl: 15); + +$lease = $registry->register(new ServiceInstance( + name: 'user-service', + host: '10.0.0.12', + port: 9527, + metadata: ['version' => '1.0.0'], +)); + +// 定时续约 +$registry->heartbeat($lease); + +// 进程退出 +$registry->deregister($lease); +``` + +etcd key 结构： + +```text +/fastd/services/{service}/{instanceId} +``` + +value 为 JSON： + +```json +{"host":"10.0.0.12","port":9527,"weight":100,"metadata":{"version":"1.0.0"}} +``` + +## 配置中心 + +```php +use FastD\Integrations\Config\EtcdConfigSource; + +$source = new EtcdConfigSource($etcd, prefix: '/fastd/config'); +$config = $source->load('user-service'); +``` + +etcd key： + +```text +/fastd/config/user-service/database/host +``` + +会转换成： + +```php +['database' => ['host' => '...']] +``` + +## 链路传播 + +在 FastD 中把中间件挂到全局中间件栈： + +```php +use FastD\Integrations\Tracing\TracingMiddleware; + +$container->get('matcher')->push(new TracingMiddleware()); +``` + +入口读取或生成 `traceparent`，写入请求属性 `fastd.trace_context`，并返回 `X-Trace-Id`。下游调用使用： + +```php +use FastD\Integrations\Tracing\TraceClient; + +$client = new TraceClient($psr18Client, $traceContext); +``` + +`TraceClient` 只负责 W3C 上下文传播，不导出 Span。Span 采集/导出请接入 OpenTelemetry SDK。 + +## 治理 + +```php +use FastD\Integrations\Governance\CircuitBreaker; +use FastD\Integrations\Governance\CircuitBreakerClient; +use FastD\Integrations\Governance\RetryClient; +use FastD\Integrations\Governance\RetryPolicy; + +$client = new RetryClient( + new CircuitBreakerClient($psr18Client, new CircuitBreaker()), + new RetryPolicy(maxAttempts: 3), +); +``` + +- 重试只对幂等方法生效 +- 熔断和限流状态默认进程内；多实例共享需替换存储实现 +- 生产环境需设置超时，避免重试放大故障 + +## 边界 + +本包不提供： + +- 服务注册中心本身 +- gRPC/Thrift 协议 +- 分布式事务 +- 完整 OpenTelemetry SDK +- 全局限流存储 + +这些应由 etcd、OpenTelemetry、网关或独立库负责。 + +## 测试 + +```bash +composer install +vendor/bin/phpunit +``` diff --git a/composer.json b/composer.json new file mode 100644 index 0000000..9ca8fcf --- /dev/null +++ b/composer.json @@ -0,0 +1,50 @@ +{ + "name": "fastd/integrations", + "type": "library", + "license": "MIT", + "description": "Optional microservice integrations for FastD: etcd discovery/config, service clients, W3C tracing and client-side governance.", + "keywords": [ + "fastd", + "microservice", + "etcd", + "service-discovery", + "tracing", + "circuit-breaker", + "rate-limit" + ], + "homepage": "https://github.com/fastdlabs/integrations", + "authors": [ + { + "name": "janhuang", + "email": "bboyjanhuang@gmail.com" + } + ], + "require": { + "php": ">=8.4.1", + "psr/http-client": "^1.0", + "psr/http-factory": "^1.0", + "psr/http-message": "^2.0", + "psr/http-server-handler": "^1.0", + "psr/http-server-middleware": "^1.0" + }, + "require-dev": { + "fastd/http": "^8.2", + "phpunit/phpunit": "^13.4" + }, + "autoload": { + "psr-4": { + "FastD\\Integrations\\": "src/" + } + }, + "autoload-dev": { + "psr-4": { + "FastD\\Integrations\\Tests\\": "tests/" + } + }, + "scripts": { + "test": "phpunit" + }, + "config": { + "sort-packages": true + } +} diff --git a/phpunit.xml.dist b/phpunit.xml.dist new file mode 100644 index 0000000..d0d7053 --- /dev/null +++ b/phpunit.xml.dist @@ -0,0 +1,12 @@ + + + + + tests + + + diff --git a/src/Config/ConfigSourceInterface.php b/src/Config/ConfigSourceInterface.php new file mode 100644 index 0000000..9c736b6 --- /dev/null +++ b/src/Config/ConfigSourceInterface.php @@ -0,0 +1,13 @@ + + */ + public function load(string $service): array; +} diff --git a/src/Config/EtcdConfigSource.php b/src/Config/EtcdConfigSource.php new file mode 100644 index 0000000..ae8a0b4 --- /dev/null +++ b/src/Config/EtcdConfigSource.php @@ -0,0 +1,72 @@ + ['host' => '...']] + */ +final class EtcdConfigSource implements ConfigSourceInterface +{ + public function __construct( + private readonly EtcdClient $client, + private readonly string $prefix = '/fastd/config', + ) { + } + + public function load(string $service): array + { + $prefix = sprintf( + '%s/%s/', + trim($this->prefix, '/'), + trim($service, '/'), + ); + + $config = []; + + foreach ($this->client->getPrefix($prefix) as $keyValue) { + $path = substr($keyValue->key, strlen($prefix)); + + if ($path === false || $path === '') { + continue; + } + + $this->assign($config, explode('/', trim($path, '/')), $keyValue->value); + } + + return $config; + } + + /** + * @param array $config + * @param string[] $path + */ + private function assign(array &$config, array $path, string $value): void + { + $cursor = &$config; + + foreach ($path as $index => $segment) { + if ($segment === '') { + continue; + } + + if ($index === array_key_last($path)) { + $cursor[$segment] = $value; + break; + } + + if (!isset($cursor[$segment]) || !is_array($cursor[$segment])) { + $cursor[$segment] = []; + } + + $cursor = &$cursor[$segment]; + } + } +} diff --git a/src/Discovery/EtcdResolver.php b/src/Discovery/EtcdResolver.php new file mode 100644 index 0000000..8cd0eb7 --- /dev/null +++ b/src/Discovery/EtcdResolver.php @@ -0,0 +1,77 @@ + */ + private array $cache = []; + + public function __construct( + private readonly EtcdClient $client, + private readonly string $prefix = '/fastd/services', + private readonly int $cacheTtl = 5, + private readonly ?\Closure $clock = null, + ) { + } + + public function resolve(string $service): array + { + $now = $this->now(); + $cached = $this->cache[$service] ?? null; + + if ($cached !== null && $cached['expires'] > $now) { + return $cached['instances']; + } + + $instances = $this->fetch($service); + $this->cache[$service] = [ + 'expires' => $now + max(0, $this->cacheTtl), + 'instances' => $instances, + ]; + + return $instances; + } + + public function refresh(string $service): array + { + unset($this->cache[$service]); + + return $this->resolve($service); + } + + /** + * @return ServiceInstance[] + */ + private function fetch(string $service): array + { + $prefix = sprintf( + '%s/%s/', + trim($this->prefix, '/'), + trim($service, '/'), + ); + + $instances = []; + + foreach ($this->client->getPrefix($prefix) as $keyValue) { + $id = basename($keyValue->key); + + try { + $instances[] = ServiceInstance::fromJson($service, $keyValue->value, $id); + } catch (\Throwable) { + continue; + } + } + + return $instances; + } + + private function now(): int + { + return $this->clock !== null ? (int) ($this->clock)() : time(); + } +} diff --git a/src/Discovery/EtcdServiceRegistry.php b/src/Discovery/EtcdServiceRegistry.php new file mode 100644 index 0000000..c455f82 --- /dev/null +++ b/src/Discovery/EtcdServiceRegistry.php @@ -0,0 +1,70 @@ +id !== '' + ? $instance->id + : bin2hex(random_bytes(8)); + + $registered = new ServiceInstance( + name: $instance->name, + host: $instance->host, + port: $instance->port, + weight: $instance->weight, + metadata: $instance->metadata, + id: $id, + ); + + $leaseId = $this->client->grantLease($this->ttl); + + if ($leaseId <= 0) { + throw new \RuntimeException('Unable to grant an etcd lease.'); + } + + $key = $this->key($instance->name, $id); + $this->client->put($key, $registered->toJson(), $leaseId); + + return new ServiceLease($registered, $key, $leaseId); + } + + public function deregister(ServiceLease $lease): void + { + $this->client->delete($lease->key); + + if ($lease->leaseId > 0) { + $this->client->revokeLease($lease->leaseId); + } + } + + public function heartbeat(ServiceLease $lease): void + { + if ($lease->leaseId > 0) { + $this->client->keepAlive($lease->leaseId); + } + } + + private function key(string $service, string $id): string + { + return sprintf( + '/%s/%s/%s', + trim($this->prefix, '/'), + trim($service, '/'), + $id, + ); + } +} diff --git a/src/Discovery/LoadBalancer/LoadBalancerInterface.php b/src/Discovery/LoadBalancer/LoadBalancerInterface.php new file mode 100644 index 0000000..fce9dd1 --- /dev/null +++ b/src/Discovery/LoadBalancer/LoadBalancerInterface.php @@ -0,0 +1,15 @@ +counter % count($instances); + $this->counter++; + + return $instances[$index]; + } +} diff --git a/src/Discovery/LoadBalancer/WeightedRandomBalancer.php b/src/Discovery/LoadBalancer/WeightedRandomBalancer.php new file mode 100644 index 0000000..4d44445 --- /dev/null +++ b/src/Discovery/LoadBalancer/WeightedRandomBalancer.php @@ -0,0 +1,35 @@ +weight); + } + + $target = random_int(1, $total); + + foreach ($instances as $instance) { + $target -= max(1, $instance->weight); + + if ($target <= 0) { + return $instance; + } + } + + return $instances[array_key_last($instances)]; + } +} diff --git a/src/Discovery/ResolverInterface.php b/src/Discovery/ResolverInterface.php new file mode 100644 index 0000000..0ae7c31 --- /dev/null +++ b/src/Discovery/ResolverInterface.php @@ -0,0 +1,13 @@ + $metadata + */ + public function __construct( + public string $name, + public string $host, + public int $port, + public int $weight = 100, + public array $metadata = [], + public string $id = '', + ) { + } + + public function endpoint(string $scheme = 'http'): string + { + $host = str_contains($this->host, ':') && !str_starts_with($this->host, '[') + ? '[' . $this->host . ']' + : $this->host; + + return sprintf('%s://%s:%d', $scheme, $host, $this->port); + } + + /** + * @return array + */ + public function toArray(): array + { + return [ + 'host' => $this->host, + 'port' => $this->port, + 'weight' => $this->weight, + 'metadata' => $this->metadata, + ]; + } + + /** + * @throws JsonException + */ + public function toJson(): string + { + return json_encode($this->toArray(), JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES); + } + + /** + * @throws JsonException + */ + public static function fromJson(string $name, string $json, string $id = ''): self + { + $data = json_decode($json, true, 512, JSON_THROW_ON_ERROR); + + if (!is_array($data) || !isset($data['host'], $data['port'])) { + throw new DiscoveryException(sprintf( + 'Invalid service instance payload for "%s".', + $name, + )); + } + + return new self( + name: $name, + host: (string) $data['host'], + port: (int) $data['port'], + weight: max(1, (int) ($data['weight'] ?? 100)), + metadata: is_array($data['metadata'] ?? null) ? $data['metadata'] : [], + id: $id, + ); + } +} diff --git a/src/Discovery/ServiceLease.php b/src/Discovery/ServiceLease.php new file mode 100644 index 0000000..d05a43a --- /dev/null +++ b/src/Discovery/ServiceLease.php @@ -0,0 +1,15 @@ + */ + private readonly array $headers = [], + ) { + } + + public function put(string $key, string $value, int $leaseId = 0): void + { + $payload = [ + 'key' => base64_encode($key), + 'value' => base64_encode($value), + ]; + + if ($leaseId > 0) { + $payload['lease'] = $leaseId; + } + + $this->request('/v3/kv/put', $payload); + } + + /** + * @return EtcdKeyValue[] + */ + public function get(string $key, string $rangeEnd = ''): array + { + return $this->range($key, $rangeEnd); + } + + /** + * @return EtcdKeyValue[] + */ + public function getPrefix(string $prefix): array + { + return $this->range($prefix, self::prefixEnd($prefix)); + } + + public function delete(string $key): void + { + $this->request('/v3/kv/deleterange', [ + 'key' => base64_encode($key), + ]); + } + + public function deletePrefix(string $prefix): void + { + $this->request('/v3/kv/deleterange', [ + 'key' => base64_encode($prefix), + 'range_end' => base64_encode(self::prefixEnd($prefix)), + ]); + } + + public function grantLease(int $ttl): int + { + $data = $this->request('/v3/lease/grant', ['TTL' => $ttl]); + + return (int) ($data['ID'] ?? 0); + } + + public function keepAlive(int $leaseId): void + { + $this->request('/v3/lease/keepalive', ['ID' => $leaseId]); + } + + public function revokeLease(int $leaseId): void + { + $this->request('/v3/lease/revoke', ['ID' => $leaseId]); + } + + /** + * @return EtcdKeyValue[] + */ + private function range(string $key, string $rangeEnd): array + { + $payload = ['key' => base64_encode($key)]; + + if ($rangeEnd !== '') { + $payload['range_end'] = base64_encode($rangeEnd); + } + + $data = $this->request('/v3/kv/range', $payload); + $kvs = $data['kvs'] ?? []; + + if (!is_array($kvs)) { + throw new EtcdException('Invalid etcd range response.'); + } + + return array_values(array_map( + static fn (array $item): EtcdKeyValue => EtcdKeyValue::fromArray($item), + array_filter($kvs, 'is_array'), + )); + } + + /** + * @param array $payload + * @return array + */ + private function request(string $path, array $payload): array + { + $uri = rtrim($this->baseUri, '/') . $path; + + $request = $this->requestFactory->createRequest('POST', $uri) + ->withHeader('Content-Type', 'application/json') + ->withBody($this->streamFactory->createStream(json_encode( + $payload, + JSON_THROW_ON_ERROR | JSON_UNESCAPED_SLASHES, + ))); + + foreach ($this->headers as $name => $value) { + $request = $request->withHeader($name, $value); + } + + $response = $this->httpClient->sendRequest($request); + $body = (string) $response->getBody(); + + if ($response->getStatusCode() >= 400) { + throw new EtcdException(sprintf( + 'etcd request failed: HTTP %d %s', + $response->getStatusCode(), + $body, + )); + } + + if ($body === '') { + return []; + } + + $data = json_decode($body, true); + + if (!is_array($data)) { + throw new EtcdException('Invalid etcd JSON response: ' . $body); + } + + return $data; + } + + public static function prefixEnd(string $prefix): string + { + if ($prefix === '') { + return "\0"; + } + + for ($i = strlen($prefix) - 1; $i >= 0; $i--) { + $byte = ord($prefix[$i]); + + if ($byte < 255) { + $prefix[$i] = chr($byte + 1); + + return substr($prefix, 0, $i + 1); + } + } + + return "\0"; + } +} diff --git a/src/Etcd/EtcdKeyValue.php b/src/Etcd/EtcdKeyValue.php new file mode 100644 index 0000000..cbd20ed --- /dev/null +++ b/src/Etcd/EtcdKeyValue.php @@ -0,0 +1,44 @@ + $data + */ + public static function fromArray(array $data): self + { + return new self( + key: self::decode($data['key'] ?? ''), + value: self::decode($data['value'] ?? ''), + lease: (int) ($data['lease'] ?? 0), + createRevision: (int) ($data['create_revision'] ?? 0), + modRevision: (int) ($data['mod_revision'] ?? 0), + version: (int) ($data['version'] ?? 0), + ); + } + + private static function decode(mixed $value): string + { + if (!is_string($value)) { + return ''; + } + + $decoded = base64_decode($value, true); + + return $decoded === false ? '' : $decoded; + } +} diff --git a/src/Exception/CircuitOpenException.php b/src/Exception/CircuitOpenException.php new file mode 100644 index 0000000..4bfb569 --- /dev/null +++ b/src/Exception/CircuitOpenException.php @@ -0,0 +1,9 @@ +clock = $clock ?? static fn (): int => time(); + } + + public function allow(): bool + { + $this->transition(); + + return $this->state !== self::OPEN; + } + + public function success(): void + { + $this->failures = 0; + $this->state = self::CLOSED; + $this->openedAt = 0; + } + + public function failure(): void + { + $this->failures++; + + if ($this->state === self::HALF_OPEN || $this->failures >= $this->failureThreshold) { + $this->state = self::OPEN; + $this->openedAt = ($this->clock)(); + } + } + + public function state(): string + { + $this->transition(); + + return $this->state; + } + + private function transition(): void + { + if ($this->state !== self::OPEN) { + return; + } + + if (($this->clock)() - $this->openedAt >= $this->cooldownSeconds) { + $this->state = self::HALF_OPEN; + } + } +} diff --git a/src/Governance/CircuitBreakerClient.php b/src/Governance/CircuitBreakerClient.php new file mode 100644 index 0000000..aa18180 --- /dev/null +++ b/src/Governance/CircuitBreakerClient.php @@ -0,0 +1,43 @@ +breaker->allow()) { + throw new CircuitOpenException('Circuit breaker is open.'); + } + + try { + $response = $this->client->sendRequest($request); + } catch (Throwable $throwable) { + $this->breaker->failure(); + + throw $throwable; + } + + if ($response->getStatusCode() >= 500) { + $this->breaker->failure(); + } else { + $this->breaker->success(); + } + + return $response; + } +} diff --git a/src/Governance/InMemoryRateLimiter.php b/src/Governance/InMemoryRateLimiter.php new file mode 100644 index 0000000..f235543 --- /dev/null +++ b/src/Governance/InMemoryRateLimiter.php @@ -0,0 +1,54 @@ + */ + private array $buckets = []; + + /** @var \Closure(): int */ + private readonly \Closure $clock; + + public function __construct( + private readonly int $capacity = 60, + private readonly int $refillPerSecond = 60, + ?\Closure $clock = null, + ) { + $this->clock = $clock ?? static fn (): int => time(); + } + + public function attempt(string $key, int $tokens = 1): bool + { + $now = ($this->clock)(); + $bucket = $this->buckets[$key] ?? [ + 'tokens' => (float) $this->capacity, + 'updated' => $now, + ]; + + $elapsed = max(0, $now - $bucket['updated']); + $bucket['tokens'] = min( + (float) $this->capacity, + $bucket['tokens'] + $elapsed * $this->refillPerSecond, + ); + $bucket['updated'] = $now; + + if ($bucket['tokens'] < $tokens) { + $this->buckets[$key] = $bucket; + + return false; + } + + $bucket['tokens'] -= $tokens; + $this->buckets[$key] = $bucket; + + return true; + } +} diff --git a/src/Governance/RateLimitMiddleware.php b/src/Governance/RateLimitMiddleware.php new file mode 100644 index 0000000..56ed384 --- /dev/null +++ b/src/Governance/RateLimitMiddleware.php @@ -0,0 +1,41 @@ +keyResolver = $keyResolver ?? static function (ServerRequestInterface $request): string { + return $request->getHeaderLine('X-Api-Key') + ?: $request->getHeaderLine('X-Forwarded-For') + ?: 'anonymous'; + }; + } + + public function process(ServerRequestInterface $request, RequestHandlerInterface $handler): ResponseInterface + { + $key = ($this->keyResolver)($request); + + if (!$this->limiter->attempt($key)) { + return $this->responseFactory->createResponse(429) + ->withHeader('Retry-After', '1'); + } + + return $handler->handle($request); + } +} diff --git a/src/Governance/RateLimiterInterface.php b/src/Governance/RateLimiterInterface.php new file mode 100644 index 0000000..7b342dd --- /dev/null +++ b/src/Governance/RateLimiterInterface.php @@ -0,0 +1,10 @@ +sleeper = $sleeper ?? static function (int $milliseconds): void { + usleep($milliseconds * 1000); + }; + } + + public function sendRequest(RequestInterface $request): ResponseInterface + { + $attempt = 0; + + while (true) { + $attempt++; + $response = null; + $error = null; + + try { + $response = $this->client->sendRequest($request); + } catch (Throwable $throwable) { + $error = $throwable; + } + + if (!$this->policy->shouldRetry($request, $response, $error, $attempt)) { + if ($error !== null) { + throw $error; + } + + return $response; + } + + ($this->sleeper)($this->policy->delayMs($attempt)); + } + } +} diff --git a/src/Governance/RetryPolicy.php b/src/Governance/RetryPolicy.php new file mode 100644 index 0000000..eb1e9f8 --- /dev/null +++ b/src/Governance/RetryPolicy.php @@ -0,0 +1,53 @@ += $this->maxAttempts) { + return false; + } + + if (!in_array(strtoupper($request->getMethod()), $this->retryMethods, true)) { + return false; + } + + if ($response !== null) { + return in_array($response->getStatusCode(), $this->retryStatusCodes, true); + } + + return $error !== null; + } + + public function delayMs(int $attempt): int + { + $delay = $this->baseDelayMs * (2 ** max(0, $attempt - 1)); + + return min($delay, $this->maxDelayMs); + } +} diff --git a/src/HttpClient/ServiceClient.php b/src/HttpClient/ServiceClient.php new file mode 100644 index 0000000..9f779e2 --- /dev/null +++ b/src/HttpClient/ServiceClient.php @@ -0,0 +1,55 @@ +getUri()->getHost(); + $instances = $this->resolver->resolve($service); + + if ($instances === []) { + throw new ServiceUnavailableException(sprintf( + 'No service instance available for "%s".', + $service, + )); + } + + $instance = $this->loadBalancer->choose($instances); + + if ($instance === null) { + throw new ServiceUnavailableException(sprintf( + 'No service instance selected for "%s".', + $service, + )); + } + + $uri = $request->getUri() + ->withHost($instance->host) + ->withPort($instance->port); + + return $this->client->sendRequest($request->withUri($uri)); + } +} diff --git a/src/Tracing/TraceClient.php b/src/Tracing/TraceClient.php new file mode 100644 index 0000000..0ff5ca0 --- /dev/null +++ b/src/Tracing/TraceClient.php @@ -0,0 +1,30 @@ +getHeaderLine('traceparent') === '') { + $request = $request->withHeader('traceparent', $this->context->traceparent()); + } + + return $this->client->sendRequest($request); + } +} diff --git a/src/Tracing/TraceContext.php b/src/Tracing/TraceContext.php new file mode 100644 index 0000000..d087162 --- /dev/null +++ b/src/Tracing/TraceContext.php @@ -0,0 +1,73 @@ +traceId, + spanId: bin2hex(random_bytes(8)), + sampled: $this->sampled, + ); + } + + public function traceparent(): string + { + return sprintf( + '00-%s-%s-%02x', + $this->traceId, + $this->spanId, + $this->sampled ? 1 : 0, + ); + } +} diff --git a/src/Tracing/TracingMiddleware.php b/src/Tracing/TracingMiddleware.php new file mode 100644 index 0000000..06fb947 --- /dev/null +++ b/src/Tracing/TracingMiddleware.php @@ -0,0 +1,38 @@ +trustIncoming + ? $request->getHeaderLine('traceparent') + : null; + + $context = TraceContext::fromTraceparent($incoming !== '' ? $incoming : null); + + $response = $handler->handle( + $request->withAttribute(TraceContext::ATTRIBUTE, $context), + ); + + return $response->withHeader('X-Trace-Id', $context->traceId); + } +} diff --git a/tests/Config/EtcdConfigSourceTest.php b/tests/Config/EtcdConfigSourceTest.php new file mode 100644 index 0000000..c2e333e --- /dev/null +++ b/tests/Config/EtcdConfigSourceTest.php @@ -0,0 +1,51 @@ + [ + [ + 'key' => base64_encode('/fastd/config/user/database/host'), + 'value' => base64_encode('db.local'), + ], + [ + 'key' => base64_encode('/fastd/config/user/database/port'), + 'value' => base64_encode('3306'), + ], + [ + 'key' => base64_encode('/fastd/config/user/name'), + 'value' => base64_encode('user-service'), + ], + ], + ]), + ]); + + $source = new EtcdConfigSource( + new EtcdClient($http, $factory, $factory), + '/fastd/config', + ); + + self::assertSame([ + 'database' => [ + 'host' => 'db.local', + 'port' => '3306', + ], + 'name' => 'user-service', + ], $source->load('user')); + } +} diff --git a/tests/Discovery/DiscoveryTest.php b/tests/Discovery/DiscoveryTest.php new file mode 100644 index 0000000..41cb63e --- /dev/null +++ b/tests/Discovery/DiscoveryTest.php @@ -0,0 +1,111 @@ + 9]), + new Json(200, []), + new Json(200, []), + new Json(200, []), + new Json(200, []), + ]); + $registry = new EtcdServiceRegistry( + new EtcdClient($http, $factory, $factory), + '/fastd/services', + 15, + ); + + $lease = $registry->register(new ServiceInstance( + name: 'user', + host: '10.0.0.1', + port: 9527, + metadata: ['version' => '1.0'], + )); + + self::assertSame(9, $lease->leaseId); + self::assertStringContainsString('/fastd/services/user/', $lease->key); + + $put = json_decode((string) $http->requests[1]->getBody(), true); + $payload = json_decode(base64_decode($put['value']), true); + self::assertSame('10.0.0.1', $payload['host']); + self::assertSame(9527, $payload['port']); + + $registry->heartbeat($lease); + $registry->deregister($lease); + self::assertCount(5, $http->requests); + } + + public function testResolverBuildsInstancesAndCaches(): void + { + $factory = new Factory(); + $response = new Json(200, [ + 'kvs' => [ + [ + 'key' => base64_encode('/fastd/services/user/a'), + 'value' => base64_encode('{"host":"10.0.0.1","port":9527,"weight":100}'), + ], + [ + 'key' => base64_encode('/fastd/services/user/b'), + 'value' => base64_encode('{"host":"10.0.0.2","port":9528,"weight":200}'), + ], + ], + ]); + $http = new FakeClient([$response, $response]); + + $now = 100; + $resolver = new EtcdResolver( + new EtcdClient($http, $factory, $factory), + cacheTtl: 30, + clock: static function () use (&$now): int { + return $now; + }, + ); + + $instances = $resolver->resolve('user'); + self::assertCount(2, $instances); + self::assertSame('a', $instances[0]->id); + self::assertSame(200, $instances[1]->weight); + + $resolver->resolve('user'); + self::assertCount(1, $http->requests); + + $now += 31; + $resolver->resolve('user'); + self::assertCount(2, $http->requests); + } + + public function testLoadBalancersChooseInstances(): void + { + $instances = [ + new ServiceInstance('user', '10.0.0.1', 1), + new ServiceInstance('user', '10.0.0.2', 2), + ]; + + $rr = new RoundRobinBalancer(); + self::assertSame($instances[0], $rr->choose($instances)); + self::assertSame($instances[1], $rr->choose($instances)); + self::assertSame($instances[0], $rr->choose($instances)); + + $weighted = new WeightedRandomBalancer(); + self::assertContains($weighted->choose($instances), $instances); + self::assertNull($weighted->choose([])); + } +} diff --git a/tests/Etcd/EtcdClientTest.php b/tests/Etcd/EtcdClientTest.php new file mode 100644 index 0000000..18ec2f5 --- /dev/null +++ b/tests/Etcd/EtcdClientTest.php @@ -0,0 +1,64 @@ + [ + [ + 'key' => base64_encode('/fastd/services/user/a'), + 'value' => base64_encode('{"host":"127.0.0.1"}'), + ], + ], + ]), + ]); + + $client = new EtcdClient($http, $factory, $factory, 'http://etcd:2379'); + $items = $client->getPrefix('/fastd/services/user/'); + + self::assertCount(1, $items); + self::assertSame('/fastd/services/user/a', $items[0]->key); + self::assertSame('{"host":"127.0.0.1"}', $items[0]->value); + self::assertStringContainsString('/v3/kv/range', (string) $http->requests[0]->getUri()); + } + + public function testPutGrantAndKeepAliveUseEtcdGatewayPayloads(): void + { + $factory = new Factory(); + $http = new FakeClient([ + new Json(200, ['ID' => 42]), + new Json(200, []), + new Json(200, []), + ]); + + $client = new EtcdClient($http, $factory, $factory, 'http://etcd:2379'); + $lease = $client->grantLease(15); + $client->put('/fastd/services/a', '{"host":"10.0.0.1"}', $lease); + $client->keepAlive($lease); + + self::assertSame(42, $lease); + + $put = json_decode((string) $http->requests[1]->getBody(), true); + self::assertSame(base64_encode('/fastd/services/a'), $put['key']); + self::assertSame(42, $put['lease']); + } + + public function testPrefixEndHandlesNormalAndFullBytePrefix(): void + { + self::assertSame('/b', EtcdClient::prefixEnd('/a')); + self::assertSame("\0", EtcdClient::prefixEnd("\xff")); + } +} diff --git a/tests/Governance/GovernanceTest.php b/tests/Governance/GovernanceTest.php new file mode 100644 index 0000000..1f857ea --- /dev/null +++ b/tests/Governance/GovernanceTest.php @@ -0,0 +1,113 @@ + true]), + ]); + $delays = []; + $client = new RetryClient( + $http, + new RetryPolicy(maxAttempts: 2, baseDelayMs: 25), + static function (int $delay) use (&$delays): void { + $delays[] = $delay; + }, + ); + + $response = $client->sendRequest($factory->createRequest('GET', 'http://user/')); + + self::assertSame(200, $response->getStatusCode()); + self::assertCount(2, $http->requests); + self::assertSame([25], $delays); + } + + public function testCircuitBreakerOpensAndHalfOpens(): void + { + $now = 100; + $breaker = new CircuitBreaker( + failureThreshold: 2, + cooldownSeconds: 30, + clock: static function () use (&$now): int { + return $now; + }, + ); + + self::assertTrue($breaker->allow()); + $breaker->failure(); + $breaker->failure(); + self::assertSame(CircuitBreaker::OPEN, $breaker->state()); + self::assertFalse($breaker->allow()); + + $now += 31; + self::assertSame(CircuitBreaker::HALF_OPEN, $breaker->state()); + self::assertTrue($breaker->allow()); + $breaker->success(); + self::assertSame(CircuitBreaker::CLOSED, $breaker->state()); + } + + public function testCircuitBreakerClientRejectsOpenCircuit(): void + { + $now = 0; + $breaker = new CircuitBreaker(1, 60, static function () use (&$now): int { + return $now; + }); + $breaker->failure(); + + $client = new CircuitBreakerClient(new FakeClient(), $breaker); + + $this->expectException(CircuitOpenException::class); + $client->sendRequest((new Factory())->createRequest('GET', 'http://user/')); + } + + public function testInMemoryRateLimiterRefills(): void + { + $now = 0; + $limiter = new InMemoryRateLimiter(2, 1, static function () use (&$now): int { + return $now; + }); + + self::assertTrue($limiter->attempt('a')); + self::assertTrue($limiter->attempt('a')); + self::assertFalse($limiter->attempt('a')); + + $now += 1; + self::assertTrue($limiter->attempt('a')); + self::assertTrue($limiter->attempt('b')); + } + + public function testRateLimitMiddlewareReturns429(): void + { + $factory = new Factory(); + $limiter = new InMemoryRateLimiter(1, 1); + $handler = new FakeHandler(new Json(200, [])); + $middleware = new RateLimitMiddleware($limiter, $factory); + + $request = $factory->createServerRequest('GET', '/'); + self::assertSame(200, $middleware->process($request, $handler)->getStatusCode()); + + $response = $middleware->process($request, $handler); + self::assertSame(429, $response->getStatusCode()); + self::assertSame('1', $response->getHeaderLine('Retry-After')); + } +} diff --git a/tests/HttpClient/ServiceClientTest.php b/tests/HttpClient/ServiceClientTest.php new file mode 100644 index 0000000..13f8fa1 --- /dev/null +++ b/tests/HttpClient/ServiceClientTest.php @@ -0,0 +1,49 @@ + true])]); + $resolver = new FakeResolver(); + $resolver->set('user', [ + new ServiceInstance('user', '10.0.0.1', 9527), + ]); + + $client = new ServiceClient($http, $resolver, new RoundRobinBalancer()); + $response = $client->sendRequest($factory->createRequest('GET', 'http://user/users/1?x=1')); + + self::assertSame(200, $response->getStatusCode()); + self::assertSame('10.0.0.1', $http->requests[0]->getUri()->getHost()); + self::assertSame(9527, $http->requests[0]->getUri()->getPort()); + self::assertSame('/users/1', $http->requests[0]->getUri()->getPath()); + self::assertSame('x=1', $http->requests[0]->getUri()->getQuery()); + } + + public function testThrowsWhenNoInstanceAvailable(): void + { + $client = new ServiceClient( + new FakeClient(), + new FakeResolver(), + new RoundRobinBalancer(), + ); + + $this->expectException(ServiceUnavailableException::class); + $client->sendRequest((new Factory())->createRequest('GET', 'http://missing/')); + } +} diff --git a/tests/Support/FakeClient.php b/tests/Support/FakeClient.php new file mode 100644 index 0000000..ff5e8e2 --- /dev/null +++ b/tests/Support/FakeClient.php @@ -0,0 +1,40 @@ + $responses + */ + public function __construct(private array $responses = []) + { + } + + public function sendRequest(RequestInterface $request): ResponseInterface + { + $this->requests[] = $request; + + $next = array_shift($this->responses); + + if ($next instanceof Throwable) { + throw $next; + } + + if (!$next instanceof ResponseInterface) { + throw new \RuntimeException('No fake response configured.'); + } + + return $next; + } +} diff --git a/tests/Support/FakeHandler.php b/tests/Support/FakeHandler.php new file mode 100644 index 0000000..3638cae --- /dev/null +++ b/tests/Support/FakeHandler.php @@ -0,0 +1,26 @@ +requests[] = $request; + + return $this->response; + } +} diff --git a/tests/Support/FakeResolver.php b/tests/Support/FakeResolver.php new file mode 100644 index 0000000..b48d962 --- /dev/null +++ b/tests/Support/FakeResolver.php @@ -0,0 +1,27 @@ + */ + public array $services = []; + + /** + * @param ServiceInstance[] $instances + */ + public function set(string $service, array $instances): void + { + $this->services[$service] = $instances; + } + + public function resolve(string $service): array + { + return $this->services[$service] ?? []; + } +} diff --git a/tests/Tracing/TracingTest.php b/tests/Tracing/TracingTest.php new file mode 100644 index 0000000..59d86da --- /dev/null +++ b/tests/Tracing/TracingTest.php @@ -0,0 +1,68 @@ +traceId); + self::assertSame('00f067aa0ba902b7', $context->spanId); + self::assertTrue($context->sampled); + + $child = $context->child(); + self::assertSame($context->traceId, $child->traceId); + self::assertNotSame($context->spanId, $child->spanId); + self::assertStringStartsWith('00-', $child->traceparent()); + } + + public function testTracingMiddlewareAddsAttributeAndResponseHeader(): void + { + $factory = new Factory(); + $handler = new FakeHandler(new Json(200, [])); + $middleware = new TracingMiddleware(); + + $request = $factory->createServerRequest('GET', '/') + ->withHeader('traceparent', '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01'); + + $response = $middleware->process($request, $handler); + + self::assertInstanceOf( + TraceContext::class, + $handler->requests[0]->getAttribute(TraceContext::ATTRIBUTE), + ); + self::assertSame( + '4bf92f3577b34da6a3ce929d0e0e4736', + $response->getHeaderLine('X-Trace-Id'), + ); + } + + public function testTraceClientInjectsTraceparent(): void + { + $factory = new Factory(); + $http = new FakeClient([new Json(200, [])]); + $context = TraceContext::fromTraceparent( + '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01', + ); + $client = new TraceClient($http, $context); + + $client->sendRequest($factory->createRequest('GET', 'http://user/')); + + self::assertSame( + '00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01', + $http->requests[0]->getHeaderLine('traceparent'), + ); + } +} From 86648320e75d2c40c6dc4de021f5498cfeecce03 Mon Sep 17 00:00:00 2001 From: JanHuang <384099566@qq.com> Date: Thu, 29 Oct 2026 16:29:39 +0800 Subject: [PATCH 2/3] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0=20FastD=20Servic?= =?UTF-8?q?eProvider=20=E4=B8=8E=E5=AE=8C=E6=95=B4=E4=BD=BF=E7=94=A8?= =?UTF-8?q?=E8=AF=B4=E6=98=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - IntegrationsServiceProvider 读取 config/integrations.php - 注册 etcd、服务发现、服务客户端、配置源 - 按配置装配 tracing、retry、circuit breaker、rate limit - 增加示例配置与 Provider 测试 - README 补充安装、配置、注册、调用、治理和限流示例 --- README.md | 295 ++++++++++++++---- composer.json | 3 +- examples/config/integrations.php | 47 +++ .../IntegrationsServiceProvider.php | 214 +++++++++++++ .../IntegrationsServiceProviderTest.php | 88 ++++++ 5 files changed, 593 insertions(+), 54 deletions(-) create mode 100644 examples/config/integrations.php create mode 100644 src/ServiceProvider/IntegrationsServiceProvider.php create mode 100644 tests/ServiceProvider/IntegrationsServiceProviderTest.php diff --git a/README.md b/README.md index c16d349..8a9fca7 100644 --- a/README.md +++ b/README.md @@ -18,6 +18,7 @@ FastD 的可选微服务集成包。 | `HttpClient` | 按服务名解析实例的 PSR-18 客户端 | | `Tracing` | W3C `traceparent` 提取、生成与下游传播 | | `Governance` | 重试、熔断、令牌桶限流 | +| `ServiceProvider` | FastD 容器装配、中间件注册、配置读取 | ## 安装 @@ -25,53 +26,164 @@ FastD 的可选微服务集成包。 composer require fastd/integrations ``` -本包只依赖 PSR 接口。`fastd/http` 仅作为开发测试依赖，用于构造 PSR-7 请求/响应。 +> 当前功能分支为 `feat/initial-integrations`，正式发布前使用 VCS 方式引用。 -## 配置示例 +如果包尚未发布到 Packagist，可临时使用 VCS。合并到 master 后可将版本改为 `dev-master`： + +```json +{ + "repositories": [ + { + "type": "vcs", + "url": "https://github.com/fastdlabs/integrations" + } + ], + "require": { + "fastd/integrations": "dev-feat/initial-integrations" + } +} +``` + +## FastD 集成 + +### 1. 配置文件 + +新增 `config/integrations.php`： + +```php + [ + 'base_uri' => getenv('ETCD_ENDPOINT') ?: 'http://127.0.0.1:2379', + 'headers' => [], + 'services_prefix' => '/fastd/services', + 'config_prefix' => '/fastd/config', + 'cache_ttl' => 5, + 'lease_ttl' => 15, + 'load_balancer' => 'round_robin', // round_robin | random | weighted_random + ], + + 'tracing' => [ + 'enabled' => true, + 'trust_incoming' => true, + ], + + 'governance' => [ + 'retry' => [ + 'enabled' => false, + 'max_attempts' => 3, + 'base_delay_ms' => 50, + 'max_delay_ms' => 1000, + ], + 'circuit_breaker' => [ + 'enabled' => false, + 'failure_threshold' => 5, + 'cooldown_seconds' => 30, + ], + 'rate_limit' => [ + 'enabled' => false, + 'capacity' => 60, + 'refill_per_second' => 60, + ], + ], + + // 当前服务实例；Provider 只注册为服务，不自动注册到 etcd。 + 'service' => [ + 'name' => 'user-service', + 'host' => '127.0.0.1', + 'port' => 9527, + 'weight' => 100, + 'metadata' => [], + ], +]; +``` + +Provider 的配置来源： + +1. 容器中的 `integrations.config` 数组； +2. 应用根目录的 `config/integrations.php`。 + +如果没有配置 `etcd.base_uri`，Provider 直接跳过，不影响应用启动。 + +### 2. 注册 ServiceProvider + +`config/service.php`： + +```php +return [ + \FastD\Integrations\ServiceProvider\IntegrationsServiceProvider::class, +]; +``` + +注册后容器中可使用： + +| 服务 ID | 类型 | +|---|---| +| `integrations.etcd` | `EtcdClient` | +| `integrations.discovery.resolver` | `EtcdResolver` | +| `integrations.discovery.registry` | `EtcdServiceRegistry` | +| `integrations.http.service_client` | 已按配置包装的 PSR-18 Client | +| `integrations.config.source` | `EtcdConfigSource` | +| `integrations.tracing.middleware` | `TracingMiddleware`（启用时） | +| `integrations.governance.rate_limiter` | `InMemoryRateLimiter`（启用时） | +| `integrations.governance.rate_limit_middleware` | `RateLimitMiddleware`（启用时） | +| `integrations.service_instance` | `ServiceInstance`（配置 service 时） | + +`matcher` 已存在时，Provider 会自动把 tracing / rate limit 中间件 push 到全局中间件栈。 + +### 3. 服务间调用 ```php use FastD\Http\Factory; -use FastD\Integrations\Discovery\EtcdResolver; -use FastD\Integrations\Discovery\LoadBalancer\RoundRobinBalancer; -use FastD\Integrations\Etcd\EtcdClient; -use FastD\Integrations\HttpClient\ServiceClient; +use FastD\Integrations\Tracing\TraceClient; +use FastD\Integrations\Tracing\TraceContext; $factory = new Factory(); +$client = container()->get('integrations.http.service_client'); -$etcd = new EtcdClient( - httpClient: $psr18Client, - requestFactory: $factory, - streamFactory: $factory, - baseUri: 'http://127.0.0.1:2379', -); - -$resolver = new EtcdResolver($etcd, prefix: '/fastd/services'); -$client = new ServiceClient($psr18Client, $resolver, new RoundRobinBalancer()); +// 当前进入的 PSR-7 ServerRequest +$context = $incomingRequest->getAttribute(TraceContext::ATTRIBUTE) + ?? TraceContext::create(); -// 请求 URI 的 host 作为服务名 +// host 是服务名，不是真实域名 $request = $factory->createRequest('GET', 'http://user-service/users/1'); -$response = $client->sendRequest($request); + +$traceClient = new TraceClient($client, $context); +$response = $traceClient->sendRequest($request); ``` -## 服务注册 +`ServiceClient` 的行为： + +```text +http://user-service/users/1 + → Resolver 从 /fastd/services/user-service 读取实例 + → LoadBalancer 选择一个实例 + → 替换为 http://10.0.0.12:9527/users/1 + → 交给底层 PSR-18 Client +``` + +### 4. 注册当前服务 + +Provider 只在容器中创建 `service_instance`，不会自动注册，避免服务尚未监听端口就提前注册。 + +在服务启动完成后的生命周期中执行： ```php -use FastD\Integrations\Discovery\EtcdServiceRegistry; -use FastD\Integrations\Discovery\ServiceInstance; +use Swoole\Timer; -$registry = new EtcdServiceRegistry($etcd, prefix: '/fastd/services', ttl: 15); +$registry = container()->get('integrations.discovery.registry'); +$instance = container()->get('integrations.service_instance'); -$lease = $registry->register(new ServiceInstance( - name: 'user-service', - host: '10.0.0.12', - port: 9527, - metadata: ['version' => '1.0.0'], -)); +$lease = $registry->register($instance); -// 定时续约 -$registry->heartbeat($lease); +Timer::tick(5000, static function () use ($registry, $lease): void { + $registry->heartbeat($lease); +}); -// 进程退出 +// 进程退出时 $registry->deregister($lease); ``` @@ -81,54 +193,84 @@ etcd key 结构： /fastd/services/{service}/{instanceId} ``` -value 为 JSON： +value： ```json -{"host":"10.0.0.12","port":9527,"weight":100,"metadata":{"version":"1.0.0"}} +{ + "host": "10.0.0.12", + "port": 9527, + "weight": 100, + "metadata": { + "version": "1.0.0" + } +} ``` -## 配置中心 +### 5. 配置中心 ```php -use FastD\Integrations\Config\EtcdConfigSource; +$source = container()->get('integrations.config.source'); -$source = new EtcdConfigSource($etcd, prefix: '/fastd/config'); $config = $source->load('user-service'); ``` -etcd key： +etcd： ```text -/fastd/config/user-service/database/host +/fastd/config/user-service/database/host = db.local +/fastd/config/user-service/database/port = 3306 ``` -会转换成： +结果： ```php -['database' => ['host' => '...']] +[ + 'database' => [ + 'host' => 'db.local', + 'port' => '3306', + ], +] ``` -## 链路传播 - -在 FastD 中把中间件挂到全局中间件栈： +本包只读取和转换配置，不自动替换 FastD 本地配置。需要时在业务 Provider 中合并： ```php -use FastD\Integrations\Tracing\TracingMiddleware; - -$container->get('matcher')->push(new TracingMiddleware()); +$container->add('remote_config', $config); ``` -入口读取或生成 `traceparent`，写入请求属性 `fastd.trace_context`，并返回 `X-Trace-Id`。下游调用使用： +### 6. 链路传播 + +`TracingMiddleware` 已由 Provider 挂载时，可直接读取请求上下文： ```php -use FastD\Integrations\Tracing\TraceClient; +use FastD\Integrations\Tracing\TraceContext; -$client = new TraceClient($psr18Client, $traceContext); +$context = $incomingRequest->getAttribute(TraceContext::ATTRIBUTE); ``` -`TraceClient` 只负责 W3C 上下文传播，不导出 Span。Span 采集/导出请接入 OpenTelemetry SDK。 +下游调用使用 `TraceClient` 注入 `traceparent`。 + +本包只做 W3C 上下文传播，不导出 Span。Span 采集/导出请接入 OpenTelemetry SDK。 + +### 7. 重试与熔断 -## 治理 +在 `config/integrations.php` 开启后，`integrations.http.service_client` 会自动包装： + +```php +'governance' => [ + 'retry' => [ + 'enabled' => true, + 'max_attempts' => 3, + ], + 'circuit_breaker' => [ + 'enabled' => true, + 'failure_threshold' => 5, + 'cooldown_seconds' => 30, + ], +], +``` + +也可以手动组合： ```php use FastD\Integrations\Governance\CircuitBreaker; @@ -142,9 +284,54 @@ $client = new RetryClient( ); ``` -- 重试只对幂等方法生效 -- 熔断和限流状态默认进程内；多实例共享需替换存储实现 -- 生产环境需设置超时，避免重试放大故障 +默认只重试 `GET`、`HEAD`、`OPTIONS`、`PUT`、`DELETE`。`POST` 需要业务确认幂等后再加入重试策略。 + +### 8. 限流 + +开启： + +```php +'governance' => [ + 'rate_limit' => [ + 'enabled' => true, + 'capacity' => 60, + 'refill_per_second' => 60, + ], +], +``` + +Provider 会创建 `RateLimitMiddleware` 并 push 到 `matcher`。默认按 `X-Api-Key` / `X-Forwarded-For` 限流。 + +进程内令牌桶只适合单 worker 或演示；多实例全局限流请替换为 Redis 或网关实现。 + +## 独立使用 + +不使用 FastD Provider 时，也可以直接组合组件： + +```php +use FastD\Http\Factory; +use FastD\Http\Request\Client; +use FastD\Integrations\Discovery\EtcdResolver; +use FastD\Integrations\Discovery\LoadBalancer\RoundRobinBalancer; +use FastD\Integrations\Etcd\EtcdClient; +use FastD\Integrations\HttpClient\ServiceClient; + +$factory = new Factory(); +$psr18Client = new Client(); + +$etcd = new EtcdClient( + httpClient: $psr18Client, + requestFactory: $factory, + streamFactory: $factory, + baseUri: 'http://127.0.0.1:2379', +); + +$resolver = new EtcdResolver($etcd, prefix: '/fastd/services'); +$client = new ServiceClient($psr18Client, $resolver, new RoundRobinBalancer()); + +$request = $factory->createRequest('GET', 'http://user-service/users/1'); +$response = $client->sendRequest($request); +``` ## 边界 @@ -164,3 +351,5 @@ $client = new RetryClient( composer install vendor/bin/phpunit ``` + +当前单元测试使用 fake PSR-18 客户端，不依赖真实 etcd。生产接入前应增加真实 etcd 的集成测试。 diff --git a/composer.json b/composer.json index 9ca8fcf..59b82fb 100644 --- a/composer.json +++ b/composer.json @@ -20,6 +20,8 @@ } ], "require": { + "fastd/container": "^8.1", + "fastd/http": "^8.2", "php": ">=8.4.1", "psr/http-client": "^1.0", "psr/http-factory": "^1.0", @@ -28,7 +30,6 @@ "psr/http-server-middleware": "^1.0" }, "require-dev": { - "fastd/http": "^8.2", "phpunit/phpunit": "^13.4" }, "autoload": { diff --git a/examples/config/integrations.php b/examples/config/integrations.php new file mode 100644 index 0000000..633e81a --- /dev/null +++ b/examples/config/integrations.php @@ -0,0 +1,47 @@ + [ + 'base_uri' => getenv('ETCD_ENDPOINT') ?: 'http://127.0.0.1:2379', + 'headers' => [], + 'services_prefix' => '/fastd/services', + 'config_prefix' => '/fastd/config', + 'cache_ttl' => 5, + 'lease_ttl' => 15, + 'load_balancer' => 'round_robin', + ], + + 'tracing' => [ + 'enabled' => true, + 'trust_incoming' => true, + ], + + 'governance' => [ + 'retry' => [ + 'enabled' => false, + 'max_attempts' => 3, + 'base_delay_ms' => 50, + 'max_delay_ms' => 1000, + ], + 'circuit_breaker' => [ + 'enabled' => false, + 'failure_threshold' => 5, + 'cooldown_seconds' => 30, + ], + 'rate_limit' => [ + 'enabled' => false, + 'capacity' => 60, + 'refill_per_second' => 60, + ], + ], + + 'service' => [ + 'name' => 'user-service', + 'host' => '127.0.0.1', + 'port' => 9527, + 'weight' => 100, + 'metadata' => [], + ], +]; diff --git a/src/ServiceProvider/IntegrationsServiceProvider.php b/src/ServiceProvider/IntegrationsServiceProvider.php new file mode 100644 index 0000000..fdebc32 --- /dev/null +++ b/src/ServiceProvider/IntegrationsServiceProvider.php @@ -0,0 +1,214 @@ +config($container); + $etcdConfig = (array) ($config['etcd'] ?? []); + + if ($etcdConfig === [] || empty($etcdConfig['base_uri'])) { + return; + } + + $factory = new Factory(); + $httpClient = new Client(); + + $etcd = new EtcdClient( + httpClient: $httpClient, + requestFactory: $factory, + streamFactory: $factory, + baseUri: (string) $etcdConfig['base_uri'], + headers: (array) ($etcdConfig['headers'] ?? []), + ); + + $resolver = new EtcdResolver( + client: $etcd, + prefix: (string) ($etcdConfig['services_prefix'] ?? '/fastd/services'), + cacheTtl: (int) ($etcdConfig['cache_ttl'] ?? 5), + ); + + $registry = new EtcdServiceRegistry( + client: $etcd, + prefix: (string) ($etcdConfig['services_prefix'] ?? '/fastd/services'), + ttl: (int) ($etcdConfig['lease_ttl'] ?? 15), + ); + + $container->add('integrations.etcd', $etcd); + $container->add('integrations.discovery.resolver', $resolver); + $container->add('integrations.discovery.registry', $registry); + + $container->add( + 'integrations.http.service_client', + $this->serviceClient($httpClient, $resolver, $config), + ); + + $container->add( + 'integrations.config.source', + new EtcdConfigSource( + $etcd, + (string) ($etcdConfig['config_prefix'] ?? '/fastd/config'), + ), + ); + + $this->registerTracing($container, $config); + $this->registerRateLimit($container, $config); + + if (isset($config['service']) && is_array($config['service']) && $config['service'] !== []) { + $service = $config['service']; + $container->add('integrations.service_instance', new ServiceInstance( + name: (string) ($service['name'] ?? ''), + host: (string) ($service['host'] ?? '127.0.0.1'), + port: (int) ($service['port'] ?? 0), + weight: (int) ($service['weight'] ?? 100), + metadata: (array) ($service['metadata'] ?? []), + )); + } + } + + /** + * @param array $config + */ + private function serviceClient( + ClientInterface $httpClient, + EtcdResolver $resolver, + array $config, + ): ClientInterface { + $balancer = match ((string) ($config['etcd']['load_balancer'] ?? 'round_robin')) { + 'random' => new RandomBalancer(), + 'weighted_random' => new WeightedRandomBalancer(), + default => new RoundRobinBalancer(), + }; + + $client = new ServiceClient($httpClient, $resolver, $balancer); + + $governance = (array) ($config['governance'] ?? []); + + if (!empty($governance['circuit_breaker']['enabled'])) { + $breaker = (array) $governance['circuit_breaker']; + $client = new CircuitBreakerClient($client, new CircuitBreaker( + failureThreshold: (int) ($breaker['failure_threshold'] ?? 5), + cooldownSeconds: (int) ($breaker['cooldown_seconds'] ?? 30), + )); + } + + if (!empty($governance['retry']['enabled'])) { + $retry = (array) $governance['retry']; + $client = new RetryClient($client, new RetryPolicy( + maxAttempts: (int) ($retry['max_attempts'] ?? 3), + baseDelayMs: (int) ($retry['base_delay_ms'] ?? 50), + maxDelayMs: (int) ($retry['max_delay_ms'] ?? 1000), + )); + } + + return $client; + } + + /** + * @param array $config + */ + private function registerTracing(Container $container, array $config): void + { + $tracing = (array) ($config['tracing'] ?? []); + + if (empty($tracing['enabled'])) { + return; + } + + $middleware = new TracingMiddleware((bool) ($tracing['trust_incoming'] ?? true)); + $container->add('integrations.tracing.middleware', $middleware); + + if ($container->has('matcher')) { + $container->get('matcher')->push($middleware); + } + } + + /** + * @param array $config + */ + private function registerRateLimit(Container $container, array $config): void + { + $governance = (array) ($config['governance'] ?? []); + $rateLimit = (array) ($governance['rate_limit'] ?? []); + + if (empty($rateLimit['enabled'])) { + return; + } + + $limiter = new InMemoryRateLimiter( + capacity: (int) ($rateLimit['capacity'] ?? 60), + refillPerSecond: (int) ($rateLimit['refill_per_second'] ?? 60), + ); + + $rateLimitMiddleware = new RateLimitMiddleware($limiter, new Factory()); + $container->add('integrations.governance.rate_limiter', $limiter); + $container->add('integrations.governance.rate_limit_middleware', $rateLimitMiddleware); + + if ($container->has('matcher')) { + $container->get('matcher')->push($rateLimitMiddleware); + } + } + + /** + * @return array + */ + private function config(Container $container): array + { + try { + if ($container->has('integrations.config')) { + $config = $container->get('integrations.config'); + + return is_array($config) ? $config : []; + } + } catch (NotFoundExceptionInterface|ContainerExceptionInterface) { + // 继续回退到配置文件。 + } + + $file = getcwd() . '/config/integrations.php'; + + if (is_file($file)) { + $config = require $file; + + return is_array($config) ? $config : []; + } + + return []; + } +} diff --git a/tests/ServiceProvider/IntegrationsServiceProviderTest.php b/tests/ServiceProvider/IntegrationsServiceProviderTest.php new file mode 100644 index 0000000..551131a --- /dev/null +++ b/tests/ServiceProvider/IntegrationsServiceProviderTest.php @@ -0,0 +1,88 @@ +add('integrations.config', [ + 'etcd' => [ + 'base_uri' => 'http://127.0.0.1:2379', + 'services_prefix' => '/fastd/services', + 'config_prefix' => '/fastd/config', + ], + 'tracing' => ['enabled' => false], + 'service' => [ + 'name' => 'user-service', + 'host' => '127.0.0.1', + 'port' => 9527, + ], + ]); + + (new IntegrationsServiceProvider())->register($container); + + self::assertTrue($container->has('integrations.etcd')); + self::assertTrue($container->has('integrations.discovery.resolver')); + self::assertTrue($container->has('integrations.discovery.registry')); + self::assertTrue($container->has('integrations.config.source')); + self::assertTrue($container->has('integrations.http.service_client')); + self::assertInstanceOf( + ClientInterface::class, + $container->get('integrations.http.service_client'), + ); + self::assertInstanceOf( + ServiceInstance::class, + $container->get('integrations.service_instance'), + ); + } + + public function testRegistersTracingAndRateLimitMiddlewareOnMatcher(): void + { + $matcher = new class { + /** @var object[] */ + public array $middleware = []; + + public function push(object $middleware): void + { + $this->middleware[] = $middleware; + } + }; + + $container = new Container(); + $container->add('integrations.config', [ + 'etcd' => ['base_uri' => 'http://127.0.0.1:2379'], + 'tracing' => ['enabled' => true], + 'governance' => [ + 'rate_limit' => ['enabled' => true, 'capacity' => 10, 'refill_per_second' => 10], + ], + ]); + $container->add('matcher', $matcher); + + (new IntegrationsServiceProvider())->register($container); + + self::assertTrue($container->has('integrations.tracing.middleware')); + self::assertTrue($container->has('integrations.governance.rate_limit_middleware')); + self::assertCount(2, $matcher->middleware); + } + + public function testSkipsWhenEtcdIsNotConfigured(): void + { + $container = new Container(); + $container->add('integrations.config', []); + + (new IntegrationsServiceProvider())->register($container); + + self::assertFalse($container->has('integrations.etcd')); + self::assertFalse($container->has('integrations.http.service_client')); + } +} From edd217154e0e91ab338a83cb9a978f3eef4ca732 Mon Sep 17 00:00:00 2001 From: JanHuang <384099566@qq.com> Date: Thu, 29 Oct 2026 16:30:54 +0800 Subject: [PATCH 3/3] =?UTF-8?q?feat:=20=E8=A7=A3=E8=80=A6=20Provider=20?= =?UTF-8?q?=E7=9A=84=20tracing/rate=20limit=20=E8=A3=85=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - tracing 与 rate limit 不再依赖 etcd 配置 - 补充 Provider 测试与示例配置说明 --- README.md | 4 +-- .../IntegrationsServiceProvider.php | 7 +++-- .../IntegrationsServiceProviderTest.php | 29 +++++++++++++++++++ 3 files changed, 35 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 8a9fca7..e9f6886 100644 --- a/README.md +++ b/README.md @@ -48,7 +48,7 @@ composer require fastd/integrations ### 1. 配置文件 -新增 `config/integrations.php`： +新增 `config/integrations.php`（完整示例见 `examples/config/integrations.php`）： ```php config($container); + + $this->registerTracing($container, $config); + $this->registerRateLimit($container, $config); + $etcdConfig = (array) ($config['etcd'] ?? []); if ($etcdConfig === [] || empty($etcdConfig['base_uri'])) { @@ -88,9 +92,6 @@ public function register(Container $container): void ), ); - $this->registerTracing($container, $config); - $this->registerRateLimit($container, $config); - if (isset($config['service']) && is_array($config['service']) && $config['service'] !== []) { $service = $config['service']; $container->add('integrations.service_instance', new ServiceInstance( diff --git a/tests/ServiceProvider/IntegrationsServiceProviderTest.php b/tests/ServiceProvider/IntegrationsServiceProviderTest.php index 551131a..69ca8c1 100644 --- a/tests/ServiceProvider/IntegrationsServiceProviderTest.php +++ b/tests/ServiceProvider/IntegrationsServiceProviderTest.php @@ -75,6 +75,35 @@ public function push(object $middleware): void self::assertCount(2, $matcher->middleware); } + public function testTracingAndRateLimitDoNotRequireEtcd(): void + { + $matcher = new class { + /** @var object[] */ + public array $middleware = []; + + public function push(object $middleware): void + { + $this->middleware[] = $middleware; + } + }; + + $container = new Container(); + $container->add('integrations.config', [ + 'tracing' => ['enabled' => true], + 'governance' => [ + 'rate_limit' => ['enabled' => true], + ], + ]); + $container->add('matcher', $matcher); + + (new IntegrationsServiceProvider())->register($container); + + self::assertFalse($container->has('integrations.etcd')); + self::assertTrue($container->has('integrations.tracing.middleware')); + self::assertTrue($container->has('integrations.governance.rate_limit_middleware')); + self::assertCount(2, $matcher->middleware); + } + public function testSkipsWhenEtcdIsNotConfigured(): void { $container = new Container();