Skip to content

chore(messenger): use Messenger instrumentation #173

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
Open

chore(messenger): use Messenger instrumentation #173

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
cb84fec
chore(messenger): add trace parent in dispatched messages (AMQP)
jvocampings Apr 16, 2025
19e51da
chore(messenger): use Messenger events to start/end span for instrume…
jvocampings Apr 17, 2025
5f8c756
Merge remote-tracking branch 'origin/messenger-propagation' into mess…
jvocampings Apr 18, 2025
d59e8e2
remove strict type, merge start/end span into 1 subscriber, close spa…
jvocampings Apr 18, 2025
0549614
propagation not related to AMQP + read incoming trace (async context)
jvocampings Apr 18, 2025
5a10f47
rename event subscriber
jvocampings Apr 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion src/DependencyInjection/Compiler/SetInstrumentationTypePass.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,15 @@ 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.trace.start_span')) {
$container->getDefinition('open_telemetry.instrumentation.messenger.trace.start_span')
->addMethodCall('setInstrumentationType', [$messengerInstrumentationType]);
}

if ($container->hasDefinition('open_telemetry.instrumentation.messenger.trace.end_span')) {
$container->getDefinition('open_telemetry.instrumentation.messenger.trace.end_span')
->addMethodCall('setInstrumentationType', [$messengerInstrumentationType]);
}
}
}
Expand Down
2 changes: 2 additions & 0 deletions src/DependencyInjection/OpenTelemetryExtension.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -260,6 +260,8 @@ 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.start_span');
$container->removeDefinition('open_telemetry.instrumentation.messenger.trace.end_span');
}

$this->setTracingInstrumentationParams($container, 'messenger', $config, $isConfigEnabled);
Expand Down
13 changes: 13 additions & 0 deletions src/Instrumentation/Symfony/Console/TraceableConsoleEventSubscriber.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,10 @@ final class TraceableConsoleEventSubscriber implements EventSubscriberInterface,
*/
private array $excludeCommands = [];

private array $notSupportedCommands = [
'messenger:consume' // designed to run indefinitely
];

public function __construct(
private readonly TracerInterface $tracer,
/** @var ServiceLocator<TracerInterface> */
Expand Down Expand Up @@ -67,6 +71,10 @@ public function startSpan(ConsoleCommandEvent $event): void

assert($command instanceof Command);

if ($this->isNotSupported($command)) {
return;
}

if (false === $this->isAutoTraceable($command) && false === $this->isAttributeTraceable($command)) {
return;
}
Expand Down Expand Up @@ -173,6 +181,11 @@ private function isAutoTraceable(Command $command): bool
return true;
}

private function isNotSupported(Command $command): bool
{
return in_array($command->getName(), $this->notSupportedCommands);
}

private function isAttributeTraceable(Command $command): bool
{
$traceable = $this->parseAttribute($command);
Expand Down
80 changes: 80 additions & 0 deletions src/Instrumentation/Symfony/Messenger/EventSubscriber/EndSpanEventSubscriber.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
<?php

declare(strict_types=1);

namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\EventSubscriber;

use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\InstrumentationTypeEnum;
use OpenTelemetry\API\Trace\StatusCode;
use OpenTelemetry\Context\Context;
use OpenTelemetry\SDK\Trace\Span;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
use Symfony\Component\Messenger\Event\WorkerRunningEvent;
use Symfony\Component\Messenger\Event\WorkerStoppedEvent;

class EndSpanEventSubscriber implements EventSubscriberInterface
{
private ?InstrumentationTypeEnum $instrumentationType = null;

public function __construct(
private readonly LoggerInterface $logger
) {

}

public function setInstrumentationType(InstrumentationTypeEnum $instrumentationType): void
{
$this->instrumentationType = $instrumentationType;
}

public static function getSubscribedEvents()
{
return [
WorkerRunningEvent::class => ['endSpan'],
WorkerStoppedEvent::class => ['endSpan'],
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should not end a span on running/stopped worker events but on WorkerMessageReceived and WorkerMessageHandled events.

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ended on WorkerMessageHandledEvent. But I don't understand why it should also be ended on WorkerMessageReceived

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am sorry I mean, start a span on WorkerMessageReceived event and end span on WorkerMessageHandled event.

WorkerMessageFailedEvent::class => ['hydrateSpanWithError'],
];
}

public function endSpan(WorkerRunningEvent|WorkerStoppedEvent $event): void
{
if ($this->instrumentationType !== InstrumentationTypeEnum::Auto) {
return;
}

$scope = Context::storage()->scope();

if (null === $scope) {
return;
}

$scope->detach();

$span = Span::fromContext($scope->context());
$span->setStatus(StatusCode::STATUS_OK);
$this->logger->debug(sprintf('Ending span "%s"', $span->getContext()->getSpanId()));
$span->end();
}

public function hydrateSpanWithError(WorkerMessageFailedEvent $event): void
{
if ($this->instrumentationType !== InstrumentationTypeEnum::Auto) {
return;
}

$scope = Context::storage()->scope();

if (null === $scope) {
return;
}

$span = Span::fromContext($scope->context());
$span->setAttribute('exception.message', $event->getThrowable()->getMessage());
$previous = $event->getThrowable()->getPrevious();
if ($previous !== null) {
$span->setAttribute('exception.previous.message', $previous->getMessage());
}
}
}
77 changes: 77 additions & 0 deletions src/Instrumentation/Symfony/Messenger/EventSubscriber/StartSpanEventSubscriber.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
<?php

namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\EventSubscriber;

use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\InstrumentationTypeEnum;
use OpenTelemetry\API\Trace\SpanKind;
use OpenTelemetry\API\Trace\TracerInterface;
use OpenTelemetry\Context\Context;
use Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\Messenger\Event\WorkerMessageReceivedEvent;
use Symfony\Component\Messenger\Stamp\BusNameStamp;

/**
* Be aware if you start a span before this subscriber, it could leads to orphan span issue.
* Be sure your span is properly ended.
*/
class StartSpanEventSubscriber implements EventSubscriberInterface
{
private ?InstrumentationTypeEnum $instrumentationType = null;

public function __construct(
private readonly TracerInterface $tracer,
private readonly LoggerInterface $logger,
)
{

}

public function setInstrumentationType(InstrumentationTypeEnum $instrumentationType): void
{
$this->instrumentationType = $instrumentationType;
}

public static function getSubscribedEvents()
{
return [
WorkerMessageReceivedEvent::class => ['startSpan'],
];
}

public function startSpan(WorkerMessageReceivedEvent $event): void
{
if ($this->instrumentationType !== InstrumentationTypeEnum::Auto) {
return;
}

$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');
}

$context = Context::getCurrent();
$span = $this->tracer
->spanBuilder($event->getReceiverName())
->setParent($context)
->setSpanKind(SpanKind::KIND_CONSUMER)
->startSpan();

$busNameStamp = $event->getEnvelope()->last(BusNameStamp::class);

if ($busNameStamp !== null) {
$span->setAttribute('bus.name', $busNameStamp->getBusName());
}

$this->logger->debug(sprintf('Starting span "%s"', $span->getContext()->getSpanId()));

Context::storage()
->attach(
$span->storeInContext($context)
)
;
}
}
3 changes: 3 additions & 0 deletions src/Instrumentation/Symfony/Messenger/TraceableMessengerMiddleware.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,9 @@
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Middleware\StackInterface;

/**
* @deprecated: span should be created from event subscriber to avoid issue with orphan span
*/
class TraceableMessengerMiddleware implements MiddlewareInterface
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think there is misunderstanding here, we still need the middleware but maybe there is some adjustment to do.
You have to thing about the fact that a message can handled synchronously and it won't trigger worker events.

According to Symfony's documentation:

The middleware are executed when the message is dispatched but also again when a message is received via the worker (for messages that were sent to a transport to be handled asynchronously). Keep this in mind if you create your own middleware.

We need to find a way to either trace from the middleware when a synchronous message is dispatched or attach the trace context with a stamp when the message dispatched to the transport and trace it when the message is received and handled.

Here is the rolling of events:

  1. a message is dispatched
  2. the message goes through middlewares -> span start if sync
  3. the message is handled if synchronous -> span end if sync
  4. the message is sent to transport if asynchronous -> attach tracing context without span creation
  5. the message is received by the worker -> span start
  6. the message goes through middlewares -> no span creation
  7. the message is successfully handled by the worker -> span end
  8. the messagge has failed to be handled by the worker -> span end

By reviewing this, I think we could allow span creation within the middleware in an asynchronous context too as it will detail the every phases of the message, the dispatch phase and the handling phase as with synchronous message the dispatch/handling phase happen at the same time.

Still, we need to do some experiments, in each case to track down spans and make sure they are relevant, well linked and ended correctly.

{
public function __construct(
Expand Down
11 changes: 11 additions & 0 deletions src/Resources/config/services_tracing_instrumentation.php
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
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\EventSubscriber\EndSpanEventSubscriber;
use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\EventSubscriber\StartSpanEventSubscriber;
use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\TraceableMessengerMiddleware;
use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\TraceableMessengerTransport;
use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger\TraceableMessengerTransportFactory;
Expand Down Expand Up @@ -103,6 +105,15 @@
->tag('monolog.logger', ['channel' => 'open_telemetry'])
->alias('messenger.middleware.open_telemetry_tracer', 'open_telemetry.instrumentation.messenger.trace.middleware')

->set('open_telemetry.instrumentation.messenger.trace.start_span', StartSpanEventSubscriber::class)
->arg('$tracer', service('open_telemetry.traces.default_tracer'))
->tag('monolog.logger', ['channel' => 'open_telemetry'])
->tag('kernel.event_subscriber')

->set('open_telemetry.instrumentation.messenger.trace.end_span', EndSpanEventSubscriber::class)
->tag('monolog.logger', ['channel' => 'open_telemetry'])
->tag('kernel.event_subscriber')

// Twig
->set('open_telemetry.instrumentation.twig.trace.extension', TraceableTwigExtension::class)
->arg('$tracer', service('open_telemetry.traces.default_tracer'))
Expand Down