test(node): Port AMQP integration tests to span streaming

Co-Authored-By: GPT-6 <codex@openai.com>

Author: nicohrubec

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

@@ -2,7 +2,6 @@ import * as Sentry from '@sentry/node';
 import { loggingTransport } from '@sentry-internal/node-integration-tests';
 
 Sentry.init({
-  traceLifecycle: 'static',
   dsn: 'https://public@dsn.ingest.sentry.io/1337',
   release: '1.0',
   tracesSampleRate: 1.0,

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

@@ -1,4 +1,18 @@
-import type { TransactionEvent } from '@sentry/core';
+import {
+  MESSAGING_DESTINATION_NAME,
+  MESSAGING_OPERATION_NAME,
+  MESSAGING_OPERATION_TYPE,
+  MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY,
+  MESSAGING_SYSTEM,
+  NETWORK_PROTOCOL_NAME,
+  NETWORK_PROTOCOL_VERSION,
+  SENTRY_KIND,
+  SENTRY_OP,
+  SENTRY_ORIGIN,
+  SERVER_ADDRESS,
+  SERVER_PORT,
+  URL_FULL,
+} from '@sentry/conventions/attributes';
 import { afterAll, describe, expect } from 'vitest';
 import { cleanupChildProcesses, createEsmAndCjsTests, describeWithDockerCompose } from '../../../utils/runner';
 
@@ -8,37 +22,35 @@ import { cleanupChildProcesses, createEsmAndCjsTests, describeWithDockerCompose
 // default (empty) exchange with the queue name as the routing key.
 const expectedProducerSpan = (routingKey: string) =>
   expect.objectContaining({
-    op: 'queue.publish',
-    data: expect.objectContaining({
-      'messaging.system': 'rabbitmq',
-      'messaging.operation.name': 'send',
-      'messaging.operation.type': 'send',
-      'messaging.destination.name': routingKey,
-      'messaging.rabbitmq.destination.routing_key': routingKey,
-      'network.protocol.name': 'AMQP',
-      'network.protocol.version': '0.9.1',
-      'server.address': 'localhost',
-      'server.port': 5672,
-      'url.full': 'amqp://sentry:***@localhost:5672/',
-      'sentry.kind': 'producer',
-      'sentry.op': 'queue.publish',
-      'sentry.origin': 'auto.amqplib.publisher',
+    attributes: expect.objectContaining({
+      [MESSAGING_SYSTEM]: { type: 'string', value: 'rabbitmq' },
+      [MESSAGING_OPERATION_NAME]: { type: 'string', value: 'send' },
+      [MESSAGING_OPERATION_TYPE]: { type: 'string', value: 'send' },
+      [MESSAGING_DESTINATION_NAME]: { type: 'string', value: routingKey },
+      [MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY]: { type: 'string', value: routingKey },
+      [NETWORK_PROTOCOL_NAME]: { type: 'string', value: 'AMQP' },
+      [NETWORK_PROTOCOL_VERSION]: { type: 'string', value: '0.9.1' },
+      [SERVER_ADDRESS]: { type: 'string', value: 'localhost' },
+      [SERVER_PORT]: { type: 'integer', value: 5672 },
+      [URL_FULL]: { type: 'string', value: 'amqp://sentry:***@localhost:5672/' },
+      [SENTRY_KIND]: { type: 'string', value: 'producer' },
+      [SENTRY_OP]: { type: 'string', value: 'queue.publish' },
+      [SENTRY_ORIGIN]: { type: 'string', value: 'auto.amqplib.publisher' },
     }),
     status: 'ok',
   });
 
 const EXPECTED_MESSAGE_SPAN_CONSUMER = expect.objectContaining({
-  op: 'queue.process',
-  data: expect.objectContaining({
-    'messaging.system': 'rabbitmq',
+  attributes: expect.objectContaining({
+    [MESSAGING_SYSTEM]: { type: 'string', value: 'rabbitmq' },
     // The delivery carries the default exchange (''), so the routing key is the destination.
-    'messaging.destination.name': 'queue1',
-    'messaging.rabbitmq.destination.routing_key': 'queue1',
-    'messaging.operation.name': 'process',
-    'messaging.operation.type': 'process',
-    'sentry.kind': 'consumer',
-    'sentry.op': 'queue.process',
-    'sentry.origin': 'auto.amqplib.consumer',
+    [MESSAGING_DESTINATION_NAME]: { type: 'string', value: 'queue1' },
+    [MESSAGING_RABBITMQ_DESTINATION_ROUTING_KEY]: { type: 'string', value: 'queue1' },
+    [MESSAGING_OPERATION_NAME]: { type: 'string', value: 'process' },
+    [MESSAGING_OPERATION_TYPE]: { type: 'string', value: 'process' },
+    [SENTRY_KIND]: { type: 'string', value: 'consumer' },
+    [SENTRY_OP]: { type: 'string', value: 'queue.process' },
+    [SENTRY_ORIGIN]: { type: 'string', value: 'auto.amqplib.consumer' },
   }),
   status: 'ok',
 });
@@ -59,40 +71,27 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_
       'instrument.mjs',
       (createTestRunner, test) => {
         test('should be able to send and receive messages', { timeout: 60_000 }, async () => {
-          // The producer ('root span') and consumer ('queue1 process') transactions can
-          // arrive in any order, so we collect them and assert after both are received.
-          const receivedTransactions: TransactionEvent[] = [];
-
           await createTestRunner()
             .expect({
-              transaction: (transaction: TransactionEvent) => {
-                receivedTransactions.push(transaction);
-              },
-            })
-            .expect({
-              transaction: (transaction: TransactionEvent) => {
-                receivedTransactions.push(transaction);
-
-                // The producer span is a child of the manually-started 'root span' transaction, so we
-                // 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.publisher'),
-                );
-                const consumer = receivedTransactions.find(
-                  t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.amqplib.consumer',
+              span: container => {
+                const producer = container.items.find(span => span.is_segment && span.name === 'root span');
+                const consumer = container.items.find(
+                  t => t.attributes[SENTRY_ORIGIN]?.value === 'auto.amqplib.consumer',
                 );
 
                 expect(producer).toBeDefined();
                 expect(consumer).toBeDefined();
 
-                expect(producer!.transaction).toBe('root span');
-                expect(consumer!.transaction).toBe('queue1 process');
+                expect(producer!.name).toBe('root span');
+                expect(consumer!.name).toBe('process queue1');
 
-                const producerSpan = producer!.spans?.find(s => s.data?.['sentry.origin'] === 'auto.amqplib.publisher');
+                const producerSpan = container.items.find(
+                  s => s.attributes[SENTRY_ORIGIN]?.value === 'auto.amqplib.publisher',
+                );
+                expect(producerSpan?.name).toBe('send queue1');
                 expect(producerSpan).toMatchObject(expectedProducerSpan('queue1'));
 
-                expect(consumer!.contexts?.trace).toMatchObject(EXPECTED_MESSAGE_SPAN_CONSUMER);
+                expect(consumer!).toMatchObject(EXPECTED_MESSAGE_SPAN_CONSUMER);
               },
             })
             .start()
@@ -108,35 +107,23 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_
       'instrument.mjs',
       (createTestRunner, test) => {
         test('marks the consumer span as errored when the message is rejected', { timeout: 60_000 }, async () => {
-          // The error scenario emits the producer ('root span') and the rejected consumer
-          // ('queue1 process') transactions in any order, so we collect both and assert on the consumer.
-          const receivedTransactions: TransactionEvent[] = [];
-
           await createTestRunner()
             .expect({
-              transaction: (transaction: TransactionEvent) => {
-                receivedTransactions.push(transaction);
-              },
-            })
-            .expect({
-              transaction: (transaction: TransactionEvent) => {
-                receivedTransactions.push(transaction);
-
-                const consumer = receivedTransactions.find(
-                  t => t.contexts?.trace?.data?.['sentry.origin'] === 'auto.amqplib.consumer',
+              span: container => {
+                const consumer = container.items.find(
+                  t => t.attributes[SENTRY_ORIGIN]?.value === 'auto.amqplib.consumer',
                 );
 
                 expect(consumer).toBeDefined();
-                expect(consumer!.transaction).toBe('queue-error process');
-                expect(consumer!.contexts?.trace).toMatchObject(
+                expect(consumer!.name).toBe('process queue-error');
+                expect(consumer!).toMatchObject(
                   expect.objectContaining({
-                    op: 'queue.process',
-                    status: 'internal_error',
-                    data: expect.objectContaining({
-                      'messaging.system': 'rabbitmq',
-                      'sentry.kind': 'consumer',
-                      'sentry.op': 'queue.process',
-                      'sentry.origin': 'auto.amqplib.consumer',
+                    status: 'error',
+                    attributes: expect.objectContaining({
+                      [MESSAGING_SYSTEM]: { type: 'string', value: 'rabbitmq' },
+                      [SENTRY_KIND]: { type: 'string', value: 'consumer' },
+                      [SENTRY_OP]: { type: 'string', value: 'queue.process' },
+                      [SENTRY_ORIGIN]: { type: 'string', value: 'auto.amqplib.consumer' },
                     }),
                   }),
                 );
@@ -160,11 +147,11 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_
           async () => {
             await createTestRunner()
               .expect({
-                transaction: (transaction: TransactionEvent) => {
-                  expect(transaction.transaction).toBe('root span');
+                span: container => {
+                  expect(container.items.find(span => span.is_segment)?.name).toBe('root span');
 
-                  const producerSpans = transaction.spans?.filter(
-                    s => s.data?.['sentry.origin'] === 'auto.amqplib.publisher',
+                  const producerSpans = container.items.filter(
+                    s => s.attributes[SENTRY_ORIGIN]?.value === 'auto.amqplib.publisher',
                   );
 
                   // The confirm channel internally calls the base publish; the instrumentation must not
@@ -181,35 +168,4 @@ describeWithDockerCompose('amqplib auto-instrumentation', { workingDirectory: [_
       { additionalDependencies },
     );
   });
-
-  createEsmAndCjsTests(__dirname, 'scenario.mjs', 'instrument-span-streaming.mjs', (createTestRunner, test) => {
-    test('names streamed spans after the messaging conventions', { timeout: 60_000 }, async () => {
-      await createTestRunner()
-        .ignore('event')
-        .expect({
-          span: container => {
-            // `sendToQueue` publishes to the default exchange, which has no name. Its routing key is the
-            // queue name, so it is the destination rather than per-message data.
-            for (const origin of ['auto.amqplib.publisher', 'auto.amqplib.consumer']) {
-              const span = container.items.find(item => item.attributes['sentry.origin']?.value === origin);
-              expect(span).toBeDefined();
-              expect(span!.attributes['messaging.destination.name']?.value).toBe('queue1');
-              expect(span!.attributes['messaging.rabbitmq.destination.routing_key']?.value).toBe('queue1');
-            }
-
-            const producerSpan = container.items.find(
-              span => span.attributes['sentry.origin']?.value === 'auto.amqplib.publisher',
-            );
-            expect(producerSpan!.name).toBe('send queue1');
-
-            const consumerSpan = container.items.find(
-              span => span.attributes['sentry.origin']?.value === 'auto.amqplib.consumer',
-            );
-            expect(consumerSpan!.name).toBe('process queue1');
-          },
-        })
-        .start()
-        .completed();
-    });
-  });
 });