From 955315bc0c645f7219645952718981e88771c19a Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Mon, 6 Jul 2026 13:42:29 +0200 Subject: [PATCH 1/6] amqplib orchestrion --- .../suites/tracing/amqplib/test.ts | 27 +- packages/deno/src/index.ts | 1 + packages/deno/src/integrations/amqplib.ts | 34 + packages/deno/src/sdk.ts | 5 +- .../integrations/tracing-channel/amqplib.ts | 589 ++++++++++++++++++ .../server-utils/src/orchestrion/channels.ts | 2 + .../src/orchestrion/config/amqplib.ts | 89 +++ .../src/orchestrion/config/index.ts | 2 + .../server-utils/src/orchestrion/index.ts | 3 + .../test/orchestrion/amqplib.test.ts | 230 +++++++ 10 files changed, 969 insertions(+), 13 deletions(-) create mode 100644 packages/deno/src/integrations/amqplib.ts create mode 100644 packages/server-utils/src/integrations/tracing-channel/amqplib.ts create mode 100644 packages/server-utils/src/orchestrion/config/amqplib.ts create mode 100644 packages/server-utils/test/orchestrion/amqplib.test.ts 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 f3381b2f5038..7380b1c807e8 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts @@ -1,7 +1,14 @@ import type { TransactionEvent } from '@sentry/core'; import { afterAll, describe, expect } from 'vitest'; +import { isOrchestrionEnabled } from '../../../utils'; import { cleanupChildProcesses, createEsmAndCjsTests } 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.orchestrion.publisher' : 'auto.amqplib.otel.publisher'; +const CONSUMER_ORIGIN = isOrchestrionEnabled() ? 'auto.amqplib.orchestrion.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. const expectedProducerSpan = (routingKey: string) => @@ -12,7 +19,7 @@ const expectedProducerSpan = (routingKey: string) => 'messaging.rabbitmq.routing_key': routingKey, 'otel.kind': 'PRODUCER', 'sentry.op': 'message', - 'sentry.origin': 'auto.amqplib.otel.publisher', + 'sentry.origin': PUBLISHER_ORIGIN, }), status: 'ok', }); @@ -24,7 +31,7 @@ const EXPECTED_MESSAGE_SPAN_CONSUMER = expect.objectContaining({ 'messaging.rabbitmq.routing_key': 'queue1', 'otel.kind': 'CONSUMER', 'sentry.op': 'message', - 'sentry.origin': 'auto.amqplib.otel.consumer', + 'sentry.origin': CONSUMER_ORIGIN, }), status: 'ok', }); @@ -65,10 +72,10 @@ describe('amqplib auto-instrumentation', () => { // 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'] === 'auto.amqplib.otel.publisher'), + t.spans?.some(s => s.data?.['sentry.origin'] === PUBLISHER_ORIGIN), ); const consumer = receivedTransactions.find( - t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.amqplib.otel.consumer', + t => t.contexts?.trace?.data?.['sentry.origin'] === CONSUMER_ORIGIN, ); expect(producer).toBeDefined(); @@ -77,9 +84,7 @@ describe('amqplib auto-instrumentation', () => { expect(producer!.transaction).toBe('root span'); expect(consumer!.transaction).toBe('queue1 process'); - const producerSpan = producer!.spans?.find( - s => s.data?.['sentry.origin'] === 'auto.amqplib.otel.publisher', - ); + const producerSpan = producer!.spans?.find(s => s.data?.['sentry.origin'] === PUBLISHER_ORIGIN); expect(producerSpan).toMatchObject(expectedProducerSpan('queue1')); expect(consumer!.contexts?.trace).toMatchObject(EXPECTED_MESSAGE_SPAN_CONSUMER); @@ -116,7 +121,7 @@ describe('amqplib auto-instrumentation', () => { receivedTransactions.push(transaction); const consumer = receivedTransactions.find( - t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.amqplib.otel.consumer', + t => t.contexts?.trace?.data?.['sentry.origin'] === CONSUMER_ORIGIN, ); expect(consumer).toBeDefined(); @@ -129,7 +134,7 @@ describe('amqplib auto-instrumentation', () => { 'messaging.system': 'rabbitmq', 'otel.kind': 'CONSUMER', 'sentry.op': 'message', - 'sentry.origin': 'auto.amqplib.otel.consumer', + 'sentry.origin': CONSUMER_ORIGIN, }), }), ); @@ -159,9 +164,7 @@ describe('amqplib auto-instrumentation', () => { transaction: (transaction: TransactionEvent) => { expect(transaction.transaction).toBe('root span'); - const producerSpans = transaction.spans?.filter( - s => s.data?.['sentry.origin'] === 'auto.amqplib.otel.publisher', - ); + const producerSpans = transaction.spans?.filter(s => s.data?.['sentry.origin'] === PUBLISHER_ORIGIN); // The confirm channel internally calls the base publish; the instrumentation must not // double-instrument, so we expect exactly one producer span. diff --git a/packages/deno/src/index.ts b/packages/deno/src/index.ts index 7361e4d28e86..e87686c84d51 100644 --- a/packages/deno/src/index.ts +++ b/packages/deno/src/index.ts @@ -113,6 +113,7 @@ export { denoRedisIntegration } from './integrations/redis'; export type { DenoRedisIntegrationOptions } from './integrations/redis'; export { denoMysqlIntegration } from './integrations/mysql'; export { denoPostgresIntegration } from './integrations/postgres'; +export { denoAmqplibIntegration } from './integrations/amqplib'; export { denoContextIntegration } from './integrations/context'; export { globalHandlersIntegration } from './integrations/globalhandlers'; export { normalizePathsIntegration } from './integrations/normalizepaths'; diff --git a/packages/deno/src/integrations/amqplib.ts b/packages/deno/src/integrations/amqplib.ts new file mode 100644 index 000000000000..2ff88fd7e939 --- /dev/null +++ b/packages/deno/src/integrations/amqplib.ts @@ -0,0 +1,34 @@ +import { amqplibChannelIntegration } from '@sentry/server-utils/orchestrion'; +import type { Integration, IntegrationFn } from '@sentry/core'; +import { defineIntegration, extendIntegration } from '@sentry/core'; +import { setAsyncLocalStorageAsyncContextStrategy } from '../async'; + +const INTEGRATION_NAME = 'DenoAmqplib' as const; + +/** + * Create spans for `amqplib` publish/consume operations under Deno. + * + * `amqplib` channels are injected by the orchestrion runtime hook at load time. + * The `@sentry/deno/import` loader must be active for this integration to + * record anything. + * + * The channel-subscription logic is shared with the other server runtimes in + * `@sentry/server-utils`. This just installs Deno's `AsyncLocalStorage` context + * strategy (so spans nest under the active span and survive amqplib's internal + * callback dispatch) before delegating. + */ +const _denoAmqplibIntegration = (() => { + const inner = amqplibChannelIntegration(); + + return extendIntegration(inner, { + name: INTEGRATION_NAME, + setupOnce() { + setAsyncLocalStorageAsyncContextStrategy(); + }, + }); +}) satisfies IntegrationFn; + +export const denoAmqplibIntegration = defineIntegration(_denoAmqplibIntegration) as () => Integration & { + name: 'DenoAmqplib'; + setupOnce: () => void; +}; diff --git a/packages/deno/src/sdk.ts b/packages/deno/src/sdk.ts index db846d6b8007..f67b545966de 100644 --- a/packages/deno/src/sdk.ts +++ b/packages/deno/src/sdk.ts @@ -23,6 +23,7 @@ import { } from './denoVersion'; import { denoServeIntegration } from './integrations/deno-serve'; import { denoHttpIntegration } from './integrations/http'; +import { denoAmqplibIntegration } from './integrations/amqplib'; import { denoMysqlIntegration } from './integrations/mysql'; import { denoPostgresIntegration } from './integrations/postgres'; import { denoRedisIntegration } from './integrations/redis'; @@ -60,7 +61,9 @@ export function getDefaultIntegrations(_options: Options): Integration[] { // It's possible that the orchestrion channels will be injected AFTER // (or in parallel to) loading the SDK, so we only gate on whether the // feature is possible. If they're never loaded, it'll just be a no-op. - ...(MODULE_REGISTER_HOOKS_SUPPORTED ? [denoMysqlIntegration(), denoPostgresIntegration()] : []), + ...(MODULE_REGISTER_HOOKS_SUPPORTED + ? [denoMysqlIntegration(), denoPostgresIntegration(), denoAmqplibIntegration()] + : []), contextLinesIntegration(), normalizePathsIntegration(), globalHandlersIntegration(), diff --git a/packages/server-utils/src/integrations/tracing-channel/amqplib.ts b/packages/server-utils/src/integrations/tracing-channel/amqplib.ts new file mode 100644 index 000000000000..08d2e158b9bb --- /dev/null +++ b/packages/server-utils/src/integrations/tracing-channel/amqplib.ts @@ -0,0 +1,589 @@ +/* eslint-disable max-lines */ +import * as diagnosticsChannel from 'node:diagnostics_channel'; +import type { IntegrationFn, Span, SpanAttributes } from '@sentry/core'; +import { + continueTrace, + debug, + defineIntegration, + getTraceData, + SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, + SPAN_KIND, + SPAN_STATUS_ERROR, + startInactiveSpan, + timestampInSeconds, + waitForTracingChannelBinding, +} from '@sentry/core'; +import { MESSAGING_SYSTEM } from '@sentry/conventions/attributes'; +import { DEBUG_BUILD } from '../../debug-build'; +import { CHANNELS } from '../../orchestrion/channels'; +import { bindTracingChannelToSpan } from '../../tracing-channel'; + +// NOTE: this uses the same name as the OTel integration by design. +// When enabled, the OTel 'Amqplib' integration is omitted from the default set. +const INTEGRATION_NAME = 'Amqplib' as const; + +const PUBLISHER_ORIGIN = 'auto.amqplib.orchestrion.publisher'; +const CONSUMER_ORIGIN = 'auto.amqplib.orchestrion.consumer'; + +// Legacy messaging semantic-conventions, inlined to keep this integration free of `@opentelemetry/*` +// deps. These mirror what the vendored OTel amqplib instrumentation has always emitted. +const ATTR_MESSAGING_OPERATION = 'messaging.operation'; +const ATTR_MESSAGING_DESTINATION = 'messaging.destination'; +const ATTR_MESSAGING_DESTINATION_KIND = 'messaging.destination_kind'; +const ATTR_MESSAGING_RABBITMQ_ROUTING_KEY = 'messaging.rabbitmq.routing_key'; +const ATTR_MESSAGING_PROTOCOL = 'messaging.protocol'; +const ATTR_MESSAGING_PROTOCOL_VERSION = 'messaging.protocol_version'; +const ATTR_MESSAGING_URL = 'messaging.url'; +const ATTR_MESSAGING_MESSAGE_ID = 'messaging.message_id'; +const ATTR_MESSAGING_CONVERSATION_ID = 'messaging.conversation_id'; +const MESSAGING_DESTINATION_KIND_VALUE_TOPIC = 'topic'; +const MESSAGING_OPERATION_VALUE_PROCESS = 'process'; +// Inlined (rather than imported) because the `@sentry/conventions` exports are deprecated; the SDK +// has always emitted these legacy `net.*` attributes for amqplib connections. +const ATTR_NET_PEER_NAME = 'net.peer.name'; +const ATTR_NET_PEER_PORT = 'net.peer.port'; + +// To prevent reference leaks from un-acked messages, their spans are closed after this timeout. The +// upstream instrumentation exposed this as the `consumeTimeoutMs` option; the SDK always used the default. +const CONSUME_TIMEOUT_MS = 1000 * 60; // 1 minute + +// The end-operation labels used in the consumer span's error status message. +const END_OP = { + Ack: 'ack', + AckAll: 'ackAll', + Reject: 'reject', + Nack: 'nack', + NackAll: 'nackAll', + ChannelClosed: 'channel closed', + ChannelError: 'channel error', + InstrumentationTimeout: 'instrumentation timeout', +} as const; +type EndOp = (typeof END_OP)[keyof typeof END_OP]; + +// State stashed on the live amqplib objects (mirrors the vendored OTel instrumentation's symbols). +const MESSAGE_STORED_SPAN: unique symbol = Symbol('sentry.amqplib.message.stored-span'); +const CHANNEL_SPANS_NOT_ENDED: unique symbol = Symbol('sentry.amqplib.channel.spans-not-ended'); +const CHANNEL_CONSUME_TIMEOUT_TIMER: unique symbol = Symbol('sentry.amqplib.channel.consume-timeout-timer'); +const CHANNEL_CONSUMER_INFO: unique symbol = Symbol('sentry.amqplib.channel.consumer-info'); +const CHANNEL_IS_CONFIRM_PUBLISHING: unique symbol = Symbol('sentry.amqplib.channel.is-confirm-publishing'); +const CONNECTION_ATTRIBUTES: unique symbol = Symbol('sentry.amqplib.connection.attributes'); + +interface MessageFields { + deliveryTag?: number; + exchange?: string; + routingKey?: string; + consumerTag?: string; +} + +interface MessageProperties { + headers?: Record; + messageId?: unknown; + correlationId?: unknown; +} + +interface ConsumeMessage { + fields?: MessageFields; + properties?: MessageProperties; + [MESSAGE_STORED_SPAN]?: Span; +} + +interface PublishOptions { + headers?: Record; + messageId?: unknown; + correlationId?: unknown; +} + +interface ConnectionLike { + serverProperties?: { product?: string }; + // Some amqplib versions nest the low-level connection one level deeper. + connection?: { serverProperties?: { product?: string } }; + [CONNECTION_ATTRIBUTES]?: SpanAttributes; +} + +interface ConsumerInfo { + noAck: boolean; + queue: string; +} + +interface ChannelLike { + connection?: ConnectionLike; + on?: (event: string, listener: (...args: unknown[]) => void) => unknown; + [CHANNEL_SPANS_NOT_ENDED]?: { msg: ConsumeMessage; timeOfConsume: number }[]; + [CHANNEL_CONSUME_TIMEOUT_TIMER]?: ReturnType; + [CHANNEL_CONSUMER_INFO]?: Map; + [CHANNEL_IS_CONFIRM_PUBLISHING]?: boolean; +} + +/** + * The shape orchestrion's transform attaches to the tracing-channel `context`. Documented here rather + * than imported because orchestrion's runtime doesn't export it. + */ +interface AmqpChannelContext { + // The live args array passed to the wrapped call. + arguments: unknown[]; + self?: ChannelLike; + result?: unknown; + error?: unknown; +} + +interface AmqpDispatchContext extends AmqpChannelContext { + // Whether the delivering consumer registered with `noAck`; set at span creation so `deferSpanEnd` + // knows whether the helper should end the span (noAck) or leave it open until ack/nack (manual ack). + _sentryNoAck?: boolean; +} + +interface AmqpConnectContext { + arguments?: unknown[]; + result?: unknown; +} + +const NOOP = (): void => {}; + +const _amqplibChannelIntegration = (() => { + return { + name: INTEGRATION_NAME, + setupOnce() { + // `tracingChannel` is unavailable before Node 18.19 so do nothing in that case. + if (!diagnosticsChannel.tracingChannel) { + return; + } + + DEBUG_BUILD && debug.log('[orchestrion:amqplib] subscribing to amqplib tracing channels'); + + waitForTracingChannelBinding(() => { + subscribeConnect(); + subscribePublish(); + subscribeConfirmPublish(); + subscribeConsume(); + subscribeDispatch(); + subscribeSettle(); + }); + }, + }; +}) satisfies IntegrationFn; + +/** + * Producer span for `Channel.prototype.publish`. Creates a PRODUCER span, injects the trace headers + * into the publish options, and ends when the (synchronous) publish call returns. Skips the confirm + * channel's internal `super.publish` call, which is already handled by {@link subscribeConfirmPublish}. + */ +function subscribePublish(): void { + bindTracingChannelToSpan(diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_PUBLISH), data => { + if (data.self?.[CHANNEL_IS_CONFIRM_PUBLISHING]) { + return undefined; + } + return startPublishSpan(data); + }); +} + +/** + * Producer span for `ConfirmChannel.prototype.publish`. The span ends on the broker-confirm callback + * (the channel's trailing `cb` arg, wrapped by orchestrion) rather than synchronously. A synchronous + * flag on the channel suppresses the base `publish` channel that `super.publish` triggers. + */ +function subscribeConfirmPublish(): void { + const channel = diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_CONFIRM_PUBLISH); + + bindTracingChannelToSpan(channel, data => { + if (data.self) { + data.self[CHANNEL_IS_CONFIRM_PUBLISHING] = true; + } + return startPublishSpan(data); + }); + + // `super.publish` runs synchronously inside the confirm publish body, so clearing the flag on the + // synchronous `end` (which fires after the body returns) is enough to guard the base publish. + channel.end.subscribe(message => { + const self = (message as AmqpChannelContext).self; + if (self) { + self[CHANNEL_IS_CONFIRM_PUBLISHING] = false; + } + }); +} + +/** + * Records `consumerTag -> { noAck, queue }` when a consumer is registered, so the per-message + * dispatch hook can name the span after the queue and know when to end it. + */ +function subscribeConsume(): void { + const channel = diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_CONSUME); + + // A `start` subscriber is required for orchestrion to wrap `consume` at all. + channel.start.subscribe(NOOP); + channel.asyncEnd.subscribe(message => { + const data = message as AmqpChannelContext; + const consumerChannel = data.self; + const result = data.result as { consumerTag?: string } | undefined; + const consumerTag = result?.consumerTag; + if (!consumerChannel || !consumerTag) { + return; + } + + ensureChannelState(consumerChannel); + const queueArg = data.arguments[0]; + const queue = typeof queueArg === 'string' ? queueArg : ''; + const options = data.arguments[2] as { noAck?: boolean } | undefined; + consumerChannel[CHANNEL_CONSUMER_INFO]?.set(consumerTag, { noAck: !!options?.noAck, queue }); + }); +} + +/** + * Per delivered message (`BaseChannel.prototype.dispatchMessage`): continues the producer's trace, + * opens a CONSUMER span, and runs the user callback under it. Manual-ack consumers keep the span open + * until `ack`/`nack`/`reject`/timeout/close; `noAck` consumers end it when dispatch returns. + */ +function subscribeDispatch(): void { + bindTracingChannelToSpan( + diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_DISPATCH), + data => { + const channel = data.self; + const fields = data.arguments[0] as MessageFields | undefined; + const msg = data.arguments[1] as ConsumeMessage | null | undefined; + // `message` is null for a consumer-cancel notification: not a real message, so no span. + if (!channel || !msg) { + return undefined; + } + + ensureChannelState(channel); + const info = fields?.consumerTag ? channel[CHANNEL_CONSUMER_INFO]?.get(fields.consumerTag) : undefined; + const queue = info?.queue ?? msg.fields?.routingKey ?? ''; + const noAck = info?.noAck ?? false; + + const headers = msg.properties?.headers; + const sentryTrace = getHeaderAsString(headers, 'sentry-trace'); + const baggage = getHeaderAsString(headers, 'baggage'); + + // Continue the producer's trace so the consumer span links back to the publishing service. + const span = continueTrace({ sentryTrace, baggage }, () => startConsumeSpan(queue, msg, channel)); + + if (!noAck) { + // Track the message so its span can be ended when the user calls ack/nack/reject (or on + // timeout/channel close). + channel[CHANNEL_SPANS_NOT_ENDED]?.push({ msg, timeOfConsume: timestampInSeconds() }); + msg[MESSAGE_STORED_SPAN] = span; + } + + data._sentryNoAck = noAck; + return span; + }, + { + // Manual-ack consumers: the span outlives the dispatch call and is ended by ack/nack/reject + // (or timeout/close), so take ownership and don't let the helper end it here. noAck consumers + // have no settle call, so let the helper end the span when dispatch returns. + deferSpanEnd({ data }) { + return !data._sentryNoAck; + }, + }, + ); +} + +/** Ends consumer spans when the user acks/nacks/rejects a message or the channel closes/errors. */ +function subscribeSettle(): void { + diagnosticsChannel + .tracingChannel(CHANNELS.AMQPLIB_ACK) + .start.subscribe(message => handleAck(message as AmqpChannelContext, false, END_OP.Ack)); + diagnosticsChannel + .tracingChannel(CHANNELS.AMQPLIB_NACK) + .start.subscribe(message => handleAck(message as AmqpChannelContext, true, END_OP.Nack)); + diagnosticsChannel + .tracingChannel(CHANNELS.AMQPLIB_REJECT) + .start.subscribe(message => handleAck(message as AmqpChannelContext, true, END_OP.Reject)); + diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_ACK_ALL).start.subscribe(message => { + const data = message as AmqpChannelContext; + if (data.self) { + endAllSpansOnChannel(data.self, false, END_OP.AckAll, undefined); + } + }); + diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_NACK_ALL).start.subscribe(message => { + const data = message as AmqpChannelContext; + if (data.self) { + endAllSpansOnChannel(data.self, true, END_OP.NackAll, data.arguments[0] as boolean | undefined); + } + }); +} + +/** Captures connection attributes on the connection object for span-time reads via `channel.connection`. */ +function subscribeConnect(): void { + const channel = diagnosticsChannel.tracingChannel(CHANNELS.AMQPLIB_CONNECT); + // A `start` subscriber is required for orchestrion to wrap the callback-style `connect` at all. + channel.start.subscribe(NOOP); + channel.asyncEnd.subscribe(message => { + const data = message as AmqpConnectContext; + const conn = data.result as ConnectionLike | undefined; + if (!conn || typeof conn !== 'object') { + return; + } + conn[CONNECTION_ATTRIBUTES] = { + ...getConnectionAttributesFromUrl(data.arguments?.[0]), + ...getConnectionAttributesFromServer(conn), + }; + }); +} + +function handleAck(data: AmqpChannelContext, isRejected: boolean, endOperation: EndOp): void { + const channel = data.self; + if (!channel) { + return; + } + const message = data.arguments[0] as ConsumeMessage | undefined; + if (!message) { + return; + } + + // `reject(message, requeue)` carries requeue in arg 1; `ack`/`nack` carry `allUpTo` in arg 1. + const allUpToOrRequeue = data.arguments[1] as boolean | undefined; + const requeue = data.arguments[2] as boolean | undefined; + const requeueResolved = endOperation === END_OP.Reject ? allUpToOrRequeue : requeue; + + const spansNotEnded = channel[CHANNEL_SPANS_NOT_ENDED] ?? []; + const msgIndex = spansNotEnded.findIndex(msgDetails => msgDetails.msg === message); + if (msgIndex < 0) { + // Not tracked (e.g. the user acked the same message twice) — end the stored span directly. + endConsumerSpan(message, isRejected, endOperation, requeueResolved); + } else if (endOperation !== END_OP.Reject && allUpToOrRequeue) { + for (let i = 0; i <= msgIndex; i++) { + endConsumerSpan(spansNotEnded[i]!.msg, isRejected, endOperation, requeueResolved); + } + spansNotEnded.splice(0, msgIndex + 1); + } else { + endConsumerSpan(message, isRejected, endOperation, requeueResolved); + spansNotEnded.splice(msgIndex, 1); + } +} + +function ensureChannelState(channel: ChannelLike): void { + if (Object.prototype.hasOwnProperty.call(channel, CHANNEL_SPANS_NOT_ENDED)) { + return; + } + + channel[CHANNEL_SPANS_NOT_ENDED] = []; + channel[CHANNEL_CONSUMER_INFO] = new Map(); + + const timer = setInterval(() => checkConsumeTimeoutOnChannel(channel), CONSUME_TIMEOUT_MS); + timer.unref?.(); + channel[CHANNEL_CONSUME_TIMEOUT_TIMER] = timer; + + // End outstanding spans and stop the timer when the channel goes away (replaces patching `emit`). + if (typeof channel.on === 'function') { + channel.on('close', () => { + endAllSpansOnChannel(channel, true, END_OP.ChannelClosed, undefined); + const activeTimer = channel[CHANNEL_CONSUME_TIMEOUT_TIMER]; + if (activeTimer) { + clearInterval(activeTimer); + } + channel[CHANNEL_CONSUME_TIMEOUT_TIMER] = undefined; + }); + channel.on('error', () => { + endAllSpansOnChannel(channel, true, END_OP.ChannelError, undefined); + }); + } +} + +function checkConsumeTimeoutOnChannel(channel: ChannelLike): void { + const currentTime = timestampInSeconds(); + const spansNotEnded = channel[CHANNEL_SPANS_NOT_ENDED] ?? []; + let i: number; + for (i = 0; i < spansNotEnded.length; i++) { + const currMessage = spansNotEnded[i]!; + const timeFromConsumeMs = (currentTime - currMessage.timeOfConsume) * 1000; + if (timeFromConsumeMs < CONSUME_TIMEOUT_MS) { + break; + } + endConsumerSpan(currMessage.msg, null, END_OP.InstrumentationTimeout, true); + } + spansNotEnded.splice(0, i); +} + +function endAllSpansOnChannel( + channel: ChannelLike, + isRejected: boolean, + operation: EndOp, + requeue: boolean | undefined, +): void { + const spansNotEnded = channel[CHANNEL_SPANS_NOT_ENDED] ?? []; + spansNotEnded.forEach(msgDetails => { + endConsumerSpan(msgDetails.msg, isRejected, operation, requeue); + }); + channel[CHANNEL_SPANS_NOT_ENDED] = []; +} + +function endConsumerSpan( + message: ConsumeMessage, + isRejected: boolean | null, + operation: EndOp, + requeue: boolean | undefined, +): void { + const storedSpan = message[MESSAGE_STORED_SPAN]; + if (!storedSpan) { + return; + } + if (isRejected !== false) { + storedSpan.setStatus({ + code: SPAN_STATUS_ERROR, + message: + operation !== END_OP.ChannelClosed && operation !== END_OP.ChannelError + ? `${operation} called on message${ + requeue === true ? ' with requeue' : requeue === false ? ' without requeue' : '' + }` + : operation, + }); + } + storedSpan.end(); + message[MESSAGE_STORED_SPAN] = undefined; +} + +/** Starts an inactive PRODUCER span and propagates its trace into the publish `options.headers`. */ +function startPublishSpan(data: AmqpChannelContext): Span { + const exchangeArg = data.arguments[0]; + const routingKeyArg = data.arguments[1]; + const exchange = typeof exchangeArg === 'string' ? exchangeArg : ''; + const routingKey = typeof routingKeyArg === 'string' ? routingKeyArg : ''; + let options = data.arguments[3] as PublishOptions | undefined; + + const span = startInactiveSpan({ + name: `publish ${normalizeExchange(exchange)}`, + op: 'message', + kind: SPAN_KIND.PRODUCER, + attributes: { + ...getStoredConnectionAttributes(data.self), + [ATTR_MESSAGING_DESTINATION]: exchange, + [ATTR_MESSAGING_DESTINATION_KIND]: MESSAGING_DESTINATION_KIND_VALUE_TOPIC, + [ATTR_MESSAGING_RABBITMQ_ROUTING_KEY]: routingKey, + [ATTR_MESSAGING_MESSAGE_ID]: options?.messageId as string | undefined, + [ATTR_MESSAGING_CONVERSATION_ID]: options?.correlationId as string | undefined, + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: PUBLISHER_ORIGIN, + }, + }); + + if (!options || typeof options !== 'object') { + options = {}; + data.arguments[3] = options; + } + const headers = options.headers && typeof options.headers === 'object' ? options.headers : (options.headers = {}); + const traceData = getTraceData({ span }); + if (traceData['sentry-trace']) { + headers['sentry-trace'] = traceData['sentry-trace']; + } + if (traceData.baggage) { + headers['baggage'] = traceData.baggage; + } + + return span; +} + +/** Starts an inactive CONSUMER (process) span carrying the amqplib messaging attributes. */ +function startConsumeSpan(queue: string, msg: ConsumeMessage, channel: ChannelLike): Span { + return startInactiveSpan({ + name: `${queue} process`, + op: 'message', + kind: SPAN_KIND.CONSUMER, + attributes: { + ...getStoredConnectionAttributes(channel), + [ATTR_MESSAGING_DESTINATION]: msg.fields?.exchange, + [ATTR_MESSAGING_DESTINATION_KIND]: MESSAGING_DESTINATION_KIND_VALUE_TOPIC, + [ATTR_MESSAGING_RABBITMQ_ROUTING_KEY]: msg.fields?.routingKey, + [ATTR_MESSAGING_OPERATION]: MESSAGING_OPERATION_VALUE_PROCESS, + [ATTR_MESSAGING_MESSAGE_ID]: msg.properties?.messageId as string | undefined, + [ATTR_MESSAGING_CONVERSATION_ID]: msg.properties?.correlationId as string | undefined, + [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: CONSUMER_ORIGIN, + }, + }); +} + +/** + * Reads the connection attributes stashed by the `connect` channel, falling back to the live + * connection's server product so `messaging.system` is populated even when the connection was + * established before the integration ran. + */ +function getStoredConnectionAttributes(channel: ChannelLike | undefined): SpanAttributes { + const connection = channel?.connection; + const stored = connection?.[CONNECTION_ATTRIBUTES]; + if (stored) { + return stored; + } + const product = connection?.serverProperties?.product ?? connection?.connection?.serverProperties?.product; + if (typeof product === 'string' && product) { + return { [MESSAGING_SYSTEM]: product.toLowerCase() }; + } + return {}; +} + +function getConnectionAttributesFromServer(conn: ConnectionLike): SpanAttributes { + const product = conn.serverProperties?.product ?? conn.connection?.serverProperties?.product; + if (typeof product === 'string' && product) { + return { [MESSAGING_SYSTEM]: product.toLowerCase() }; + } + return {}; +} + +function getConnectionAttributesFromUrl(url: unknown): SpanAttributes { + const attributes: SpanAttributes = { + // The only protocol supported by the instrumented library. + [ATTR_MESSAGING_PROTOCOL_VERSION]: '0.9.1', + }; + + const resolvedUrl = url || 'amqp://localhost'; + if (typeof resolvedUrl === 'object') { + const connectOptions = resolvedUrl as { protocol?: string; hostname?: string; port?: number }; + const protocol = getProtocol(connectOptions.protocol); + attributes[ATTR_MESSAGING_PROTOCOL] = protocol; + attributes[ATTR_NET_PEER_NAME] = getHostname(connectOptions.hostname); + attributes[ATTR_NET_PEER_PORT] = getPort(connectOptions.port, protocol); + } else if (typeof resolvedUrl === 'string') { + const censoredUrl = censorPassword(resolvedUrl); + attributes[ATTR_MESSAGING_URL] = censoredUrl; + try { + const urlParts = new URL(censoredUrl); + const protocol = getProtocol(urlParts.protocol); + attributes[ATTR_MESSAGING_PROTOCOL] = protocol; + attributes[ATTR_NET_PEER_NAME] = getHostname(urlParts.hostname); + attributes[ATTR_NET_PEER_PORT] = getPort(urlParts.port ? parseInt(urlParts.port, 10) : undefined, protocol); + } catch { + // best-effort: a malformed url simply yields fewer connection attributes + } + } + return attributes; +} + +function normalizeExchange(exchangeName: string): string { + return exchangeName !== '' ? exchangeName : ''; +} + +function censorPassword(url: string): string { + return url.replace(/:[^:@/]*@/, ':***@'); +} + +function getPort(portFromUrl: number | undefined, resolvedProtocol: string): number { + // Mimics amqplib's own defaulting; the resolved protocol is upper-cased. + return portFromUrl || (resolvedProtocol === 'AMQP' ? 5672 : 5671); +} + +function getProtocol(protocolFromUrl: string | undefined): string { + const resolvedProtocol = protocolFromUrl || 'amqp'; + const noEndingColon = resolvedProtocol.endsWith(':') + ? resolvedProtocol.substring(0, resolvedProtocol.length - 1) + : resolvedProtocol; + return noEndingColon.toUpperCase(); +} + +function getHostname(hostnameFromUrl: string | undefined): string { + // An empty hostname is forwarded to `net`, which defaults it to localhost. + return hostnameFromUrl || 'localhost'; +} + +function getHeaderAsString(headers: Record | undefined, key: string): string | undefined { + const value = headers?.[key]; + if (value == null) { + return undefined; + } + return Array.isArray(value) ? String(value[0]) : String(value); +} + +/** + * EXPERIMENTAL: orchestrion-driven `amqplib` integration. + * + * Subscribes to the `orchestrion:amqplib:*` diagnostics_channels that the orchestrion code transform + * injects into `amqplib`'s channel/connection methods. Requires the orchestrion runtime hook or + * bundler plugin to be active. + */ +export const amqplibChannelIntegration = defineIntegration(_amqplibChannelIntegration); diff --git a/packages/server-utils/src/orchestrion/channels.ts b/packages/server-utils/src/orchestrion/channels.ts index b008aad618a2..cb28c8e747b4 100644 --- a/packages/server-utils/src/orchestrion/channels.ts +++ b/packages/server-utils/src/orchestrion/channels.ts @@ -5,6 +5,7 @@ import { pgChannels } from './config/pg'; import { openaiChannels } from './config/openai'; import { anthropicAiChannels } from './config/anthropic-ai'; import { vercelAiChannels } from './config/vercel-ai'; +import { amqplibChannels } from './config/amqplib'; /** * Fully-qualified `diagnostics_channel` names that orchestrion publishes to. @@ -27,6 +28,7 @@ export const CHANNELS = { ...openaiChannels, ...anthropicAiChannels, ...vercelAiChannels, + ...amqplibChannels, } as const; export type ChannelName = (typeof CHANNELS)[keyof typeof CHANNELS]; diff --git a/packages/server-utils/src/orchestrion/config/amqplib.ts b/packages/server-utils/src/orchestrion/config/amqplib.ts new file mode 100644 index 000000000000..75b5ced78cae --- /dev/null +++ b/packages/server-utils/src/orchestrion/config/amqplib.ts @@ -0,0 +1,89 @@ +import type { InstrumentationConfig } from '@apm-js-collab/code-transformer'; + +// `amqplib` splits its API across three files: +// - `lib/channel_model.js` holds `class Channel` (publish/consume/ack/nack/reject/…) and +// `class ConfirmChannel extends Channel` (a `publish` that takes a broker-confirm callback). +// - `lib/channel.js` holds `class BaseChannel` whose `dispatchMessage` invokes the registered +// consumer callback once per delivered message — the natural per-message hook for consumer spans +// (`Channel.consume` itself only registers the callback, it isn't called per message). +// - `lib/connect.js` holds the `connect` function whose callback receives the open connection, used +// to capture connection attributes. +// +// The version range mirrors `supportedVersions` in the vendored OTel instrumentation. +const module = { name: 'amqplib', versionRange: '>=0.5.5 <2' } as const; + +export const amqplibConfig = [ + // Producer span + trace-header injection. `sendToQueue` delegates to `publish`, so it's covered. + { + channelName: 'publish', + module: { ...module, filePath: 'lib/channel_model.js' }, + functionQuery: { className: 'Channel', methodName: 'publish', kind: 'Sync' }, + }, + // Confirm-channel producer span; the trailing broker-confirm callback ends the span when the + // broker acks/nacks. It internally calls `super.publish`, so the subscriber guards against the + // base `publish` channel double-instrumenting. + { + channelName: 'confirmPublish', + module: { ...module, filePath: 'lib/channel_model.js' }, + functionQuery: { className: 'ConfirmChannel', methodName: 'publish', kind: 'Callback' }, + }, + // Records `consumerTag -> { noAck, queue }` so the per-message dispatch hook knows how to name and + // when to end the consumer span. + { + channelName: 'consume', + module: { ...module, filePath: 'lib/channel_model.js' }, + functionQuery: { className: 'Channel', methodName: 'consume', kind: 'Async' }, + }, + // Per delivered message: creates the consumer span and runs the user callback under it. + { + channelName: 'dispatch', + module: { ...module, filePath: 'lib/channel.js' }, + functionQuery: { className: 'BaseChannel', methodName: 'dispatchMessage', kind: 'Sync' }, + }, + // End the consumer span when the user settles the message. + { + channelName: 'ack', + module: { ...module, filePath: 'lib/channel_model.js' }, + functionQuery: { className: 'Channel', methodName: 'ack', kind: 'Sync' }, + }, + { + channelName: 'nack', + module: { ...module, filePath: 'lib/channel_model.js' }, + functionQuery: { className: 'Channel', methodName: 'nack', kind: 'Sync' }, + }, + { + channelName: 'reject', + module: { ...module, filePath: 'lib/channel_model.js' }, + functionQuery: { className: 'Channel', methodName: 'reject', kind: 'Sync' }, + }, + { + channelName: 'ackAll', + module: { ...module, filePath: 'lib/channel_model.js' }, + functionQuery: { className: 'Channel', methodName: 'ackAll', kind: 'Sync' }, + }, + { + channelName: 'nackAll', + module: { ...module, filePath: 'lib/channel_model.js' }, + functionQuery: { className: 'Channel', methodName: 'nackAll', kind: 'Sync' }, + }, + // Stashes connection attributes (url/host/port/protocol/server product) on the connection object + // for span-time reads via `channel.connection`. + { + channelName: 'connect', + module: { ...module, filePath: 'lib/connect.js' }, + functionQuery: { functionName: 'connect', kind: 'Callback' }, + }, +] satisfies InstrumentationConfig[]; + +export const amqplibChannels = { + AMQPLIB_PUBLISH: 'orchestrion:amqplib:publish', + AMQPLIB_CONFIRM_PUBLISH: 'orchestrion:amqplib:confirmPublish', + AMQPLIB_CONSUME: 'orchestrion:amqplib:consume', + AMQPLIB_DISPATCH: 'orchestrion:amqplib:dispatch', + AMQPLIB_ACK: 'orchestrion:amqplib:ack', + AMQPLIB_NACK: 'orchestrion:amqplib:nack', + AMQPLIB_REJECT: 'orchestrion:amqplib:reject', + AMQPLIB_ACK_ALL: 'orchestrion:amqplib:ackAll', + AMQPLIB_NACK_ALL: 'orchestrion:amqplib:nackAll', + AMQPLIB_CONNECT: 'orchestrion:amqplib:connect', +} as const; diff --git a/packages/server-utils/src/orchestrion/config/index.ts b/packages/server-utils/src/orchestrion/config/index.ts index 218cbf7e8875..d83337a794a7 100644 --- a/packages/server-utils/src/orchestrion/config/index.ts +++ b/packages/server-utils/src/orchestrion/config/index.ts @@ -6,6 +6,7 @@ import { openaiConfig } from './openai'; import { pgConfig } from './pg'; import { anthropicAiConfig } from './anthropic-ai'; import { vercelAiConfig } from './vercel-ai'; +import { amqplibConfig } from './amqplib'; export const SENTRY_INSTRUMENTATIONS: InstrumentationConfig[] = [ ...mysqlConfig, @@ -15,6 +16,7 @@ export const SENTRY_INSTRUMENTATIONS: InstrumentationConfig[] = [ ...pgConfig, ...anthropicAiConfig, ...vercelAiConfig, + ...amqplibConfig, ]; /** diff --git a/packages/server-utils/src/orchestrion/index.ts b/packages/server-utils/src/orchestrion/index.ts index 3682ca752232..c534ba9eeeb0 100644 --- a/packages/server-utils/src/orchestrion/index.ts +++ b/packages/server-utils/src/orchestrion/index.ts @@ -1,3 +1,4 @@ +import { amqplibChannelIntegration } from '../integrations/tracing-channel/amqplib'; import { anthropicChannelIntegration } from '../integrations/tracing-channel/anthropic'; import { ioredisChannelIntegration } from '../integrations/tracing-channel/ioredis'; import { lruMemoizerChannelIntegration } from '../integrations/tracing-channel/lru-memoizer'; @@ -8,6 +9,7 @@ import { vercelAiChannelIntegration } from '../integrations/tracing-channel/verc export { detectOrchestrionSetup, isOrchestrionInjected } from './detect'; export { + amqplibChannelIntegration, anthropicChannelIntegration, ioredisChannelIntegration, lruMemoizerChannelIntegration, @@ -38,4 +40,5 @@ export const channelIntegrations = { openaiIntegration: openaiChannelIntegration, anthropicIntegration: anthropicChannelIntegration, vercelAiIntegration: vercelAiChannelIntegration, + amqplibIntegration: amqplibChannelIntegration, } as const; diff --git a/packages/server-utils/test/orchestrion/amqplib.test.ts b/packages/server-utils/test/orchestrion/amqplib.test.ts new file mode 100644 index 000000000000..616c1c23a4b5 --- /dev/null +++ b/packages/server-utils/test/orchestrion/amqplib.test.ts @@ -0,0 +1,230 @@ +import { AsyncLocalStorage } from 'node:async_hooks'; +import { tracingChannel } from 'node:diagnostics_channel'; +import type { Scope, Span } from '@sentry/core'; +import * as SentryCore from '@sentry/core'; +import { + _INTERNAL_setSpanForScope, + getDefaultCurrentScope, + getDefaultIsolationScope, + setAsyncContextStrategy, +} from '@sentry/core'; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, type MockInstance, vi } from 'vitest'; +import { amqplibChannelIntegration } from '../../src/orchestrion'; +import { CHANNELS } from '../../src/orchestrion/channels'; + +interface TestStore { + scope: Scope; + isolationScope: Scope; +} + +// `bindTracingChannelToSpan` only binds (and `setupOnce` only subscribes via +// `waitForTracingChannelBinding`) when an async-context strategy exposes a +// `getTracingChannelBinding`. Install a minimal one so the channel +// subscriptions actually register in this unit-test context (no SDK `init`). +function installTestAsyncContextStrategy(): void { + const asyncStorage = new AsyncLocalStorage(); + + function getScopes(): TestStore { + return asyncStorage.getStore() || { scope: getDefaultCurrentScope(), isolationScope: getDefaultIsolationScope() }; + } + + setAsyncContextStrategy({ + withScope: callback => { + const scope = getScopes().scope.clone(); + const isolationScope = getScopes().isolationScope; + return asyncStorage.run({ scope, isolationScope }, () => callback(scope)); + }, + withSetScope: (scope, callback) => { + const isolationScope = getScopes().isolationScope; + return asyncStorage.run({ scope, isolationScope }, () => callback(scope)); + }, + withIsolationScope: callback => { + const scope = getScopes().scope; + const isolationScope = getScopes().isolationScope.clone(); + return asyncStorage.run({ scope, isolationScope }, () => callback(isolationScope)); + }, + withSetIsolationScope: (isolationScope, callback) => { + const scope = getScopes().scope; + return asyncStorage.run({ scope, isolationScope }, () => callback(isolationScope)); + }, + getCurrentScope: () => getScopes().scope, + getIsolationScope: () => getScopes().isolationScope, + getTracingChannelBinding: () => ({ + asyncLocalStorage: asyncStorage, + getStoreWithActiveSpan: span => { + const scope = getScopes().scope.clone(); + const isolationScope = getScopes().isolationScope; + _INTERNAL_setSpanForScope(scope, span); + return { scope, isolationScope }; + }, + }), + }); +} + +function makeSpan(): Span { + return { end: vi.fn(), setStatus: vi.fn(), setAttributes: vi.fn() } as unknown as Span; +} + +// A channel instance whose `serverProperties.product` yields `messaging.system: 'rabbitmq'`. +function makeChannel(): { connection: { serverProperties: { product: string } } } { + return { connection: { serverProperties: { product: 'RabbitMQ' } } }; +} + +describe('amqplibChannelIntegration', () => { + let startInactiveSpanSpy: MockInstance; + let span: Span; + + beforeAll(() => { + installTestAsyncContextStrategy(); + amqplibChannelIntegration().setupOnce?.(); + }); + + afterAll(() => { + setAsyncContextStrategy(undefined); + }); + + beforeEach(() => { + span = makeSpan(); + startInactiveSpanSpy = vi.spyOn(SentryCore, 'startInactiveSpan').mockReturnValue(span); + // Continue the callback synchronously so the consumer span is built inline. + vi.spyOn(SentryCore, 'continueTrace').mockImplementation(((_options: unknown, cb: () => unknown) => + cb()) as unknown as typeof SentryCore.continueTrace); + }); + + afterEach(() => { + vi.restoreAllMocks(); + }); + + it('publish: builds a PRODUCER span with messaging attributes and the orchestrion origin', () => { + const ctx = { + arguments: ['my-exchange', 'my-routing-key', Buffer.from('hi'), { messageId: 'm1', correlationId: 'c1' }], + self: makeChannel(), + }; + + tracingChannel(CHANNELS.AMQPLIB_PUBLISH).traceSync(() => true, ctx); + + expect(startInactiveSpanSpy).toHaveBeenCalledWith( + expect.objectContaining({ + name: 'publish my-exchange', + op: 'message', + attributes: expect.objectContaining({ + 'messaging.system': 'rabbitmq', + 'messaging.destination': 'my-exchange', + 'messaging.rabbitmq.routing_key': 'my-routing-key', + 'messaging.message_id': 'm1', + 'messaging.conversation_id': 'c1', + 'sentry.origin': 'auto.amqplib.orchestrion.publisher', + }), + }), + ); + // Ends synchronously once the publish call returns. + expect(span.end).toHaveBeenCalledTimes(1); + }); + + it('publish: injects the trace headers into the publish options', () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ 'sentry-trace': 'trace-abc', baggage: 'baggage-abc' }); + const options: { headers?: Record } = {}; + const ctx = { arguments: ['ex', 'rk', Buffer.from('hi'), options], self: makeChannel() }; + + tracingChannel(CHANNELS.AMQPLIB_PUBLISH).traceSync(() => true, ctx); + + expect(options.headers).toEqual({ 'sentry-trace': 'trace-abc', baggage: 'baggage-abc' }); + }); + + it('publish: creates the options object when the caller omitted it', () => { + vi.spyOn(SentryCore, 'getTraceData').mockReturnValue({ 'sentry-trace': 'trace-xyz' }); + const ctx: { arguments: unknown[]; self: unknown } = { + arguments: ['ex', 'rk', Buffer.from('hi')], + self: makeChannel(), + }; + + tracingChannel(CHANNELS.AMQPLIB_PUBLISH).traceSync(() => true, ctx); + + expect(ctx.arguments[3]).toEqual({ headers: { 'sentry-trace': 'trace-xyz' } }); + }); + + it('consume + ack: builds a CONSUMER span that stays open until the message is acked', async () => { + const channel = makeChannel(); + + // Register the consumer so the dispatch hook knows the queue + ack mode. amqplib resolves + // `consume` (registering the consumerTag) before any message can be dispatched. + await tracingChannel(CHANNELS.AMQPLIB_CONSUME).tracePromise(async () => ({ consumerTag: 'ct-1' }), { + arguments: ['queue1', () => {}, { noAck: false }], + self: channel, + }); + + const message = { fields: { exchange: '', routingKey: 'queue1' }, properties: { headers: {} } }; + tracingChannel(CHANNELS.AMQPLIB_DISPATCH).traceSync(() => undefined, { + arguments: [{ consumerTag: 'ct-1' }, message], + self: channel, + }); + + expect(startInactiveSpanSpy).toHaveBeenCalledWith( + expect.objectContaining({ + name: 'queue1 process', + op: 'message', + attributes: expect.objectContaining({ + 'messaging.system': 'rabbitmq', + 'messaging.operation': 'process', + 'sentry.origin': 'auto.amqplib.orchestrion.consumer', + }), + }), + ); + // Manual ack: the span must stay open after the dispatch returns. + expect(span.end).not.toHaveBeenCalled(); + + tracingChannel(CHANNELS.AMQPLIB_ACK).traceSync(() => undefined, { arguments: [message], self: channel }); + + expect(span.end).toHaveBeenCalledTimes(1); + }); + + it('consume (noAck): ends the CONSUMER span when dispatch returns', async () => { + const channel = makeChannel(); + + await tracingChannel(CHANNELS.AMQPLIB_CONSUME).tracePromise(async () => ({ consumerTag: 'ct-2' }), { + arguments: ['queue-noack', () => {}, { noAck: true }], + self: channel, + }); + + const message = { fields: { exchange: '', routingKey: 'queue-noack' }, properties: { headers: {} } }; + tracingChannel(CHANNELS.AMQPLIB_DISPATCH).traceSync(() => undefined, { + arguments: [{ consumerTag: 'ct-2' }, message], + self: channel, + }); + + expect(span.end).toHaveBeenCalledTimes(1); + }); + + it('reject: sets an error status on the consumer span', async () => { + const channel = makeChannel(); + + await tracingChannel(CHANNELS.AMQPLIB_CONSUME).tracePromise(async () => ({ consumerTag: 'ct-3' }), { + arguments: ['queue-reject', () => {}, { noAck: false }], + self: channel, + }); + + const message = { fields: { exchange: '', routingKey: 'queue-reject' }, properties: { headers: {} } }; + tracingChannel(CHANNELS.AMQPLIB_DISPATCH).traceSync(() => undefined, { + arguments: [{ consumerTag: 'ct-3' }, message], + self: channel, + }); + + tracingChannel(CHANNELS.AMQPLIB_REJECT).traceSync(() => undefined, { + arguments: [message, false], + self: channel, + }); + + expect(span.setStatus).toHaveBeenCalledWith(expect.objectContaining({ code: expect.anything() })); + expect(span.end).toHaveBeenCalledTimes(1); + }); + + it('dispatch: does not create a span for a consumer-cancel notification (null message)', () => { + const channel = makeChannel(); + tracingChannel(CHANNELS.AMQPLIB_DISPATCH).traceSync(() => undefined, { + arguments: [{ consumerTag: 'ct-x' }, null], + self: channel, + }); + + expect(startInactiveSpanSpy).not.toHaveBeenCalled(); + }); +}); From da9ed11838d425154162bde3a43c57a03c3424a5 Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Mon, 6 Jul 2026 14:00:30 +0200 Subject: [PATCH 2/6] add more tests --- .../suites/tracing/amqplib/test.ts | 5 + .../test/orchestrion/amqplib.test.ts | 212 ++++++++++++++++++ 2 files changed, 217 insertions(+) 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 7380b1c807e8..fc8799026c6a 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts @@ -17,6 +17,11 @@ const expectedProducerSpan = (routingKey: string) => data: expect.objectContaining({ 'messaging.system': 'rabbitmq', 'messaging.rabbitmq.routing_key': routingKey, + 'messaging.url': 'amqp://sentry:***@localhost:5672/', + 'messaging.protocol': 'AMQP', + 'messaging.protocol_version': '0.9.1', + 'net.peer.name': 'localhost', + 'net.peer.port': 5672, 'otel.kind': 'PRODUCER', 'sentry.op': 'message', 'sentry.origin': PUBLISHER_ORIGIN, diff --git a/packages/server-utils/test/orchestrion/amqplib.test.ts b/packages/server-utils/test/orchestrion/amqplib.test.ts index 616c1c23a4b5..bfc8d47a228e 100644 --- a/packages/server-utils/test/orchestrion/amqplib.test.ts +++ b/packages/server-utils/test/orchestrion/amqplib.test.ts @@ -227,4 +227,216 @@ describe('amqplibChannelIntegration', () => { expect(startInactiveSpanSpy).not.toHaveBeenCalled(); }); + + // A confirm channel's `publish` internally calls the base `Channel.prototype.publish`, which fires + // its own tracing channel. The subscriber must produce a single producer span, not two. + describe('confirm channel guard', () => { + // Drives a `ConfirmChannel.publish` the way orchestrion does — as a callback channel whose body + // synchronously runs the base `publish` channel (mirroring `super.publish`) and whose trailing + // callback fires when the broker confirms. + function publishOnConfirmChannel( + channel: unknown, + { onBrokerConfirm }: { onBrokerConfirm: (cb: (err?: unknown) => void) => void }, + ): void { + tracingChannel(CHANNELS.AMQPLIB_CONFIRM_PUBLISH).traceCallback( + function (_options: unknown, cb: (err?: unknown) => void) { + tracingChannel(CHANNELS.AMQPLIB_PUBLISH).traceSync(() => true, { + arguments: ['orders', 'confirm-key', Buffer.from('payload'), {}], + self: channel, + }); + onBrokerConfirm(cb); + }, + 1, + { arguments: ['orders', 'confirm-key', Buffer.from('payload'), {}], self: channel }, + undefined, + {}, + () => {}, + ); + } + + it('creates exactly one producer span despite the internal base publish', async () => { + const channel = makeChannel(); + + await new Promise(resolve => { + publishOnConfirmChannel(channel, { + onBrokerConfirm: cb => + setImmediate(() => { + cb(null); + resolve(); + }), + }); + }); + + expect(startInactiveSpanSpy).toHaveBeenCalledTimes(1); + expect(span.end).toHaveBeenCalledTimes(1); + }); + + it('clears the guard so later publishes on the same channel are still instrumented', async () => { + const channel = makeChannel(); + + await new Promise(resolve => { + publishOnConfirmChannel(channel, { + onBrokerConfirm: cb => + setImmediate(() => { + cb(null); + resolve(); + }), + }); + }); + + tracingChannel(CHANNELS.AMQPLIB_PUBLISH).traceSync(() => true, { + arguments: ['orders', 'plain-key', Buffer.from('payload'), {}], + self: channel, + }); + + // 1 span for the confirm publish + 1 for the later plain publish. A stuck guard would suppress + // the second, leaving only 1. + expect(startInactiveSpanSpy).toHaveBeenCalledTimes(2); + }); + + it('does not end the producer span until the broker confirm callback fires', () => { + const channel = makeChannel(); + let brokerConfirm: (err?: unknown) => void = () => {}; + + publishOnConfirmChannel(channel, { + onBrokerConfirm: cb => { + brokerConfirm = cb; + }, + }); + + expect(span.end).not.toHaveBeenCalled(); + + brokerConfirm(null); + + expect(span.end).toHaveBeenCalledTimes(1); + }); + }); + + describe('consumer span lifecycle', () => { + async function registerConsumer(channel: unknown, consumerTag: string, noAck: boolean): Promise { + await tracingChannel(CHANNELS.AMQPLIB_CONSUME).tracePromise(async () => ({ consumerTag }), { + arguments: ['task-queue', () => {}, { noAck }], + self: channel, + }); + } + + it('runs the consumer callback under the consumer span', async () => { + const channel = makeChannel(); + await registerConsumer(channel, 'ct-active', false); + const message = { fields: { exchange: '', routingKey: 'task-queue' }, properties: { headers: {} } }; + + let activeDuringCallback: Span | undefined; + tracingChannel(CHANNELS.AMQPLIB_DISPATCH).traceSync( + () => { + activeDuringCallback = SentryCore.getActiveSpan(); + }, + { arguments: [{ consumerTag: 'ct-active' }, message], self: channel }, + ); + + expect(activeDuringCallback).toBe(span); + }); + + it('noAck consumer: ends the span when dispatch returns, even if the callback is async', async () => { + const channel = makeChannel(); + await registerConsumer(channel, 'ct-async', true); + const message = { fields: { exchange: '', routingKey: 'task-queue' }, properties: { headers: {} } }; + + let resolveWork: () => void = () => {}; + const asyncWork = new Promise(resolve => { + resolveWork = resolve; + }); + + tracingChannel(CHANNELS.AMQPLIB_DISPATCH).traceSync(() => asyncWork, { + arguments: [{ consumerTag: 'ct-async' }, message], + self: channel, + }); + + // The span closes when the synchronous dispatch returns, not when the async work settles. + expect(span.end).toHaveBeenCalledTimes(1); + + resolveWork(); + await asyncWork; + }); + + it('manual-ack consumer: keeps the span open across the dispatch and ends it on ack', async () => { + const channel = makeChannel(); + await registerConsumer(channel, 'ct-manual', false); + const message = { fields: { exchange: '', routingKey: 'task-queue' }, properties: { headers: {} } }; + + tracingChannel(CHANNELS.AMQPLIB_DISPATCH).traceSync(() => undefined, { + arguments: [{ consumerTag: 'ct-manual' }, message], + self: channel, + }); + + expect(span.end).not.toHaveBeenCalled(); + + tracingChannel(CHANNELS.AMQPLIB_ACK).traceSync(() => undefined, { arguments: [message], self: channel }); + + expect(span.end).toHaveBeenCalledTimes(1); + }); + }); + + // The `connect` channel stashes connection attributes derived from the connect URL onto the + // connection object; the producer span then reads them via `channel.connection`. A silent + // regression here would drop everything except `messaging.system` (which has a live-object + // fallback), so these assert the full URL-derived attribute set flows onto the span. + describe('connection attributes', () => { + function connect(url: unknown, connection: unknown): void { + tracingChannel(CHANNELS.AMQPLIB_CONNECT).traceCallback( + function (_url: unknown, _opts: unknown, cb: (err: unknown, conn: unknown) => void) { + cb(null, connection); + }, + 2, + { arguments: [url] }, + undefined, + url, + {}, + () => {}, + ); + } + + it.each([ + { + name: 'string url with credentials (password censored)', + url: 'amqp://user:secret@rabbit.example.com:5672/vhost', + expected: { + 'messaging.url': 'amqp://user:***@rabbit.example.com:5672/vhost', + 'messaging.protocol': 'AMQP', + 'messaging.protocol_version': '0.9.1', + 'net.peer.name': 'rabbit.example.com', + 'net.peer.port': 5672, + }, + }, + { + name: 'amqps string url defaulting to port 5671', + url: 'amqps://rabbit.example.com/', + expected: { + 'messaging.protocol': 'AMQPS', + 'net.peer.name': 'rabbit.example.com', + 'net.peer.port': 5671, + }, + }, + { + name: 'object connect options', + url: { protocol: 'amqp', hostname: 'broker.internal', port: 5673 }, + expected: { + 'messaging.protocol': 'AMQP', + 'net.peer.name': 'broker.internal', + 'net.peer.port': 5673, + }, + }, + ])('carries $name onto the producer span', ({ url, expected }) => { + const connection = { serverProperties: { product: 'RabbitMQ' } }; + connect(url, connection); + + tracingChannel(CHANNELS.AMQPLIB_PUBLISH).traceSync(() => true, { + arguments: ['orders', 'routing-key', Buffer.from('payload'), {}], + self: { connection }, + }); + + expect(startInactiveSpanSpy).toHaveBeenCalledTimes(1); + const options = startInactiveSpanSpy.mock.calls[0]![0] as { attributes: Record }; + expect(options.attributes).toMatchObject({ ...expected, 'messaging.system': 'rabbitmq' }); + }); + }); }); From 5f886d404491aed1a6f2e02e048e21b579718f86 Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Wed, 8 Jul 2026 10:34:05 +0200 Subject: [PATCH 3/6] import attributes --- .../integrations/tracing-channel/amqplib.ts | 37 ++++++++++++++----- 1 file changed, 28 insertions(+), 9 deletions(-) diff --git a/packages/server-utils/src/integrations/tracing-channel/amqplib.ts b/packages/server-utils/src/integrations/tracing-channel/amqplib.ts index 08d2e158b9bb..2329bb8b7798 100644 --- a/packages/server-utils/src/integrations/tracing-channel/amqplib.ts +++ b/packages/server-utils/src/integrations/tracing-channel/amqplib.ts @@ -13,7 +13,14 @@ import { timestampInSeconds, waitForTracingChannelBinding, } from '@sentry/core'; -import { MESSAGING_SYSTEM } from '@sentry/conventions/attributes'; +// eslint-disable-next-line typescript/no-deprecated -- NET_PEER_* emitted alongside SERVER_* for backwards compatibility (TODO(v11): remove) +import { + MESSAGING_SYSTEM, + NET_PEER_NAME, + NET_PEER_PORT, + SERVER_ADDRESS, + SERVER_PORT, +} from '@sentry/conventions/attributes'; import { DEBUG_BUILD } from '../../debug-build'; import { CHANNELS } from '../../orchestrion/channels'; import { bindTracingChannelToSpan } from '../../tracing-channel'; @@ -38,10 +45,6 @@ const ATTR_MESSAGING_MESSAGE_ID = 'messaging.message_id'; const ATTR_MESSAGING_CONVERSATION_ID = 'messaging.conversation_id'; const MESSAGING_DESTINATION_KIND_VALUE_TOPIC = 'topic'; const MESSAGING_OPERATION_VALUE_PROCESS = 'process'; -// Inlined (rather than imported) because the `@sentry/conventions` exports are deprecated; the SDK -// has always emitted these legacy `net.*` attributes for amqplib connections. -const ATTR_NET_PEER_NAME = 'net.peer.name'; -const ATTR_NET_PEER_PORT = 'net.peer.port'; // To prevent reference leaks from un-acked messages, their spans are closed after this timeout. The // upstream instrumentation exposed this as the `consumeTimeoutMs` option; the SDK always used the default. @@ -526,18 +529,34 @@ function getConnectionAttributesFromUrl(url: unknown): SpanAttributes { if (typeof resolvedUrl === 'object') { const connectOptions = resolvedUrl as { protocol?: string; hostname?: string; port?: number }; const protocol = getProtocol(connectOptions.protocol); + const hostname = getHostname(connectOptions.hostname); + const port = getPort(connectOptions.port, protocol); + attributes[ATTR_MESSAGING_PROTOCOL] = protocol; - attributes[ATTR_NET_PEER_NAME] = getHostname(connectOptions.hostname); - attributes[ATTR_NET_PEER_PORT] = getPort(connectOptions.port, protocol); + attributes[SERVER_ADDRESS] = hostname; + attributes[SERVER_PORT] = port; + // TODO(v11): remove deprecated options + // eslint-disable-next-line typescript/no-deprecated -- emitted alongside SERVER_ADDRESS/SERVER_PORT for backwards compatibility + attributes[NET_PEER_NAME] = hostname; + // eslint-disable-next-line typescript/no-deprecated -- emitted alongside SERVER_ADDRESS/SERVER_PORT for backwards compatibility + attributes[NET_PEER_PORT] = port; } else if (typeof resolvedUrl === 'string') { const censoredUrl = censorPassword(resolvedUrl); attributes[ATTR_MESSAGING_URL] = censoredUrl; + try { const urlParts = new URL(censoredUrl); const protocol = getProtocol(urlParts.protocol); + const hostname = getHostname(urlParts.hostname); + const port = getPort(urlParts.port ? parseInt(urlParts.port, 10) : undefined, protocol); + attributes[ATTR_MESSAGING_PROTOCOL] = protocol; - attributes[ATTR_NET_PEER_NAME] = getHostname(urlParts.hostname); - attributes[ATTR_NET_PEER_PORT] = getPort(urlParts.port ? parseInt(urlParts.port, 10) : undefined, protocol); + attributes[SERVER_ADDRESS] = hostname; + attributes[SERVER_PORT] = port; + // eslint-disable-next-line typescript/no-deprecated -- emitted alongside SERVER_ADDRESS/SERVER_PORT for backwards compatibility + attributes[NET_PEER_NAME] = hostname; + // eslint-disable-next-line typescript/no-deprecated -- emitted alongside SERVER_ADDRESS/SERVER_PORT for backwards compatibility + attributes[NET_PEER_PORT] = port; } catch { // best-effort: a malformed url simply yields fewer connection attributes } From 1909a2d7395e526e08fc1902b701631ea2dd731e Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Wed, 8 Jul 2026 11:40:31 +0200 Subject: [PATCH 4/6] add legacy alongside new attributes --- .../suites/tracing/amqplib/test.ts | 3 + .../integrations/tracing-channel/amqplib.ts | 62 ++++++++++++++----- .../test/orchestrion/amqplib.test.ts | 31 +++++++++- 3 files changed, 79 insertions(+), 17 deletions(-) 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 fc8799026c6a..bb9c34ed38f2 100644 --- a/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts @@ -22,6 +22,9 @@ const expectedProducerSpan = (routingKey: string) => 'messaging.protocol_version': '0.9.1', 'net.peer.name': 'localhost', 'net.peer.port': 5672, + // Only the orchestrion integration emits the current `messaging.operation.type`; the OTel + // instrumentation never set it on the producer. + ...(isOrchestrionEnabled() ? { 'messaging.operation.type': 'send' } : {}), 'otel.kind': 'PRODUCER', 'sentry.op': 'message', 'sentry.origin': PUBLISHER_ORIGIN, diff --git a/packages/server-utils/src/integrations/tracing-channel/amqplib.ts b/packages/server-utils/src/integrations/tracing-channel/amqplib.ts index 2329bb8b7798..2b34d47da2c7 100644 --- a/packages/server-utils/src/integrations/tracing-channel/amqplib.ts +++ b/packages/server-utils/src/integrations/tracing-channel/amqplib.ts @@ -16,10 +16,16 @@ import { // eslint-disable-next-line typescript/no-deprecated -- NET_PEER_* emitted alongside SERVER_* for backwards compatibility (TODO(v11): remove) import { MESSAGING_SYSTEM, + MESSAGING_MESSAGE_ID, + MESSAGING_OPERATION_TYPE, + MESSAGING_DESTINATION_NAME, NET_PEER_NAME, NET_PEER_PORT, + NETWORK_PROTOCOL_NAME, + NETWORK_PROTOCOL_VERSION, SERVER_ADDRESS, SERVER_PORT, + URL_FULL, } from '@sentry/conventions/attributes'; import { DEBUG_BUILD } from '../../debug-build'; import { CHANNELS } from '../../orchestrion/channels'; @@ -33,18 +39,26 @@ const PUBLISHER_ORIGIN = 'auto.amqplib.orchestrion.publisher'; const CONSUMER_ORIGIN = 'auto.amqplib.orchestrion.consumer'; // Legacy messaging semantic-conventions, inlined to keep this integration free of `@opentelemetry/*` -// deps. These mirror what the vendored OTel amqplib instrumentation has always emitted. +// deps. These mirror what the vendored OTel amqplib instrumentation has always emitted. We keep +// emitting them alongside the current `@sentry/conventions` attributes for backwards compatibility. +// TODO(v11): remove these legacy attributes. const ATTR_MESSAGING_OPERATION = 'messaging.operation'; const ATTR_MESSAGING_DESTINATION = 'messaging.destination'; const ATTR_MESSAGING_DESTINATION_KIND = 'messaging.destination_kind'; const ATTR_MESSAGING_RABBITMQ_ROUTING_KEY = 'messaging.rabbitmq.routing_key'; const ATTR_MESSAGING_PROTOCOL = 'messaging.protocol'; -const ATTR_MESSAGING_PROTOCOL_VERSION = 'messaging.protocol_version'; +const ATTR_MESSAGING_PROTOCOL_VERSION_LEGACY = 'messaging.protocol_version'; const ATTR_MESSAGING_URL = 'messaging.url'; const ATTR_MESSAGING_MESSAGE_ID = 'messaging.message_id'; -const ATTR_MESSAGING_CONVERSATION_ID = 'messaging.conversation_id'; +const ATTR_MESSAGING_CONVERSATION_ID_LEGACY = 'messaging.conversation_id'; + +// TODO(v11): replace with the corresponding attribute from `@sentry/conventions` once it is added there. +const ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY = 'messaging.rabbitmq.destination.routing_key'; +const ATTR_MESSAGING_CONVERSATION_ID = 'messaging.message.conversation_id'; + const MESSAGING_DESTINATION_KIND_VALUE_TOPIC = 'topic'; const MESSAGING_OPERATION_VALUE_PROCESS = 'process'; +const MESSAGING_OPERATION_VALUE_SEND = 'send'; // To prevent reference leaks from un-acked messages, their spans are closed after this timeout. The // upstream instrumentation exposed this as the `consumeTimeoutMs` option; the SDK always used the default. @@ -449,10 +463,15 @@ function startPublishSpan(data: AmqpChannelContext): Span { kind: SPAN_KIND.PRODUCER, attributes: { ...getStoredConnectionAttributes(data.self), - [ATTR_MESSAGING_DESTINATION]: exchange, - [ATTR_MESSAGING_DESTINATION_KIND]: MESSAGING_DESTINATION_KIND_VALUE_TOPIC, - [ATTR_MESSAGING_RABBITMQ_ROUTING_KEY]: routingKey, - [ATTR_MESSAGING_MESSAGE_ID]: options?.messageId as string | undefined, + [ATTR_MESSAGING_DESTINATION]: exchange, // TODO(v11) remove this attribute + [MESSAGING_DESTINATION_NAME]: exchange, + [ATTR_MESSAGING_DESTINATION_KIND]: MESSAGING_DESTINATION_KIND_VALUE_TOPIC, // TODO(v11) remove this attribute + [ATTR_MESSAGING_RABBITMQ_ROUTING_KEY]: routingKey, // TODO(v11) remove this attribute + [ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY]: routingKey, + [MESSAGING_OPERATION_TYPE]: MESSAGING_OPERATION_VALUE_SEND, + [ATTR_MESSAGING_MESSAGE_ID]: options?.messageId as string | undefined, // todo(v11) remove this attribute + [MESSAGING_MESSAGE_ID]: options?.messageId as string | undefined, + [ATTR_MESSAGING_CONVERSATION_ID_LEGACY]: options?.correlationId as string | undefined, // todo(v11) remove this attribute [ATTR_MESSAGING_CONVERSATION_ID]: options?.correlationId as string | undefined, [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: PUBLISHER_ORIGIN, }, @@ -482,11 +501,16 @@ function startConsumeSpan(queue: string, msg: ConsumeMessage, channel: ChannelLi kind: SPAN_KIND.CONSUMER, attributes: { ...getStoredConnectionAttributes(channel), - [ATTR_MESSAGING_DESTINATION]: msg.fields?.exchange, - [ATTR_MESSAGING_DESTINATION_KIND]: MESSAGING_DESTINATION_KIND_VALUE_TOPIC, - [ATTR_MESSAGING_RABBITMQ_ROUTING_KEY]: msg.fields?.routingKey, - [ATTR_MESSAGING_OPERATION]: MESSAGING_OPERATION_VALUE_PROCESS, - [ATTR_MESSAGING_MESSAGE_ID]: msg.properties?.messageId as string | undefined, + [ATTR_MESSAGING_DESTINATION]: msg.fields?.exchange, // TODO(v11) remove this attribute + [MESSAGING_DESTINATION_NAME]: msg.fields?.exchange, + [ATTR_MESSAGING_DESTINATION_KIND]: MESSAGING_DESTINATION_KIND_VALUE_TOPIC, // TODO(v11) remove this attribute + [ATTR_MESSAGING_RABBITMQ_ROUTING_KEY]: msg.fields?.routingKey, // TODO(v11) remove this attribute + [ATTR_MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY]: msg.fields?.routingKey, + [ATTR_MESSAGING_OPERATION]: MESSAGING_OPERATION_VALUE_PROCESS, // TODO(v11) remove this attribute + [MESSAGING_OPERATION_TYPE]: MESSAGING_OPERATION_VALUE_PROCESS, + [ATTR_MESSAGING_MESSAGE_ID]: msg.properties?.messageId as string | undefined, // todo(v11) remove this attribute + [MESSAGING_MESSAGE_ID]: msg.properties?.messageId as string | undefined, + [ATTR_MESSAGING_CONVERSATION_ID_LEGACY]: msg.properties?.correlationId as string | undefined, // todo(v11) remove this attribute [ATTR_MESSAGING_CONVERSATION_ID]: msg.properties?.correlationId as string | undefined, [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: CONSUMER_ORIGIN, }, @@ -522,7 +546,8 @@ function getConnectionAttributesFromServer(conn: ConnectionLike): SpanAttributes function getConnectionAttributesFromUrl(url: unknown): SpanAttributes { const attributes: SpanAttributes = { // The only protocol supported by the instrumented library. - [ATTR_MESSAGING_PROTOCOL_VERSION]: '0.9.1', + [ATTR_MESSAGING_PROTOCOL_VERSION_LEGACY]: '0.9.1', // TODO(v11): remove this attribute + [NETWORK_PROTOCOL_VERSION]: '0.9.1', }; const resolvedUrl = url || 'amqp://localhost'; @@ -532,7 +557,9 @@ function getConnectionAttributesFromUrl(url: unknown): SpanAttributes { const hostname = getHostname(connectOptions.hostname); const port = getPort(connectOptions.port, protocol); - attributes[ATTR_MESSAGING_PROTOCOL] = protocol; + attributes[ATTR_MESSAGING_PROTOCOL] = protocol; // TODO(v11) remove this attribute + attributes[NETWORK_PROTOCOL_NAME] = protocol; + attributes[SERVER_ADDRESS] = hostname; attributes[SERVER_PORT] = port; // TODO(v11): remove deprecated options @@ -542,7 +569,8 @@ function getConnectionAttributesFromUrl(url: unknown): SpanAttributes { attributes[NET_PEER_PORT] = port; } else if (typeof resolvedUrl === 'string') { const censoredUrl = censorPassword(resolvedUrl); - attributes[ATTR_MESSAGING_URL] = censoredUrl; + attributes[ATTR_MESSAGING_URL] = censoredUrl; // todo(v11) remove this attribute + attributes[URL_FULL] = censoredUrl; try { const urlParts = new URL(censoredUrl); @@ -550,7 +578,9 @@ function getConnectionAttributesFromUrl(url: unknown): SpanAttributes { const hostname = getHostname(urlParts.hostname); const port = getPort(urlParts.port ? parseInt(urlParts.port, 10) : undefined, protocol); - attributes[ATTR_MESSAGING_PROTOCOL] = protocol; + attributes[ATTR_MESSAGING_PROTOCOL] = protocol; // TODO(v11) remove this attribute + attributes[NETWORK_PROTOCOL_NAME] = protocol; + attributes[SERVER_ADDRESS] = hostname; attributes[SERVER_PORT] = port; // eslint-disable-next-line typescript/no-deprecated -- emitted alongside SERVER_ADDRESS/SERVER_PORT for backwards compatibility diff --git a/packages/server-utils/test/orchestrion/amqplib.test.ts b/packages/server-utils/test/orchestrion/amqplib.test.ts index bfc8d47a228e..116bf290fe79 100644 --- a/packages/server-utils/test/orchestrion/amqplib.test.ts +++ b/packages/server-utils/test/orchestrion/amqplib.test.ts @@ -109,8 +109,14 @@ describe('amqplibChannelIntegration', () => { op: 'message', attributes: expect.objectContaining({ 'messaging.system': 'rabbitmq', + // Legacy attributes (TODO(v11): remove) emitted alongside the current ones. 'messaging.destination': 'my-exchange', + 'messaging.destination_kind': 'topic', 'messaging.rabbitmq.routing_key': 'my-routing-key', + // Current `@sentry/conventions` attributes. + 'messaging.destination.name': 'my-exchange', + 'messaging.rabbitmq.destination.routing_key': 'my-routing-key', + 'messaging.operation.type': 'send', 'messaging.message_id': 'm1', 'messaging.conversation_id': 'c1', 'sentry.origin': 'auto.amqplib.orchestrion.publisher', @@ -165,7 +171,15 @@ describe('amqplibChannelIntegration', () => { op: 'message', attributes: expect.objectContaining({ 'messaging.system': 'rabbitmq', + // Legacy attributes (TODO(v11): remove) emitted alongside the current ones. + 'messaging.destination': '', + 'messaging.destination_kind': 'topic', + 'messaging.rabbitmq.routing_key': 'queue1', 'messaging.operation': 'process', + // Current `@sentry/conventions` attributes. + 'messaging.destination.name': '', + 'messaging.rabbitmq.destination.routing_key': 'queue1', + 'messaging.operation.type': 'process', 'sentry.origin': 'auto.amqplib.orchestrion.consumer', }), }), @@ -401,28 +415,43 @@ describe('amqplibChannelIntegration', () => { url: 'amqp://user:secret@rabbit.example.com:5672/vhost', expected: { 'messaging.url': 'amqp://user:***@rabbit.example.com:5672/vhost', - 'messaging.protocol': 'AMQP', 'messaging.protocol_version': '0.9.1', + // Legacy attributes (TODO(v11): remove) emitted alongside the current ones. + 'messaging.protocol': 'AMQP', 'net.peer.name': 'rabbit.example.com', 'net.peer.port': 5672, + // Current `@sentry/conventions` attributes. + 'network.protocol.name': 'AMQP', + 'server.address': 'rabbit.example.com', + 'server.port': 5672, }, }, { name: 'amqps string url defaulting to port 5671', url: 'amqps://rabbit.example.com/', expected: { + // Legacy attributes (TODO(v11): remove) emitted alongside the current ones. 'messaging.protocol': 'AMQPS', 'net.peer.name': 'rabbit.example.com', 'net.peer.port': 5671, + // Current `@sentry/conventions` attributes. + 'network.protocol.name': 'AMQPS', + 'server.address': 'rabbit.example.com', + 'server.port': 5671, }, }, { name: 'object connect options', url: { protocol: 'amqp', hostname: 'broker.internal', port: 5673 }, expected: { + // Legacy attributes (TODO(v11): remove) emitted alongside the current ones. 'messaging.protocol': 'AMQP', 'net.peer.name': 'broker.internal', 'net.peer.port': 5673, + // Current `@sentry/conventions` attributes. + 'network.protocol.name': 'AMQP', + 'server.address': 'broker.internal', + 'server.port': 5673, }, }, ])('carries $name onto the producer span', ({ url, expected }) => { From 19c41106ae19e5c1106051c7213be0723d4fbc32 Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Wed, 8 Jul 2026 12:01:20 +0200 Subject: [PATCH 5/6] add `clearConsumeTimeoutTimer` --- .../integrations/tracing-channel/amqplib.ts | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/packages/server-utils/src/integrations/tracing-channel/amqplib.ts b/packages/server-utils/src/integrations/tracing-channel/amqplib.ts index 2b34d47da2c7..100425795b56 100644 --- a/packages/server-utils/src/integrations/tracing-channel/amqplib.ts +++ b/packages/server-utils/src/integrations/tracing-channel/amqplib.ts @@ -381,21 +381,29 @@ function ensureChannelState(channel: ChannelLike): void { channel[CHANNEL_CONSUME_TIMEOUT_TIMER] = timer; // End outstanding spans and stop the timer when the channel goes away (replaces patching `emit`). + // amqplib emits 'close' after 'error', but we clear in both to avoid leaking the interval (which + // pins the channel via its closure) should a version or edge case ever skip the trailing 'close'. if (typeof channel.on === 'function') { channel.on('close', () => { endAllSpansOnChannel(channel, true, END_OP.ChannelClosed, undefined); - const activeTimer = channel[CHANNEL_CONSUME_TIMEOUT_TIMER]; - if (activeTimer) { - clearInterval(activeTimer); - } - channel[CHANNEL_CONSUME_TIMEOUT_TIMER] = undefined; + clearConsumeTimeoutTimer(channel); }); channel.on('error', () => { endAllSpansOnChannel(channel, true, END_OP.ChannelError, undefined); + clearConsumeTimeoutTimer(channel); }); } } +/** Stops and clears the per-channel consume-timeout interval. Idempotent. */ +function clearConsumeTimeoutTimer(channel: ChannelLike): void { + const activeTimer = channel[CHANNEL_CONSUME_TIMEOUT_TIMER]; + if (activeTimer) { + clearInterval(activeTimer); + channel[CHANNEL_CONSUME_TIMEOUT_TIMER] = undefined; + } +} + function checkConsumeTimeoutOnChannel(channel: ChannelLike): void { const currentTime = timestampInSeconds(); const spansNotEnded = channel[CHANNEL_SPANS_NOT_ENDED] ?? []; From d2f8cd70ce07d29765341abfb5ec4be7e2eeb098 Mon Sep 17 00:00:00 2001 From: s1gr1d <32902192+s1gr1d@users.noreply.github.com> Date: Wed, 8 Jul 2026 12:03:59 +0200 Subject: [PATCH 6/6] fix snapshot --- packages/deno/test/__snapshots__/mod.test.ts.snap | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/packages/deno/test/__snapshots__/mod.test.ts.snap b/packages/deno/test/__snapshots__/mod.test.ts.snap index 92173dfec650..3b344fa68eff 100644 --- a/packages/deno/test/__snapshots__/mod.test.ts.snap +++ b/packages/deno/test/__snapshots__/mod.test.ts.snap @@ -117,6 +117,7 @@ snapshot[`captureException 1`] = ` "DenoRedis", "DenoMysql", "DenoPostgres", + "DenoAmqplib", "ContextLines", "NormalizePaths", "GlobalHandlers", @@ -194,6 +195,7 @@ snapshot[`captureMessage 1`] = ` "DenoRedis", "DenoMysql", "DenoPostgres", + "DenoAmqplib", "ContextLines", "NormalizePaths", "GlobalHandlers", @@ -278,6 +280,7 @@ snapshot[`captureMessage twice 1`] = ` "DenoRedis", "DenoMysql", "DenoPostgres", + "DenoAmqplib", "ContextLines", "NormalizePaths", "GlobalHandlers", @@ -369,6 +372,7 @@ snapshot[`captureMessage twice 2`] = ` "DenoRedis", "DenoMysql", "DenoPostgres", + "DenoAmqplib", "ContextLines", "NormalizePaths", "GlobalHandlers",