feat(server-utils)!: Emit queue.* span ops for messaging spans

dev-packages/deno-integration-tests/suites/orchestrion-amqplib/test.ts

@@ -15,7 +15,7 @@ Deno.test('amqplib instrumentation: included in default integrations (Deno 2.8.0
   assert(names.includes('Amqplib'), `Amqplib should be in defaults, got ${names.join(', ')}`);
 });
 
-Deno.test('amqplib instrumentation: orchestrion:amqplib:publish channel produces a nested message span', async () => {
+Deno.test('amqplib instrumentation: orchestrion:amqplib:publish channel produces a nested queue.publish span', async () => {
   resetGlobals();
   const sink = transactionSink();
   init({
@@ -49,8 +49,8 @@ Deno.test('amqplib instrumentation: orchestrion:amqplib:publish channel produces
     "'parent' transaction",
   );
 
-  const publishSpan = parent.spans?.find(s => s.op === 'message');
-  assertExists(publishSpan, `expected a message child span, got ops: ${parent.spans?.map(s => s.op).join(', ')}`);
+  const publishSpan = parent.spans?.find(s => s.op === 'queue.publish');
+  assertExists(publishSpan, `expected a queue.publish child span, got ops: ${parent.spans?.map(s => s.op).join(', ')}`);
   assertEquals(publishSpan!.description, 'publish my-exchange');
   assertEquals(publishSpan!.data?.['messaging.destination.name'], 'my-exchange');
   assertEquals(publishSpan!.data?.['messaging.system'], 'rabbitmq');

dev-packages/deno-integration-tests/suites/orchestrion-kafkajs/test.ts

@@ -41,8 +41,8 @@ Deno.test('kafkajs instrumentation: orchestrion:kafkajs:send_batch channel produ
     "'parent' transaction",
   );
 
-  const kafkaSpan = parent.spans?.find(s => s.op === 'message');
-  assertExists(kafkaSpan, `expected a message child span, got ops: ${parent.spans?.map(s => s.op).join(', ')}`);
+  const kafkaSpan = parent.spans?.find(s => s.op === 'queue.publish');
+  assertExists(kafkaSpan, `expected a queue.publish child span, got ops: ${parent.spans?.map(s => s.op).join(', ')}`);
   assertEquals(kafkaSpan!.description, 'send my-topic');
   assertEquals(kafkaSpan!.data?.['messaging.system'], 'kafka');
   assertEquals(kafkaSpan!.data?.['messaging.destination.name'], 'my-topic');

dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts

@@ -1,21 +1,14 @@
 import type { TransactionEvent } from '@sentry/core';
 import { afterAll, describe, expect } from 'vitest';
-import { isOrchestrionEnabled } from '../../../utils';
 import { cleanupChildProcesses, createEsmAndCjsTests, describeWithDockerCompose } from '../../../utils/runner';
 
-// The span origin depends on which instrumentation is active. These blocks drive the SDK's default
-// integrations, so when the generic orchestrion run is enabled (via INJECT_ORCHESTRION) the OTel
-// `Amqplib` integration is swapped for the diagnostics-channel one, changing the origin.
-const PUBLISHER_ORIGIN = isOrchestrionEnabled() ? 'auto.amqplib.publisher' : 'auto.amqplib.otel.publisher';
-const CONSUMER_ORIGIN = isOrchestrionEnabled() ? 'auto.amqplib.consumer' : 'auto.amqplib.otel.consumer';
-
 // Each scenario uses its own queue name to keep them isolated on the shared broker, so the
 // expected producer span is parameterized by the routing key (queue name) it publishes to.
 // The scenarios all publish via `sendToQueue`, which delegates to `publish('', queue, ...)` — i.e. the
 // default (empty) exchange with the queue name as the routing key.
 const expectedProducerSpan = (routingKey: string) =>
   expect.objectContaining({
-    op: 'message',
+    op: 'queue.publish',
     data: expect.objectContaining({
       'messaging.system': 'rabbitmq',
       // Legacy messaging attributes emitted by both the OTel and orchestrion integrations.
@@ -27,29 +20,23 @@ const expectedProducerSpan = (routingKey: string) =>
       'messaging.protocol_version': '0.9.1',
       'net.peer.name': 'localhost',
       'net.peer.port': 5672,
-      // Current `@sentry/conventions` attributes are only emitted by the orchestrion integration; the
-      // vendored OTel instrumentation never set them.
-      ...(isOrchestrionEnabled()
-        ? {
-            'messaging.operation.type': 'send',
-            'messaging.destination.name': '',
-            'messaging.rabbitmq.destination.routing_key': routingKey,
-            'network.protocol.name': 'AMQP',
-            'network.protocol.version': '0.9.1',
-            'server.address': 'localhost',
-            'server.port': 5672,
-            'url.full': 'amqp://sentry:***@localhost:5672/',
-          }
-        : {}),
+      'messaging.operation.type': 'send',
+      'messaging.destination.name': '',
+      'messaging.rabbitmq.destination.routing_key': routingKey,
+      'network.protocol.name': 'AMQP',
+      'network.protocol.version': '0.9.1',
+      'server.address': 'localhost',
+      'server.port': 5672,
+      'url.full': 'amqp://sentry:***@localhost:5672/',
       'sentry.kind': 'producer',
-      'sentry.op': 'message',
-      'sentry.origin': PUBLISHER_ORIGIN,
+      'sentry.op': 'queue.publish',
+      'sentry.origin': 'auto.amqplib.publisher',
     }),
     status: 'ok',
   });
 
 const EXPECTED_MESSAGE_SPAN_CONSUMER = expect.objectContaining({
-  op: 'message',
+  op: 'queue.process',
   data: expect.objectContaining({
     'messaging.system': 'rabbitmq',
     // Legacy messaging attributes emitted by both the OTel and orchestrion integrations. The consumer
@@ -58,17 +45,12 @@ const EXPECTED_MESSAGE_SPAN_CONSUMER = expect.objectContaining({
     'messaging.destination_kind': 'topic',
     'messaging.rabbitmq.routing_key': 'queue1',
     'messaging.operation': 'process',
-    // Current `@sentry/conventions` attributes are only emitted by the orchestrion integration.
-    ...(isOrchestrionEnabled()
-      ? {
-          'messaging.destination.name': '',
-          'messaging.rabbitmq.destination.routing_key': 'queue1',
-          'messaging.operation.type': 'process',
-        }
-      : {}),
+    'messaging.destination.name': '',
+    'messaging.rabbitmq.destination.routing_key': 'queue1',
+    'messaging.operation.type': 'process',
     'sentry.kind': 'consumer',
-    'sentry.op': 'message',
-    'sentry.origin': CONSUMER_ORIGIN,
+    'sentry.op': 'queue.process',
+    'sentry.origin': 'auto.amqplib.consumer',
   }),
   status: 'ok',
 });
@@ -106,10 +88,10 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_
                 // identify it by its origin rather than by transaction name. The consumer span is its
                 // own transaction, identified by the origin on its trace context.
                 const producer = receivedTransactions.find(t =>
-                  t.spans?.some(s => s.data?.['sentry.origin'] === PUBLISHER_ORIGIN),
+                  t.spans?.some(s => s.data?.['sentry.origin'] === 'auto.amqplib.publisher'),
                 );
                 const consumer = receivedTransactions.find(
-                  t => t.contexts?.trace?.data?.['sentry.origin'] === CONSUMER_ORIGIN,
+                  t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.amqplib.consumer',
                 );
 
                 expect(producer).toBeDefined();
@@ -118,7 +100,7 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_
                 expect(producer!.transaction).toBe('root span');
                 expect(consumer!.transaction).toBe('queue1 process');
 
-                const producerSpan = producer!.spans?.find(s => s.data?.['sentry.origin'] === PUBLISHER_ORIGIN);
+                const producerSpan = producer!.spans?.find(s => s.data?.['sentry.origin'] === 'auto.amqplib.publisher');
                 expect(producerSpan).toMatchObject(expectedProducerSpan('queue1'));
 
                 expect(consumer!.contexts?.trace).toMatchObject(EXPECTED_MESSAGE_SPAN_CONSUMER);
@@ -152,20 +134,20 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_
                 receivedTransactions.push(transaction);
 
                 const consumer = receivedTransactions.find(
-                  t => t.contexts?.trace?.data?.['sentry.origin'] === CONSUMER_ORIGIN,
+                  t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.amqplib.consumer',
                 );
 
                 expect(consumer).toBeDefined();
                 expect(consumer!.transaction).toBe('queue-error process');
                 expect(consumer!.contexts?.trace).toMatchObject(
                   expect.objectContaining({
-                    op: 'message',
+                    op: 'queue.process',
                     status: 'internal_error',
                     data: expect.objectContaining({
                       'messaging.system': 'rabbitmq',
                       'sentry.kind': 'consumer',
-                      'sentry.op': 'message',
-                      'sentry.origin': CONSUMER_ORIGIN,
+                      'sentry.op': 'queue.process',
+                      'sentry.origin': 'auto.amqplib.consumer',
                     }),
                   }),
                 );
@@ -192,7 +174,9 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_
                 transaction: (transaction: TransactionEvent) => {
                   expect(transaction.transaction).toBe('root span');
 
-                  const producerSpans = transaction.spans?.filter(s => s.data?.['sentry.origin'] === PUBLISHER_ORIGIN);
+                  const producerSpans = transaction.spans?.filter(
+                    s => s.data?.['sentry.origin'] === 'auto.amqplib.publisher',
+                  );
 
                   // The confirm channel internally calls the base publish; the instrumentation must not
                   // double-instrument, so we expect exactly one producer span.

dev-packages/node-integration-tests/suites/tracing/kafkajs/test.ts

@@ -53,27 +53,27 @@ describeWithDockerCompose('kafkajs', { workingDirectory: [__dirname] }, () => {
 
             expect(producer!.contexts?.trace).toMatchObject(
               expect.objectContaining({
-                op: 'message',
+                op: 'queue.publish',
                 status: 'ok',
                 data: expect.objectContaining({
                   'messaging.system': 'kafka',
                   'messaging.destination.name': 'test-topic',
                   'sentry.kind': 'producer',
-                  'sentry.op': 'message',
+                  'sentry.op': 'queue.publish',
                   'sentry.origin': producerOrigin,
                 }),
               }),
             );
 
             expect(consumer!.contexts?.trace).toMatchObject(
               expect.objectContaining({
-                op: 'message',
+                op: 'queue.process',
                 status: 'ok',
                 data: expect.objectContaining({
                   'messaging.system': 'kafka',
                   'messaging.destination.name': 'test-topic',
                   'sentry.kind': 'consumer',
-                  'sentry.op': 'message',
+                  'sentry.op': 'queue.process',
                   'sentry.origin': consumerOrigin,
                 }),
               }),
@@ -93,13 +93,13 @@ describeWithDockerCompose('kafkajs', { workingDirectory: [__dirname] }, () => {
             expect(transaction.transaction).toBe('send invalid topic name');
             expect(transaction.contexts?.trace).toMatchObject(
               expect.objectContaining({
-                op: 'message',
+                op: 'queue.publish',
                 status: 'internal_error',
                 data: expect.objectContaining({
                   'messaging.system': 'kafka',
                   'messaging.destination.name': 'invalid topic name',
                   'sentry.kind': 'producer',
-                  'sentry.op': 'message',
+                  'sentry.op': 'queue.publish',
                   'sentry.origin': producerOrigin,
                   'error.type': 'KafkaJSNonRetriableError',
                 }),

packages/opentelemetry/src/utils/parseSpanDescription.ts

@@ -7,13 +7,20 @@ import {
   HTTP_REQUEST_METHOD,
   HTTP_ROUTE,
   HTTP_TARGET,
+  MESSAGING_OPERATION_TYPE,
   MESSAGING_SYSTEM,
   RPC_SERVICE,
   SENTRY_KIND,
   URL_FRAGMENT,
   URL_FULL,
   URL_QUERY,
 } from '@sentry/conventions/attributes';
+import {
+  MESSAGING_QUEUE_PROCESS_SPAN_OP,
+  MESSAGING_QUEUE_PUBLISH_SPAN_OP,
+  MESSAGING_QUEUE_RECEIVE_SPAN_OP,
+  MESSAGING_QUEUE_SPAN_OP,
+} from '@sentry/conventions/op';
 import type { Span, SpanAttributes } from '@sentry/core';
 import {
   getSanitizedUrlString,
@@ -65,11 +72,11 @@ export function inferSpanData(attributes: SpanAttributes): SpanDescription {
   }
 
   // If messaging.system exists then this is a messaging system span.
-  // eslint-disable-next-line typescript/no-deprecated
+  // Derive the queue op from the messaging operation type.
   const messagingSystem = attributes[MESSAGING_SYSTEM];
   if (messagingSystem) {
     return {
-      op: 'message',
+      op: getMessagingOp(attributes[MESSAGING_OPERATION_TYPE]),
     };
   }
 
@@ -85,6 +92,25 @@ export function inferSpanData(attributes: SpanAttributes): SpanDescription {
   return { op: undefined };
 }
 
+/**
+ * Maps an OTel `messaging.operation.type` to the corresponding `queue.*` span op. `send` is the
+ * pre-1.0 spelling of `publish`; both map to `queue.publish`. Unknown or missing types fall back to
+ * the generic `queue` op.
+ */
+function getMessagingOp(operationType: unknown): string {
+  switch (operationType) {
+    case 'publish':
+    case 'send':
+      return MESSAGING_QUEUE_PUBLISH_SPAN_OP;
+    case 'receive':
+      return MESSAGING_QUEUE_RECEIVE_SPAN_OP;
+    case 'process':
+      return MESSAGING_QUEUE_PROCESS_SPAN_OP;
+    default:
+      return MESSAGING_QUEUE_SPAN_OP;
+  }
+}
+
 /**
  * Extract better op/description from an otel span.
  *

packages/opentelemetry/test/utils/parseSpanDescription.test.ts

@@ -193,7 +193,7 @@ describe('inferSpanData', () => {
         [MESSAGING_SYSTEM]: 'test-messaging-system',
       },
       {
-        op: 'message',
+        op: 'queue',
       },
     ],
     [
@@ -203,7 +203,7 @@ describe('inferSpanData', () => {
         [MESSAGING_SYSTEM]: 'test-messaging-system',
       },
       {
-        op: 'message',
+        op: 'queue',
       },
     ],
     [
@@ -214,7 +214,7 @@ describe('inferSpanData', () => {
         [SEMANTIC_ATTRIBUTE_SENTRY_CUSTOM_SPAN_NAME]: 'custom name',
       },
       {
-        op: 'message',
+        op: 'queue',
       },
     ],
     [
@@ -225,7 +225,7 @@ describe('inferSpanData', () => {
         [SEMANTIC_ATTRIBUTE_SENTRY_CUSTOM_SPAN_NAME]: 'custom name',
       },
       {
-        op: 'message',
+        op: 'queue',
       },
     ],
     [

packages/server-utils/src/integrations/tracing-channel/amqplib.ts

@@ -22,10 +22,12 @@ import {
   NETWORK_PROTOCOL_NAME,
   NETWORK_PROTOCOL_VERSION,
   SENTRY_KIND,
+  SENTRY_OP,
   SERVER_ADDRESS,
   SERVER_PORT,
   URL_FULL,
 } from '@sentry/conventions/attributes';
+import { MESSAGING_QUEUE_PROCESS_SPAN_OP, MESSAGING_QUEUE_PUBLISH_SPAN_OP } from '@sentry/conventions/op';
 import { amqplibModuleNames } from '../../orchestrion/config/amqplib';
 import { invokeOrchestrionInstrumentation } from '../../orchestrion/instrumentation';
 import { CHANNELS } from '../../orchestrion/channels';
@@ -462,8 +464,8 @@ function startPublishSpan(data: AmqpChannelContext): Span {
 
   const span = startInactiveSpan({
     name: `publish ${normalizeExchange(exchange)}`,
-    op: 'message',
     attributes: {
+      [SENTRY_OP]: MESSAGING_QUEUE_PUBLISH_SPAN_OP,
       [SENTRY_KIND]: 'producer',
       ...getStoredConnectionAttributes(data.self),
       [ATTR_MESSAGING_DESTINATION]: exchange, // TODO(v11) remove this attribute
@@ -500,8 +502,8 @@ function startPublishSpan(data: AmqpChannelContext): Span {
 function startConsumeSpan(queue: string, msg: ConsumeMessage, channel: ChannelLike): Span {
   return startInactiveSpan({
     name: `${queue} process`,
-    op: 'message',
     attributes: {
+      [SENTRY_OP]: MESSAGING_QUEUE_PROCESS_SPAN_OP,
       [SENTRY_KIND]: 'consumer',
       [SEMANTIC_ATTRIBUTE_SENTRY_SOURCE]: 'component',
       ...getStoredConnectionAttributes(channel),

packages/server-utils/src/integrations/tracing-channel/kafkajs/spans.ts

@@ -16,7 +16,13 @@ import {
   MESSAGING_OPERATION_TYPE,
   MESSAGING_SYSTEM,
   SENTRY_KIND,
+  SENTRY_OP,
 } from '@sentry/conventions/attributes';
+import {
+  MESSAGING_QUEUE_PROCESS_SPAN_OP,
+  MESSAGING_QUEUE_PUBLISH_SPAN_OP,
+  MESSAGING_QUEUE_RECEIVE_SPAN_OP,
+} from '@sentry/conventions/op';
 import type { Span, SpanAttributes, SpanLink } from '@sentry/core';
 import {
   getTraceData,
@@ -99,13 +105,14 @@ export function startConsumerSpan({ topic, message, operationType, links, attrib
   // The batch "receive" span is named `poll`; per-message spans use the operation type verbatim.
   const operationName = operationType === MESSAGING_OPERATION_TYPE_VALUE_RECEIVE ? 'poll' : operationType;
 
+  const isBatchReceive = operationType === MESSAGING_OPERATION_TYPE_VALUE_RECEIVE;
+
   return startInactiveSpan({
     name: `${operationName} ${topic}`,
-    // todo(v11): Use https://getsentry.github.io/sentry-conventions/ops/#messaging
-    op: 'message',
     links,
     attributes: {
-      [SENTRY_KIND]: operationType === MESSAGING_OPERATION_TYPE_VALUE_RECEIVE ? 'client' : 'consumer',
+      [SENTRY_OP]: isBatchReceive ? MESSAGING_QUEUE_RECEIVE_SPAN_OP : MESSAGING_QUEUE_PROCESS_SPAN_OP,
+      [SENTRY_KIND]: isBatchReceive ? 'client' : 'consumer',
       ...attributes,
       [MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA,
       [MESSAGING_DESTINATION_NAME]: topic,
@@ -125,8 +132,8 @@ export function startConsumerSpan({ topic, message, operationType, links, attrib
 export function startProducerSpan(topic: string, message: Message): Span {
   const span = startInactiveSpan({
     name: `send ${topic}`,
-    op: 'message',
     attributes: {
+      [SENTRY_OP]: MESSAGING_QUEUE_PUBLISH_SPAN_OP,
       [SENTRY_KIND]: 'producer',
       [MESSAGING_SYSTEM]: MESSAGING_SYSTEM_VALUE_KAFKA,
       [MESSAGING_DESTINATION_NAME]: topic,