ref(node): Vendor amqplib instrumentation

.oxlintrc.base.json

@@ -147,7 +147,8 @@
         "**/integrations/fs/vendored/**/*.ts",
         "**/integrations/tracing/knex/vendored/**/*.ts",
         "**/integrations/tracing/mongo/vendored/**/*.ts",
-        "**/integrations/tracing/connect/vendored/**/*.ts"
+        "**/integrations/tracing/connect/vendored/**/*.ts",
+        "**/integrations/tracing/amqplib/vendored/**/*.ts"
       ],
       "rules": {
         "typescript/no-explicit-any": "off"

dev-packages/node-integration-tests/suites/tracing/amqplib-v1/docker-compose.yml

@@ -0,0 +1,22 @@
+version: '3'
+
+services:
+  rabbitmq:
+    image: rabbitmq:management
+    container_name: rabbitmq-v1
+    environment:
+      - RABBITMQ_DEFAULT_USER=sentry
+      - RABBITMQ_DEFAULT_PASS=sentry
+    ports:
+      - '5673:5672'
+      - '15673:15672'
+    healthcheck:
+      test: ['CMD-SHELL', 'rabbitmq-diagnostics -q ping']
+      interval: 2s
+      timeout: 10s
+      retries: 30
+      start_period: 15s
+
+networks:
+  default:
+    driver: bridge

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

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

dev-packages/node-integration-tests/suites/tracing/amqplib-v1/package.json

@@ -0,0 +1,7 @@
+{
+  "name": "sentry-amqplib-v1-test",
+  "version": "1.0.0",
+  "dependencies": {
+    "amqplib": "^1.0.0"
+  }
+}

dev-packages/node-integration-tests/suites/tracing/amqplib-v1/scenario.mjs

@@ -0,0 +1,72 @@
+import * as Sentry from '@sentry/node';
+import amqp from 'amqplib';
+
+const queueName = 'queue1';
+const amqpUsername = 'sentry';
+const amqpPassword = 'sentry';
+
+const AMQP_URL = `amqp://${amqpUsername}:${amqpPassword}@localhost:5673/`;
+const ACKNOWLEDGEMENT = { noAck: false };
+
+const QUEUE_OPTIONS = {
+  durable: true, // Make the queue durable
+  exclusive: false, // Not exclusive
+  autoDelete: false, // Don't auto-delete the queue
+  arguments: {
+    'x-message-ttl': 30000, // Message TTL of 30 seconds
+    'x-max-length': 1000, // Maximum queue length of 1000 messages
+  },
+};
+
+(async () => {
+  const { connection, channel } = await connectToRabbitMQ();
+  await createQueue(queueName, channel);
+
+  const consumeMessagePromise = consumeMessageFromQueue(queueName, channel);
+
+  await Sentry.startSpan({ name: 'root span' }, async () => {
+    sendMessageToQueue(queueName, channel, JSON.stringify({ foo: 'bar01' }));
+  });
+
+  await consumeMessagePromise;
+
+  await channel.close();
+  await connection.close();
+})();
+
+async function connectToRabbitMQ() {
+  const connection = await amqp.connect(AMQP_URL);
+  const channel = await connection.createChannel();
+  return { connection, channel };
+}
+
+async function createQueue(queueName, channel) {
+  await channel.assertQueue(queueName, QUEUE_OPTIONS);
+}
+
+function sendMessageToQueue(queueName, channel, message) {
+  channel.sendToQueue(queueName, Buffer.from(message));
+}
+
+async function consumer(queueName, channel) {
+  return new Promise((resolve, reject) => {
+    channel
+      .consume(
+        queueName,
+        message => {
+          if (message) {
+            channel.ack(message);
+            resolve();
+          } else {
+            reject(new Error('No message received'));
+          }
+        },
+        ACKNOWLEDGEMENT,
+      )
+      .catch(reject);
+  });
+}
+
+async function consumeMessageFromQueue(queueName, channel) {
+  await consumer(queueName, channel);
+}

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

@@ -0,0 +1,55 @@
+import type { TransactionEvent } from '@sentry/core';
+import { afterAll, describe, expect } from 'vitest';
+import { cleanupChildProcesses, createEsmAndCjsTests } from '../../../utils/runner';
+
+const EXPECTED_MESSAGE_SPAN_PRODUCER = expect.objectContaining({
+  op: 'message',
+  data: expect.objectContaining({
+    'messaging.system': 'rabbitmq',
+    'otel.kind': 'PRODUCER',
+    'sentry.op': 'message',
+    'sentry.origin': 'auto.amqplib.otel.publisher',
+  }),
+  status: 'ok',
+});
+
+const EXPECTED_MESSAGE_SPAN_CONSUMER = expect.objectContaining({
+  op: 'message',
+  data: expect.objectContaining({
+    'messaging.system': 'rabbitmq',
+    'otel.kind': 'CONSUMER',
+    'sentry.op': 'message',
+    'sentry.origin': 'auto.amqplib.otel.consumer',
+  }),
+  status: 'ok',
+});
+
+describe('amqplib v1 auto-instrumentation', () => {
+  afterAll(async () => {
+    cleanupChildProcesses();
+  });
+
+  createEsmAndCjsTests(__dirname, 'scenario.mjs', 'instrument.mjs', (createTestRunner, test) => {
+    test('should be able to send and receive messages with amqplib v1', { timeout: 60_000 }, async () => {
+      await createTestRunner()
+        .withDockerCompose({
+          workingDirectory: [__dirname],
+        })
+        .expect({
+          transaction: (transaction: TransactionEvent) => {
+            expect(transaction.transaction).toEqual('root span');
+            expect(transaction.spans?.length).toEqual(1);
+            expect(transaction.spans![0]).toMatchObject(EXPECTED_MESSAGE_SPAN_PRODUCER);
+          },
+        })
+        .expect({
+          transaction: (transaction: TransactionEvent) => {
+            expect(transaction.transaction).toEqual('queue1 process');
+            expect(transaction.contexts?.trace).toMatchObject(EXPECTED_MESSAGE_SPAN_CONSUMER);
+          },
+        })
+        .start()
+        .completed();
+    });
+  });
+});

packages/node/package.json

@@ -68,7 +68,6 @@
     "@opentelemetry/api": "^1.9.1",
     "@opentelemetry/core": "^2.6.1",
     "@opentelemetry/instrumentation": "^0.214.0",
-    "@opentelemetry/instrumentation-amqplib": "0.61.0",
     "@opentelemetry/instrumentation-graphql": "0.62.0",
     "@opentelemetry/instrumentation-hapi": "0.60.0",
     "@opentelemetry/instrumentation-http": "0.214.0",

packages/node/src/integrations/tracing/amqplib.ts → packages/node/src/integrations/tracing/amqplib/index.ts

@@ -1,5 +1,6 @@
 import type { Span } from '@opentelemetry/api';
-import { AmqplibInstrumentation, type AmqplibInstrumentationConfig } from '@opentelemetry/instrumentation-amqplib';
+import { AmqplibInstrumentation } from './vendored/amqplib';
+import type { AmqplibInstrumentationConfig } from './vendored/types';
 import type { IntegrationFn } from '@sentry/core';
 import { defineIntegration } from '@sentry/core';
 import { addOriginToSpan, generateInstrumentOnce } from '@sentry/node-core';

packages/node/src/integrations/tracing/amqplib/vendored/amqplib-types.ts

@@ -0,0 +1,47 @@
+/*
+ * 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;
+  }
+}