77use OpenTelemetry \API \Globals ;
88use OpenTelemetry \API \Instrumentation \CachedInstrumentation ;
99use OpenTelemetry \API \Trace \Span ;
10+ use OpenTelemetry \API \Trace \SpanBuilderInterface ;
1011use OpenTelemetry \API \Trace \SpanKind ;
1112use OpenTelemetry \API \Trace \StatusCode ;
1213use OpenTelemetry \Context \Context ;
2627use Symfony \Component \Messenger \Transport \Receiver \ReceiverInterface ;
2728use Symfony \Component \Messenger \Transport \Sender \SenderInterface ;
2829use Symfony \Component \Messenger \Worker ;
29- use OpenTelemetry \API \Trace \SpanBuilderInterface ;
3030
3131// Add Amazon SQS stamp class if available
3232if (\class_exists ('Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsReceivedStamp ' )) {
@@ -42,10 +42,6 @@ class_alias(
4242 */
4343final class MessengerInstrumentation
4444{
45- const ATTRIBUTE_MESSAGING_SYSTEM = 'messaging.system ' ;
46- const ATTRIBUTE_MESSAGING_OPERATION = 'messaging.operation ' ;
47- const ATTRIBUTE_MESSAGING_DESTINATION = 'messaging.destination ' ;
48- const ATTRIBUTE_MESSAGING_MESSAGE_ID = 'messaging.message_id ' ;
4945 const ATTRIBUTE_MESSAGING_MESSAGE = 'messaging.message ' ;
5046 const ATTRIBUTE_MESSAGING_BUS = 'messaging.symfony.bus ' ;
5147 const ATTRIBUTE_MESSAGING_HANDLER = 'messaging.symfony.handler ' ;
@@ -54,10 +50,11 @@ final class MessengerInstrumentation
5450 const ATTRIBUTE_MESSAGING_DELAY = 'messaging.symfony.delay ' ;
5551 const ATTRIBUTE_MESSAGING_RETRY_COUNT = 'messaging.symfony.retry_count ' ;
5652 const ATTRIBUTE_MESSAGING_STAMPS = 'messaging.symfony.stamps ' ;
53+ const ATTRIBUTE_MESSAGING_MIDDLEWARE = 'symfony.messenger.middleware ' ;
54+ const ATTRIBUTE_MESSAGING_CONSUMED_BY_WORKER = 'messaging.symfony.consumed_by_worker ' ;
5755
5856 // Constants used in tests
5957 const ATTRIBUTE_MESSENGER_BUS = self ::ATTRIBUTE_MESSAGING_BUS ;
60- const ATTRIBUTE_MESSENGER_TRANSPORT = self ::ATTRIBUTE_MESSAGING_DESTINATION ;
6158 const ATTRIBUTE_MESSENGER_MESSAGE = self ::ATTRIBUTE_MESSAGING_MESSAGE ;
6259
6360 /** @psalm-suppress PossiblyUnusedMethod */
@@ -96,8 +93,8 @@ public static function register(): void
9693 ->setAttribute (TraceAttributes::CODE_NAMESPACE , $ class )
9794 ->setAttribute (TraceAttributes::CODE_FILEPATH , $ filename )
9895 ->setAttribute (TraceAttributes::CODE_LINE_NUMBER , $ lineno )
99- ->setAttribute (self :: ATTRIBUTE_MESSAGING_SYSTEM , 'symfony ' )
100- ->setAttribute (self :: ATTRIBUTE_MESSAGING_OPERATION , 'dispatch ' )
96+ ->setAttribute (TraceAttributes:: MESSAGING_SYSTEM , 'symfony ' )
97+ ->setAttribute (TraceAttributes:: MESSAGING_OPERATION_TYPE , 'dispatch ' )
10198 ->setAttribute (self ::ATTRIBUTE_MESSAGING_MESSAGE , $ messageClass )
10299 ->setAttribute (self ::ATTRIBUTE_MESSAGING_BUS , $ class )
103100 ;
@@ -174,10 +171,10 @@ public static function register(): void
174171 ->setAttribute (TraceAttributes::CODE_NAMESPACE , $ class )
175172 ->setAttribute (TraceAttributes::CODE_FILEPATH , $ filename )
176173 ->setAttribute (TraceAttributes::CODE_LINE_NUMBER , $ lineno )
177- ->setAttribute (self :: ATTRIBUTE_MESSAGING_SYSTEM , 'symfony ' )
178- ->setAttribute (self :: ATTRIBUTE_MESSAGING_OPERATION , 'send ' )
174+ ->setAttribute (TraceAttributes:: MESSAGING_SYSTEM , 'symfony ' )
175+ ->setAttribute (TraceAttributes:: MESSAGING_OPERATION_TYPE , 'send ' )
179176 ->setAttribute (self ::ATTRIBUTE_MESSAGING_MESSAGE , $ messageClass )
180- ->setAttribute (self :: ATTRIBUTE_MESSAGING_DESTINATION , $ class )
177+ ->setAttribute (TraceAttributes:: MESSAGING_DESTINATION_NAME , $ class )
181178 ;
182179
183180 self ::addMessageStampsToSpan ($ builder , $ envelope );
@@ -257,9 +254,9 @@ public static function register(): void
257254 ->setAttribute (TraceAttributes::CODE_NAMESPACE , $ class )
258255 ->setAttribute (TraceAttributes::CODE_FILEPATH , $ filename )
259256 ->setAttribute (TraceAttributes::CODE_LINE_NUMBER , $ lineno )
260- ->setAttribute (self :: ATTRIBUTE_MESSAGING_SYSTEM , 'symfony ' )
261- ->setAttribute (self :: ATTRIBUTE_MESSAGING_OPERATION , 'receive ' )
262- ->setAttribute (self :: ATTRIBUTE_MESSAGING_DESTINATION , $ transportName )
257+ ->setAttribute (TraceAttributes:: MESSAGING_SYSTEM , 'symfony ' )
258+ ->setAttribute (TraceAttributes:: MESSAGING_OPERATION_TYPE , 'receive ' )
259+ ->setAttribute (TraceAttributes:: MESSAGING_DESTINATION_NAME , $ transportName )
263260 ;
264261
265262 self ::addMessageStampsToSpan ($ builder , $ envelope );
@@ -324,8 +321,8 @@ public static function register(): void
324321 ->setAttribute (TraceAttributes::CODE_NAMESPACE , $ class )
325322 ->setAttribute (TraceAttributes::CODE_FILEPATH , $ filename )
326323 ->setAttribute (TraceAttributes::CODE_LINE_NUMBER , $ lineno )
327- ->setAttribute (self :: ATTRIBUTE_MESSAGING_SYSTEM , 'symfony ' )
328- ->setAttribute (self :: ATTRIBUTE_MESSAGING_OPERATION , 'process ' )
324+ ->setAttribute (TraceAttributes:: MESSAGING_SYSTEM , 'symfony ' )
325+ ->setAttribute (TraceAttributes:: MESSAGING_OPERATION_TYPE , 'process ' )
329326 ->setAttribute (self ::ATTRIBUTE_MESSAGING_MESSAGE , $ messageClass )
330327 ;
331328
@@ -391,8 +388,8 @@ public static function register(): void
391388 ->setAttribute (TraceAttributes::CODE_NAMESPACE , $ class )
392389 ->setAttribute (TraceAttributes::CODE_FILEPATH , $ filename )
393390 ->setAttribute (TraceAttributes::CODE_LINE_NUMBER , $ lineno )
394- ->setAttribute (self :: ATTRIBUTE_MESSAGING_SYSTEM , 'symfony ' )
395- ->setAttribute (self :: ATTRIBUTE_MESSAGING_OPERATION , 'process ' )
391+ ->setAttribute (TraceAttributes:: MESSAGING_SYSTEM , 'symfony ' )
392+ ->setAttribute (TraceAttributes:: MESSAGING_OPERATION_TYPE , 'process ' )
396393 ->setAttribute (self ::ATTRIBUTE_MESSAGING_MESSAGE , $ messageClass )
397394 ->setAttribute (self ::ATTRIBUTE_MESSAGING_HANDLER , $ handlerClass )
398395 ;
@@ -460,10 +457,10 @@ public static function register(): void
460457 ->setAttribute (TraceAttributes::CODE_NAMESPACE , $ class )
461458 ->setAttribute (TraceAttributes::CODE_FILEPATH , $ filename )
462459 ->setAttribute (TraceAttributes::CODE_LINE_NUMBER , $ lineno )
463- ->setAttribute (self :: ATTRIBUTE_MESSAGING_SYSTEM , 'symfony ' )
464- ->setAttribute (self :: ATTRIBUTE_MESSAGING_OPERATION , 'middleware ' )
460+ ->setAttribute (TraceAttributes:: MESSAGING_SYSTEM , 'symfony ' )
461+ ->setAttribute (TraceAttributes:: MESSAGING_OPERATION_TYPE , 'middleware ' )
465462 ->setAttribute (self ::ATTRIBUTE_MESSAGING_MESSAGE , $ messageClass )
466- ->setAttribute (' messaging.symfony.middleware ' , $ middlewareClass )
463+ ->setAttribute (self :: ATTRIBUTE_MESSAGING_MIDDLEWARE , $ middlewareClass )
467464 ;
468465
469466 self ::addMessageStampsToSpan ($ builder , $ envelope );
@@ -514,14 +511,14 @@ private static function addMessageStampsToSpan(SpanBuilderInterface $builder, En
514511 $ sentStamp = $ envelope ->last (SentStamp::class);
515512 $ transportMessageIdStamp = $ envelope ->last (TransportMessageIdStamp::class);
516513
517- $ messageClass = \get_class ($ envelope ->getMessage ());
518- $ transportName = null ;
519- $ operation = null ;
520-
521514 if ($ busStamp ) {
522515 $ builder ->setAttribute (self ::ATTRIBUTE_MESSAGING_BUS , $ busStamp ->getBusName ());
523516 }
524517
518+ if ($ consumedByWorkerStamp ) {
519+ $ builder ->setAttribute (self ::ATTRIBUTE_MESSAGING_CONSUMED_BY_WORKER , true );
520+ }
521+
525522 if ($ handledStamp ) {
526523 $ builder ->setAttribute (self ::ATTRIBUTE_MESSAGING_HANDLER , $ handledStamp ->getHandlerName ());
527524 }
@@ -533,17 +530,13 @@ private static function addMessageStampsToSpan(SpanBuilderInterface $builder, En
533530
534531 if ($ sentStamp ) {
535532 $ builder ->setAttribute (self ::ATTRIBUTE_MESSAGING_SENDER , $ sentStamp ->getSenderClass ());
536- $ builder ->setAttribute (self ::ATTRIBUTE_MESSAGING_DESTINATION , $ sentStamp ->getSenderAlias ());
537- $ transportName = $ sentStamp ->getSenderAlias ();
538- $ operation = 'send ' ;
533+ $ builder ->setAttribute (TraceAttributes::MESSAGING_DESTINATION_NAME , $ sentStamp ->getSenderAlias ());
539534 } elseif ($ receivedStamp ) {
540- $ builder ->setAttribute (self ::ATTRIBUTE_MESSAGING_DESTINATION , $ receivedStamp ->getTransportName ());
541- $ transportName = $ receivedStamp ->getTransportName ();
542- $ operation = 'receive ' ;
535+ $ builder ->setAttribute (TraceAttributes::MESSAGING_DESTINATION_NAME , $ receivedStamp ->getTransportName ());
543536 }
544537
545538 if ($ transportMessageIdStamp ) {
546- $ builder ->setAttribute (self :: ATTRIBUTE_MESSAGING_MESSAGE_ID , $ transportMessageIdStamp ->getId ());
539+ $ builder ->setAttribute (TraceAttributes:: MESSAGING_MESSAGE_ID , $ transportMessageIdStamp ->getId ());
547540 }
548541
549542 if ($ delayStamp ) {
@@ -564,7 +557,7 @@ private static function addMessageStampsToSpan(SpanBuilderInterface $builder, En
564557 /** @var \Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsReceivedStamp|null $amazonSqsReceivedStamp */
565558 $ amazonSqsReceivedStamp = $ envelope ->last ('Symfony\Component\Messenger\Bridge\AmazonSqs\Transport\AmazonSqsReceivedStamp ' );
566559 if ($ amazonSqsReceivedStamp && !$ transportMessageIdStamp && method_exists ($ amazonSqsReceivedStamp , 'getId ' )) {
567- $ builder ->setAttribute (self :: ATTRIBUTE_MESSAGING_MESSAGE_ID , $ amazonSqsReceivedStamp ->getId ());
560+ $ builder ->setAttribute (TraceAttributes:: MESSAGING_MESSAGE_ID , $ amazonSqsReceivedStamp ->getId ());
568561 }
569562 }
570563 }
0 commit comments