diff --git a/dev-packages/node-integration-tests/suites/tracing/bullmq/test.ts b/dev-packages/node-integration-tests/suites/tracing/bullmq/test.ts index 49871dcc9786..bf2e343e39d7 100644 --- a/dev-packages/node-integration-tests/suites/tracing/bullmq/test.ts +++ b/dev-packages/node-integration-tests/suites/tracing/bullmq/test.ts @@ -1,6 +1,6 @@ import type { Event, SerializedMetric } from '@sentry/core'; import { SDK_VERSION } from '@sentry/core'; -import { afterAll, expect } from 'vitest'; +import { afterAll, describe, expect } from 'vitest'; import { EXPECTED_SDK_NAME } from '../../../utils'; import { cleanupChildProcesses, createEsmAndCjsTests, describeWithDockerCompose } from '../../../utils/runner'; @@ -210,96 +210,121 @@ describeWithDockerCompose('bullmq', { workingDirectory: [__dirname] }, () => { }); }); - createEsmAndCjsTests(__dirname, 'scenario-failed-job.mjs', 'instrument.mjs', (createRunner, test) => { - test( - 'captures an error for a failed job and sets the error status on its process span', - { timeout: 90_000 }, - async () => { - let errorEvent: Event | undefined; - const runner = createRunner().ignore('trace_metric'); - const consumerSpansPromise = runner.collectStreamedSpansUntilSegment('process test-failed-queue'); + // BullMQ passes the job error to the telemetry adapter since 6.7.5, so only then does the error have a stack. + describe.each([ + ['bullmq v5', {}, undefined], + [ + 'bullmq v6', + { bullmq: '^6.7.5' }, + { + frames: expect.arrayContaining([ + expect.objectContaining({ filename: expect.stringContaining('scenario-failed-job') }), + ]), + }, + ], + ])('%s', (_version, additionalDependencies, expectedStacktrace) => { + createEsmAndCjsTests( + __dirname, + 'scenario-failed-job.mjs', + 'instrument.mjs', + (createRunner, test) => { + test( + 'captures an error for a failed job and sets the error status on its process span', + { timeout: 90_000 }, + async () => { + let errorEvent: Event | undefined; + const runner = createRunner().ignore('trace_metric'); + const consumerSpansPromise = runner.collectStreamedSpansUntilSegment('process test-failed-queue'); - await runner - .expect({ - event: event => { - errorEvent = event; - }, - }) - .start() - .completed(); + await runner + .expect({ + event: event => { + errorEvent = event; + }, + }) + .start() + .completed(); - const consumerSpans = await consumerSpansPromise; - const processSpan = consumerSpans.find(span => span.is_segment && span.name === 'process test-failed-queue')!; + const consumerSpans = await consumerSpansPromise; + const processSpan = consumerSpans.find( + span => span.is_segment && span.name === 'process test-failed-queue', + )!; - expect( - consumerSpans - .filter(span => span.attributes['messaging.system']?.value === 'bullmq') - .sort((a, b) => a.start_timestamp - b.start_timestamp) - .map(span => ({ - name: span.name, - status: span.status, - attributes: span.attributes, - })), - ).toEqual([ - { - name: 'process test-failed-queue', - status: 'error', - attributes: { - 'sentry.op': { type: 'string', value: 'queue.process' }, - 'sentry.origin': { type: 'string', value: 'auto.queue.bullmq.consumer' }, - 'messaging.system': { type: 'string', value: 'bullmq' }, - 'messaging.destination.name': { type: 'string', value: 'test-failed-queue' }, - 'messaging.operation.name': { type: 'string', value: 'process' }, - 'messaging.message.id': { type: 'string', value: expect.any(String) }, - 'messaging.message.retry.count': { type: 'integer', value: 0 }, - 'messaging.client.id': { type: 'string', value: expect.any(String) }, - 'messaging.bullmq.job.name': { type: 'string', value: 'failing-job' }, - 'messaging.bullmq.job.processed_timestamp': { type: 'integer', value: expect.any(Number) }, - 'messaging.bullmq.job.attempt_finished_timestamp': { type: 'integer', value: expect.any(Number) }, - 'sentry.environment': { type: 'string', value: 'production' }, - 'sentry.is_localhost': { type: 'boolean', value: false }, - 'sentry.release': { type: 'string', value: '1.0' }, - 'sentry.sdk.name': { type: 'string', value: EXPECTED_SDK_NAME }, - 'sentry.sdk.version': { type: 'string', value: SDK_VERSION }, - 'sentry.segment.id': { type: 'string', value: processSpan.span_id }, - 'sentry.segment.name': { type: 'string', value: 'process test-failed-queue' }, - 'sentry.trace_lifecycle': { type: 'string', value: 'stream' }, - 'sentry.sample_rate': { type: 'integer', value: 1 }, - 'sentry.sdk.integrations': { type: 'array', value: expect.any(Array) }, - 'sentry.segment.name.source': { type: 'string', value: 'custom' }, - 'sentry.status.message': { type: 'string', value: 'internal_error' }, - }, - }, - { - name: 'fail test-failed-queue', - status: 'ok', - attributes: { - 'sentry.op': { type: 'string', value: 'queue' }, - 'sentry.origin': { type: 'string', value: 'auto.queue.bullmq' }, - 'messaging.system': { type: 'string', value: 'bullmq' }, - 'messaging.destination.name': { type: 'string', value: 'test-failed-queue' }, - 'messaging.operation.name': { type: 'string', value: 'fail' }, - 'messaging.message.id': { type: 'string', value: expect.any(String) }, - 'messaging.bullmq.job.name': { type: 'string', value: 'failing-job' }, - 'sentry.environment': { type: 'string', value: 'production' }, - 'sentry.is_localhost': { type: 'boolean', value: false }, - 'sentry.release': { type: 'string', value: '1.0' }, - 'sentry.sdk.name': { type: 'string', value: EXPECTED_SDK_NAME }, - 'sentry.sdk.version': { type: 'string', value: SDK_VERSION }, - 'sentry.segment.id': { type: 'string', value: processSpan.span_id }, - 'sentry.segment.name': { type: 'string', value: 'process test-failed-queue' }, - 'sentry.trace_lifecycle': { type: 'string', value: 'stream' }, - }, - }, - ]); - expect(errorEvent?.exception?.values).toHaveLength(1); - expect(errorEvent?.exception?.values?.[0]?.type).toBe('Error'); - expect(errorEvent?.exception?.values?.[0]?.value).toBe('Job failed on purpose'); - expect(errorEvent?.exception?.values?.[0]?.mechanism).toEqual({ type: 'auto.queue.bullmq', handled: false }); - expect(errorEvent?.exception?.values?.[0]?.stacktrace).toBeUndefined(); - expect(errorEvent?.contexts?.trace?.trace_id).toBe(processSpan.trace_id); - expect(errorEvent?.contexts?.trace?.span_id).toBe(processSpan.span_id); + expect( + consumerSpans + .filter(span => span.attributes['messaging.system']?.value === 'bullmq') + .sort((a, b) => a.start_timestamp - b.start_timestamp) + .map(span => ({ + name: span.name, + status: span.status, + attributes: span.attributes, + })), + ).toEqual([ + { + name: 'process test-failed-queue', + status: 'error', + attributes: { + 'sentry.op': { type: 'string', value: 'queue.process' }, + 'sentry.origin': { type: 'string', value: 'auto.queue.bullmq.consumer' }, + 'messaging.system': { type: 'string', value: 'bullmq' }, + 'messaging.destination.name': { type: 'string', value: 'test-failed-queue' }, + 'messaging.operation.name': { type: 'string', value: 'process' }, + 'messaging.message.id': { type: 'string', value: expect.any(String) }, + 'messaging.message.retry.count': { type: 'integer', value: 0 }, + 'messaging.client.id': { type: 'string', value: expect.any(String) }, + 'messaging.bullmq.job.name': { type: 'string', value: 'failing-job' }, + 'messaging.bullmq.job.processed_timestamp': { type: 'integer', value: expect.any(Number) }, + 'messaging.bullmq.job.attempt_finished_timestamp': { type: 'integer', value: expect.any(Number) }, + 'sentry.environment': { type: 'string', value: 'production' }, + 'sentry.is_localhost': { type: 'boolean', value: false }, + 'sentry.release': { type: 'string', value: '1.0' }, + 'sentry.sdk.name': { type: 'string', value: EXPECTED_SDK_NAME }, + 'sentry.sdk.version': { type: 'string', value: SDK_VERSION }, + 'sentry.segment.id': { type: 'string', value: processSpan.span_id }, + 'sentry.segment.name': { type: 'string', value: 'process test-failed-queue' }, + 'sentry.trace_lifecycle': { type: 'string', value: 'stream' }, + 'sentry.sample_rate': { type: 'integer', value: 1 }, + 'sentry.sdk.integrations': { type: 'array', value: expect.any(Array) }, + 'sentry.segment.name.source': { type: 'string', value: 'custom' }, + 'sentry.status.message': { type: 'string', value: 'internal_error' }, + }, + }, + { + name: 'fail test-failed-queue', + status: 'ok', + attributes: { + 'sentry.op': { type: 'string', value: 'queue' }, + 'sentry.origin': { type: 'string', value: 'auto.queue.bullmq' }, + 'messaging.system': { type: 'string', value: 'bullmq' }, + 'messaging.destination.name': { type: 'string', value: 'test-failed-queue' }, + 'messaging.operation.name': { type: 'string', value: 'fail' }, + 'messaging.message.id': { type: 'string', value: expect.any(String) }, + 'messaging.bullmq.job.name': { type: 'string', value: 'failing-job' }, + 'sentry.environment': { type: 'string', value: 'production' }, + 'sentry.is_localhost': { type: 'boolean', value: false }, + 'sentry.release': { type: 'string', value: '1.0' }, + 'sentry.sdk.name': { type: 'string', value: EXPECTED_SDK_NAME }, + 'sentry.sdk.version': { type: 'string', value: SDK_VERSION }, + 'sentry.segment.id': { type: 'string', value: processSpan.span_id }, + 'sentry.segment.name': { type: 'string', value: 'process test-failed-queue' }, + 'sentry.trace_lifecycle': { type: 'string', value: 'stream' }, + }, + }, + ]); + expect(errorEvent?.exception?.values).toHaveLength(1); + expect(errorEvent?.exception?.values?.[0]?.type).toBe('Error'); + expect(errorEvent?.exception?.values?.[0]?.value).toBe('Job failed on purpose'); + expect(errorEvent?.exception?.values?.[0]?.mechanism).toEqual({ + type: 'auto.queue.bullmq', + handled: false, + }); + expect(errorEvent?.exception?.values?.[0]?.stacktrace).toEqual(expectedStacktrace); + expect(errorEvent?.contexts?.trace?.trace_id).toBe(processSpan.trace_id); + expect(errorEvent?.contexts?.trace?.span_id).toBe(processSpan.span_id); + }, + ); }, + { additionalDependencies }, ); }); diff --git a/packages/node/src/integrations/tracing/bullmq/span.ts b/packages/node/src/integrations/tracing/bullmq/span.ts index 10ec13bee4c2..6de52055b99e 100644 --- a/packages/node/src/integrations/tracing/bullmq/span.ts +++ b/packages/node/src/integrations/tracing/bullmq/span.ts @@ -142,6 +142,7 @@ export function toSentryAttributes(attributes: Record): export class SentryBullMQSpan implements TelemetrySpan { private _span: Span; private _scope: Scope; + private _recordedError: Error | undefined; public constructor(span: Span, scope: Scope) { this._span = span; @@ -164,11 +165,16 @@ export class SentryBullMQSpan implements TelemetrySpan { if (name === 'job failed') { this._span.setStatus({ code: SPAN_STATUS_ERROR, message: 'internal_error' }); - const reason = attributes?.['bullmq.job.failed.reason']; - const error = new Error(String(reason || 'Unknown error')); - // BullMQ passes only the error message (https://github.com/taskforcesh/bullmq/issues/4902), so this stack has only - // BullMQ and SDK frames, the same for every job. Without it, Sentry groups the errors by their message. - error.stack = undefined; + // Since 6.7.5, BullMQ passes the job error to `recordException` right before this event + // (https://github.com/taskforcesh/bullmq/pull/4915). Older versions pass only the error message. + let error = this._recordedError; + this._recordedError = undefined; + if (!error) { + error = new Error(String(attributes?.['bullmq.job.failed.reason'] || 'Unknown error')); + // This stack has only BullMQ and SDK frames, the same for every job. Without it, Sentry groups the errors by + // their message. + error.stack = undefined; + } captureException(error, { mechanism: { handled: false, @@ -180,13 +186,15 @@ export class SentryBullMQSpan implements TelemetrySpan { /** * Does not capture to Sentry. BullMQ calls this for Redis, lock and serialization errors that it - * retries itself, often several times for one fault. Job failures are captured in `addEvent`. + * retries itself, often several times for one fault, and for the job error right before the `job failed` event. + * Job failures are captured in `addEvent`. */ public recordException(exception: Error | string | { code?: number; message?: string; name?: string }): void { + // BullMQ passes the retried errors as strings, so only the job error is kept. + this._recordedError = exception instanceof Error ? exception : undefined; const error = - exception instanceof Error - ? exception - : new Error(typeof exception === 'string' ? exception : exception.message || 'Unknown error'); + this._recordedError ?? + new Error(typeof exception === 'string' ? exception : exception.message || 'Unknown error'); this._span.recordException(error); } diff --git a/packages/node/test/integrations/tracing/bullmq.test.ts b/packages/node/test/integrations/tracing/bullmq.test.ts index fc0e62d6ae5b..98cfd0097422 100644 --- a/packages/node/test/integrations/tracing/bullmq.test.ts +++ b/packages/node/test/integrations/tracing/bullmq.test.ts @@ -397,6 +397,32 @@ describe('SentryBullMQSpan', () => { }); }); + it('captures the job error that BullMQ recorded before the "job failed" event', () => { + const { span } = createSpan(); + const error = new TypeError('Cannot read properties of undefined'); + + span.recordException(error); + span.addEvent('job failed', { 'bullmq.job.failed.reason': error.message }); + + expect(SentryCore.captureException).toHaveBeenCalledTimes(1); + expect(SentryCore.captureException).toHaveBeenCalledWith( + expect.objectContaining({ name: 'TypeError', stack: error.stack }), + { mechanism: { handled: false, type: 'auto.queue.bullmq' } }, + ); + }); + + it('captures an error without a stack from the reason when BullMQ recorded only a string', () => { + const { span } = createSpan(); + + (span as any).recordException('Connection timed out'); + span.addEvent('job failed', { 'bullmq.job.failed.reason': 'Connection refused' }); + + expect(SentryCore.captureException).toHaveBeenCalledWith( + expect.objectContaining({ message: 'Connection refused', stack: undefined }), + { mechanism: { handled: false, type: 'auto.queue.bullmq' } }, + ); + }); + it('sets the error status on the span when the event name is "job failed"', () => { const { span, otelSpan } = createSpan();