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

$this->setTracingInstrumentationParams($container, 'messenger', $config, $isConfigEnabled);
Expand Down
10 changes: 8 additions & 2 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,8 +30,11 @@ final class TraceableConsoleEventSubscriber implements EventSubscriberInterface,
*/
private array $excludeCommands = [];

/**
* @var string[]
*/
private array $notSupportedCommands = [
'messenger:consume' // designed to run indefinitely
'messenger:consume', // designed to run indefinitely
];

public function __construct(
Expand Down Expand Up @@ -60,6 +63,9 @@ public static function getSubscribedEvents(): array
];
}

/**
* @return class-string[]
*/
public static function getSubscribedServices(): array
{
return [TracerInterface::class];
Expand Down Expand Up @@ -183,7 +189,7 @@ private function isAutoTraceable(Command $command): bool

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

private function isAttributeTraceable(Command $command): bool
Expand Down
9 changes: 4 additions & 5 deletions src/Instrumentation/Symfony/Messenger/Amqp/AddStampForPropagationMiddleware.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
<?php

declare(strict_types=1);

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

use FriendsOfOpenTelemetry\OpenTelemetryBundle\OpenTelemetry\Context\Propagator\TraceStampPropagator;
Expand All @@ -21,11 +19,12 @@
public function __construct(
private MultiTextMapPropagator $propagator,
private ?LoggerInterface $logger = null,
) {}
) {
}

public function handle(Envelope $envelope, StackInterface $stack): Envelope
{
if ($envelope->last(AmqpStamp::class) !== null) {
if (null !== $envelope->last(AmqpStamp::class)) {
$this->onMessageSent($envelope);
}

Expand All @@ -36,7 +35,7 @@ private function onMessageSent(Envelope &$envelope): void
{
$scope = Context::storage()->scope();

if ($scope === null) {
if (null === $scope) {
$this->logger?->debug('No active scope');
}

Expand Down
80 changes: 0 additions & 80 deletions src/Instrumentation/Symfony/Messenger/EventSubscriber/EndSpanEventSubscriber.php
View file
Open in desktop

This file was deleted.

77 changes: 0 additions & 77 deletions src/Instrumentation/Symfony/Messenger/EventSubscriber/StartSpanEventSubscriber.php
View file
Open in desktop

This file was deleted.

128 changes: 128 additions & 0 deletions src/Instrumentation/Symfony/Messenger/InstrumentationEventSubscriber.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,128 @@
<?php

namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger;

use FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\InstrumentationTypeEnum;
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 Psr\Log\LoggerInterface;
use Symfony\Component\EventDispatcher\EventSubscriberInterface;
use Symfony\Component\Messenger\Event\WorkerMessageFailedEvent;
use Symfony\Component\Messenger\Event\WorkerMessageHandledEvent;
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 InstrumentationEventSubscriber implements EventSubscriberInterface
Copy link
Contributor

Choose a reason for hiding this comment

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

Since this subscriber is about worker events, I think we should name it WorkerMessageEventSubscriber.

{
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'],
WorkerMessageFailedEvent::class => ['endSpanOnError'],
WorkerMessageHandledEvent::class => ['endSpanWithSuccess'],
];
}

public function startSpan(WorkerMessageReceivedEvent $event): void
{
if (InstrumentationTypeEnum::Auto !== $this->instrumentationType) {
Copy link
Contributor

Choose a reason for hiding this comment

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

I'd like to adapt this to handle manual instrumentation, for example if the message has the #[Traceable] attribute, it should be instrumented.

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 (null !== $busNameStamp) {
$span->setAttribute('bus.name', $busNameStamp->getBusName());
}

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

Context::storage()
->attach(
$span->storeInContext($context)
)
;
}

public function endSpanWithSuccess(WorkerMessageHandledEvent $event): void
{
if (InstrumentationTypeEnum::Auto !== $this->instrumentationType) {
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 endSpanOnError(WorkerMessageFailedEvent $event): void
{
if (InstrumentationTypeEnum::Auto !== $this->instrumentationType) {
return;
}

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

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

$scope->detach();

$span = Span::fromContext($scope->context());
$span->setStatus(StatusCode::STATUS_ERROR);
$span->setAttribute('exception.message', $event->getThrowable()->getMessage());
$previous = $event->getThrowable()->getPrevious();

if (null !== $previous) {
$span->setAttribute('exception.previous.message', $previous->getMessage());
}

$this->logger->debug(sprintf('Ending span "%s"', $span->getContext()->getSpanId()));
$span->end();
}
}
2 changes: 0 additions & 2 deletions src/Instrumentation/Symfony/Messenger/TraceStamp.php
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,5 @@
<?php

declare(strict_types=1);

namespace FriendsOfOpenTelemetry\OpenTelemetryBundle\Instrumentation\Symfony\Messenger;

use Symfony\Component\Messenger\Stamp\StampInterface;
Expand Down
3 changes: 0 additions & 3 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,9 +8,6 @@
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
Loading