feat(server-utils): Migrate @opentelemetry/instrumentation-amqplib to orchestrion

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) =>
@@ -10,9 +17,17 @@ 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,
+      // 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': 'auto.amqplib.otel.publisher',
+      'sentry.origin': PUBLISHER_ORIGIN,
     }),
     status: 'ok',
   });
@@ -24,7 +39,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 +80,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 +92,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 +129,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 +142,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 +172,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.

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';

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;
+};

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(),

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",