chore(messenger): add trace parent in dispatched messages (AMQP)

Author: jvocampings

composer.json

@@ -33,9 +33,11 @@
         "open-telemetry/sdk": "^1",
         "open-telemetry/sem-conv": "^1",
         "open-telemetry/symfony-sdk-bundle": "^0",
+        "symfony/amqp-messenger": "^7.2",
         "symfony/config": "^7.2",
         "symfony/dependency-injection": "^7.2",
         "symfony/event-dispatcher": "^7.2",
+        "symfony/messenger": "^7.2",
         "zenstruck/dsn": "^0.2",
         "zenstruck/uri": "^2.3"
     },
@@ -67,7 +69,6 @@
         "symfony/http-client": "^7.2",
         "symfony/http-kernel": "^7.2",
         "symfony/mailer": "^7.2",
-        "symfony/messenger": "^7.2",
         "symfony/monolog-bundle": "^3.10",
         "symfony/phpunit-bridge": "^7.2",
         "symfony/runtime": "^7.2",

src/Instrumentation/Symfony/Messenger/Amqp/AddStampForPropagationMiddleware.php

@@ -0,0 +1,46 @@
+<?php
+
+declare(strict_types=1);
+
+namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\Amqp;
+
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator\TraceStampPropagator;
+use OpenTelemetry\Context\Context;
+use OpenTelemetry\Context\Propagation\MultiTextMapPropagator;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\Messenger\Bridge\Amqp\Transport\AmqpStamp;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
+use Symfony\Component\Messenger\Middleware\StackInterface;
+
+/**
+ * Be aware the app consuming the message must be able to denormalize the stamp.
+ */
+readonly class AddStampForPropagationMiddleware implements MiddlewareInterface
+{
+    public function __construct(
+        private MultiTextMapPropagator $propagator,
+        private ?LoggerInterface $logger = null,
+    ) {}
+
+    public function handle(Envelope $envelope, StackInterface $stack): Envelope
+    {
+        if ($envelope->last(AmqpStamp::class) !== null) {
+            $this->onMessageSent($envelope);
+        }
+
+        return $stack->next()->handle($envelope, $stack);
+    }
+
+    private function onMessageSent(Envelope &$envelope): void
+    {
+        $scope = Context::storage()->scope();
+
+        if ($scope === null) {
+            $this->logger?->debug('No active scope');
+        }
+
+        $this->propagator->inject($envelope, new TraceStampPropagator(), Context::getCurrent());
+        $this->logger?->debug('Trace stamp added to envelope for propagation');
+    }
+}

src/Instrumentation/Symfony/Messenger/TraceStamp.php

@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger;
+
+use Symfony\Component\Messenger\Stamp\StampInterface;
+
+/**
+ * @doc: https://www.w3.org/TR/trace-context/
+ *
+ * You can see how the trace parent generated here: https://github.com/open-telemetry/opentelemetry-php/blob/main/src/API/Trace/Propagation/TraceContextPropagator.php
+ */
+readonly class TraceStamp implements StampInterface
+{
+    public function __construct(
+        private string $traceParent,
+    ) {
+    }
+
+    public function getTraceParent(): string
+    {
+        return $this->traceParent;
+    }
+}

src/OpenTelemetry/Context/Propagator/PropagatorFactory.php

@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator;
+
+use OpenTelemetry\API\Baggage\Propagation\BaggagePropagator;
+use OpenTelemetry\API\Trace\Propagation\TraceContextPropagator;
+use OpenTelemetry\Context\Propagation\MultiTextMapPropagator;
+use OpenTelemetry\Context\Propagation\NoopTextMapPropagator;
+use OpenTelemetry\Context\Propagation\TextMapPropagatorInterface;
+use OpenTelemetry\SDK\Registry;
+
+class PropagatorFactory
+{
+    /**
+     * Default propagators from https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#general-sdk-configuration
+     */
+    public static function createDefault(): MultiTextMapPropagator
+    {
+        return new MultiTextMapPropagator([
+            BaggagePropagator::getInstance(),
+            TraceContextPropagator::getInstance()
+        ]);
+    }
+}

src/OpenTelemetry/Context/Propagator/TraceStampPropagator.php

@@ -0,0 +1,29 @@
+<?php
+
+declare(strict_types=1);
+
+namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator;
+
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\TraceStamp;
+use OpenTelemetry\API\Trace\Propagation\TraceContextPropagator;
+use OpenTelemetry\Context\Propagation\PropagationSetterInterface;
+use Symfony\Component\Messenger\Envelope;
+
+class TraceStampPropagator implements PropagationSetterInterface
+{
+    /**
+     * @param mixed $carrier
+     */
+    public function set(&$carrier, string $key, string $value): void
+    {
+        if (!$carrier instanceof Envelope) {
+            throw new \InvalidArgumentException(sprintf('The carrier for trace stamp propagation must be instance of %s', Envelope::class));
+        }
+
+        if ($key !== TraceContextPropagator::TRACEPARENT) {
+            return;
+        }
+
+        $carrier = $carrier->with(new TraceStamp($value));
+    }
+}

src/Resources/config/services.php

@@ -1,6 +1,7 @@
 <?php
 
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator\HeadersPropagator as HeadersPropagationGetter;
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator\PropagatorFactory;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Exporter\ExporterDsn;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Exporter\OtlpExporterOptions;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Resource\ResourceInfoFactory;
@@ -33,6 +34,7 @@
 
         ->set('open_telemetry.propagator_text_map.noop', NoopTextMapPropagator::class)
         ->set('open_telemetry.propagator_text_map.multi', MultiTextMapPropagator::class)
+            ->factory([PropagatorFactory::class, 'createDefault'])
 
         ->set('open_telemetry.propagation_getter.headers', HeadersPropagationGetter::class)
         ->set('open_telemetry.propagation_getter.sanitize_combined_headers', SanitizeCombinedHeadersPropagationGetter::class)

src/Resources/config/services_tracing_instrumentation.php

@@ -9,6 +9,7 @@
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\HttpKernel\TraceableHttpKernelEventSubscriber;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Mailer\TraceableMailer;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Mailer\TraceableMailerTransport;
+use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\Amqp\AddStampForPropagationMiddleware;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\TraceableMessengerMiddleware;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\TraceableMessengerTransport;
 use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\TraceableMessengerTransportFactory;
@@ -103,6 +104,11 @@
             ->tag('monolog.logger', ['channel' => 'open_telemetry'])
         ->alias('messenger.middleware.open_telemetry_tracer', 'open_telemetry.instrumentation.messenger.trace.middleware')
 
+        ->set('open_telemetry.instrumentation.messenger.propagation.amqp.middleware', AddStampForPropagationMiddleware::class)
+            ->arg('$propagator', service('open_telemetry.propagator_text_map.multi'))
+            ->tag('messenger.middleware')
+            ->tag('monolog.logger', ['channel' => 'open_telemetry'])
+
         // Twig
         ->set('open_telemetry.instrumentation.twig.trace.extension', TraceableTwigExtension::class)
             ->arg('$tracer', service('open_telemetry.traces.default_tracer'))