From 3075bbaf1b19f227fb633f1d6b487d1d8fd3367b Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Wed, 26 Aug 2026 14:15:45 +0200 Subject: [PATCH 1/2] feat: Emit low-cardinality span names for messaging spans (amqplib, kafka, ...) --- MIGRATION.md | 21 +++++--- .../aws-integration-streamed/scenario.mjs | 15 ++++++ .../aws-integration-streamed/test.ts | 14 ++++++ .../aws-integration/scenario.mjs | 15 ++++++ .../aws-serverless/aws-integration/test.ts | 12 +++++ .../amqplib/instrument-span-streaming.mjs | 10 ++++ .../suites/tracing/amqplib/test.ts | 31 ++++++++++++ packages/nestjs/src/integrations/helpers.ts | 16 +++++-- .../integrations/orchestrion-subscriber.ts | 2 +- .../nestjs/src/integrations/wrap-handlers.ts | 8 ++-- .../orchestrion-subscriber.test.ts | 48 +++++++++++++++++-- .../server-utils/src/integrations/amqplib.ts | 21 ++++++-- .../src/integrations/aws-sdk/services/sns.ts | 10 +++- .../src/integrations/aws-sdk/services/sqs.ts | 14 +++--- .../src/integrations/kafkajs/spans.ts | 11 +++-- .../test/integrations/aws-sdk-sqs.test.ts | 31 ++++++++++++ .../test/integrations/kafkajs.test.ts | 47 ++++++++++++++++++ 17 files changed, 293 insertions(+), 33 deletions(-) create mode 100644 dev-packages/node-integration-tests/suites/tracing/amqplib/instrument-span-streaming.mjs create mode 100644 packages/server-utils/test/integrations/aws-sdk-sqs.test.ts diff --git a/MIGRATION.md b/MIGRATION.md index 46d39ebf1085..ee5b3ae012e8 100644 --- a/MIGRATION.md +++ b/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,10 @@ 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 ` ` 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`. + +An AWS SNS `Publish` to a platform endpoint no longer puts the endpoint ARN in the span name, because that ARN ends in a per-device id (`endpoint/GCM/myapp/ send` becomes `send`). Publishing to a topic is unchanged (`my-topic send`), and the full ARN remains on `messaging.destination.name`. + 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. diff --git a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/scenario.mjs b/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/scenario.mjs index 1c0fc382c620..624a10e9523b 100644 --- a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/scenario.mjs +++ b/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, + 'message-id-2request-id-2', + { '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() { diff --git a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/test.ts b/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/test.ts index 7964b04c96a9..b528d679d6d7 100644 --- a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/test.ts +++ b/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/test.ts @@ -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', diff --git a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration/scenario.mjs b/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration/scenario.mjs index 0e58050adc9c..def8000f3381 100644 --- a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration/scenario.mjs +++ b/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, + 'message-id-2request-id-2', + { '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() { diff --git a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration/test.ts b/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration/test.ts index 59b0ab3c6a49..71eb7d8623ab 100644 --- a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration/test.ts +++ b/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', diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/instrument-span-streaming.mjs b/dev-packages/node-integration-tests/suites/tracing/amqplib/instrument-span-streaming.mjs new file mode 100644 index 000000000000..54c513068360 --- /dev/null +++ b/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, +}); diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts index 2c5262bc2195..cd7bc1ea9ed8 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts @@ -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, so both names drop the + // destination instead of falling back to the routing key. + 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(''); + 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'); + + const consumerSpan = container.items.find( + span => span.attributes['sentry.origin']?.value === 'auto.amqplib.consumer', + ); + expect(consumerSpan!.name).toBe('process'); + }, + }) + .start() + .completed(); + }); + }); }); diff --git a/packages/nestjs/src/integrations/helpers.ts b/packages/nestjs/src/integrations/helpers.ts index ed2bcd9abb93..2db72f6c0d21 100644 --- a/packages/nestjs/src/integrations/helpers.ts +++ b/packages/nestjs/src/integrations/helpers.ts @@ -3,6 +3,8 @@ import { FUNCTION, MIDDLEWARE } from '@sentry/conventions/op'; import type { Span } from '@sentry/core'; import { addNonEnumerableProperty, + getClient, + hasSpanStreamingEnabled, SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, withActiveSpan, @@ -116,20 +118,26 @@ export function getEventSpanOptions(event: string): { } /** - * 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; forceTransaction: boolean; } { + const client = getClient(); + + const destination = queueName ?? 'unknown'; + const streamedName = queueName ? `process ${queueName}` : 'process'; + return { - name: `${queueName} process`, + name: client && hasSpanStreamingEnabled(client) ? streamedName : `${destination} process`, attributes: { [SEMANTIC_ATTRIBUTE_SENTRY_OP]: 'queue.process', [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: 'auto.queue.nestjs.bullmq', 'messaging.system': 'bullmq', - 'messaging.destination.name': queueName, + 'messaging.destination.name': destination, }, forceTransaction: true, }; diff --git a/packages/nestjs/src/integrations/orchestrion-subscriber.ts b/packages/nestjs/src/integrations/orchestrion-subscriber.ts index 07a0b437f4e0..e2ac9e7524f2 100644 --- a/packages/nestjs/src/integrations/orchestrion-subscriber.ts +++ b/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); diff --git a/packages/nestjs/src/integrations/wrap-handlers.ts b/packages/nestjs/src/integrations/wrap-handlers.ts index 8888428832f5..62f677635eeb 100644 --- a/packages/nestjs/src/integrations/wrap-handlers.ts +++ b/packages/nestjs/src/integrations/wrap-handlers.ts @@ -116,7 +116,7 @@ export function wrapEventHandler(handler: AnyFn, fallbackEvent: unknown): AnyFn * Wrap a BullMQ `process` method: fork the isolation scope, open a * `queue.process` transaction, and capture errors. */ -export function wrapBullMQProcess(process: AnyFn, queueName: string): AnyFn { +export function wrapBullMQProcess(process: AnyFn, queueName: string | undefined): AnyFn { return function (this: unknown, ...args: unknown[]): unknown { return withIsolationScope(() => startSpan(getBullMQProcessSpanOptions(queueName), async () => { @@ -157,21 +157,21 @@ export function patchMethodDescriptor( /** * Extract the queue name from `@Processor('name')` or `@Processor({ name })`. */ -export function extractQueueName(arg: unknown): string { +export function extractQueueName(arg: unknown): string | undefined { if (typeof arg === 'string') { return arg; } if (arg && typeof arg === 'object' && 'name' in arg && typeof (arg as { name?: unknown }).name === 'string') { return (arg as { name: string }).name; } - return 'unknown'; + return undefined; } /** * Patch a `@Processor`-decorated class's `prototype.process` with a wrapped * version. */ -export function patchProcessorTarget(target: ProcessorTarget | undefined, queueName: string): void { +export function patchProcessorTarget(target: ProcessorTarget | undefined, queueName: string | undefined): void { const prototype = target?.prototype; const process = prototype?.process; if (prototype && process && typeof process === 'function' && !target?.__SENTRY_INTERNAL__ && !isWrapped(process)) { diff --git a/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts b/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts index 93d42ededb4a..040d009b2ddc 100644 --- a/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts +++ b/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts @@ -36,7 +36,7 @@ class TestClient extends Client { } } -function initTestClient(): void { +function initTestClient(traceLifecycle?: 'stream' | 'static'): void { //@ts-expect-error - just a mock for the test, this is fine initAndBind(TestClient, { dsn: 'https://username@domain/123', @@ -44,6 +44,7 @@ function initTestClient(): void { sendClientReports: false, stackParser: () => [], tracesSampleRate: 1, + traceLifecycle, transport: () => createTransport({ recordDroppedEvent: () => undefined }, () => resolvedSyncPromise({})), }); } @@ -883,7 +884,7 @@ describe('NestJS orchestrion subscriber: schedule / event / bullmq', () => { await new EmailProcessor().process({}); const json = spanToJSON(spanInside!); - expect(json.name).toBe('emails process'); + expect(json.name).toBe('process emails'); expect(json.attributes['sentry.op']).toBe('queue.process'); expect(json.attributes['sentry.origin']).toBe('auto.queue.nestjs.bullmq'); expect(json.attributes).toMatchObject({ @@ -907,10 +908,51 @@ describe('NestJS orchestrion subscriber: schedule / event / bullmq', () => { } wrappedDecorator(ReportsProcessor); return new ReportsProcessor().process().then(() => { - expect(spanToJSON(spanInside!).name).toBe('reports process'); + expect(spanToJSON(spanInside!).name).toBe('process reports'); }); }); + it('bullmq @Processor: keeps the transaction-mode span name when span streaming is off', async () => { + installTestAsyncContextStrategy(); + initTestClient('static'); + subscribeToNestChannels(); + + const wrappedDecorator = driveFactory(CHANNELS.NESTJS_PROCESSOR, ['emails'], () => undefined); + + let spanInside: Span | undefined; + class EmailProcessor { + public async process(): Promise { + spanInside = getActiveSpan(); + } + } + wrappedDecorator(EmailProcessor); + + await new EmailProcessor().process(); + expect(spanToJSON(spanInside!).name).toBe('emails process'); + }); + + it('bullmq @Processor: drops the destination from a streamed span name when the queue is unresolved', async () => { + installTestAsyncContextStrategy(); + initTestClient(); + subscribeToNestChannels(); + + const wrappedDecorator = driveFactory(CHANNELS.NESTJS_PROCESSOR, [undefined], () => undefined); + + let spanInside: Span | undefined; + class AnonymousProcessor { + public async process(): Promise { + spanInside = getActiveSpan(); + } + } + wrappedDecorator(AnonymousProcessor); + + await new AnonymousProcessor().process(); + const json = spanToJSON(spanInside!); + expect(json.name).toBe('process'); + // Only the name drops the placeholder. The attribute keeps it. + expect(json.attributes).toMatchObject({ 'messaging.destination.name': 'unknown' }); + }); + it('schedule @Timeout: captures sync errors with the timeout mechanism', () => { installTestAsyncContextStrategy(); initTestClient(); diff --git a/packages/server-utils/src/integrations/amqplib.ts b/packages/server-utils/src/integrations/amqplib.ts index 7a2ee1eb4afc..54dcc01af28a 100644 --- a/packages/server-utils/src/integrations/amqplib.ts +++ b/packages/server-utils/src/integrations/amqplib.ts @@ -4,7 +4,9 @@ import type { IntegrationFn, Span, SpanAttributes } from '@sentry/core'; import { continueTrace, defineIntegration, + getClient, getTraceData, + hasSpanStreamingEnabled, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, SPAN_STATUS_ERROR, startInactiveSpan, @@ -444,8 +446,13 @@ function startPublishSpan(data: AmqpChannelContext): Span { const routingKey = typeof routingKeyArg === 'string' ? routingKeyArg : ''; let options = data.arguments[3] as PublishOptions | undefined; + const client = getClient(); + + // The default exchange has no name, so the streamed name is the operation on its own. + const streamedName = exchange ? `${MESSAGING_OPERATION_VALUE_SEND} ${exchange}` : MESSAGING_OPERATION_VALUE_SEND; + const span = startInactiveSpan({ - name: `publish ${normalizeExchange(exchange)}`, + name: client && hasSpanStreamingEnabled(client) ? streamedName : `publish ${normalizeExchange(exchange)}`, attributes: { [SENTRY_OP]: QUEUE_PUBLISH, [SENTRY_KIND]: 'producer', @@ -478,14 +485,22 @@ function startPublishSpan(data: AmqpChannelContext): Span { /** Starts an inactive CONSUMER (process) span carrying the amqplib messaging attributes. */ function startConsumeSpan(queue: string, msg: ConsumeMessage, channel: ChannelLike): Span { + const client = getClient(); + const exchange = msg.fields?.exchange; + + // `queue` falls back to the routing key (`order.created.12345`), so the streamed name uses the exchange. + const streamedName = exchange + ? `${MESSAGING_OPERATION_VALUE_PROCESS} ${exchange}` + : MESSAGING_OPERATION_VALUE_PROCESS; + return startInactiveSpan({ - name: `${queue} process`, + name: client && hasSpanStreamingEnabled(client) ? streamedName : `${queue} process`, attributes: { [SENTRY_OP]: QUEUE_PROCESS, [SENTRY_KIND]: 'consumer', [SENTRY_SEGMENT_NAME_SOURCE]: 'component', ...getStoredConnectionAttributes(channel), - [MESSAGING_DESTINATION_NAME]: msg.fields?.exchange, + [MESSAGING_DESTINATION_NAME]: exchange, [ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY]: msg.fields?.routingKey, [MESSAGING_OPERATION_NAME]: MESSAGING_OPERATION_VALUE_PROCESS, [MESSAGING_OPERATION_TYPE]: MESSAGING_OPERATION_VALUE_PROCESS, diff --git a/packages/server-utils/src/integrations/aws-sdk/services/sns.ts b/packages/server-utils/src/integrations/aws-sdk/services/sns.ts index 2efe96d721d4..95cd8b2c12ee 100644 --- a/packages/server-utils/src/integrations/aws-sdk/services/sns.ts +++ b/packages/server-utils/src/integrations/aws-sdk/services/sns.ts @@ -1,5 +1,5 @@ import type { Span } from '@sentry/core'; -import { getTraceData } from '@sentry/core'; +import { getClient, getTraceData, hasSpanStreamingEnabled } from '@sentry/core'; import { AWS_SNS_TOPIC_ARN as ATTR_AWS_SNS_TOPIC_ARN, MESSAGING_DESTINATION as ATTR_MESSAGING_DESTINATION, @@ -30,7 +30,13 @@ export class SnsServiceExtension implements ServiceExtension { spanAttributes[ATTR_MESSAGING_DESTINATION] = destinationName; spanAttributes[MESSAGING_DESTINATION_NAME] = TopicArn || TargetArn || PhoneNumber || 'unknown'; - spanName = `${PhoneNumber ? 'phone_number' : destinationName} send`; + const client = getClient(); + + // A `/` in the suffix means a platform-endpoint ARN (`endpoint/GCM/myapp/`), not a topic name. + const isStreamedOpaqueDestination = + !PhoneNumber && destinationName.includes('/') && !!client && hasSpanStreamingEnabled(client); + + spanName = isStreamedOpaqueDestination ? 'send' : `${PhoneNumber ? 'phone_number' : destinationName} send`; } const topicArn = request.commandInput?.TopicArn; diff --git a/packages/server-utils/src/integrations/aws-sdk/services/sqs.ts b/packages/server-utils/src/integrations/aws-sdk/services/sqs.ts index 6b4ae2c0e474..15e588e7a364 100644 --- a/packages/server-utils/src/integrations/aws-sdk/services/sqs.ts +++ b/packages/server-utils/src/integrations/aws-sdk/services/sqs.ts @@ -22,7 +22,7 @@ export class SqsServiceExtension implements ServiceExtension { public requestPreSpanHook(request: NormalizedRequest): RequestMetadata { const queueUrl = extractQueueUrl(request.commandInput); const queueName = extractQueueNameFromUrl(queueUrl); - let spanName: string | undefined; + let operation: string | undefined; const spanAttributes: Record = { [MESSAGING_SYSTEM]: 'aws_sqs', @@ -35,8 +35,8 @@ export class SqsServiceExtension implements ServiceExtension { switch (request.commandName) { case 'ReceiveMessage': { - spanName = `${queueName} receive`; - spanAttributes[MESSAGING_OPERATION_TYPE] = 'receive'; + operation = 'receive'; + spanAttributes[MESSAGING_OPERATION_TYPE] = operation; spanAttributes[SENTRY_KIND] = 'consumer'; request.commandInput.MessageAttributeNames = addPropagationFieldsToAttributeNames( @@ -47,14 +47,15 @@ export class SqsServiceExtension implements ServiceExtension { case 'SendMessage': case 'SendMessageBatch': + operation = 'send'; spanAttributes[SENTRY_KIND] = 'producer'; - spanName = `${queueName} send`; break; } return { spanAttributes, - spanName, + // Without a queue name the subscriber falls back to `.`. + spanName: operation && queueName ? `${queueName} ${operation}` : undefined, }; } @@ -146,5 +147,6 @@ function extractQueueNameFromUrl(queueUrl: string): string | undefined { const segments = queueUrl.split('/'); if (segments.length === 0) return undefined; - return segments[segments.length - 1]; + // A trailing slash leaves an empty last segment, which is not a queue name + return segments[segments.length - 1] || undefined; } diff --git a/packages/server-utils/src/integrations/kafkajs/spans.ts b/packages/server-utils/src/integrations/kafkajs/spans.ts index fdec629fd80b..e6d46f4c32e6 100644 --- a/packages/server-utils/src/integrations/kafkajs/spans.ts +++ b/packages/server-utils/src/integrations/kafkajs/spans.ts @@ -6,7 +6,8 @@ * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-kafkajs * - Upstream version: @opentelemetry/instrumentation-kafkajs@0.27.0 * - Span builders migrated to the `@sentry/core` span API. Kept byte-identical in span name/attributes - * for parity with the OTel integration this replaces. + * for parity with the OTel integration this replaces, except for the span name under span streaming, + * which follows the Sentry messaging conventions. */ import { @@ -21,7 +22,9 @@ import { import { QUEUE_PROCESS, QUEUE_PUBLISH, QUEUE_RECEIVE } from '@sentry/conventions/op'; import type { Span, SpanAttributes, SpanLink } from '@sentry/core'; import { + getClient, getTraceData, + hasSpanStreamingEnabled, propagationContextFromHeaders, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, SPAN_STATUS_ERROR, @@ -49,7 +52,7 @@ const TRACE_FLAG_NONE = 0; interface ConsumerSpanOptions { topic: string; message: KafkaMessage | undefined; - operationType: string; + operationType: 'process' | 'receive'; attributes: SpanAttributes; links?: SpanLink[]; } @@ -103,8 +106,10 @@ export function startConsumerSpan({ topic, message, operationType, links, attrib const isBatchReceive = operationType === MESSAGING_OPERATION_TYPE_VALUE_RECEIVE; + const client = getClient(); + return startInactiveSpan({ - name: `${operationName} ${topic}`, + name: client && hasSpanStreamingEnabled(client) ? `${operationType} ${topic}` : `${operationName} ${topic}`, links, attributes: { [SENTRY_OP]: isBatchReceive ? QUEUE_RECEIVE : QUEUE_PROCESS, diff --git a/packages/server-utils/test/integrations/aws-sdk-sqs.test.ts b/packages/server-utils/test/integrations/aws-sdk-sqs.test.ts new file mode 100644 index 000000000000..ef76fc544dbc --- /dev/null +++ b/packages/server-utils/test/integrations/aws-sdk-sqs.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from 'vitest'; +import { SqsServiceExtension } from '../../src/integrations/aws-sdk/services/sqs'; +import type { RequestMetadata } from '../../src/integrations/aws-sdk/types'; + +function preSpanHook(commandName: string, commandInput: Record): RequestMetadata { + return new SqsServiceExtension().requestPreSpanHook({ serviceName: 'SQS', commandName, commandInput }); +} + +const QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue'; + +describe('SqsServiceExtension span naming', () => { + it.each([ + ['ReceiveMessage', 'my-queue receive'], + ['SendMessage', 'my-queue send'], + ['SendMessageBatch', 'my-queue send'], + ])('names a %s span after the queue', (commandName, expected) => { + expect(preSpanHook(commandName, { QueueUrl: QUEUE_URL }).spanName).toBe(expected); + }); + + // Interpolating the missing queue would produce the literal name 'undefined receive'. + it.each(['ReceiveMessage', 'SendMessage'])('leaves a %s span unnamed when the QueueUrl is missing', commandName => { + const { spanName, spanAttributes } = preSpanHook(commandName, {}); + + expect(spanName).toBeUndefined(); + expect(spanAttributes?.['messaging.destination.name']).toBeUndefined(); + }); + + it('treats a trailing slash as no queue name rather than an empty one', () => { + expect(preSpanHook('ReceiveMessage', { QueueUrl: `${QUEUE_URL}/` }).spanName).toBeUndefined(); + }); +}); diff --git a/packages/server-utils/test/integrations/kafkajs.test.ts b/packages/server-utils/test/integrations/kafkajs.test.ts index bfde55d7e568..79b6faa51c2e 100644 --- a/packages/server-utils/test/integrations/kafkajs.test.ts +++ b/packages/server-utils/test/integrations/kafkajs.test.ts @@ -1,5 +1,14 @@ +import { setCurrentClient, spanToJSON } from '@sentry/core'; import { describe, expect, it } from 'vitest'; import { isWrappedConsumerCallback, wrapEachBatch, wrapEachMessage } from '../../src/integrations/kafkajs/consumer'; +import { startConsumerSpan, startProducerSpan } from '../../src/integrations/kafkajs/spans'; +import { getDefaultTestClientOptions, TestClient } from '../mocks/client'; + +function setUpClient(traceLifecycle: 'stream' | 'static'): void { + const client = new TestClient(getDefaultTestClientOptions({ traceLifecycle, tracesSampleRate: 1 })); + setCurrentClient(client); + client.init(); +} // End-to-end span output (producer/consumer, error status, origins) is covered by the node integration // suite, which runs against a real broker in both OTel and orchestrion modes. This only covers the @@ -14,3 +23,41 @@ describe('kafkajs consumer callback wrapping', () => { expect(isWrappedConsumerCallback(wrapEachBatch(userCallback))).toBe(true); }); }); + +describe('kafkajs span names', () => { + it('names the streamed batch span after the operation type rather than the upstream `poll`', () => { + setUpClient('stream'); + + const span = startConsumerSpan({ topic: 'my-topic', message: undefined, operationType: 'receive', attributes: {} }); + + expect(spanToJSON(span).name).toBe('receive my-topic'); + // The attribute keeps the upstream name. Only the span name changes. + expect(spanToJSON(span).attributes).toMatchObject({ 'messaging.operation.name': 'poll' }); + }); + + it('keeps the upstream `poll` batch span name when span streaming is off', () => { + setUpClient('static'); + + const span = startConsumerSpan({ topic: 'my-topic', message: undefined, operationType: 'receive', attributes: {} }); + + expect(spanToJSON(span).name).toBe('poll my-topic'); + }); + + it.each(['stream', 'static'] as const)( + 'names per-message and producer spans identically in %s mode', + traceLifecycle => { + setUpClient(traceLifecycle); + + const processSpan = startConsumerSpan({ + topic: 'my-topic', + message: undefined, + operationType: 'process', + attributes: {}, + }); + const producerSpan = startProducerSpan('my-topic', { value: 'hi' }); + + expect(spanToJSON(processSpan).name).toBe('process my-topic'); + expect(spanToJSON(producerSpan).name).toBe('send my-topic'); + }, + ); +}); From ccec06183c14b65615abe78882833f10526d380a Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Thu, 27 Aug 2026 15:49:28 +0200 Subject: [PATCH 2/2] review comments (span naming) --- MIGRATION.md | 6 +- .../aws-integration-streamed/test.ts | 6 +- .../suites/tracing/amqplib/test.ts | 16 ++-- packages/nestjs/src/integrations/helpers.ts | 32 ++++--- .../orchestrion-subscriber.test.ts | 40 +++++---- .../server-utils/src/integrations/amqplib.ts | 43 ++++++--- .../src/integrations/aws-sdk/services/sns.ts | 32 +++++-- .../src/integrations/aws-sdk/services/sqs.ts | 33 ++++++- .../test/integrations/amqplib.test.ts | 29 ++++++ .../test/integrations/aws-sdk-sns.test.ts | 84 ++++++++++++++++++ .../test/integrations/aws-sdk-sqs.test.ts | 88 ++++++++++++++++--- 11 files changed, 332 insertions(+), 77 deletions(-) create mode 100644 packages/server-utils/test/integrations/amqplib.test.ts create mode 100644 packages/server-utils/test/integrations/aws-sdk-sns.test.ts diff --git a/MIGRATION.md b/MIGRATION.md index ee5b3ae012e8..9e82a01df25e 100644 --- a/MIGRATION.md +++ b/MIGRATION.md @@ -809,7 +809,11 @@ Only the Express, Koa and Hapi integrations resolve a route template for `router Messaging span names now read ` ` 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`. -An AWS SNS `Publish` to a platform endpoint no longer puts the endpoint ARN in the span name, because that ARN ends in a per-device id (`endpoint/GCM/myapp/ send` becomes `send`). Publishing to a topic is unchanged (`my-topic send`), and the full ARN remains on `messaging.destination.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/ 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. diff --git a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/test.ts b/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/test.ts index b528d679d6d7..29d67b07f98d 100644 --- a/dev-packages/node-integration-tests/suites/aws-serverless/aws-integration-streamed/test.ts +++ b/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' }, diff --git a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts index cd7bc1ea9ed8..301f5520f553 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/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', @@ -187,24 +187,24 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_ .ignore('event') .expect({ span: container => { - // `sendToQueue` publishes to the default exchange, which has no name, so both names drop the - // destination instead of falling back to the routing key. + // `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(''); + 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'); + 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'); + expect(consumerSpan!.name).toBe('process queue1'); }, }) .start() diff --git a/packages/nestjs/src/integrations/helpers.ts b/packages/nestjs/src/integrations/helpers.ts index 2db72f6c0d21..5539ce9cb2c3 100644 --- a/packages/nestjs/src/integrations/helpers.ts +++ b/packages/nestjs/src/integrations/helpers.ts @@ -1,11 +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, getClient, hasSpanStreamingEnabled, - SEMANTIC_ATTRIBUTE_SENTRY_OP, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, withActiveSpan, } from '@sentry/core'; @@ -117,27 +121,35 @@ export function getEventSpanOptions(event: string): { }; } +const PROCESS_OPERATION = 'process'; + /** * Returns span options for nest bullmq process spans. `queueName` is undefined when the `@Processor` * decorator has no queue name. */ export function getBullMQProcessSpanOptions(queueName: string | undefined): { name: string; - attributes: Record; + attributes: Record; forceTransaction: boolean; } { const client = getClient(); + const isStreamed = !!client && hasSpanStreamingEnabled(client); - const destination = queueName ?? 'unknown'; - const streamedName = queueName ? `process ${queueName}` : 'process'; + // 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: client && hasSpanStreamingEnabled(client) ? streamedName : `${destination} 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': destination, + [MESSAGING_SYSTEM]: 'bullmq', + [MESSAGING_OPERATION_TYPE]: PROCESS_OPERATION, + [MESSAGING_DESTINATION_NAME]: queueName, }, forceTransaction: true, }; diff --git a/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts b/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts index 040d009b2ddc..fef86fbfedd4 100644 --- a/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts +++ b/packages/nestjs/test/integrations/orchestrion-subscriber.test.ts @@ -931,27 +931,31 @@ describe('NestJS orchestrion subscriber: schedule / event / bullmq', () => { expect(spanToJSON(spanInside!).name).toBe('emails process'); }); - it('bullmq @Processor: drops the destination from a streamed span name when the queue is unresolved', async () => { - installTestAsyncContextStrategy(); - initTestClient(); - subscribeToNestChannels(); + it.each(['stream', 'static'] as const)( + 'bullmq @Processor: names an unresolved queue `process` in %s mode', + async traceLifecycle => { + installTestAsyncContextStrategy(); + initTestClient(traceLifecycle); + subscribeToNestChannels(); - const wrappedDecorator = driveFactory(CHANNELS.NESTJS_PROCESSOR, [undefined], () => undefined); + const wrappedDecorator = driveFactory(CHANNELS.NESTJS_PROCESSOR, [undefined], () => undefined); - let spanInside: Span | undefined; - class AnonymousProcessor { - public async process(): Promise { - spanInside = getActiveSpan(); + let spanInside: Span | undefined; + class AnonymousProcessor { + public async process(): Promise { + spanInside = getActiveSpan(); + } } - } - wrappedDecorator(AnonymousProcessor); - - await new AnonymousProcessor().process(); - const json = spanToJSON(spanInside!); - expect(json.name).toBe('process'); - // Only the name drops the placeholder. The attribute keeps it. - expect(json.attributes).toMatchObject({ 'messaging.destination.name': 'unknown' }); - }); + wrappedDecorator(AnonymousProcessor); + + await new AnonymousProcessor().process(); + const json = spanToJSON(spanInside!); + expect(json.name).toBe('process'); + // An unresolved queue is an absent destination, so the attribute is omitted rather than sent as + // a placeholder. + expect(json.attributes['messaging.destination.name']).toBeUndefined(); + }, + ); it('schedule @Timeout: captures sync errors with the timeout mechanism', () => { installTestAsyncContextStrategy(); diff --git a/packages/server-utils/src/integrations/amqplib.ts b/packages/server-utils/src/integrations/amqplib.ts index 54dcc01af28a..864022d8d291 100644 --- a/packages/server-utils/src/integrations/amqplib.ts +++ b/packages/server-utils/src/integrations/amqplib.ts @@ -244,7 +244,7 @@ function subscribeDispatch(): void { ensureChannelState(channel); const info = fields?.consumerTag ? channel[CHANNEL_CONSUMER_INFO]?.get(fields.consumerTag) : undefined; - const queue = info?.queue ?? msg.fields?.routingKey ?? ''; + const queue = info?.queue ?? msg.fields?.routingKey; const noAck = info?.noAck ?? false; const headers = msg.properties?.headers; @@ -447,9 +447,10 @@ function startPublishSpan(data: AmqpChannelContext): Span { let options = data.arguments[3] as PublishOptions | undefined; const client = getClient(); - - // The default exchange has no name, so the streamed name is the operation on its own. - const streamedName = exchange ? `${MESSAGING_OPERATION_VALUE_SEND} ${exchange}` : MESSAGING_OPERATION_VALUE_SEND; + const destination = resolveDestination(exchange, routingKey); + const streamedName = destination + ? `${MESSAGING_OPERATION_VALUE_SEND} ${destination}` + : MESSAGING_OPERATION_VALUE_SEND; const span = startInactiveSpan({ name: client && hasSpanStreamingEnabled(client) ? streamedName : `publish ${normalizeExchange(exchange)}`, @@ -457,7 +458,7 @@ function startPublishSpan(data: AmqpChannelContext): Span { [SENTRY_OP]: QUEUE_PUBLISH, [SENTRY_KIND]: 'producer', ...getStoredConnectionAttributes(data.self), - [MESSAGING_DESTINATION_NAME]: exchange, + [MESSAGING_DESTINATION_NAME]: destination, [ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY]: routingKey, [MESSAGING_OPERATION_NAME]: MESSAGING_OPERATION_VALUE_SEND, [MESSAGING_OPERATION_TYPE]: MESSAGING_OPERATION_VALUE_SEND, @@ -484,23 +485,26 @@ function startPublishSpan(data: AmqpChannelContext): Span { } /** Starts an inactive CONSUMER (process) span carrying the amqplib messaging attributes. */ -function startConsumeSpan(queue: string, msg: ConsumeMessage, channel: ChannelLike): Span { +function startConsumeSpan(queue: string | undefined, msg: ConsumeMessage, channel: ChannelLike): Span { const client = getClient(); - const exchange = msg.fields?.exchange; - - // `queue` falls back to the routing key (`order.created.12345`), so the streamed name uses the exchange. - const streamedName = exchange - ? `${MESSAGING_OPERATION_VALUE_PROCESS} ${exchange}` + const destination = resolveDestination(msg.fields?.exchange, msg.fields?.routingKey); + const streamedName = destination + ? `${MESSAGING_OPERATION_VALUE_PROCESS} ${destination}` : MESSAGING_OPERATION_VALUE_PROCESS; return startInactiveSpan({ - name: client && hasSpanStreamingEnabled(client) ? streamedName : `${queue} process`, + name: + client && hasSpanStreamingEnabled(client) + ? streamedName + : queue + ? `${queue} ${MESSAGING_OPERATION_VALUE_PROCESS}` + : MESSAGING_OPERATION_VALUE_PROCESS, attributes: { [SENTRY_OP]: QUEUE_PROCESS, [SENTRY_KIND]: 'consumer', [SENTRY_SEGMENT_NAME_SOURCE]: 'component', ...getStoredConnectionAttributes(channel), - [MESSAGING_DESTINATION_NAME]: exchange, + [MESSAGING_DESTINATION_NAME]: destination, [ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY]: msg.fields?.routingKey, [MESSAGING_OPERATION_NAME]: MESSAGING_OPERATION_VALUE_PROCESS, [MESSAGING_OPERATION_TYPE]: MESSAGING_OPERATION_VALUE_PROCESS, @@ -576,6 +580,19 @@ function getConnectionAttributesFromUrl(url: unknown): SpanAttributes { return attributes; } +/** + * The default exchange has no name and binds every queue under a key equal to the queue's own name, so + * a routing key used with it is the queue name. On a named exchange the routing key is per-message + * (`order.created.12345`), so only the exchange is used. + * + * @see https://www.rabbitmq.com/docs/exchanges#default-exchange + * @internal Exported for tests; every scenario publishes through `sendToQueue`. + */ +export function resolveDestination(exchange: string | undefined, routingKey: string | undefined): string | undefined { + // `undefined` so an absent destination is omitted rather than reported as an empty string. + return exchange || routingKey || undefined; +} + function normalizeExchange(exchangeName: string): string { return exchangeName !== '' ? exchangeName : ''; } diff --git a/packages/server-utils/src/integrations/aws-sdk/services/sns.ts b/packages/server-utils/src/integrations/aws-sdk/services/sns.ts index 95cd8b2c12ee..5f56b05be852 100644 --- a/packages/server-utils/src/integrations/aws-sdk/services/sns.ts +++ b/packages/server-utils/src/integrations/aws-sdk/services/sns.ts @@ -4,6 +4,7 @@ import { AWS_SNS_TOPIC_ARN as ATTR_AWS_SNS_TOPIC_ARN, MESSAGING_DESTINATION as ATTR_MESSAGING_DESTINATION, MESSAGING_DESTINATION_NAME, + MESSAGING_OPERATION_TYPE, MESSAGING_SYSTEM, SENTRY_KIND, } from '@sentry/conventions/attributes'; @@ -12,6 +13,21 @@ import type { NormalizedRequest, NormalizedResponse } from '../types'; import { injectPropagationContext } from './MessageAttributes'; import type { RequestMetadata, ServiceExtension } from './ServiceExtension'; +/** Stands in for a destination the command does not name. */ +const UNKNOWN_DESTINATION = 'unknown'; +const SEND_OPERATION = 'send'; + +/** + * Streamed names follow the messaging conventions, ` `. Transaction-mode + * names keep the order they had. A destination the command does not name drops out of both. + */ +function buildSpanName(operation: string, destination: string | undefined, isStreamed: boolean): string { + if (!destination) { + return operation; + } + return isStreamed ? `${operation} ${destination}` : `${destination} ${operation}`; +} + export class SnsServiceExtension implements ServiceExtension { public requestPreSpanHook(request: NormalizedRequest): RequestMetadata { let spanName = `SNS ${request.commandName}`; @@ -28,15 +44,19 @@ export class SnsServiceExtension implements ServiceExtension { const destinationName = extractDestinationName(TopicArn, TargetArn, PhoneNumber); // oxlint-disable-next-line typescript/no-deprecated -- old-semconv messaging.destination, matched to the OTel aws-sdk integration spanAttributes[ATTR_MESSAGING_DESTINATION] = destinationName; - spanAttributes[MESSAGING_DESTINATION_NAME] = TopicArn || TargetArn || PhoneNumber || 'unknown'; + spanAttributes[MESSAGING_DESTINATION_NAME] = TopicArn || TargetArn || PhoneNumber || UNKNOWN_DESTINATION; + spanAttributes[MESSAGING_OPERATION_TYPE] = SEND_OPERATION; const client = getClient(); - // A `/` in the suffix means a platform-endpoint ARN (`endpoint/GCM/myapp/`), not a topic name. - const isStreamedOpaqueDestination = - !PhoneNumber && destinationName.includes('/') && !!client && hasSpanStreamingEnabled(client); + const isStreamed = !!client && hasSpanStreamingEnabled(client); + // The raw phone number never goes in a name. A `/` in the suffix means a platform-endpoint ARN + // (`endpoint/GCM/myapp/`) rather than a topic, so only a streamed name drops it. + const named = PhoneNumber ? 'phone_number' : destinationName; + const destinationForName = + named === UNKNOWN_DESTINATION || (isStreamed && !PhoneNumber && named.includes('/')) ? undefined : named; - spanName = isStreamedOpaqueDestination ? 'send' : `${PhoneNumber ? 'phone_number' : destinationName} send`; + spanName = buildSpanName(SEND_OPERATION, destinationForName, isStreamed); } const topicArn = request.commandInput?.TopicArn; @@ -76,6 +96,6 @@ function extractDestinationName(topicArn: string, targetArn: string, phoneNumber } else if (phoneNumber) { return phoneNumber; } else { - return 'unknown'; + return UNKNOWN_DESTINATION; } } diff --git a/packages/server-utils/src/integrations/aws-sdk/services/sqs.ts b/packages/server-utils/src/integrations/aws-sdk/services/sqs.ts index 15e588e7a364..45d3700128d3 100644 --- a/packages/server-utils/src/integrations/aws-sdk/services/sqs.ts +++ b/packages/server-utils/src/integrations/aws-sdk/services/sqs.ts @@ -1,5 +1,5 @@ import type { Span } from '@sentry/core'; -import { getTraceData, propagationContextFromHeaders } from '@sentry/core'; +import { getClient, getTraceData, hasSpanStreamingEnabled, propagationContextFromHeaders } from '@sentry/core'; import { MESSAGING_BATCH_MESSAGE_COUNT, MESSAGING_DESTINATION_NAME, @@ -36,7 +36,6 @@ export class SqsServiceExtension implements ServiceExtension { case 'ReceiveMessage': { operation = 'receive'; - spanAttributes[MESSAGING_OPERATION_TYPE] = operation; spanAttributes[SENTRY_KIND] = 'consumer'; request.commandInput.MessageAttributeNames = addPropagationFieldsToAttributeNames( @@ -52,10 +51,16 @@ export class SqsServiceExtension implements ServiceExtension { break; } + if (operation) { + spanAttributes[MESSAGING_OPERATION_TYPE] = operation; + } + + const client = getClient(); + const isStreamed = !!client && hasSpanStreamingEnabled(client); + return { spanAttributes, - // Without a queue name the subscriber falls back to `.`. - spanName: operation && queueName ? `${queueName} ${operation}` : undefined, + spanName: buildSpanName(operation, queueName, isStreamed), }; } @@ -137,6 +142,26 @@ function linkReceivedMessageToProducer(span: Span, message: SQS.Message): void { } } +/** + * Streamed names follow the messaging conventions, ` `. Transaction-mode + * names keep the order they had. Either way a command with no `QueueUrl` drops to the operation alone, + * which wins over the aws-sdk house style (`SQS.ReceiveMessage`). That style still names the commands + * with no messaging operation, which is why this returns undefined for them. + */ +function buildSpanName( + operation: string | undefined, + queueName: string | undefined, + isStreamed: boolean, +): string | undefined { + if (!operation) { + return undefined; + } + if (!queueName) { + return operation; + } + return isStreamed ? `${operation} ${queueName}` : `${queueName} ${operation}`; +} + function extractQueueUrl(commandInput: CommandInput): string { return commandInput?.QueueUrl; } diff --git a/packages/server-utils/test/integrations/amqplib.test.ts b/packages/server-utils/test/integrations/amqplib.test.ts new file mode 100644 index 000000000000..393067539a03 --- /dev/null +++ b/packages/server-utils/test/integrations/amqplib.test.ts @@ -0,0 +1,29 @@ +import { describe, expect, it } from 'vitest'; +import { resolveDestination } from '../../src/integrations/amqplib'; + +describe('resolveDestination', () => { + it.each([ + ['orders', 'order.created.12345', 'orders'], + ['logs', '', 'logs'], + ['amq.fanout', undefined, 'amq.fanout'], + ])( + 'keeps the per-message routing key out of the destination on the %s exchange', + (exchange, routingKey, expected) => { + expect(resolveDestination(exchange, routingKey)).toBe(expected); + }, + ); + + // The default exchange binds every queue under a key equal to the queue's own name, so there the + // routing key names the queue and is as bounded as an exchange name. + it('uses the routing key on the default exchange, where it is the queue name', () => { + expect(resolveDestination('', 'orders-worker')).toBe('orders-worker'); + }); + + it.each([ + ['', ''], + ['', undefined], + [undefined, undefined], + ])('resolves no destination from exchange %o and routing key %o', (exchange, routingKey) => { + expect(resolveDestination(exchange, routingKey)).toBeUndefined(); + }); +}); diff --git a/packages/server-utils/test/integrations/aws-sdk-sns.test.ts b/packages/server-utils/test/integrations/aws-sdk-sns.test.ts new file mode 100644 index 000000000000..40dddcfb91a0 --- /dev/null +++ b/packages/server-utils/test/integrations/aws-sdk-sns.test.ts @@ -0,0 +1,84 @@ +import { setCurrentClient } from '@sentry/core'; +import { afterEach, describe, expect, it } from 'vitest'; +import { SnsServiceExtension } from '../../src/integrations/aws-sdk/services/sns'; +import type { RequestMetadata } from '../../src/integrations/aws-sdk/types'; +import { getDefaultTestClientOptions, TestClient } from '../mocks/client'; + +function setUpClient(traceLifecycle: 'stream' | 'static'): void { + const client = new TestClient(getDefaultTestClientOptions({ traceLifecycle, tracesSampleRate: 1 })); + setCurrentClient(client); + client.init(); +} + +function publishSpanName(commandInput: Record): string | undefined { + const metadata: RequestMetadata = new SnsServiceExtension().requestPreSpanHook({ + serviceName: 'SNS', + commandName: 'Publish', + commandInput, + }); + return metadata.spanName; +} + +const TOPIC_ARN = 'arn:aws:sns:us-east-1:123456789012:my-topic'; +// A platform endpoint, whose ARN ends in a per-device id. +const ENDPOINT_ARN = 'arn:aws:sns:us-east-1:123456789012:endpoint/GCM/myapp/5e3e9847-3183-3f18-a7e8-671c3a57d4b3'; + +describe('SnsServiceExtension span naming', () => { + afterEach(() => { + setCurrentClient(undefined as never); + }); + + describe('with span streaming', () => { + it('names a topic publish with the operation before the destination', () => { + setUpClient('stream'); + + expect(publishSpanName({ TopicArn: TOPIC_ARN, Message: 'Hello' })).toBe('send my-topic'); + }); + + it('drops a platform-endpoint destination, which ends in a per-device id', () => { + setUpClient('stream'); + + expect(publishSpanName({ TargetArn: ENDPOINT_ARN, Message: 'Hello' })).toBe('send'); + }); + + it('drops the destination when the command names none', () => { + setUpClient('stream'); + + expect(publishSpanName({ Message: 'Hello' })).toBe('send'); + }); + + it('never puts the raw phone number in the name', () => { + setUpClient('stream'); + + const spanName = publishSpanName({ PhoneNumber: '+15551234567', Message: 'Hello' }); + + expect(spanName).toBe('send phone_number'); + expect(spanName).not.toContain('5551234567'); + }); + }); + + it('reports the operation type it names the span after', () => { + setUpClient('stream'); + + const metadata = new SnsServiceExtension().requestPreSpanHook({ + serviceName: 'SNS', + commandName: 'Publish', + commandInput: { TopicArn: TOPIC_ARN, Message: 'Hello' }, + }); + + expect(metadata.spanAttributes?.['messaging.operation.type']).toBe('send'); + }); + + describe('without span streaming', () => { + it.each([ + [{ TopicArn: TOPIC_ARN }, 'my-topic send'], + [{ TargetArn: ENDPOINT_ARN }, 'endpoint/GCM/myapp/5e3e9847-3183-3f18-a7e8-671c3a57d4b3 send'], + [{ PhoneNumber: '+15551234567' }, 'phone_number send'], + [{}, 'send'], + ])('keeps the transaction-mode name for %o', (commandInput, expected) => { + setUpClient('static'); + + expect(publishSpanName({ ...commandInput, Message: 'Hello' })).toBe(expected); + }); + }); +}); diff --git a/packages/server-utils/test/integrations/aws-sdk-sqs.test.ts b/packages/server-utils/test/integrations/aws-sdk-sqs.test.ts index ef76fc544dbc..f7652e88964e 100644 --- a/packages/server-utils/test/integrations/aws-sdk-sqs.test.ts +++ b/packages/server-utils/test/integrations/aws-sdk-sqs.test.ts @@ -1,6 +1,14 @@ -import { describe, expect, it } from 'vitest'; +import { setCurrentClient } from '@sentry/core'; +import { afterEach, describe, expect, it } from 'vitest'; import { SqsServiceExtension } from '../../src/integrations/aws-sdk/services/sqs'; import type { RequestMetadata } from '../../src/integrations/aws-sdk/types'; +import { getDefaultTestClientOptions, TestClient } from '../mocks/client'; + +function setUpClient(traceLifecycle: 'stream' | 'static'): void { + const client = new TestClient(getDefaultTestClientOptions({ traceLifecycle, tracesSampleRate: 1 })); + setCurrentClient(client); + client.init(); +} function preSpanHook(commandName: string, commandInput: Record): RequestMetadata { return new SqsServiceExtension().requestPreSpanHook({ serviceName: 'SQS', commandName, commandInput }); @@ -9,23 +17,75 @@ function preSpanHook(commandName: string, commandInput: Record) const QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue'; describe('SqsServiceExtension span naming', () => { - it.each([ - ['ReceiveMessage', 'my-queue receive'], - ['SendMessage', 'my-queue send'], - ['SendMessageBatch', 'my-queue send'], - ])('names a %s span after the queue', (commandName, expected) => { - expect(preSpanHook(commandName, { QueueUrl: QUEUE_URL }).spanName).toBe(expected); + afterEach(() => { + setCurrentClient(undefined as never); }); - // Interpolating the missing queue would produce the literal name 'undefined receive'. - it.each(['ReceiveMessage', 'SendMessage'])('leaves a %s span unnamed when the QueueUrl is missing', commandName => { - const { spanName, spanAttributes } = preSpanHook(commandName, {}); + describe('with span streaming', () => { + it.each([ + ['ReceiveMessage', 'receive my-queue'], + ['SendMessage', 'send my-queue'], + ['SendMessageBatch', 'send my-queue'], + ])('names a %s span with the operation before the destination', (commandName, expected) => { + setUpClient('stream'); + + expect(preSpanHook(commandName, { QueueUrl: QUEUE_URL }).spanName).toBe(expected); + }); + + it.each([ + ['ReceiveMessage', 'receive'], + ['SendMessage', 'send'], + ])('drops the destination from a %s span when the QueueUrl is missing', (commandName, expected) => { + setUpClient('stream'); + + const { spanName, spanAttributes } = preSpanHook(commandName, {}); + + expect(spanName).toBe(expected); + expect(spanAttributes?.['messaging.destination.name']).toBeUndefined(); + }); + + // The aws-sdk house style (`SQS.DeleteMessage`) still names these. + it('leaves a command with no messaging operation unnamed', () => { + setUpClient('stream'); - expect(spanName).toBeUndefined(); - expect(spanAttributes?.['messaging.destination.name']).toBeUndefined(); + expect(preSpanHook('DeleteMessage', { QueueUrl: QUEUE_URL }).spanName).toBeUndefined(); + }); + + it.each([ + ['ReceiveMessage', 'receive'], + ['SendMessage', 'send'], + ['SendMessageBatch', 'send'], + ])('reports the operation type it names a %s span after', (commandName, expected) => { + setUpClient('stream'); + + expect(preSpanHook(commandName, { QueueUrl: QUEUE_URL }).spanAttributes?.['messaging.operation.type']).toBe( + expected, + ); + }); }); - it('treats a trailing slash as no queue name rather than an empty one', () => { - expect(preSpanHook('ReceiveMessage', { QueueUrl: `${QUEUE_URL}/` }).spanName).toBeUndefined(); + describe('without span streaming', () => { + it.each([ + ['ReceiveMessage', 'my-queue receive'], + ['SendMessage', 'my-queue send'], + ['SendMessageBatch', 'my-queue send'], + ])('keeps the transaction-mode name for a %s span', (commandName, expected) => { + setUpClient('static'); + + expect(preSpanHook(commandName, { QueueUrl: QUEUE_URL }).spanName).toBe(expected); + }); + + // Interpolating the missing queue would produce the literal name 'undefined receive'. + it('drops the destination when the QueueUrl is missing', () => { + setUpClient('static'); + + expect(preSpanHook('ReceiveMessage', {}).spanName).toBe('receive'); + }); + + it('treats a trailing slash as no queue name rather than an empty one', () => { + setUpClient('static'); + + expect(preSpanHook('ReceiveMessage', { QueueUrl: `${QUEUE_URL}/` }).spanName).toBe('receive'); + }); }); });