feat(node): Add amqplibIntegration

.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",
@@ -66,6 +67,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://[email protected]/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,54 @@
+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',
+ '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 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

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

packages/bun/src/index.ts

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

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

@@ -111,6 +111,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-dataloader": "0.12.0",
  "@opentelemetry/instrumentation-express": "0.42.0",

packages/node/src/index.ts

@@ -29,6 +29,7 @@ export { connectIntegration, setupConnectErrorHandler } from './integrations/tra
 export { spotlightIntegration } from './integrations/spotlight';
 export { genericPoolIntegration } from './integrations/tracing/genericPool';
 export { dataloaderIntegration } from './integrations/tracing/dataloader';
+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');
+ },
+ publishHook: (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 { dataloaderIntegration, instrumentDataloader } from './dataloader';
 import { expressIntegration, instrumentExpress } from './express';
@@ -43,6 +44,7 @@ export function getAutoPerformanceIntegrations(): Integration[] {
  genericPoolIntegration(),
  kafkaIntegration(),
  dataloaderIntegration(),
+ amqplibIntegration(),
  ];
 }
 
@@ -70,5 +72,6 @@ export function getOpenTelemetryInstrumentationToPreload(): (((options?: any) =>
  instrumentRedis,
  instrumentGenericPool,
  instrumentDataloader,
+ 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,