From aee31a3b65317126953a760a881440c36ca10c77 Mon Sep 17 00:00:00 2001 From: Lucas Angeli Date: Thu, 23 Apr 2026 12:09:53 -0300 Subject: [PATCH 1/3] fix(grpc): resilient SwooleGrpcTransport with mutex, retry and reconnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add Channel(1) mutex to serialize concurrent coroutine sends; shutdown closes the channel to unblock waiting coroutines without deadlock - Extract attemptSend() retry loop (send failure → reset + retry up to maxRetries; recv failure → ErrorFuture immediately to avoid duplicate DELTA metric exports) - Extract doRequest() to reduce cyclomatic complexity below threshold - Reset HTTP/2 client on send/recv failure so next call reconnects - Fix getClient(): close stale (disconnected) client before reconnecting - Suppress PHP 8.2 E_DEPRECATED from Swoole's dynamic $serverLastStreamId property in recv() using @ operator (Hyperf ErrorExceptionHandler converts deprecation notices to ErrorException, aborting the recv) - Validate unsupported compression types in factory (prevents compressed flag=1 with uncompressed payload) - Wire retryDelay and maxRetries from factory to transport constructor - Add endpoint fallback in Trace and Metric gRPC exporter factories - Replace 'application/x-protobuf' literal with ContentTypes::PROTOBUF - Cast grpc-status to int before comparison (more robust against whitespace) - Add tests: default endpoint fallback, retry parameters, invalid compression Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../Exporter/OtlpGrpcLogExporterFactory.php | 3 +- .../OtlpGrpcMetricExporterFactory.php | 5 +- .../Exporter/OtlpGrpcTraceExporterFactory.php | 5 +- src/Transport/SwooleGrpcTransport.php | 166 ++++++++++++++---- src/Transport/SwooleGrpcTransportFactory.php | 16 ++ .../OtlpGrpcMetricExporterFactoryTest.php | 13 ++ .../OtlpGrpcTraceExporterFactoryTest.php | 13 ++ .../SwooleGrpcTransportFactoryTest.php | 31 ++++ 8 files changed, 211 insertions(+), 41 deletions(-) diff --git a/src/Factory/Log/Exporter/OtlpGrpcLogExporterFactory.php b/src/Factory/Log/Exporter/OtlpGrpcLogExporterFactory.php index 4de03c3..e1505a8 100644 --- a/src/Factory/Log/Exporter/OtlpGrpcLogExporterFactory.php +++ b/src/Factory/Log/Exporter/OtlpGrpcLogExporterFactory.php @@ -6,6 +6,7 @@ use Hyperf\Contract\ConfigInterface; use Hyperf\OpenTelemetry\Transport\SwooleGrpcTransportFactory; +use OpenTelemetry\Contrib\Otlp\ContentTypes; use OpenTelemetry\Contrib\Otlp\LogsExporter; use OpenTelemetry\SDK\Common\Export\TransportFactoryInterface; use OpenTelemetry\SDK\Logs\LogRecordExporterInterface; @@ -28,7 +29,7 @@ public function make(): LogRecordExporterInterface return new LogsExporter( (new SwooleGrpcTransportFactory())->create( endpoint: $endpoint, - contentType: 'application/x-protobuf', + contentType: ContentTypes::PROTOBUF, headers: $options['headers'] ?? [], compression: $options['compression'] ?? TransportFactoryInterface::COMPRESSION_GZIP, timeout: $options['timeout'] ?? 10, diff --git a/src/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactory.php b/src/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactory.php index 8f1cf1a..0a5c673 100644 --- a/src/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactory.php +++ b/src/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactory.php @@ -6,6 +6,7 @@ use Hyperf\Contract\ConfigInterface; use Hyperf\OpenTelemetry\Transport\SwooleGrpcTransportFactory; +use OpenTelemetry\Contrib\Otlp\ContentTypes; use OpenTelemetry\Contrib\Otlp\MetricExporter; use OpenTelemetry\SDK\Common\Export\TransportFactoryInterface; use OpenTelemetry\SDK\Metrics\Data\Temporality; @@ -24,12 +25,12 @@ public function make(): MetricExporterInterface { $options = $this->config->get('open-telemetry.metrics.exporters.otlp_grpc.options', []); - $endpoint = rtrim($options['endpoint'], '/') . self::GRPC_METHOD; + $endpoint = rtrim($options['endpoint'] ?? 'http://localhost:4317', '/') . self::GRPC_METHOD; return new MetricExporter( transport: (new SwooleGrpcTransportFactory())->create( endpoint: $endpoint, - contentType: 'application/x-protobuf', + contentType: ContentTypes::PROTOBUF, headers: $options['headers'] ?? [], compression: $options['compression'] ?? TransportFactoryInterface::COMPRESSION_GZIP, timeout: $options['timeout'] ?? 10, diff --git a/src/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactory.php b/src/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactory.php index 930f025..0df0e50 100644 --- a/src/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactory.php +++ b/src/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactory.php @@ -6,6 +6,7 @@ use Hyperf\Contract\ConfigInterface; use Hyperf\OpenTelemetry\Transport\SwooleGrpcTransportFactory; +use OpenTelemetry\Contrib\Otlp\ContentTypes; use OpenTelemetry\Contrib\Otlp\SpanExporter; use OpenTelemetry\SDK\Common\Export\TransportFactoryInterface; use OpenTelemetry\SDK\Trace\SpanExporterInterface; @@ -23,12 +24,12 @@ public function make(): SpanExporterInterface { $options = $this->config->get('open-telemetry.traces.exporters.otlp_grpc.options', []); - $endpoint = rtrim($options['endpoint'], '/') . self::GRPC_METHOD; + $endpoint = rtrim($options['endpoint'] ?? 'http://localhost:4317', '/') . self::GRPC_METHOD; return new SpanExporter( (new SwooleGrpcTransportFactory())->create( endpoint: $endpoint, - contentType: 'application/x-protobuf', + contentType: ContentTypes::PROTOBUF, headers: $options['headers'] ?? [], compression: $options['compression'] ?? TransportFactoryInterface::COMPRESSION_GZIP, timeout: $options['timeout'] ?? 10, diff --git a/src/Transport/SwooleGrpcTransport.php b/src/Transport/SwooleGrpcTransport.php index 76f5476..29883ab 100644 --- a/src/Transport/SwooleGrpcTransport.php +++ b/src/Transport/SwooleGrpcTransport.php @@ -12,6 +12,8 @@ use OpenTelemetry\SDK\Common\Future\ErrorFuture; use OpenTelemetry\SDK\Common\Future\FutureInterface; use RuntimeException; +use Swoole\Coroutine; +use Swoole\Coroutine\Channel; use Swoole\Coroutine\Http2\Client; use Swoole\Http2\Request; use Throwable; @@ -22,6 +24,8 @@ final class SwooleGrpcTransport implements TransportInterface private ?Client $client = null; + private ?Channel $mutex = null; + /** * @SuppressWarnings(PHPMD.BooleanArgumentFlag) */ @@ -33,6 +37,8 @@ public function __construct( private readonly float $timeout = 10.0, private readonly bool $ssl = false, private readonly ?string $compression = null, + private readonly int $retryDelay = 100, + private readonly int $maxRetries = 3, ) { } @@ -47,42 +53,21 @@ public function send(string $payload, ?CancellationInterface $cancellation = nul return new ErrorFuture(new RuntimeException('Transport is closed')); } - try { - $client = $this->getClient(); - - $data = $this->compress($payload); - - $request = new Request(); - $request->method = 'POST'; - $request->path = $this->method; - $request->headers = $this->buildHeaders(); - $request->data = $this->packMessage($data); - - $streamId = $client->send($request); - - if ($streamId === false || $streamId <= 0) { - return new ErrorFuture(new RuntimeException( - 'Failed to send gRPC request: ' . ($client->errMsg ?: 'unknown error') - )); - } - - $response = $client->recv($this->timeout); - - if ($response === false) { - return new ErrorFuture(new RuntimeException( - 'Failed to receive gRPC response: ' . ($client->errMsg ?: 'timeout') - )); - } + $mutex = $this->getMutex(); + if ($mutex->pop() === false) { + return new ErrorFuture(new RuntimeException('Transport is closed')); + } - $grpcStatus = $response->headers['grpc-status'] ?? '0'; - if ($grpcStatus !== '0') { - $grpcMessage = $response->headers['grpc-message'] ?? 'Unknown error'; - return new ErrorFuture(new RuntimeException("gRPC error: {$grpcMessage}", (int) $grpcStatus)); + try { + if ($this->closed) { + return new ErrorFuture(new RuntimeException('Transport is closed')); } - return new CompletedFuture(null); + return $this->attemptSend($payload); } catch (Throwable $e) { return new ErrorFuture($e); + } finally { + $mutex->push(true); // release — returns false silently if channel was closed by shutdown } } @@ -93,6 +78,7 @@ public function shutdown(?CancellationInterface $cancellation = null): bool } $this->closed = true; + $this->mutex?->close(); // unblock any coroutines waiting to acquire the mutex if ($this->client !== null) { $this->client->close(); @@ -107,19 +93,126 @@ public function forceFlush(?CancellationInterface $cancellation = null): bool return ! $this->closed; } + /** + * @SuppressWarnings(PHPMD.ErrorControlOperator) + */ + private function attemptSend(string $payload): FutureInterface + { + $data = $this->compress($payload); + $lastError = null; + + for ($attempt = 0; $attempt <= $this->maxRetries; ++$attempt) { + if ($this->closed) { + return new ErrorFuture(new RuntimeException('Transport is closed')); + } + + if ($attempt > 0 && $this->retryDelay > 0) { + Coroutine::sleep($this->retryDelay / 1000.0); + } + + try { + return $this->doRequest($this->getClient(), $data); + } catch (Throwable $e) { + $this->resetClient(); + $lastError = $e; + } + } + + return new ErrorFuture($lastError ?? new RuntimeException('Unknown transport error after retries')); + } + + /** + * Executes a single gRPC send+recv cycle. + * + * Throws RuntimeException on send failure so the caller can reset the client and retry. + * Returns ErrorFuture on recv failure (delivery uncertain — no retry). + * + * Note: @$client->recv() suppresses E_DEPRECATED from Swoole setting $serverLastStreamId + * as a dynamic property in PHP 8.2+. Hyperf's ErrorExceptionHandler would otherwise convert + * the deprecation notice into an ErrorException, aborting the recv() call. + * + * @throws RuntimeException on retryable send failure + * @SuppressWarnings(PHPMD.ErrorControlOperator) + */ + private function doRequest(Client $client, string $data): FutureInterface + { + $request = new Request(); + $request->method = 'POST'; + $request->path = $this->method; + $request->headers = $this->buildHeaders(); + $request->data = $this->packMessage($data); + + $streamId = $client->send($request); + + if ($streamId === false || $streamId <= 0) { + // Throw so the retry loop can reset the client and try again; + // data was never transmitted so retrying is safe. + throw new RuntimeException( + 'Failed to send gRPC request: ' . ($client->errMsg ?: 'unknown error') + ); + } + + $response = @$client->recv($this->timeout); + + if ($response === false) { + $this->resetClient($client); + // Do not retry: send succeeded, so delivery is uncertain — retrying could duplicate exports. + return new ErrorFuture(new RuntimeException( + 'Failed to receive gRPC response: ' . ($client->errMsg ?: 'timeout') + )); + } + + $grpcStatus = (int) ($response->headers['grpc-status'] ?? 0); + if ($grpcStatus !== 0) { + $grpcMessage = $response->headers['grpc-message'] ?? 'Unknown error'; + // gRPC application error — not retried + return new ErrorFuture(new RuntimeException("gRPC error: {$grpcMessage}", $grpcStatus)); + } + + return new CompletedFuture(null); + } + + private function getMutex(): Channel + { + if ($this->mutex === null) { + $this->mutex = new Channel(1); + $this->mutex->push(true); // initially unlocked + } + + return $this->mutex; + } + + private function resetClient(?Client $client = null): void + { + $target = $client ?? $this->client; + $target?->close(); + if ($target === $this->client) { + $this->client = null; + } + } + private function getClient(): Client { - if ($this->client === null || ! $this->client->connected) { - $this->client = new Client($this->host, $this->port, $this->ssl); - $this->client->set([ + if ($this->client !== null && ! $this->client->connected) { + $this->client->close(); + $this->client = null; + } + + if ($this->client === null) { + $client = new Client($this->host, $this->port, $this->ssl); + $client->set([ 'timeout' => $this->timeout, ]); - if (! $this->client->connect()) { + if (! $client->connect()) { + $errMsg = $client->errMsg; + $client->close(); throw new RuntimeException( - "Failed to connect to {$this->host}:{$this->port}: " . $this->client->errMsg + "Failed to connect to {$this->host}:{$this->port}: " . $errMsg ); } + + $this->client = $client; } return $this->client; @@ -165,6 +258,7 @@ private function buildHeaders(): array private function packMessage(string $data): string { $compressed = $this->compression !== null ? 1 : 0; + // gRPC message frame: [1-byte compressed flag][4-byte big-endian message length][message] return pack('CN', $compressed, strlen($data)) . $data; } } diff --git a/src/Transport/SwooleGrpcTransportFactory.php b/src/Transport/SwooleGrpcTransportFactory.php index edaf074..2e24e14 100644 --- a/src/Transport/SwooleGrpcTransportFactory.php +++ b/src/Transport/SwooleGrpcTransportFactory.php @@ -11,6 +11,13 @@ final class SwooleGrpcTransportFactory implements TransportFactoryInterface { + /** + * Note: $cacert, $cert, and $key are accepted to satisfy the TransportFactoryInterface + * contract but are not implemented in this transport. TLS certificates are not supported; + * use the https:// or grpcs:// scheme for basic TLS. + * + * @param null|mixed $compression + */ public function create( string $endpoint, string $contentType = ContentTypes::PROTOBUF, @@ -45,6 +52,13 @@ public function create( $compressionType = null; } + $supportedCompression = [null, TransportFactoryInterface::COMPRESSION_GZIP, TransportFactoryInterface::COMPRESSION_DEFLATE]; + if (! in_array($compressionType, $supportedCompression, true)) { + throw new InvalidArgumentException( + sprintf('Unsupported compression type "%s"', $compressionType) + ); + } + return new SwooleGrpcTransport( host: $parsed['host'], port: $parsed['port'], @@ -53,6 +67,8 @@ public function create( timeout: $timeout, ssl: $parsed['ssl'], compression: $compressionType, + retryDelay: $retryDelay, + maxRetries: $maxRetries, ); } diff --git a/tests/Unit/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactoryTest.php b/tests/Unit/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactoryTest.php index 1a22fd2..c0dd301 100644 --- a/tests/Unit/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactoryTest.php +++ b/tests/Unit/Factory/Metric/Exporter/OtlpGrpcMetricExporterFactoryTest.php @@ -36,4 +36,17 @@ public function testMake(): void $this->assertInstanceOf(MetricExporterInterface::class, $exporter); } + + public function testMakeWithDefaultEndpoint(): void + { + $config = $this->createMock(ConfigInterface::class); + $config->method('get') + ->with('open-telemetry.metrics.exporters.otlp_grpc.options', []) + ->willReturn([]); + + $factory = new OtlpGrpcMetricExporterFactory($config); + $exporter = $factory->make(); + + $this->assertInstanceOf(MetricExporterInterface::class, $exporter); + } } diff --git a/tests/Unit/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactoryTest.php b/tests/Unit/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactoryTest.php index 0b8324c..0cc36b9 100644 --- a/tests/Unit/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactoryTest.php +++ b/tests/Unit/Factory/Trace/Exporter/OtlpGrpcTraceExporterFactoryTest.php @@ -34,4 +34,17 @@ public function testMake(): void $this->assertInstanceOf(SpanExporterInterface::class, $exporter); } + + public function testMakeWithDefaultEndpoint(): void + { + $config = $this->createMock(ConfigInterface::class); + $config->method('get') + ->with('open-telemetry.traces.exporters.otlp_grpc.options', []) + ->willReturn([]); + + $factory = new OtlpGrpcTraceExporterFactory($config); + $exporter = $factory->make(); + + $this->assertInstanceOf(SpanExporterInterface::class, $exporter); + } } diff --git a/tests/Unit/Transport/SwooleGrpcTransportFactoryTest.php b/tests/Unit/Transport/SwooleGrpcTransportFactoryTest.php index 508922b..0c100dc 100644 --- a/tests/Unit/Transport/SwooleGrpcTransportFactoryTest.php +++ b/tests/Unit/Transport/SwooleGrpcTransportFactoryTest.php @@ -119,4 +119,35 @@ public function testCreateWithCompressionNoneTreatedAsNull(): void $compressionProp = $reflection->getProperty('compression'); $this->assertNull($compressionProp->getValue($transport)); } + + public function testCreateWithRetryParameters(): void + { + $factory = new SwooleGrpcTransportFactory(); + + $transport = $factory->create( + endpoint: 'http://localhost:4317/opentelemetry.proto.collector.trace.v1.TraceService/Export', + contentType: ContentTypes::PROTOBUF, + retryDelay: 200, + maxRetries: 5, + ); + + $reflection = new ReflectionClass($transport); + + $this->assertSame(200, $reflection->getProperty('retryDelay')->getValue($transport)); + $this->assertSame(5, $reflection->getProperty('maxRetries')->getValue($transport)); + } + + public function testCreateThrowsExceptionForUnsupportedCompression(): void + { + $factory = new SwooleGrpcTransportFactory(); + + $this->expectException(InvalidArgumentException::class); + $this->expectExceptionMessage('Unsupported compression type "br"'); + + $factory->create( + endpoint: 'http://localhost:4317/opentelemetry.proto.collector.trace.v1.TraceService/Export', + contentType: ContentTypes::PROTOBUF, + compression: 'br', + ); + } } From db2648ae1be463971d740482b376844c9e93065f Mon Sep 17 00:00:00 2001 From: Lucas Angeli Date: Thu, 23 Apr 2026 12:10:03 -0300 Subject: [PATCH 2/3] feat(shutdown): flush OTel on AfterHandle and AfterExecute events Ensures traces and metrics are flushed when a Hyperf CLI command exits, in addition to the existing OnWorkerExit handler. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/Listener/OtelShutdownListener.php | 4 ++++ tests/Unit/Listener/OtelShutdownListenerTest.php | 4 +++- 2 files changed, 7 insertions(+), 1 deletion(-) diff --git a/src/Listener/OtelShutdownListener.php b/src/Listener/OtelShutdownListener.php index ef760d8..bc453d1 100644 --- a/src/Listener/OtelShutdownListener.php +++ b/src/Listener/OtelShutdownListener.php @@ -4,6 +4,8 @@ namespace Hyperf\OpenTelemetry\Listener; +use Hyperf\Command\Event\AfterExecute; +use Hyperf\Command\Event\AfterHandle; use Hyperf\Contract\StdoutLoggerInterface; use Hyperf\Coroutine\Coroutine; use Hyperf\Event\Contract\ListenerInterface; @@ -25,6 +27,8 @@ public function listen(): array { return [ OnWorkerExit::class, + AfterHandle::class, + AfterExecute::class, ]; } diff --git a/tests/Unit/Listener/OtelShutdownListenerTest.php b/tests/Unit/Listener/OtelShutdownListenerTest.php index 547b331..4cb3cd5 100644 --- a/tests/Unit/Listener/OtelShutdownListenerTest.php +++ b/tests/Unit/Listener/OtelShutdownListenerTest.php @@ -4,6 +4,8 @@ namespace Tests\Unit\Listener; +use Hyperf\Command\Event\AfterExecute; +use Hyperf\Command\Event\AfterHandle; use Hyperf\Contract\StdoutLoggerInterface; use Hyperf\Event\Contract\ListenerInterface; use Hyperf\Framework\Event\OnWorkerExit; @@ -36,7 +38,7 @@ public function testListensToOnWorkerExit(): void $this->createMock(TracerProviderInterface::class), $this->createMock(StdoutLoggerInterface::class) ); - $this->assertSame([OnWorkerExit::class], $listener->listen()); + $this->assertSame([OnWorkerExit::class, AfterHandle::class, AfterExecute::class], $listener->listen()); } public function testProcessCallsShutdownInsideCoroutine(): void From 314d9489e81af6f5631bfd832b83b09990d826e9 Mon Sep 17 00:00:00 2001 From: Lucas Angeli Date: Thu, 23 Apr 2026 12:10:14 -0300 Subject: [PATCH 3/3] style: fix method ordering in ChannelBatchSpanProcessor Move ensureInitialized() after public methods per PHP-CS-Fixer ordered_class_elements rule. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../ChannelBatchSpanProcessor.php | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/src/SpanProcessor/ChannelBatchSpanProcessor.php b/src/SpanProcessor/ChannelBatchSpanProcessor.php index 826a265..08045c0 100644 --- a/src/SpanProcessor/ChannelBatchSpanProcessor.php +++ b/src/SpanProcessor/ChannelBatchSpanProcessor.php @@ -49,19 +49,6 @@ public function __construct( ) { } - private function ensureInitialized(): void - { - if ($this->initialized) { - return; - } - - $this->initialized = true; - $this->exportContext = Context::getCurrent(); - $this->channel = new Channel($this->channelCapacity); - $this->startConsumer(); - $this->startFlushTimer(); - } - public function onStart(ReadWriteSpanInterface $span, ContextInterface $parentContext): void { } @@ -116,6 +103,19 @@ public function shutdown(?CancellationInterface $cancellation = null): bool return $this->exporter->shutdown($cancellation); } + private function ensureInitialized(): void + { + if ($this->initialized) { + return; + } + + $this->initialized = true; + $this->exportContext = Context::getCurrent(); + $this->channel = new Channel($this->channelCapacity); + $this->startConsumer(); + $this->startFlushTimer(); + } + private function pushBatch(): void { if ($this->batch === [] || $this->channel === null) {