diff --git a/dev-packages/deno-integration-tests/suites/orchestrion-amqplib/test.ts b/dev-packages/deno-integration-tests/suites/orchestrion-amqplib/test.ts index f319af4a00ad..d1131ccbad36 100644 --- a/dev-packages/deno-integration-tests/suites/orchestrion-amqplib/test.ts +++ b/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'); diff --git a/dev-packages/deno-integration-tests/suites/orchestrion-kafkajs/test.ts b/dev-packages/deno-integration-tests/suites/orchestrion-kafkajs/test.ts index 9ca813557ba3..0f3eab7ecf8d 100644 --- a/dev-packages/deno-integration-tests/suites/orchestrion-kafkajs/test.ts +++ b/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'); 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 dcb41143510f..90b6a7c932b9 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/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. diff --git a/dev-packages/node-integration-tests/suites/tracing/kafkajs/test.ts b/dev-packages/node-integration-tests/suites/tracing/kafkajs/test.ts index d947e3050a3e..49b91406b437 100644 --- a/dev-packages/node-integration-tests/suites/tracing/kafkajs/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/kafkajs/test.ts @@ -53,13 +53,13 @@ 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, }), }), @@ -67,13 +67,13 @@ describeWithDockerCompose('kafkajs', { workingDirectory: [__dirname] }, () => { 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', }), diff --git a/packages/opentelemetry/src/utils/parseSpanDescription.ts b/packages/opentelemetry/src/utils/parseSpanDescription.ts index e50364fb1f3d..5ac1582a568f 100644 --- a/packages/opentelemetry/src/utils/parseSpanDescription.ts +++ b/packages/opentelemetry/src/utils/parseSpanDescription.ts @@ -7,6 +7,7 @@ import { HTTP_REQUEST_METHOD, HTTP_ROUTE, HTTP_TARGET, + MESSAGING_OPERATION_TYPE, MESSAGING_SYSTEM, RPC_SERVICE, SENTRY_KIND, @@ -14,6 +15,12 @@ import { 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. * diff --git a/packages/opentelemetry/test/utils/parseSpanDescription.test.ts b/packages/opentelemetry/test/utils/parseSpanDescription.test.ts index 63308d2ebdfa..1c59da07076d 100644 --- a/packages/opentelemetry/test/utils/parseSpanDescription.test.ts +++ b/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', }, ], [ diff --git a/packages/server-utils/src/integrations/tracing-channel/amqplib.ts b/packages/server-utils/src/integrations/tracing-channel/amqplib.ts index e3ef27976661..27066a9c5ac7 100644 --- a/packages/server-utils/src/integrations/tracing-channel/amqplib.ts +++ b/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), diff --git a/packages/server-utils/src/integrations/tracing-channel/kafkajs/spans.ts b/packages/server-utils/src/integrations/tracing-channel/kafkajs/spans.ts index 66dca8269c42..90ca5cbdf9d2 100644 --- a/packages/server-utils/src/integrations/tracing-channel/kafkajs/spans.ts +++ b/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,