feat(amqplibIntegration): Add amqplibIntegration to Node

Author: obecny

.gitignore

@@ -58,3 +58,6 @@ packages/deno/lib.deno.d.ts
 
 # gatsby
 packages/gatsby/gatsby-node.d.ts
+
+# intellij
+*.iml

dev-packages/node-integration-tests/package.json

@@ -38,6 +38,7 @@
     "@types/mongodb": "^3.6.20",
     "@types/mysql": "^2.15.21",
     "@types/pg": "^8.6.5",
+    "amqplib": "^0.10.4",
     "apollo-server": "^3.11.1",
     "axios": "^1.6.7",
     "connect": "^3.7.0",
@@ -65,6 +66,7 @@
     "yargs": "^16.2.0"
   },
   "devDependencies": {
+    "@types/amqplib": "^0.10.5",
     "@types/node-cron": "^3.0.11",
     "@types/node-schedule": "^2.1.7",
     "globby": "11"

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

@@ -0,0 +1,15 @@
+const amqpUsername = 'sentry';
+const amqpPassword = 'sentry';
+
+export const AMQP_URL = `amqp://${amqpUsername}:${amqpPassword}@localhost:5672/`;
+export const ACKNOWLEDGEMENT = { noAck: false };
+
+export 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
+  },
+};

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

@@ -0,0 +1,16 @@
+version: '3'
+
+services:
+  rabbitmq:
+    image: rabbitmq:management
+    container_name: rabbitmq
+    environment:
+      - RABBITMQ_DEFAULT_USER=sentry
+      - RABBITMQ_DEFAULT_PASS=sentry
+    ports:
+      - "5672:5672"
+      - "15672:15672"
+
+networks:
+  default:
+    driver: bridge

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

@@ -0,0 +1,9 @@
+import { loggingTransport } from '@sentry-internal/node-integration-tests';
+import * as Sentry from '@sentry/node';
+
+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/scenario-message.ts

@@ -0,0 +1,19 @@
+import * as Sentry from '@sentry/node';
+import './init';
+import { connectToRabbitMQ, consumeMessageFromQueue, createQueue, sendMessageToQueue } from './utils';
+
+const queueName = 'queue1';
+
+// eslint-disable-next-line @typescript-eslint/no-floating-promises
+(async () => {
+  const { connection, channel } = await connectToRabbitMQ();
+  await createQueue(queueName, channel);
+
+  await Sentry.startSpan({ name: 'root span' }, async () => {
+    sendMessageToQueue(queueName, channel, JSON.stringify({ foo: 'bar01' }));
+  });
+
+  await consumeMessageFromQueue(queueName, channel);
+  await channel.close();
+  await connection.close();
+})();

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

@@ -0,0 +1,53 @@
+import type { TransactionEvent } from '@sentry/types';
+import { cleanupChildProcesses, createRunner } from '../../../utils/runner';
+
+jest.setTimeout(30_000);
+
+const EXPECTED_MESSAGE_SPAN_PRODUCER = expect.objectContaining({
+  op: 'message',
+  data: expect.objectContaining({
+    'messaging.system': 'rabbitmq',
+    'otel.kind': 'PRODUCER',
+    'sentry.op': 'message',
+  }),
+  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 auto-instrumentation', () => {
+  afterAll(async () => {
+    cleanupChildProcesses();
+  });
+
+  test('should be able to send and receive messages', done => {
+    createRunner(__dirname, 'scenario-message.ts')
+      .withDockerCompose({
+        workingDirectory: [__dirname],
+        readyMatches: ['Time to start RabbitMQ'],
+      })
+      .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(done);
+  });
+});

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

@@ -0,0 +1,38 @@
+import amqp from 'amqplib';
+import type { Channel, Connection } from 'amqplib';
+import { ACKNOWLEDGEMENT, AMQP_URL, QUEUE_OPTIONS } from './constants';
+
+export type RabbitMQData = {
+  connection: Connection;
+  channel: Channel;
+};
+
+export async function connectToRabbitMQ(): Promise<RabbitMQData> {
+  const connection = await amqp.connect(AMQP_URL);
+  const channel = await connection.createChannel();
+  return { connection, channel };
+}
+
+export async function createQueue(queueName: string, channel: Channel): Promise<void> {
+  await channel.assertQueue(queueName, QUEUE_OPTIONS);
+}
+
+export function sendMessageToQueue(queueName: string, channel: Channel, message: string): void {
+  channel.sendToQueue(queueName, Buffer.from(message));
+}
+
+async function consumer(queueName: string, channel: Channel): Promise<void> {
+  await channel.consume(
+    queueName,
+    message => {
+      if (message) {
+        channel.ack(message);
+      }
+    },
+    ACKNOWLEDGEMENT,
+  );
+}
+
+export async function consumeMessageFromQueue(queueName: string, channel: Channel): Promise<void> {
+  await consumer(queueName, channel);
+}

dev-packages/node-integration-tests/utils/runner.ts

@@ -11,6 +11,7 @@ import type {
   SerializedCheckIn,
   SerializedSession,
   SessionAggregates,
+  TransactionEvent,
 } from '@sentry/types';
 import axios from 'axios';
 import { createBasicSentryServer } from './server';
@@ -151,7 +152,7 @@ type Expected =
       event: Partial<Event> | ((event: Event) => void);
     }
   | {
-      transaction: Partial<Event> | ((event: Event) => void);
+      transaction: Partial<TransactionEvent> | ((event: TransactionEvent) => void);
     }
   | {
       session: Partial<SerializedSession> | ((event: SerializedSession) => void);
@@ -317,7 +318,7 @@ export function createRunner(...paths: string[]) {
             }
 
             if ('transaction' in expected) {
-              const event = item[1] as Event;
+              const event = item[1] as TransactionEvent;
               if (typeof expected.transaction === 'function') {
                 expected.transaction(event);
               } else {
@@ -483,6 +484,7 @@ export function createRunner(...paths: string[]) {
           method: 'get' | 'post',
           path: string,
           headers: Record<string, string> = {},
+          data?: any, // axios accept any as data
         ): Promise<T | undefined> {
           try {
             await waitFor(() => scenarioServerPort !== undefined);
@@ -497,7 +499,7 @@ export function createRunner(...paths: string[]) {
               if (method === 'get') {
                 await axios.get(url, { headers });
               } else {
-                await axios.post(url, { headers });
+                await axios.post(url, data, { headers });
               }
             } catch (e) {
               return;
@@ -506,7 +508,7 @@ export function createRunner(...paths: string[]) {
           } else if (method === 'get') {
             return (await axios.get(url, { headers })).data;
           } else {
-            return (await axios.post(url, { headers })).data;
+            return (await axios.post(url, data, { headers })).data;
           }
         },
       };

packages/astro/src/index.server.ts

@@ -13,6 +13,7 @@ export {
   addIntegration,
   addOpenTelemetryInstrumentation,
   addRequestDataToEvent,
+  amqplibIntegration,
   anrIntegration,
   captureCheckIn,
   captureConsoleIntegration,

packages/aws-serverless/src/index.ts

@@ -110,6 +110,7 @@ export {
   addOpenTelemetryInstrumentation,
   zodErrorsIntegration,
   profiler,
+  amqplibIntegration,
 } from '@sentry/node';
 
 export {

packages/bun/src/index.ts

@@ -131,6 +131,7 @@ export {
   addOpenTelemetryInstrumentation,
   zodErrorsIntegration,
   profiler,
+  amqplibIntegration,
 } from '@sentry/node';
 
 export {

packages/google-cloud-serverless/src/index.ts

@@ -110,6 +110,7 @@ export {
   addOpenTelemetryInstrumentation,
   zodErrorsIntegration,
   profiler,
+  amqplibIntegration,
 } from '@sentry/node';
 
 export {

packages/node/package.json

@@ -69,6 +69,7 @@
     "@opentelemetry/context-async-hooks": "^1.25.1",
     "@opentelemetry/core": "^1.25.1",
     "@opentelemetry/instrumentation": "^0.53.0",
+    "@opentelemetry/instrumentation-amqplib": "^0.42.0",
     "@opentelemetry/instrumentation-connect": "0.39.0",
     "@opentelemetry/instrumentation-express": "0.42.0",
     "@opentelemetry/instrumentation-fastify": "0.39.0",

packages/node/src/index.ts

@@ -28,6 +28,7 @@ export { koaIntegration, setupKoaErrorHandler } from './integrations/tracing/koa
 export { connectIntegration, setupConnectErrorHandler } from './integrations/tracing/connect';
 export { spotlightIntegration } from './integrations/spotlight';
 export { genericPoolIntegration } from './integrations/tracing/genericPool';
+export { amqplibIntegration } from './integrations/tracing/amqplib';
 
 export { SentryContextManager } from './otel/contextManager';
 export { generateInstrumentOnce } from './otel/instrument';

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

@@ -0,0 +1,30 @@
+import type { Span } from '@opentelemetry/api';
+import { AmqplibInstrumentation, type AmqplibInstrumentationConfig } from '@opentelemetry/instrumentation-amqplib';
+import { defineIntegration } from '@sentry/core';
+import type { IntegrationFn } from '@sentry/types';
+import { generateInstrumentOnce } from '../../otel/instrument';
+import { addOriginToSpan } from '../../utils/addOriginToSpan';
+
+const INTEGRATION_NAME = 'Amqplib';
+
+const config: AmqplibInstrumentationConfig = {
+  consumeEndHook: (span: Span) => {
+    addOriginToSpan(span, 'auto.amqplib.otel.consumer');
+  },
+  publishConfirmHook: (span: Span) => {
+    addOriginToSpan(span, 'auto.amqplib.otel.publisher');
+  },
+};
+
+export const instrumentAmqplib = generateInstrumentOnce(INTEGRATION_NAME, () => new AmqplibInstrumentation(config));
+
+const _amqplibIntegration = (() => {
+  return {
+    name: INTEGRATION_NAME,
+    setupOnce() {
+      instrumentAmqplib();
+    },
+  };
+}) satisfies IntegrationFn;
+
+export const amqplibIntegration = defineIntegration(_amqplibIntegration);

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

@@ -1,6 +1,7 @@
 import type { Integration } from '@sentry/types';
 import { instrumentHttp } from '../http';
 
+import { amqplibIntegration, instrumentAmqplib } from './amqplib';
 import { connectIntegration, instrumentConnect } from './connect';
 import { expressIntegration, instrumentExpress } from './express';
 import { fastifyIntegration, instrumentFastify } from './fastify';
@@ -41,6 +42,7 @@ export function getAutoPerformanceIntegrations(): Integration[] {
     connectIntegration(),
     genericPoolIntegration(),
     kafkaIntegration(),
+    amqplibIntegration(),
   ];
 }
 
@@ -67,5 +69,6 @@ export function getOpenTelemetryInstrumentationToPreload(): (((options?: any) =>
     instrumentGraphql,
     instrumentRedis,
     instrumentGenericPool,
+    instrumentAmqplib,
   ];
 }

packages/remix/src/index.server.ts

@@ -18,6 +18,7 @@ export {
   addIntegration,
   addOpenTelemetryInstrumentation,
   addRequestDataToEvent,
+  amqplibIntegration,
   anrIntegration,
   captureCheckIn,
   captureConsoleIntegration,

packages/solidstart/src/server/index.ts

@@ -9,6 +9,7 @@ export {
   addIntegration,
   addOpenTelemetryInstrumentation,
   addRequestDataToEvent,
+  amqplibIntegration,
   anrIntegration,
   captureCheckIn,
   captureConsoleIntegration,

packages/sveltekit/src/server/index.ts

@@ -9,6 +9,7 @@ export {
   addIntegration,
   addOpenTelemetryInstrumentation,
   addRequestDataToEvent,
+  amqplibIntegration,
   anrIntegration,
   captureCheckIn,
   captureConsoleIntegration,