Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
nicohrubec merged 2 commits into from
Oct 8, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d8f8243
test(node): Port AMQP integration tests to span streaming
nicohrubec Oct 8, 2026
f680dfc
test(node): Use conventions op constants in AMQP assertions
nicohrubec Oct 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 0 additions & 10 deletions dev-packages/node-integration-tests/suites/tracing/amqplib/instrument-span-streaming.mjs
View file
Open in desktop

This file was deleted.

1 change: 0 additions & 1 deletion dev-packages/node-integration-tests/suites/tracing/amqplib/instrument.mjs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
173 changes: 65 additions & 108 deletions dev-packages/node-integration-tests/suites/tracing/amqplib/test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,19 @@
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 { QUEUE_PROCESS, QUEUE_PUBLISH } from '@sentry/conventions/op';
import { afterAll, describe, expect } from 'vitest';
import { cleanupChildProcesses, createEsmAndCjsTests, describeWithDockerCompose } from '../../../utils/runner';

Expand All @@ -8,37 +23,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({

@JPeer264 JPeer264 Oct 8, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

q: Should we be more explicit with attributes to adhere to the /write-tests skill guidelines? Same for the other parts where this is being touched

@nicohrubec nicohrubec Oct 8, 2026 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't wanna switch to full toEqual because the tests would explode and this would be basically impossible to review. We could switch to check individual attributes but I think in most cases not really worth it if the check is specific enough.

[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',
});
Expand All @@ -59,40 +72,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);
Comment thread
nicohrubec marked this conversation as resolved.
Resolved
expect(consumer!).toMatchObject(EXPECTED_MESSAGE_SPAN_CONSUMER);
},
})
.start()
Expand All @@ -108,35 +108,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' },
}),
}),
);
Expand All @@ -160,11 +148,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
Expand All @@ -181,35 +169,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();
});
});
});
Loading