Feat/span processors (#22)

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: lucas-angeli-gimenes

.gitignore

@@ -4,3 +4,4 @@ vendor/
 .idea/
 .phpunit.result.cache
 .phpunit.cache
+.vscode/

Makefile

@@ -13,3 +13,5 @@ update:
 test:
 	@docker-compose run --rm hyperf-opentelemetry -c "composer test"
 
+ci:
+	@docker-compose run --rm hyperf-opentelemetry -c "composer ci"

publish/open-telemetry.php

@@ -14,8 +14,11 @@
 use Hyperf\OpenTelemetry\Factory\Trace\Exporter\OtlpHttpTraceExporterFactory;
 use Hyperf\OpenTelemetry\Factory\Trace\Exporter\StdoutTraceExporterFactory;
 use Hyperf\OpenTelemetry\Factory\Trace\Processor\BatchSpanProcessorFactory;
+use Hyperf\OpenTelemetry\Factory\Trace\Processor\ChannelBatchSpanProcessorFactory;
+use Hyperf\OpenTelemetry\Factory\Trace\Processor\DeferSpanProcessorFactory;
 use Hyperf\OpenTelemetry\Factory\Trace\Processor\SimpleSpanProcessorFactory;
 use Hyperf\OpenTelemetry\Factory\Trace\Sampler\AlwaysOnSamplerFactory;
+use Hyperf\OpenTelemetry\SpanProcessor\ChannelBatchSpanProcessor;
 use OpenTelemetry\SDK\Common\Export\TransportFactoryInterface;
 use OpenTelemetry\SDK\Logs\Processor\BatchLogRecordProcessor;
 use OpenTelemetry\SDK\Metrics\Data\Temporality;
@@ -77,12 +80,23 @@
                     'schedule_delay_ms' => BatchSpanProcessor::DEFAULT_SCHEDULE_DELAY,
                     'export_timeout_ms' => BatchSpanProcessor::DEFAULT_EXPORT_TIMEOUT,
                     'max_export_batch_size' => BatchSpanProcessor::DEFAULT_MAX_EXPORT_BATCH_SIZE,
-                    'auto_flush' => false,
+                    'auto_flush' => true,
                 ],
             ],
             'simple' => [
                 'driver' => SimpleSpanProcessorFactory::class,
             ],
+            'defer' => [
+                'driver' => DeferSpanProcessorFactory::class,
+            ],
+            'channel' => [
+                'driver' => ChannelBatchSpanProcessorFactory::class,
+                'options' => [
+                    'max_batch_size' => ChannelBatchSpanProcessor::DEFAULT_MAX_EXPORT_BATCH_SIZE,
+                    'channel_capacity' => ChannelBatchSpanProcessor::DEFAULT_CHANNEL_CAPACITY,
+                    'flush_interval' => ChannelBatchSpanProcessor::DEFAULT_FLUSH_INTERVAL,
+                ],
+            ],
         ],
         'samplers' => [
             'always_on' => [

src/ConfigProvider.php

@@ -42,7 +42,6 @@ public function __invoke(): array
             ],
             'listeners' => [
                 Listener\MetricFlushListener::class,
-                Listener\TraceFlushListener::class,
                 Listener\OtelShutdownListener::class,
                 Listener\DbQueryExecutedListener::class,
             ],

src/Factory/Trace/Processor/ChannelBatchSpanProcessorFactory.php

@@ -0,0 +1,30 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hyperf\OpenTelemetry\Factory\Trace\Processor;
+
+use Hyperf\Contract\ConfigInterface;
+use Hyperf\OpenTelemetry\SpanProcessor\ChannelBatchSpanProcessor;
+use OpenTelemetry\SDK\Trace\SpanExporterInterface;
+use OpenTelemetry\SDK\Trace\SpanProcessorInterface;
+
+class ChannelBatchSpanProcessorFactory implements TraceProcessorFactoryInterface
+{
+    public function __construct(
+        protected readonly ConfigInterface $config,
+    ) {
+    }
+
+    public function make(SpanExporterInterface $exporter): SpanProcessorInterface
+    {
+        $options = $this->config->get('open-telemetry.traces.processors.channel_batch.options', []);
+
+        return new ChannelBatchSpanProcessor(
+            exporter: $exporter,
+            maxBatchSize: $options['max_batch_size'] ?? ChannelBatchSpanProcessor::DEFAULT_MAX_EXPORT_BATCH_SIZE,
+            channelCapacity: $options['channel_capacity'] ?? ChannelBatchSpanProcessor::DEFAULT_CHANNEL_CAPACITY,
+            flushInterval: $options['flush_interval'] ?? ChannelBatchSpanProcessor::DEFAULT_FLUSH_INTERVAL,
+        );
+    }
+}

src/Factory/Trace/Processor/DeferSpanProcessorFactory.php

@@ -0,0 +1,17 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hyperf\OpenTelemetry\Factory\Trace\Processor;
+
+use Hyperf\OpenTelemetry\SpanProcessor\DeferSpanProcessor;
+use OpenTelemetry\SDK\Trace\SpanExporterInterface;
+use OpenTelemetry\SDK\Trace\SpanProcessorInterface;
+
+class DeferSpanProcessorFactory implements TraceProcessorFactoryInterface
+{
+    public function make(SpanExporterInterface $exporter): SpanProcessorInterface
+    {
+        return new DeferSpanProcessor($exporter);
+    }
+}

src/Listener/DbQueryExecutedListener.php

@@ -39,6 +39,10 @@ protected function featureName(): string
         return 'db_query';
     }
 
+    /**
+     * @SuppressWarnings(PHPMD.CyclomaticComplexity)
+     * @SuppressWarnings(PHPMD.NPathComplexity)
+     */
     protected function onQueryExecuted(QueryExecuted $event): void
     {
         if (! $this->isTelemetryEnabled()) {

src/Listener/OtelShutdownListener.php

@@ -30,7 +30,7 @@ public function listen(): array
 
     public function process(object $event): void
     {
-        Coroutine::create(function () {
+        Coroutine::create(function (): void {
             try {
                 $this->tracerProvider->shutdown();
             } catch (Throwable $e) {

src/Listener/TraceFlushListener.php

@@ -33,8 +33,13 @@ public function process(ServerRequestInterface $request, RequestHandlerInterface
 
         $context = $this->instrumentation->propagator()->extract($request->getHeaders());
 
+        $name = $request->getMethod() . ' ' . Uri::sanitize(
+            $path,
+            $this->config->get('open-telemetry.traces.uri_mask', [])
+        );
+
         $scope = $this->instrumentation->startSpan(
-            name: $request->getMethod() . ' ' . Uri::sanitize($path, $this->config->get('open-telemetry.traces.uri_mask', [])),
+            name: $name,
             spanKind: SpanKind::KIND_SERVER,
             attributes: [
                 HttpAttributes::HTTP_REQUEST_METHOD => $request->getMethod(),