From f4b54a572c7ef5d7bce2997047ca71ec01f6a436 Mon Sep 17 00:00:00 2001 From: Andrei Borza Date: Tue, 28 Jul 2026 12:14:05 +0200 Subject: [PATCH] ref(node): Remove vendored amqplib instrumentation This instrumentation is dead code: the amqplib integration is provided by the channel-based implementation in `@sentry/server-utils` (default since #22501), and the vendored OpenTelemetry instrumentation here is no longer wired up. Ref: #22346 Co-Authored-By: Claude Opus 4.8 --- .../src/integrations/tracing/amqplib/index.ts | 6 - .../tracing/amqplib/vendored/amqplib-types.ts | 47 --- .../amqplib/vendored/instrumentation.ts | 143 --------- .../tracing/amqplib/vendored/patches.ts | 299 ------------------ .../tracing/amqplib/vendored/semconv.ts | 45 --- .../tracing/amqplib/vendored/types.ts | 68 ---- .../tracing/amqplib/vendored/utils.ts | 206 ------------ 7 files changed, 814 deletions(-) delete mode 100644 packages/node/src/integrations/tracing/amqplib/index.ts delete mode 100644 packages/node/src/integrations/tracing/amqplib/vendored/amqplib-types.ts delete mode 100644 packages/node/src/integrations/tracing/amqplib/vendored/instrumentation.ts delete mode 100644 packages/node/src/integrations/tracing/amqplib/vendored/patches.ts delete mode 100644 packages/node/src/integrations/tracing/amqplib/vendored/semconv.ts delete mode 100644 packages/node/src/integrations/tracing/amqplib/vendored/types.ts delete mode 100644 packages/node/src/integrations/tracing/amqplib/vendored/utils.ts diff --git a/packages/node/src/integrations/tracing/amqplib/index.ts b/packages/node/src/integrations/tracing/amqplib/index.ts deleted file mode 100644 index 5b6181291c71..000000000000 --- a/packages/node/src/integrations/tracing/amqplib/index.ts +++ /dev/null @@ -1,6 +0,0 @@ -import { generateInstrumentOnce } from '../../../otel/instrument'; -import { AmqplibInstrumentation } from './vendored/instrumentation'; - -const INTEGRATION_NAME = 'Amqplib' as const; - -export const instrumentAmqplib = generateInstrumentOnce(INTEGRATION_NAME, () => new AmqplibInstrumentation()); diff --git a/packages/node/src/integrations/tracing/amqplib/vendored/amqplib-types.ts b/packages/node/src/integrations/tracing/amqplib/vendored/amqplib-types.ts deleted file mode 100644 index 71a03a7e998d..000000000000 --- a/packages/node/src/integrations/tracing/amqplib/vendored/amqplib-types.ts +++ /dev/null @@ -1,47 +0,0 @@ -/* - * Simplified types inlined from @types/amqplib (DefinitelyTyped). - * Only includes members accessed by this instrumentation. - * Other amqplib types (Message, ConsumeMessage, Options.Publish, etc.) are already - * vendored in types.ts by the upstream OTel instrumentation. - */ - -export interface Connection { - connection: { serverProperties: { product?: string; [key: string]: any } }; - [key: string]: any; -} - -export interface Channel { - connection: Connection; - [key: string]: any; -} - -export interface ConfirmChannel extends Channel {} - -export namespace Options { - export interface Connect { - protocol?: string; - hostname?: string; - port?: number; - username?: string; - vhost?: string; - } - export interface Consume { - consumerTag?: string; - noLocal?: boolean; - noAck?: boolean; - exclusive?: boolean; - priority?: number; - arguments?: any; - } - export interface Publish { - headers?: any; - [key: string]: any; - } -} - -export namespace Replies { - export interface Empty {} - export interface Consume { - consumerTag: string; - } -} diff --git a/packages/node/src/integrations/tracing/amqplib/vendored/instrumentation.ts b/packages/node/src/integrations/tracing/amqplib/vendored/instrumentation.ts deleted file mode 100644 index e4a7d630e2f5..000000000000 --- a/packages/node/src/integrations/tracing/amqplib/vendored/instrumentation.ts +++ /dev/null @@ -1,143 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - * - * NOTICE from the Sentry authors: - * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-amqplib - * - Upstream version: @opentelemetry/instrumentation-amqplib@0.65.0 - * - Refactored to use Sentry's span APIs instead of OpenTelemetry tracing APIs; the channel/connection - * patches live in `patches.ts` and span creation in `utils.ts` - * - Dropped the instrumentation config and all hooks; origin is folded into span creation instead of - * `index.ts` hooks (see `patches.ts`/`utils.ts` for details) - */ - -import type { InstrumentationConfig } from '@opentelemetry/instrumentation'; -import { InstrumentationBase, InstrumentationNodeModuleDefinition, isWrapped } from '@opentelemetry/instrumentation'; -import { SDK_VERSION } from '@sentry/core'; -import { InstrumentationNodeModuleFile } from '../../InstrumentationNodeModuleFile'; -import { - getAckAllPatch, - getAckPatch, - getChannelEmitPatch, - getConfirmedPublishPatch, - getConnectPatch, - getConsumePatch, - getPublishPatch, -} from './patches'; -import { EndOperation } from './types'; - -const PACKAGE_NAME = '@sentry/instrumentation-amqplib'; -const supportedVersions = ['>=0.5.5 <2']; - -export class AmqplibInstrumentation extends InstrumentationBase { - public constructor(config: InstrumentationConfig = {}) { - super(PACKAGE_NAME, SDK_VERSION, config); - } - - protected init(): InstrumentationNodeModuleDefinition { - const channelModelModuleFile = new InstrumentationNodeModuleFile( - 'amqplib/lib/channel_model.js', - supportedVersions, - this.patchChannelModel.bind(this), - this.unpatchChannelModel.bind(this), - ); - - const callbackModelModuleFile = new InstrumentationNodeModuleFile( - 'amqplib/lib/callback_model.js', - supportedVersions, - this.patchChannelModel.bind(this), - this.unpatchChannelModel.bind(this), - ); - - const connectModuleFile = new InstrumentationNodeModuleFile( - 'amqplib/lib/connect.js', - supportedVersions, - this.patchConnect.bind(this), - this.unpatchConnect.bind(this), - ); - - const module = new InstrumentationNodeModuleDefinition('amqplib', supportedVersions, undefined, undefined, [ - channelModelModuleFile, - connectModuleFile, - callbackModelModuleFile, - ]); - return module; - } - - private patchConnect(moduleExports: any): any { - const unpatchedExports = this.unpatchConnect(moduleExports); - if (!isWrapped(unpatchedExports.connect)) { - this._wrap(unpatchedExports, 'connect', getConnectPatch); - } - return unpatchedExports; - } - - private unpatchConnect(moduleExports: any): any { - if (isWrapped(moduleExports.connect)) { - this._unwrap(moduleExports, 'connect'); - } - return moduleExports; - } - - private patchChannelModel(moduleExports: any): any { - if (!isWrapped(moduleExports.Channel.prototype.publish)) { - this._wrap(moduleExports.Channel.prototype, 'publish', getPublishPatch); - } - if (!isWrapped(moduleExports.Channel.prototype.consume)) { - this._wrap(moduleExports.Channel.prototype, 'consume', getConsumePatch); - } - if (!isWrapped(moduleExports.Channel.prototype.ack)) { - this._wrap(moduleExports.Channel.prototype, 'ack', getAckPatch(false, EndOperation.Ack)); - } - if (!isWrapped(moduleExports.Channel.prototype.nack)) { - this._wrap(moduleExports.Channel.prototype, 'nack', getAckPatch(true, EndOperation.Nack)); - } - if (!isWrapped(moduleExports.Channel.prototype.reject)) { - this._wrap(moduleExports.Channel.prototype, 'reject', getAckPatch(true, EndOperation.Reject)); - } - if (!isWrapped(moduleExports.Channel.prototype.ackAll)) { - this._wrap(moduleExports.Channel.prototype, 'ackAll', getAckAllPatch(false, EndOperation.AckAll)); - } - if (!isWrapped(moduleExports.Channel.prototype.nackAll)) { - this._wrap(moduleExports.Channel.prototype, 'nackAll', getAckAllPatch(true, EndOperation.NackAll)); - } - if (!isWrapped(moduleExports.Channel.prototype.emit)) { - this._wrap(moduleExports.Channel.prototype, 'emit', getChannelEmitPatch); - } - if (!isWrapped(moduleExports.ConfirmChannel.prototype.publish)) { - this._wrap(moduleExports.ConfirmChannel.prototype, 'publish', getConfirmedPublishPatch); - } - return moduleExports; - } - - private unpatchChannelModel(moduleExports: any): any { - if (isWrapped(moduleExports.Channel.prototype.publish)) { - this._unwrap(moduleExports.Channel.prototype, 'publish'); - } - if (isWrapped(moduleExports.Channel.prototype.consume)) { - this._unwrap(moduleExports.Channel.prototype, 'consume'); - } - if (isWrapped(moduleExports.Channel.prototype.ack)) { - this._unwrap(moduleExports.Channel.prototype, 'ack'); - } - if (isWrapped(moduleExports.Channel.prototype.nack)) { - this._unwrap(moduleExports.Channel.prototype, 'nack'); - } - if (isWrapped(moduleExports.Channel.prototype.reject)) { - this._unwrap(moduleExports.Channel.prototype, 'reject'); - } - if (isWrapped(moduleExports.Channel.prototype.ackAll)) { - this._unwrap(moduleExports.Channel.prototype, 'ackAll'); - } - if (isWrapped(moduleExports.Channel.prototype.nackAll)) { - this._unwrap(moduleExports.Channel.prototype, 'nackAll'); - } - if (isWrapped(moduleExports.Channel.prototype.emit)) { - this._unwrap(moduleExports.Channel.prototype, 'emit'); - } - if (isWrapped(moduleExports.ConfirmChannel.prototype.publish)) { - this._unwrap(moduleExports.ConfirmChannel.prototype, 'publish'); - } - return moduleExports; - } -} diff --git a/packages/node/src/integrations/tracing/amqplib/vendored/patches.ts b/packages/node/src/integrations/tracing/amqplib/vendored/patches.ts deleted file mode 100644 index cf1233f47bf7..000000000000 --- a/packages/node/src/integrations/tracing/amqplib/vendored/patches.ts +++ /dev/null @@ -1,299 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - * - * NOTICE from the Sentry authors: - * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-amqplib - * - Upstream version: @opentelemetry/instrumentation-amqplib@0.65.0 - * - The channel/connection patches were extracted from the instrumentation class into standalone factories - * and migrated to Sentry's span APIs; origin is folded into span creation instead of `index.ts` hooks - * - Cross-service trace propagation uses Sentry's `getTraceData`/`continueTrace` instead of the OTel propagator - * - Replaced the OTel context-key confirm-channel marker with a synchronous flag on the channel instance - * - Dropped the instrumentation config and all hooks (publish/publishConfirm/consume/consumeEnd) and the - * `useLinksForConsume` path; the SDK never used them - */ - -import { continueTrace, SPAN_STATUS_ERROR, timestampInSeconds, withActiveSpan } from '@sentry/core'; -import type { Connection, Options, Replies } from './amqplib-types'; -import { EndOperation, type ConsumeMessage, type Message } from './types'; -import type { - InstrumentationConnection, - InstrumentationConsumeChannel, - InstrumentationConsumeMessage, - InstrumentationMessage, - InstrumentationPublishChannel, -} from './utils'; -import { - CHANNEL_CONSUME_TIMEOUT_TIMER, - CHANNEL_IS_CONFIRM_PUBLISHING, - CHANNEL_SPANS_NOT_ENDED, - CONNECTION_ATTRIBUTES, - getConnectionAttributesFromServer, - getConnectionAttributesFromUrl, - getHeaderAsString, - MESSAGE_STORED_SPAN, - startConsumeSpan, - startPublishSpan, -} from './utils'; - -// 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 - -function endConsumerSpan( - message: InstrumentationMessage, - isRejected: boolean | null, - operation: EndOperation, - requeue: boolean | undefined, -): void { - const storedSpan = message[MESSAGE_STORED_SPAN]; - if (!storedSpan) { - return; - } - if (isRejected !== false) { - storedSpan.setStatus({ - code: SPAN_STATUS_ERROR, - message: - operation !== EndOperation.ChannelClosed && operation !== EndOperation.ChannelError - ? `${operation} called on message${ - requeue === true ? ' with requeue' : requeue === false ? ' without requeue' : '' - }` - : operation, - }); - } - storedSpan.end(); - message[MESSAGE_STORED_SPAN] = undefined; -} - -function endAllSpansOnChannel( - channel: InstrumentationConsumeChannel, - isRejected: boolean, - operation: EndOperation, - requeue: boolean | undefined, -): void { - const spansNotEnded: { msg: Message }[] = channel[CHANNEL_SPANS_NOT_ENDED] ?? []; - spansNotEnded.forEach(msgDetails => { - endConsumerSpan(msgDetails.msg, isRejected, operation, requeue); - }); - channel[CHANNEL_SPANS_NOT_ENDED] = []; -} - -function checkConsumeTimeoutOnChannel(channel: InstrumentationConsumeChannel): 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, EndOperation.InstrumentationTimeout, true); - } - spansNotEnded.splice(0, i); -} - -export function getConnectPatch( - original: ( - url: string | Options.Connect, - socketOptions: any, - openCallback: (err: any, connection: Connection) => void, - ) => Connection, -) { - return function patchedConnect( - this: unknown, - url: string | Options.Connect, - socketOptions: any, - openCallback: Function, - ): Connection { - return original.call(this, url, socketOptions, function (this: unknown, err: any, conn: InstrumentationConnection) { - if (err == null) { - const urlAttributes = getConnectionAttributesFromUrl(url); - const serverAttributes = getConnectionAttributesFromServer(conn); - conn[CONNECTION_ATTRIBUTES] = { - ...urlAttributes, - ...serverAttributes, - }; - } - openCallback.apply(this, arguments); - }); - }; -} - -export function getChannelEmitPatch(original: Function) { - return function emit(this: InstrumentationConsumeChannel, eventName: string): void { - if (eventName === 'close') { - endAllSpansOnChannel(this, true, EndOperation.ChannelClosed, undefined); - const activeTimer = this[CHANNEL_CONSUME_TIMEOUT_TIMER]; - if (activeTimer) { - clearInterval(activeTimer); - } - this[CHANNEL_CONSUME_TIMEOUT_TIMER] = undefined; - } else if (eventName === 'error') { - endAllSpansOnChannel(this, true, EndOperation.ChannelError, undefined); - } - return original.apply(this, arguments); - }; -} - -export function getAckAllPatch(isRejected: boolean, endOperation: EndOperation) { - return (original: Function) => - function ackAll(this: InstrumentationConsumeChannel, requeueOrEmpty?: boolean): void { - endAllSpansOnChannel(this, isRejected, endOperation, requeueOrEmpty); - return original.apply(this, arguments); - }; -} - -export function getAckPatch(isRejected: boolean, endOperation: EndOperation) { - return (original: Function) => - function ack( - this: InstrumentationConsumeChannel, - message: Message, - allUpToOrRequeue?: boolean, - requeue?: boolean, - ): void { - const channel = this; - // we use this patch in the reject function as well, but it has a different signature - const requeueResolved = endOperation === EndOperation.Reject ? allUpToOrRequeue : requeue; - - const spansNotEnded: { msg: Message }[] = channel[CHANNEL_SPANS_NOT_ENDED] ?? []; - const msgIndex = spansNotEnded.findIndex(msgDetails => msgDetails.msg === message); - if (msgIndex < 0) { - // should not happen in the happy flow, but possible if the user calls ack twice with the same message - endConsumerSpan(message, isRejected, endOperation, requeueResolved); - } else if (endOperation !== EndOperation.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); - } - return original.apply(this, arguments); - }; -} - -export function getConsumePatch(original: Function) { - return function consume( - this: InstrumentationConsumeChannel, - queue: string, - onMessage: (msg: ConsumeMessage | null) => void, - options?: Options.Consume, - ): Promise { - const channel = this; - if (!Object.prototype.hasOwnProperty.call(channel, CHANNEL_SPANS_NOT_ENDED)) { - const timer = setInterval(() => { - checkConsumeTimeoutOnChannel(channel); - }, CONSUME_TIMEOUT_MS); - timer.unref(); - channel[CHANNEL_CONSUME_TIMEOUT_TIMER] = timer; - channel[CHANNEL_SPANS_NOT_ENDED] = []; - } - - const patchedOnMessage = function (this: unknown, msg: InstrumentationConsumeMessage | null): void { - // msg is expected to be null for a consumer cancel notification - // https://www.rabbitmq.com/consumer-cancel.html - // in this case, we do not start a span, as this is not a real message. - if (!msg) { - return onMessage.call(this, msg); - } - - 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 is parented to the message's producer. - continueTrace({ sentryTrace, baggage }, () => { - const span = startConsumeSpan(queue, msg, channel); - - if (!options?.noAck) { - // store the message on the channel so we can close the span on ackAll etc - channel[CHANNEL_SPANS_NOT_ENDED]!.push({ msg, timeOfConsume: timestampInSeconds() }); - // store the span on the message so we can end it when the user calls 'ack' on it - msg[MESSAGE_STORED_SPAN] = span; - } - withActiveSpan(span, () => { - onMessage.call(this, msg); - }); - - if (options?.noAck) { - span.end(); - } - }); - }; - - // Copy `arguments` instead of mutating it: in CJS builds it's aliased to the named parameters, - // so `arguments[1] = ...` would also reassign `onMessage` to `patchedOnMessage`, making the wrapper - // call itself and recurse infinitely. - const callArgs = Array.prototype.slice.call(arguments); - callArgs[1] = patchedOnMessage; - return original.apply(this, callArgs); - }; -} - -export function getConfirmedPublishPatch(original: Function) { - return function confirmedPublish( - this: InstrumentationPublishChannel, - exchange: string, - routingKey: string, - content: Buffer, - options?: Options.Publish, - callback?: (err: any, ok: Replies.Empty) => void, - ): boolean { - const channel = this; - const { span, modifiedOptions } = startPublishSpan(exchange, routingKey, channel, options); - - const patchedOnConfirm = function (this: unknown, err: any, ok: Replies.Empty): void { - try { - withActiveSpan(span, () => { - callback?.call(this, err, ok); - }); - } finally { - if (err) { - span.setStatus({ code: SPAN_STATUS_ERROR, message: "message confirmation has been nack'ed" }); - } - span.end(); - } - }; - - // The confirm channel publish stores the message and registers a broker-confirm callback; the span - // ends in that callback. The confirm publish internally delegates to the base channel publish, so we - // flag the channel to stop the base publish patch from creating a second span. The inner call is - // synchronous, so a flag on the instance is enough and avoids the OTel context machinery. - const argumentsCopy = [...arguments]; - argumentsCopy[3] = modifiedOptions; - argumentsCopy[4] = patchedOnConfirm; - channel[CHANNEL_IS_CONFIRM_PUBLISHING] = true; - try { - return original.apply(this, argumentsCopy); - } finally { - channel[CHANNEL_IS_CONFIRM_PUBLISHING] = false; - } - }; -} - -export function getPublishPatch(original: Function) { - return function publish( - this: InstrumentationPublishChannel, - exchange: string, - routingKey: string, - content: Buffer, - options?: Options.Publish, - ): boolean { - if (this[CHANNEL_IS_CONFIRM_PUBLISHING]) { - // already instrumented by the confirm-channel publish patch - return original.apply(this, arguments); - } - const channel = this; - const { span, modifiedOptions } = startPublishSpan(exchange, routingKey, channel, options); - - // calling the normal channel publish function only stores the message in the queue; it does not send - // it and wait for an ack, so the span duration is expected to be very short. - const argumentsCopy = [...arguments]; - argumentsCopy[3] = modifiedOptions; - const originalRes = original.apply(this, argumentsCopy); - span.end(); - return originalRes; - }; -} diff --git a/packages/node/src/integrations/tracing/amqplib/vendored/semconv.ts b/packages/node/src/integrations/tracing/amqplib/vendored/semconv.ts deleted file mode 100644 index ffabc460056c..000000000000 --- a/packages/node/src/integrations/tracing/amqplib/vendored/semconv.ts +++ /dev/null @@ -1,45 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - * - * NOTICE from the Sentry authors: - * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-amqplib - * - Upstream version: @opentelemetry/instrumentation-amqplib@0.65.0 - * - Merged the upstream `semconv.ts` and `semconv-obsolete.ts` into a single file containing only the - * constants this instrumentation emits. These mirror the (now legacy) messaging semantic conventions - * that the SDK has always emitted; the `@deprecated` annotations were dropped since these vendored - * copies are intentionally the chosen output and we don't want to flag every usage site. - */ - -/** A string identifying the kind of message consumption. */ -export const ATTR_MESSAGING_OPERATION = 'messaging.operation' as const; - -/** The message destination name (the exchange for amqplib). */ -export const ATTR_MESSAGING_DESTINATION = 'messaging.destination' as const; - -/** The kind of message destination. */ -export const ATTR_MESSAGING_DESTINATION_KIND = 'messaging.destination_kind' as const; - -/** RabbitMQ message routing key. */ -export const ATTR_MESSAGING_RABBITMQ_ROUTING_KEY = 'messaging.rabbitmq.routing_key' as const; - -/** The name of the transport protocol. */ -export const ATTR_MESSAGING_PROTOCOL = 'messaging.protocol' as const; - -/** The version of the transport protocol. */ -export const ATTR_MESSAGING_PROTOCOL_VERSION = 'messaging.protocol_version' as const; - -/** Connection string. */ -export const ATTR_MESSAGING_URL = 'messaging.url' as const; - -/** A value used by the messaging system as an identifier for the message, represented as a string. */ -export const OLD_ATTR_MESSAGING_MESSAGE_ID = 'messaging.message_id' as const; - -/** The conversation ID (a.k.a. correlation ID) identifying the conversation the message belongs to. */ -export const ATTR_MESSAGING_CONVERSATION_ID = 'messaging.conversation_id' as const; - -/** Value for `messaging.destination_kind` when the destination is a topic. */ -export const MESSAGING_DESTINATION_KIND_VALUE_TOPIC = 'topic' as const; - -/** Value for `messaging.operation` when the message is being processed by a consumer. */ -export const MESSAGING_OPERATION_VALUE_PROCESS = 'process' as const; diff --git a/packages/node/src/integrations/tracing/amqplib/vendored/types.ts b/packages/node/src/integrations/tracing/amqplib/vendored/types.ts deleted file mode 100644 index 1b444d6723c1..000000000000 --- a/packages/node/src/integrations/tracing/amqplib/vendored/types.ts +++ /dev/null @@ -1,68 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - * - * NOTICE from the Sentry authors: - * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-amqplib - * - Upstream version: @opentelemetry/instrumentation-amqplib@0.65.0 - * - Some types vendored from @types/amqplib with simplifications - * - Dropped the instrumentation config and all hooks; the SDK folds origin into span creation instead - */ - -export enum EndOperation { - AutoAck = 'auto ack', - Ack = 'ack', - AckAll = 'ackAll', - Reject = 'reject', - Nack = 'nack', - NackAll = 'nackAll', - ChannelClosed = 'channel closed', - ChannelError = 'channel error', - InstrumentationTimeout = 'instrumentation timeout', -} - -// The following types are vendored from `@types/amqplib@0.10.1` - commit SHA: 4205e03127692a40b4871709a7134fe4e2ed5510 - -// Vendored from: https://github.com/DefinitelyTyped/DefinitelyTyped/blob/4205e03127692a40b4871709a7134fe4e2ed5510/types/amqplib/properties.d.ts#L142 -export interface Message { - content: Buffer; - fields: MessageFields; - properties: MessageProperties; -} - -export interface ConsumeMessage extends Message { - fields: ConsumeMessageFields; -} - -export interface CommonMessageFields { - deliveryTag: number; - redelivered: boolean; - exchange: string; - routingKey: string; -} - -export interface MessageFields extends CommonMessageFields { - messageCount?: number; - consumerTag?: string; -} - -export interface ConsumeMessageFields extends CommonMessageFields { - deliveryTag: number; -} - -export interface MessageProperties { - contentType: any | undefined; - contentEncoding: any | undefined; - headers: any; - deliveryMode: any | undefined; - priority: any | undefined; - correlationId: any | undefined; - replyTo: any | undefined; - expiration: any | undefined; - messageId: any | undefined; - timestamp: any | undefined; - type: any | undefined; - userId: any | undefined; - appId: any | undefined; - clusterId: any | undefined; -} diff --git a/packages/node/src/integrations/tracing/amqplib/vendored/utils.ts b/packages/node/src/integrations/tracing/amqplib/vendored/utils.ts deleted file mode 100644 index eec57b724a8c..000000000000 --- a/packages/node/src/integrations/tracing/amqplib/vendored/utils.ts +++ /dev/null @@ -1,206 +0,0 @@ -/* - * Copyright The OpenTelemetry Authors - * SPDX-License-Identifier: Apache-2.0 - * - * NOTICE from the Sentry authors: - * - Vendored from: https://github.com/open-telemetry/opentelemetry-js-contrib/tree/15ef7506553f631ea4181391e0c5725a56f0d082/packages/instrumentation-amqplib - * - Upstream version: @opentelemetry/instrumentation-amqplib@0.65.0 - * - Some types vendored from @types/amqplib with simplifications - * - Span creation extracted here and migrated to the @sentry/core API; origin folded into span creation - * - Cross-service trace propagation uses Sentry's `getTraceData` instead of the OTel propagator - * - Dropped the env-gated stable-semconv dual emission; only the (default) old semantic conventions are emitted - * - Replaced the OTel context-key confirm-channel marker with a synchronous flag on the channel instance - */ - -import type { Span, SpanAttributes } from '@sentry/core'; -import { getTraceData, SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN, startInactiveSpan } from '@sentry/core'; -import { MESSAGING_SYSTEM, NET_PEER_NAME, NET_PEER_PORT, SENTRY_KIND } from '@sentry/conventions/attributes'; -import type { Channel, ConfirmChannel, Connection, Options } from './amqplib-types'; -import { - ATTR_MESSAGING_CONVERSATION_ID, - ATTR_MESSAGING_DESTINATION, - ATTR_MESSAGING_DESTINATION_KIND, - ATTR_MESSAGING_OPERATION, - ATTR_MESSAGING_PROTOCOL, - ATTR_MESSAGING_PROTOCOL_VERSION, - ATTR_MESSAGING_RABBITMQ_ROUTING_KEY, - ATTR_MESSAGING_URL, - MESSAGING_DESTINATION_KIND_VALUE_TOPIC, - MESSAGING_OPERATION_VALUE_PROCESS, - OLD_ATTR_MESSAGING_MESSAGE_ID, -} from './semconv'; -import type { ConsumeMessage, Message } from './types'; - -const PUBLISHER_ORIGIN = 'auto.amqplib.otel.publisher'; -const CONSUMER_ORIGIN = 'auto.amqplib.otel.consumer'; - -export const MESSAGE_STORED_SPAN: unique symbol = Symbol('opentelemetry.amqplib.message.stored-span'); -export const CHANNEL_SPANS_NOT_ENDED: unique symbol = Symbol('opentelemetry.amqplib.channel.spans-not-ended'); -export const CHANNEL_CONSUME_TIMEOUT_TIMER: unique symbol = Symbol( - 'opentelemetry.amqplib.channel.consumer-timeout-timer', -); -export const CONNECTION_ATTRIBUTES: unique symbol = Symbol('opentelemetry.amqplib.connection.attributes'); -export const CHANNEL_IS_CONFIRM_PUBLISHING: unique symbol = Symbol('sentry.amqplib.channel.is-confirm-publishing'); - -export type InstrumentationConnection = Connection & { - [CONNECTION_ATTRIBUTES]?: SpanAttributes; -}; -export type InstrumentationPublishChannel = (Channel | ConfirmChannel) & { - connection: InstrumentationConnection; - [CHANNEL_IS_CONFIRM_PUBLISHING]?: boolean; -}; -export type InstrumentationConsumeChannel = Channel & { - connection: InstrumentationConnection; - [CHANNEL_SPANS_NOT_ENDED]?: { - msg: ConsumeMessage; - timeOfConsume: number; - }[]; - [CHANNEL_CONSUME_TIMEOUT_TIMER]?: NodeJS.Timeout; -}; -export type InstrumentationMessage = Message & { - [MESSAGE_STORED_SPAN]?: Span; -}; -export type InstrumentationConsumeMessage = ConsumeMessage & { - [MESSAGE_STORED_SPAN]?: Span; -}; - -export const normalizeExchange = (exchangeName: string): string => (exchangeName !== '' ? exchangeName : ''); - -const censorPassword = (url: string): string => { - return url.replace(/:[^:@/]*@/, ':***@'); -}; - -const getPort = (portFromUrl: number | undefined, resolvedProtocol: string): number => { - // we are using the resolved protocol which is upper case - // this code mimics the behavior of amqplib which is used to set connection params - return portFromUrl || (resolvedProtocol === 'AMQP' ? 5672 : 5671); -}; - -const getProtocol = (protocolFromUrl: string | undefined): string => { - const resolvedProtocol = protocolFromUrl || 'amqp'; - // the substring removes the ':' part of the protocol ('amqp:' -> 'amqp') - const noEndingColon = resolvedProtocol.endsWith(':') - ? resolvedProtocol.substring(0, resolvedProtocol.length - 1) - : resolvedProtocol; - // upper case to match spec - return noEndingColon.toUpperCase(); -}; - -const getHostname = (hostnameFromUrl: string | undefined): string => { - // if user supplies empty hostname, it gets forwarded to 'net' package which defaults it to localhost. - // https://nodejs.org/docs/latest-v12.x/api/net.html#net_socket_connect_options_connectlistener - return hostnameFromUrl || 'localhost'; -}; - -export const getConnectionAttributesFromServer = (conn: Connection): SpanAttributes => { - const product = conn.serverProperties.product?.toLowerCase?.(); - if (product) { - return { - [MESSAGING_SYSTEM]: product, - }; - } else { - return {}; - } -}; - -export const getConnectionAttributesFromUrl = (url: string | Options.Connect): SpanAttributes => { - const attributes: SpanAttributes = { - [ATTR_MESSAGING_PROTOCOL_VERSION]: '0.9.1', // this is the only protocol supported by the instrumented library - }; - - const resolvedUrl = url || 'amqp://localhost'; - if (typeof resolvedUrl === 'object') { - const connectOptions = resolvedUrl; - - const protocol = getProtocol(connectOptions?.protocol); - attributes[ATTR_MESSAGING_PROTOCOL] = protocol; - // oxlint-disable-next-line typescript/no-deprecated - attributes[NET_PEER_NAME] = getHostname(connectOptions?.hostname); - // oxlint-disable-next-line typescript/no-deprecated - attributes[NET_PEER_PORT] = getPort(connectOptions.port, protocol); - } else { - 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; - // oxlint-disable-next-line typescript/no-deprecated - attributes[NET_PEER_NAME] = getHostname(urlParts.hostname); - // oxlint-disable-next-line typescript/no-deprecated - attributes[NET_PEER_PORT] = getPort(urlParts.port ? parseInt(urlParts.port) : undefined, protocol); - } catch { - // best-effort: a malformed url simply yields fewer connection attributes - } - } - return attributes; -}; - -/** Reads a propagation header value off an amqplib message as a string. */ -export 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); -} - -/** Starts an inactive producer span and propagates its trace into the publish `options.headers`. */ -export function startPublishSpan( - exchange: string, - routingKey: string, - channel: InstrumentationPublishChannel, - options?: Options.Publish, -): { span: Span; modifiedOptions: Options.Publish } { - const normalizedExchange = normalizeExchange(exchange); - - const span = startInactiveSpan({ - name: `publish ${normalizedExchange}`, - attributes: { - [SENTRY_KIND]: 'producer', - ...channel.connection[CONNECTION_ATTRIBUTES], - [ATTR_MESSAGING_DESTINATION]: exchange, - [ATTR_MESSAGING_DESTINATION_KIND]: MESSAGING_DESTINATION_KIND_VALUE_TOPIC, - [ATTR_MESSAGING_RABBITMQ_ROUTING_KEY]: routingKey, - [OLD_ATTR_MESSAGING_MESSAGE_ID]: options?.messageId, - [ATTR_MESSAGING_CONVERSATION_ID]: options?.correlationId, - [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: PUBLISHER_ORIGIN, - }, - }); - - const modifiedOptions = options ?? {}; - modifiedOptions.headers = modifiedOptions.headers ?? {}; - - const traceData = getTraceData({ span }); - if (traceData['sentry-trace']) { - modifiedOptions.headers['sentry-trace'] = traceData['sentry-trace']; - } - if (traceData.baggage) { - modifiedOptions.headers['baggage'] = traceData.baggage; - } - - return { span, modifiedOptions }; -} - -/** Starts an inactive consumer (process) span carrying the amqplib messaging attributes. */ -export function startConsumeSpan( - queue: string, - msg: InstrumentationConsumeMessage, - channel: InstrumentationConsumeChannel, -): Span { - return startInactiveSpan({ - name: `${queue} process`, - attributes: { - [SENTRY_KIND]: 'consumer', - ...channel?.connection?.[CONNECTION_ATTRIBUTES], - [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, - [OLD_ATTR_MESSAGING_MESSAGE_ID]: msg?.properties.messageId, - [ATTR_MESSAGING_CONVERSATION_ID]: msg?.properties.correlationId, - [SEMANTIC_ATTRIBUTE_SENTRY_ORIGIN]: CONSUMER_ORIGIN, - }, - }); -}