feat(Instrumentation/Messenger): implement worker and middleware instrumentation

.gitignore

@@ -25,4 +25,3 @@ CLAUDE.md
 .claude
 GEMINI.md
 .gemini
-

composer.json

@@ -62,6 +62,7 @@
         "phpunit/phpunit": "^13.0",
         "pyrech/composer-changelogs": "^2.2",
         "roave/security-advisories": "dev-master",
+        "symfony/amqp-messenger": "^7.4",
         "symfony/browser-kit": "^7.4",
         "symfony/cache": "^7.4",
         "symfony/framework-bundle": "^7.4",
@@ -127,9 +128,13 @@
     },
     "scripts": {
         "check-reqs": "@php tools/composer-require-checker/vendor/bin/composer-require-checker check",
-        "coverage": [
+        "coverage:html": [
             "@putenv XDEBUG_MODE=coverage",
-            "@phpunit --coverage-html=coverage"
+            "@phpunit --coverage-html=coverage/html"
+        ],
+        "coverage:xml": [
+            "@putenv XDEBUG_MODE=coverage",
+            "@phpunit --coverage-clover=coverage/coverage.xml"
         ],
         "format": [
             "@php-cs-fixer:fix",

phpstan-baseline.neon

@@ -12,6 +12,12 @@ parameters:
 			count: 1
 			path: src/Instrumentation/Doctrine/Middleware/TraceableConnectionV4.php
 
+		-
+			message: '#^Property FriendsOfOpenTelemetry\\OpenTelemetryBundle\\Instrumentation\\Symfony\\Messenger\\AddStampForPropagationMiddleware\:\:\$logger is never read, only written\.$#'
+			identifier: property.onlyWritten
+			count: 1
+			path: src/Instrumentation/Symfony/Messenger/AddStampForPropagationMiddleware.php
+
 		-
 			message: '#^Property FriendsOfOpenTelemetry\\OpenTelemetryBundle\\Instrumentation\\Doctrine\\Middleware\\TraceableStatementV4\:\:\$logger is never read, only written\.$#'
 			identifier: property.onlyWritten
@@ -90,6 +96,24 @@ parameters:
 			count: 1
 			path: tests/Unit/OpenTelemetry/Trace/TracerProvider/NoopTracerProviderFactoryTest.php
 
+		-
+			message: '#^Method FriendsOfOpenTelemetry\\OpenTelemetryBundle\\OpenTelemetry\\Context\\Propagator\\TraceStampPropagator\:\:set\(\) has parameter \$carrier with no type specified\.$#'
+			identifier: missingType.parameter
+			count: 1
+			path: src/OpenTelemetry/Context/Propagator/TraceStampPropagator.php
+
+		-
+			message: '#^Method FriendsOfOpenTelemetry\\OpenTelemetryBundle\\OpenTelemetry\\Context\\Propagator\\TraceStampPropagator\:\:keys\(\) has parameter \$carrier with no type specified\.$#'
+			identifier: missingType.parameter
+			count: 1
+			path: src/OpenTelemetry/Context/Propagator/TraceStampPropagator.php
+
+		-
+			message: '#^Method FriendsOfOpenTelemetry\\OpenTelemetryBundle\\OpenTelemetry\\Context\\Propagator\\TraceStampPropagator\:\:get\(\) has parameter \$carrier with no type specified\.$#'
+			identifier: missingType.parameter
+			count: 1
+			path: src/OpenTelemetry/Context/Propagator/TraceStampPropagator.php
+
 		-
 			message: '#^Call to an undefined static method ReflectionMethod\:\:createFromMethodName\(\)\.$#'
 			identifier: staticMethod.notFound

src/DependencyInjection/Compiler/SetInstrumentationTypePass.php

@@ -27,7 +27,10 @@ public function process(ContainerBuilder $container): void
 
         if ($container->hasParameter('open_telemetry.instrumentation.messenger.type')) {
             $messengerInstrumentationType = $container->getParameter('open_telemetry.instrumentation.messenger.type');
-            if ($container->hasDefinition('open_telemetry.instrumentation.http_kernel.trace.event_subscriber')) {
+
+            if ($container->hasDefinition('open_telemetry.instrumentation.messenger.worker')) {
+                $container->getDefinition('open_telemetry.instrumentation.messenger.worker')
+                    ->addMethodCall('setInstrumentationType', [$messengerInstrumentationType]);
             }
         }
     }

src/DependencyInjection/Compiler/TracerLocatorPass.php

@@ -23,6 +23,10 @@ public function process(ContainerBuilder $container): void
                 $traceableHttpKernelEventSubscriber = $container->getDefinition('open_telemetry.instrumentation.http_kernel.trace.event_subscriber');
                 $this->setTracerLocatorArgument($container, $traceableHttpKernelEventSubscriber, $tracers);
             }
+            if ($container->hasDefinition('open_telemetry.instrumentation.messenger.worker')) {
+                $messengerWorkerSubscriber = $container->getDefinition('open_telemetry.instrumentation.messenger.worker');
+                $this->setTracerLocatorArgument($container, $messengerWorkerSubscriber, $tracers);
+            }
         }
     }
 

src/DependencyInjection/Configuration.php

@@ -104,6 +104,7 @@ private function addInstrumentationSection(ArrayNodeDefinition $node): void
                                 (new ArrayNodeDefinition('exclude_commands'))
                                     ->info('Exclude commands from auto instrumentation')
                                     ->scalarPrototype()->cannotBeEmpty()->end()
+                                    ->defaultValue(['messenger:consume'])
                             ))
                             ->append($this->getMeteringInstrumentationNode())
                         ->end()

src/DependencyInjection/OpenTelemetryExtension.php

@@ -283,6 +283,9 @@ private function registerMessengerTracingInstrumentationConfiguration(ContainerB
             $container->removeDefinition('open_telemetry.instrumentation.messenger.trace.transport');
             $container->removeDefinition('open_telemetry.instrumentation.messenger.trace.transport_factory');
             $container->removeDefinition('open_telemetry.instrumentation.messenger.trace.middleware');
+            $container->removeDefinition('open_telemetry.instrumentation.messenger.trace_stamp_propagator');
+            $container->removeDefinition('open_telemetry.instrumentation.messenger.propagation.middleware');
+            $container->removeDefinition('open_telemetry.instrumentation.messenger.worker');
         }
 
         $this->setTracingInstrumentationParams($container, 'messenger', $config, $isConfigEnabled);

src/Instrumentation/Symfony/Console/TraceableConsoleEventSubscriber.php

@@ -56,6 +56,9 @@ public static function getSubscribedEvents(): array
         ];
     }
 
+    /**
+     * @return class-string[]
+     */
     public static function getSubscribedServices(): array
     {
         return [TracerInterface::class];

src/Instrumentation/Symfony/HttpClient/TraceableResponse.php

@@ -138,8 +138,8 @@ private function endSpan(): void
         $statusCode = $this->response->getStatusCode();
         if (0 !== $statusCode && $this->span->isRecording()) {
             $headers = $this->response->getHeaders(false);
-            if (isset($headers['Content-Length'])) {
-                $this->span->setAttribute(HttpIncubatingAttributes::HTTP_RESPONSE_BODY_SIZE, $headers['Content-Length']);
+            if (isset($headers['content-length'][0])) {
+                $this->span->setAttribute(HttpIncubatingAttributes::HTTP_RESPONSE_BODY_SIZE, $headers['content-length'][0]);
             }
 
             $this->span->setAttribute(HttpAttributes::HTTP_RESPONSE_STATUS_CODE, $statusCode);

src/Instrumentation/Symfony/Messenger/AddStampForPropagationMiddleware.php

@@ -0,0 +1,44 @@
+<?php
+
+namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger;
+
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator\TraceStampPropagator;
+use OpenTelemetry\Context\Context;
+use OpenTelemetry\Context\Propagation\MultiTextMapPropagator;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Middleware\StackInterface;
+
+/**
+ * Injects the current trace context into the Messenger envelope as a TraceStamp
+ * so that trace propagation is maintained across asynchronous message boundaries.
+ */
+final readonly class AddStampForPropagationMiddleware implements MiddlewareInterface
+{
+    public function __construct(
+        private MultiTextMapPropagator $propagator,
+        private TraceStampPropagator $traceStampPropagator,
+        private ?LoggerInterface $logger = null,
+    ) {
+    }
+
+    public function handle(Envelope $envelope, StackInterface $stack): Envelope
+    {
+        $traceStamp = $envelope->last(TraceStamp::class);
+
+        if (null !== $traceStamp) {
+            return $stack->next()->handle($envelope, $stack);
+        }
+
+        $scope = Context::storage()->scope();
+
+        if (null !== $scope) {
+            // inject() mutates $envelope by reference through the TraceStampPropagator setter,
+            // because Envelope is immutable and with() returns a new instance.
+            $this->propagator->inject($envelope, $this->traceStampPropagator, Context::getCurrent());
+        }
+
+        return $stack->next()->handle($envelope, $stack);
+    }
+}

src/Instrumentation/Symfony/Messenger/TraceStamp.php

@@ -0,0 +1,27 @@
+<?php
+
+namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger;
+
+use Symfony\Component\Messenger\Stamp\StampInterface;
+
+/**
+ * @see https://www.w3.org/TR/trace-context/
+ */
+final readonly class TraceStamp implements StampInterface
+{
+    public function __construct(
+        private string $traceParent,
+        private ?string $traceState = null,
+    ) {
+    }
+
+    public function getTraceParent(): string
+    {
+        return $this->traceParent;
+    }
+
+    public function getTraceState(): ?string
+    {
+        return $this->traceState;
+    }
+}

src/Instrumentation/Symfony/Messenger/TraceableMessengerMiddleware.php

@@ -30,6 +30,9 @@ public function handle(Envelope $envelope, StackInterface $stack): Envelope
 
         try {
             return $stack->next()->handle($envelope, $stack);
+        } catch (\Throwable $exception) {
+            $stack->stop($exception);
+            throw $exception;
         } finally {
             $stack->stop();
         }

src/Instrumentation/Symfony/Messenger/TraceableMessengerStack.php

@@ -2,18 +2,35 @@
 
 namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger;
 
+use OpenTelemetry\API\Trace\SpanInterface;
 use OpenTelemetry\API\Trace\SpanKind;
 use OpenTelemetry\API\Trace\StatusCode;
 use OpenTelemetry\API\Trace\TracerInterface;
 use OpenTelemetry\Context\Context;
-use OpenTelemetry\SDK\Trace\Span;
+use OpenTelemetry\Context\ContextInterface;
+use OpenTelemetry\Context\ContextStorageScopeInterface;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
 use Symfony\Component\Messenger\Middleware\StackInterface;
 
+/**
+ * Wraps a middleware stack to create one span per middleware in the chain.
+ *
+ * Follows a "stop previous, start new" pattern (like Symfony's own TraceableStack
+ * with its Stopwatch): each call to next() closes the span/scope from the prior
+ * call before opening a new one, so at most one scope is active at any time.
+ *
+ * A simpler alternative would be to drop this class entirely and wrap the whole
+ * $stack->next()->handle() chain in a single span inside TraceableMessengerMiddleware
+ * (one activate/detach pair in try/finally). That removes per-middleware timing
+ * granularity but eliminates scope management complexity.
+ */
 class TraceableMessengerStack implements StackInterface
 {
     private ?string $currentEvent = null;
+    private ?ContextStorageScopeInterface $currentScope = null;
+    private ?SpanInterface $currentSpan = null;
+    private ?ContextInterface $parentContext = null;
 
     public function __construct(
         private TracerInterface $tracer,
@@ -26,36 +43,31 @@ public function __construct(
 
     public function next(): MiddlewareInterface
     {
-        $scope = Context::storage()->scope();
-        if (null !== $scope) {
-            $this->logger?->debug(sprintf('Using scope "%s"', spl_object_id($scope)));
-        } else {
-            $this->logger?->debug('No active scope');
+        // "Stop previous" — close the span/scope from the prior next() call
+        if (null !== $this->currentScope) {
+            $this->logger?->debug(sprintf('Detaching scope "%s"', spl_object_id($this->currentScope)));
+            $this->currentScope->detach();
+            $this->currentScope = null;
+
+            if (null !== $this->currentSpan) {
+                $this->currentSpan->setStatus(StatusCode::STATUS_OK);
+                $this->logger?->debug(sprintf('Ending span "%s"', $this->currentSpan->getContext()->getSpanId()));
+                $this->currentSpan->end();
+                $this->currentSpan = null;
+            }
         }
 
-        /*        if (null !== $scope) {
-                    $span = Span::fromContext($scope->context());
-
-                    if ($span->isRecording()) {
-                        $scope->detach();
+        // Capture the parent context once (on the first call)
+        $this->parentContext ??= Context::getCurrent();
 
-                        $span->setStatus(StatusCode::STATUS_OK);
-                        $this->logger?->debug(sprintf('Ending span "%s"', $span->getContext()->getSpanId()));
-                        $span->end();
-                    }
-                }*/
-
-        $spanBuilder = $this->tracer
+        // "Start new"
+        $span = $this->tracer
             ->spanBuilder('messenger.middleware')
             ->setSpanKind(SpanKind::KIND_INTERNAL)
-            ->setParent($scope?->context())
-            ->setAttribute('event.category', $this->eventCategory)
-            ->setAttribute('bus.name', $this->busName)
-        ;
-
-        $parent = Context::getCurrent();
-
-        $span = $spanBuilder->setParent($parent)->startSpan();
+            ->setParent($this->parentContext)
+            ->setAttribute('symfony.messenger.event.category', $this->eventCategory)
+            ->setAttribute('symfony.messenger.bus.name', $this->busName)
+            ->startSpan();
 
         $this->logger?->debug(sprintf('Starting span "%s"', $span->getContext()->getSpanId()));
 
@@ -66,32 +78,44 @@ public function next(): MiddlewareInterface
         }
         $this->currentEvent .= sprintf(' on "%s"', $this->busName);
 
-        $span->setAttribute('event.current', $this->currentEvent);
+        $span->setAttribute('symfony.messenger.event.current', $this->currentEvent);
 
-        $context = $span->storeInContext($parent);
-        Context::storage()->attach($context);
+        $context = $span->storeInContext($this->parentContext);
+        $this->currentScope = Context::storage()->attach($context);
+        $this->currentSpan = $span;
 
         return $nextMiddleware;
     }
 
-    public function stop(): void
+    public function stop(?\Throwable $throwable = null): void
     {
-        $scope = Context::storage()->scope();
-        if (null === $scope) {
-            return;
+        if (null !== $this->currentScope) {
+            $this->logger?->debug(sprintf('Detaching scope "%s"', spl_object_id($this->currentScope)));
+            $this->currentScope->detach();
+            $this->currentScope = null;
         }
 
-        $scope->detach();
+        if (null !== $this->currentSpan) {
+            if (null !== $throwable) {
+                $this->currentSpan->recordException($throwable);
+                $this->currentSpan->setStatus(StatusCode::STATUS_ERROR, $throwable->getMessage());
+            } else {
+                $this->currentSpan->setStatus(StatusCode::STATUS_OK);
+            }
+            $this->logger?->debug(sprintf('Ending span "%s"', $this->currentSpan->getContext()->getSpanId()));
+            $this->currentSpan->end();
+            $this->currentSpan = null;
+        }
 
-        $span = Span::fromContext($scope->context());
-        $span->setStatus(StatusCode::STATUS_OK);
-        $this->logger?->debug(sprintf('Ending span "%s"', $span->getContext()->getSpanId()));
-        $span->end();
         $this->currentEvent = null;
     }
 
     public function __clone()
     {
         $this->stack = clone $this->stack;
+        $this->currentScope = null;
+        $this->currentSpan = null;
+        $this->parentContext = null;
+        $this->currentEvent = null;
     }
 }