feat: Emit low-cardinality span names for messaging spans (amqplib, kafka, ...)

MIGRATION.md

@@ -784,13 +784,16 @@ If you [opt out of span streaming](#opting-out-of-span-streaming), span names re
 
 The following span names were adjusted:
 
-| Span op                                                                  | Before                                                                                                                      | After                                                                                                                |
-| ------------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------- |
-| `pageload`                                                               | The parameterized route, or the raw URL path if the SDK couldn't resolve one (`/users/123`)                                 | The parameterized route, or `Pageload` if the SDK has none                                                           |
-| `router`                                                                 | Framework-specific, sometimes containing the raw URL (`/users/123`, `SvelteKit Route Change`)                               | The span's `http.route`, or `Router` if the SDK has none                                                             |
-| `graphql`                                                                | The graphql phase and, for operations, the operation name (`query GetUser`, `graphql.parse`, `graphql.resolve user.0.name`) | The operation type, or the processing type where there is none (`GraphQL query`, `GraphQL parse`, `GraphQL resolve`) |
-| `resource.*`                                                             | The resource URL, relative to the page origin for same-origin resources (`/assets/app.js`)                                  | The resource domain (`cdn.example.com`), or `Resource` if the SDK has none                                           |
-| `mcp.notification.client_to_server`, `mcp.notification.server_to_client` | The notification method name (`notifications/tools/list_changed`)                                                           | The notification method name, or `MCP notification` if the message carries none                                      |
+| Span op                                                                  | Before                                                                                                                      | After                                                                                                                                             |
+| ------------------------------------------------------------------------ | --------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `pageload`                                                               | The parameterized route, or the raw URL path if the SDK couldn't resolve one (`/users/123`)                                 | The parameterized route, or `Pageload` if the SDK has none                                                                                        |
+| `router`                                                                 | Framework-specific, sometimes containing the raw URL (`/users/123`, `SvelteKit Route Change`)                               | The span's `http.route`, or `Router` if the SDK has none                                                                                          |
+| `graphql`                                                                | The graphql phase and, for operations, the operation name (`query GetUser`, `graphql.parse`, `graphql.resolve user.0.name`) | The operation type, or the processing type where there is none (`GraphQL query`, `GraphQL parse`, `GraphQL resolve`)                              |
+| `resource.*`                                                             | The resource URL, relative to the page origin for same-origin resources (`/assets/app.js`)                                  | The resource domain (`cdn.example.com`), or `Resource` if the SDK has none                                                                        |
+| `mcp.notification.client_to_server`, `mcp.notification.server_to_client` | The notification method name (`notifications/tools/list_changed`)                                                           | The notification method name, or `MCP notification` if the message carries none                                                                   |
+| `queue.publish`                                                          | Integration-specific (`publish my-exchange`, `send my-topic`)                                                               | The messaging operation type and the destination (`send my-exchange`), or just the operation type when the destination has no name (`send`)       |
+| `queue.process`                                                          | Integration-specific, sometimes containing per-message data (`my-queue process`, `order.created.12345 process`)             | The messaging operation type and the destination (`process my-exchange`), or just the operation type when the destination has no name (`process`) |
+| `queue.receive`                                                          | The kafkajs operation name (`poll my-topic`)                                                                                | The messaging operation type and the destination (`receive my-topic`)                                                                             |
 
 Resource spans now also carry a `url.domain` attribute holding that domain. The full URL remains available on `url.full`.
 
@@ -804,6 +807,14 @@ For the same reason, `useOperationNameForRootSpan` no longer renames the enclosi
 
 Only the Express, Koa and Hapi integrations resolve a route template for `router` spans. Angular, Ember and SvelteKit have none when the span starts, so their router spans are named `Router`.
 
+Messaging span names now read `<operation type> <destination>` in every integration. The amqplib, kafkajs and NestJS BullMQ integrations used their own word order or verb, so their names change: `my-queue process` became `process my-queue`, amqplib's `publish` became `send`, and the kafkajs batch span's `poll` became `receive`. Cloudflare Queues and the kafkajs producer already matched the conventions, so their names are the same in both trace lifecycles. The operation name an integration reports upstream stays on `messaging.operation.name`.
+
+AWS SQS and SNS span names follow the messaging conventions too, so the operation comes first (`my-queue receive` becomes `receive my-queue`, `my-topic send` becomes `send my-topic`). A streamed SNS `Publish` to a platform endpoint is named `send`, because the endpoint ARN it used to carry ends in a per-device id (`endpoint/GCM/myapp/<uuid> send`). The full ARN remains on `messaging.destination.name`.
+
+An amqplib span's destination is the exchange it uses, or the routing key when it uses the default exchange. RabbitMQ binds every queue to the default exchange under a key equal to the queue's own name, so `sendToQueue` spans are named after their queue (`send my-queue`) instead of dropping the destination. `messaging.destination.name` reports the same value, and the routing key remains on `messaging.rabbitmq.destination.routing_key` in full.
+
+Because a name built from an operation type has to be able to say which operation it means, the NestJS BullMQ, AWS SNS and AWS SQS `SendMessage` spans now carry the `messaging.operation.type` attribute they name themselves after. The other messaging integrations already set it. The attribute is set in both trace lifecycles.
+
 Child spans of a service or root span carry its name in their `sentry.segment.name` attribute, so that changes with it. If you group or filter spans by segment name in dashboards or alerts, update those references.
 
 `ignoreSpans` is evaluated when a span **starts**, at which point a span might not yet have its final name. For example, an unresolved pageload span name is named `'Pageload'` and might receive its final, resolved route name later.

dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/scenario.mjs

@@ -101,6 +101,21 @@ async function sns() {
       { 'content-type': 'text/xml' },
     );
   await client.send(new PublishCommand({ TopicArn: 'arn:aws:sns:us-east-1:123456789012:my-topic', Message: 'Hello' }));
+
+  // Publish to a platform endpoint, whose ARN ends in a per-device id.
+  nock(`https://sns.${region}.amazonaws.com`)
+    .post('/')
+    .reply(
+      200,
+      '<PublishResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/"><PublishResult><MessageId>message-id-2</MessageId></PublishResult><ResponseMetadata><RequestId>request-id-2</RequestId></ResponseMetadata></PublishResponse>',
+      { 'content-type': 'text/xml' },
+    );
+  await client.send(
+    new PublishCommand({
+      TargetArn: 'arn:aws:sns:us-east-1:123456789012:endpoint/GCM/myapp/5e3e9847-3183-3f18-a7e8-671c3a57d4b3',
+      Message: 'Hello',
+    }),
+  );
 }
 
 async function lambda() {

dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/test.ts

@@ -120,7 +120,7 @@ function assertAwsServiceSpans(spanCcontainer: SerializedStreamedSpanContainer):
 
   // SQS - SendMessage (producer)
   expectSpan('SQS SendMessage', {
-    name: 'my-queue send',
+    name: 'send my-queue',
     attributes: expect.objectContaining({
       'rpc.method': { value: 'SendMessage', type: 'string' },
       'rpc.service': { value: 'SQS', type: 'string' },
@@ -134,7 +134,7 @@ function assertAwsServiceSpans(spanCcontainer: SerializedStreamedSpanContainer):
 
   // SQS - ReceiveMessage (consumer)
   expectSpan('SQS ReceiveMessage', {
-    name: 'my-queue receive',
+    name: 'receive my-queue',
     attributes: expect.objectContaining({
       'rpc.method': { value: 'ReceiveMessage', type: 'string' },
       'messaging.system': { value: 'aws_sqs', type: 'string' },
@@ -146,7 +146,7 @@ function assertAwsServiceSpans(spanCcontainer: SerializedStreamedSpanContainer):
 
   // SNS - Publish (producer)
   expectSpan('SNS Publish', {
-    name: 'my-topic send',
+    name: 'send my-topic',
     attributes: expect.objectContaining({
       'rpc.method': { value: 'Publish', type: 'string' },
       'rpc.service': { value: 'SNS', type: 'string' },
@@ -157,6 +157,20 @@ function assertAwsServiceSpans(spanCcontainer: SerializedStreamedSpanContainer):
     }),
   });
 
+  // The ARN suffix is a per-device id, so the streamed name drops the destination.
+  expectSpan('SNS Publish (platform endpoint)', {
+    name: 'send',
+    attributes: expect.objectContaining({
+      'rpc.method': { value: 'Publish', type: 'string' },
+      'rpc.service': { value: 'SNS', type: 'string' },
+      'messaging.destination': { value: 'endpoint/GCM/myapp/5e3e9847-3183-3f18-a7e8-671c3a57d4b3', type: 'string' },
+      'messaging.destination.name': {
+        value: 'arn:aws:sns:us-east-1:123456789012:endpoint/GCM/myapp/5e3e9847-3183-3f18-a7e8-671c3a57d4b3',
+        type: 'string',
+      },
+    }),
+  });
+
   // Lambda - Invoke
   expectSpan('Lambda Invoke', {
     name: 'my-function Invoke',

dev-packages/node-integration-tests/suites/aws-serverless/aws-integration/scenario.mjs

@@ -101,6 +101,21 @@ async function sns() {
       { 'content-type': 'text/xml' },
     );
   await client.send(new PublishCommand({ TopicArn: 'arn:aws:sns:us-east-1:123456789012:my-topic', Message: 'Hello' }));
+
+  // Publish to a platform endpoint, whose ARN ends in a per-device id.
+  nock(`https://sns.${region}.amazonaws.com`)
+    .post('/')
+    .reply(
+      200,
+      '<PublishResponse xmlns="http://sns.amazonaws.com/doc/2010-03-31/"><PublishResult><MessageId>message-id-2</MessageId></PublishResult><ResponseMetadata><RequestId>request-id-2</RequestId></ResponseMetadata></PublishResponse>',
+      { 'content-type': 'text/xml' },
+    );
+  await client.send(
+    new PublishCommand({
+      TargetArn: 'arn:aws:sns:us-east-1:123456789012:endpoint/GCM/myapp/5e3e9847-3183-3f18-a7e8-671c3a57d4b3',
+      Message: 'Hello',
+    }),
+  );
 }
 
 async function lambda() {

dev-packages/node-integration-tests/suites/aws-serverless/aws-integration/test.ts

@@ -151,6 +151,18 @@ function assertAwsServiceSpans(transaction: TransactionEvent): void {
     }),
   });
 
+  // Without span streaming the name keeps the raw ARN suffix, including the per-device id.
+  expectSpan('SNS Publish (platform endpoint)', {
+    description: 'endpoint/GCM/myapp/5e3e9847-3183-3f18-a7e8-671c3a57d4b3 send',
+    data: expect.objectContaining({
+      'rpc.method': 'Publish',
+      'rpc.service': 'SNS',
+      'messaging.destination': 'endpoint/GCM/myapp/5e3e9847-3183-3f18-a7e8-671c3a57d4b3',
+      'messaging.destination.name':
+        'arn:aws:sns:us-east-1:123456789012:endpoint/GCM/myapp/5e3e9847-3183-3f18-a7e8-671c3a57d4b3',
+    }),
+  });
+
   // Lambda - Invoke
   expectSpan('Lambda Invoke', {
     description: 'my-function Invoke',

dev-packages/node-integration-tests/suites/tracing/amqplib/instrument-span-streaming.mjs

@@ -0,0 +1,10 @@
+import * as Sentry from '@sentry/node';
+import { loggingTransport } from '@sentry-internal/node-integration-tests';
+
+Sentry.init({
+  traceLifecycle: 'stream',
+  dsn: 'https://public@dsn.ingest.sentry.io/1337',
+  release: '1.0',
+  tracesSampleRate: 1.0,
+  transport: loggingTransport,
+});

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

@@ -13,7 +13,7 @@ const expectedProducerSpan = (routingKey: string) =>
       'messaging.system': 'rabbitmq',
       'messaging.operation.name': 'send',
       'messaging.operation.type': 'send',
-      'messaging.destination.name': '',
+      'messaging.destination.name': routingKey,
       'messaging.rabbitmq.destination.routing_key': routingKey,
       'network.protocol.name': 'AMQP',
       'network.protocol.version': '0.9.1',
@@ -31,8 +31,8 @@ const EXPECTED_MESSAGE_SPAN_CONSUMER = expect.objectContaining({
   op: 'queue.process',
   data: expect.objectContaining({
     'messaging.system': 'rabbitmq',
-    // The consumer reads the default exchange ('') off the delivered message and the queue name as the routing key.
-    'messaging.destination.name': '',
+    // The delivery carries the default exchange (''), so the routing key is the destination.
+    'messaging.destination.name': 'queue1',
     'messaging.rabbitmq.destination.routing_key': 'queue1',
     'messaging.operation.name': 'process',
     'messaging.operation.type': 'process',
@@ -180,4 +180,35 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_
       { additionalDependencies },
     );
   });
+
+  createEsmAndCjsTests(__dirname, 'scenario.mjs', 'instrument-span-streaming.mjs', (createTestRunner, test) => {
+    test('names streamed spans after the messaging conventions', { timeout: 60_000 }, async () => {
+      await createTestRunner()
+        .ignore('event')
+        .expect({
+          span: container => {
+            // `sendToQueue` publishes to the default exchange, which has no name. Its routing key is the
+            // queue name, so it is the destination rather than per-message data.
+            for (const origin of ['auto.amqplib.publisher', 'auto.amqplib.consumer']) {
+              const span = container.items.find(item => item.attributes['sentry.origin']?.value === origin);
+              expect(span).toBeDefined();
+              expect(span!.attributes['messaging.destination.name']?.value).toBe('queue1');
+              expect(span!.attributes['messaging.rabbitmq.destination.routing_key']?.value).toBe('queue1');
+            }
+
+            const producerSpan = container.items.find(
+              span => span.attributes['sentry.origin']?.value === 'auto.amqplib.publisher',
+            );
+            expect(producerSpan!.name).toBe('send queue1');
+
+            const consumerSpan = container.items.find(
+              span => span.attributes['sentry.origin']?.value === 'auto.amqplib.consumer',
+            );
+            expect(consumerSpan!.name).toBe('process queue1');
+          },
+        })
+        .start()
+        .completed();
+    });
+  });
 });

packages/nestjs/src/integrations/helpers.ts

@@ -1,9 +1,15 @@
-import { SENTRY_OP } from '@sentry/conventions/attributes';
-import { FUNCTION, MIDDLEWARE } from '@sentry/conventions/op';
+import {
+  MESSAGING_DESTINATION_NAME,
+  MESSAGING_OPERATION_TYPE,
+  MESSAGING_SYSTEM,
+  SENTRY_OP,
+} from '@sentry/conventions/attributes';
+import { FUNCTION, MIDDLEWARE, QUEUE_PROCESS } from '@sentry/conventions/op';
 import type { Span } from '@sentry/core';
 import {
   addNonEnumerableProperty,
-  SEMANTIC_ATTRIBUTE_SENTRY_OP,
+  getClient,
+  hasSpanStreamingEnabled,
   SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN,
   withActiveSpan,
 } from '@sentry/core';
@@ -115,21 +121,35 @@ export function getEventSpanOptions(event: string): {
   };
 }
 
+const PROCESS_OPERATION = 'process';
+
 /**
- * Returns span options for nest bullmq process spans.
+ * Returns span options for nest bullmq process spans. `queueName` is undefined when the `@Processor`
+ * decorator has no queue name.
  */
-export function getBullMQProcessSpanOptions(queueName: string): {
+export function getBullMQProcessSpanOptions(queueName: string | undefined): {
   name: string;
-  attributes: Record<string, string>;
+  attributes: Record<string, string | undefined>;
   forceTransaction: boolean;
 } {
+  const client = getClient();
+  const isStreamed = !!client && hasSpanStreamingEnabled(client);
+
+  // Only the word order differs between lifecycles (to keep the old naming pattern).
+  const name = queueName
+    ? isStreamed
+      ? `${PROCESS_OPERATION} ${queueName}`
+      : `${queueName} ${PROCESS_OPERATION}`
+    : PROCESS_OPERATION;
+
   return {
-    name: `${queueName} process`,
+    name,
     attributes: {
-      [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'queue.process',
+      [SENTRY_OP]: QUEUE_PROCESS,
       [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.queue.nestjs.bullmq',
-      'messaging.system': 'bullmq',
-      'messaging.destination.name': queueName,
+      [MESSAGING_SYSTEM]: 'bullmq',
+      [MESSAGING_OPERATION_TYPE]: PROCESS_OPERATION,
+      [MESSAGING_DESTINATION_NAME]: queueName,
     },
     forceTransaction: true,
   };

packages/nestjs/src/integrations/orchestrion-subscriber.ts

@@ -77,7 +77,7 @@ function makeMethodDecorator(original: AnyFn, wrapHandler: (handler: AnyFn) => A
  * Wrap the class decorator `@Processor` returns so it patches
  * `target.prototype.process` before delegating.
  */
-function makeProcessorDecorator(original: AnyFn, queueName: string): AnyFn {
+function makeProcessorDecorator(original: AnyFn, queueName: string | undefined): AnyFn {
   return function (this: unknown, ...args: unknown[]): unknown {
     patchProcessorTarget(args[0] as { __SENTRY_INTERNAL__?: boolean; prototype?: { process?: AnyFn } }, queueName);
     return original.apply(this, args);