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..e9f6886 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,355 @@ -# 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` | 重试、熔断、令牌桶限流 | +| `ServiceProvider` | FastD 容器装配、中间件注册、配置读取 | + +## 安装 + +```bash +composer require fastd/integrations +``` + +> 当前功能分支为 `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`(完整示例见 `examples/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`,etcd 相关服务会跳过;tracing 和 rate limit 仍可独立启用。 + +### 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\Tracing\TraceClient; +use FastD\Integrations\Tracing\TraceContext; + +$factory = new Factory(); +$client = container()->get('integrations.http.service_client'); + +// 当前进入的 PSR-7 ServerRequest +$context = $incomingRequest->getAttribute(TraceContext::ATTRIBUTE) + ?? TraceContext::create(); + +// host 是服务名,不是真实域名 +$request = $factory->createRequest('GET', 'http://user-service/users/1'); + +$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 Swoole\Timer; + +$registry = container()->get('integrations.discovery.registry'); +$instance = container()->get('integrations.service_instance'); + +$lease = $registry->register($instance); + +Timer::tick(5000, static function () use ($registry, $lease): void { + $registry->heartbeat($lease); +}); + +// 进程退出时 +$registry->deregister($lease); +``` + +etcd key 结构: + +```text +/fastd/services/{service}/{instanceId} +``` + +value: + +```json +{ + "host": "10.0.0.12", + "port": 9527, + "weight": 100, + "metadata": { + "version": "1.0.0" + } +} +``` + +### 5. 配置中心 + +```php +$source = container()->get('integrations.config.source'); + +$config = $source->load('user-service'); +``` + +etcd: + +```text +/fastd/config/user-service/database/host = db.local +/fastd/config/user-service/database/port = 3306 +``` + +结果: + +```php +[ + 'database' => [ + 'host' => 'db.local', + 'port' => '3306', + ], +] +``` + +本包只读取和转换配置,不自动替换 FastD 本地配置。需要时在业务 Provider 中合并: + +```php +$container->add('remote_config', $config); +``` + +### 6. 链路传播 + +`TracingMiddleware` 已由 Provider 挂载时,可直接读取请求上下文: + +```php +use FastD\Integrations\Tracing\TraceContext; + +$context = $incomingRequest->getAttribute(TraceContext::ATTRIBUTE); +``` + +下游调用使用 `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; +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), +); +``` + +默认只重试 `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); +``` + +## 边界 + +本包不提供: + +- 服务注册中心本身 +- gRPC/Thrift 协议 +- 分布式事务 +- 完整 OpenTelemetry SDK +- 全局限流存储 + +这些应由 etcd、OpenTelemetry、网关或独立库负责。 + +## 测试 + +```bash +composer install +vendor/bin/phpunit +``` + +当前单元测试使用 fake PSR-18 客户端,不依赖真实 etcd。生产接入前应增加真实 etcd 的集成测试。 diff --git a/composer.json b/composer.json new file mode 100644 index 0000000..59b82fb --- /dev/null +++ b/composer.json @@ -0,0 +1,51 @@ +{ + "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": { + "fastd/container": "^8.1", + "fastd/http": "^8.2", + "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": { + "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/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/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/ServiceProvider/IntegrationsServiceProvider.php b/src/ServiceProvider/IntegrationsServiceProvider.php new file mode 100644 index 0000000..652fd37 --- /dev/null +++ b/src/ServiceProvider/IntegrationsServiceProvider.php @@ -0,0 +1,215 @@ +config($container); + + $this->registerTracing($container, $config); + $this->registerRateLimit($container, $config); + + $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'), + ), + ); + + 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/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/ServiceProvider/IntegrationsServiceProviderTest.php b/tests/ServiceProvider/IntegrationsServiceProviderTest.php new file mode 100644 index 0000000..69ca8c1 --- /dev/null +++ b/tests/ServiceProvider/IntegrationsServiceProviderTest.php @@ -0,0 +1,117 @@ +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 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(); + $container->add('integrations.config', []); + + (new IntegrationsServiceProvider())->register($container); + + self::assertFalse($container->has('integrations.etcd')); + self::assertFalse($container->has('integrations.http.service_client')); + } +} 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'), + ); + } +}