diff --git a/README.md b/README.md index 6bc6462..b268054 100644 --- a/README.md +++ b/README.md @@ -1011,6 +1011,8 @@ renewal that fails is logged at the `warn` level. `@boringnode/queue` ships with built-in OpenTelemetry instrumentation that creates **PRODUCER** spans for job dispatch and **CONSUMER** spans for job execution, following [OTel messaging semantic conventions](https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/). +It requires OpenTelemetry JS SDK 3 (`@opentelemetry/api` 1.10, `@opentelemetry/core` 3, `@opentelemetry/instrumentation` 0.300+). Applications still on SDK 2 should stay on `@boringnode/queue` 0.9.x. + ### Quick Setup ```typescript @@ -1045,6 +1047,7 @@ The instrumentation uses standard [OTel messaging semantic conventions](https:// | `messaging.job.group_id` | Custom | Queue-specific group identifier | | `messaging.job.priority` | Custom | Queue-specific job priority | | `messaging.job.delay_ms` | Custom | Delay before the job becomes available | +| `messaging.job.dedup_outcome` | Custom | Outcome of `.dedup()` (e.g. `'skipped'`) | | `messaging.job.queue_time_ms` | Custom | Time spent waiting in queue before processing | ### Trace Context Propagation diff --git a/package.json b/package.json index 622fd4c..6cebe20 100644 --- a/package.json +++ b/package.json @@ -45,11 +45,11 @@ "@japa/expect-type": "^2.0.4", "@japa/file-system": "^3.0.0", "@japa/runner": "^5.4.0", - "@opentelemetry/api": "^1.9.1", - "@opentelemetry/context-async-hooks": "^2.12.0", - "@opentelemetry/core": "^2.12.0", - "@opentelemetry/instrumentation": "^0.223.0", - "@opentelemetry/sdk-trace-base": "^2.12.0", + "@opentelemetry/api": "^1.10.0", + "@opentelemetry/context-async-hooks": "^3.1.1", + "@opentelemetry/core": "^3.1.1", + "@opentelemetry/instrumentation": "^0.301.1", + "@opentelemetry/sdk-trace": "^3.1.1", "@poppinss/ts-exec": "^1.5.1", "@release-it/conventional-changelog": "^12.0.2", "@types/better-sqlite3": "^9.6.0", @@ -73,9 +73,9 @@ "typescript": "^6.0.3" }, "peerDependencies": { - "@opentelemetry/api": "^1.9.0", - "@opentelemetry/core": "^1.30.0 || ^2.0.0", - "@opentelemetry/instrumentation": ">=0.200.0 <0.300.0", + "@opentelemetry/api": "^1.10.0", + "@opentelemetry/core": "^3.0.0", + "@opentelemetry/instrumentation": ">=0.300.0 <0.400.0", "ioredis": "^5.0.0 || ^6.0.0", "knex": "^3.0.0", "kysely": "^0.29.0" diff --git a/src/job_dispatch_runtime.ts b/src/job_dispatch_runtime.ts index 33b5aa9..d00dc02 100644 --- a/src/job_dispatch_runtime.ts +++ b/src/job_dispatch_runtime.ts @@ -111,6 +111,7 @@ export class JobDispatchRuntime { if (result) { pushResult = result message.dedupOutcome = result.outcome + message.dedupJobId = result.jobId } }, message) diff --git a/src/otel.ts b/src/otel.ts index 78c891d..58e0291 100644 --- a/src/otel.ts +++ b/src/otel.ts @@ -15,6 +15,7 @@ import type { InstrumentationConfig } from '@opentelemetry/instrumentation' import { dispatchChannel, executeChannel } from './tracing_channels.js' import type { AcquiredJob } from './contracts/adapter.js' import type { JobDispatchMessage, JobExecuteMessage } from './types/tracing_channels.js' +import packageJson from '../package.json' with { type: 'json' } export interface QueueInstrumentationConfig extends InstrumentationConfig { /** @@ -53,7 +54,7 @@ export class QueueInstrumentation extends InstrumentationBase any } constructor(config: QueueInstrumentationConfig = {}) { - super('@boringnode/queue', '0.1.0', config) + super(packageJson.name, packageJson.version, config) } get #messagingSystem(): string { @@ -192,6 +193,11 @@ export class QueueInstrumentation extends InstrumentationBase(fn: () => Promise, job: AcquiredJob, queue: string): Promise { + // The queue manager keeps this wrapper after disable(); without the + // execute subscription, nothing would end the span. + if (!this.subscribed) return fn() + const extractedContext = this.#extractParentContext(job.traceContext) const parentSpanContext = trace.getSpanContext(extractedContext) diff --git a/src/types/tracing_channels.ts b/src/types/tracing_channels.ts index 77d2057..4df80cc 100644 --- a/src/types/tracing_channels.ts +++ b/src/types/tracing_channels.ts @@ -28,6 +28,12 @@ export type JobDispatchMessage = { */ dedupOutcome?: DedupOutcome + /** + * ID of the job that holds the deduplication key, set with `dedupOutcome`. + * It differs from `jobs[0].id` when the dispatch did not add a new job. + */ + dedupJobId?: string + /** Error that caused the dispatch to fail */ error?: Error } diff --git a/tests/helpers/setup_tracing.ts b/tests/helpers/setup_tracing.ts index 6677e52..da3f136 100644 --- a/tests/helpers/setup_tracing.ts +++ b/tests/helpers/setup_tracing.ts @@ -1,15 +1,11 @@ import { trace, context, propagation } from '@opentelemetry/api' -import { - BasicTracerProvider, - InMemorySpanExporter, - SimpleSpanProcessor, -} from '@opentelemetry/sdk-trace-base' +import { TracerProvider, InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace' import { AsyncLocalStorageContextManager } from '@opentelemetry/context-async-hooks' import { W3CTraceContextPropagator } from '@opentelemetry/core' let initialized = false let exporter: InMemorySpanExporter -let provider: BasicTracerProvider +let provider: TracerProvider export function setupTracing() { if (!initialized) { @@ -18,7 +14,7 @@ export function setupTracing() { context.setGlobalContextManager(contextManager.enable()) propagation.setGlobalPropagator(new W3CTraceContextPropagator()) - provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }) + provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] }) trace.setGlobalTracerProvider(provider) initialized = true diff --git a/tests/otel.spec.ts b/tests/otel.spec.ts index 078ef7c..1ed1847 100644 --- a/tests/otel.spec.ts +++ b/tests/otel.spec.ts @@ -6,6 +6,7 @@ import { dispatchChannel, executeChannel } from '../src/tracing_channels.js' import type { JobExecuteMessage, JobDispatchMessage } from '../src/types/tracing_channels.js' import type { AcquiredJob } from '../src/contracts/adapter.js' import type { JobData } from '../src/types/main.js' +import packageJson from '../package.json' with { type: 'json' } function makeJob(overrides: Partial = {}): AcquiredJob { return { @@ -113,6 +114,8 @@ test.group('QueueInstrumentation | dispatch via DC', (group) => { assert.equal(spans[0].attributes['messaging.destination.name'], 'emails') assert.equal(spans[0].attributes['messaging.message.id'], 'job-1') assert.equal(spans[0].attributes['messaging.job.name'], 'SendEmailJob') + assert.equal(spans[0].instrumentationScope.name, '@boringnode/queue') + assert.equal(spans[0].instrumentationScope.version, packageJson.version) instrumentation.disable() }) @@ -156,6 +159,23 @@ test.group('QueueInstrumentation | dispatch via DC', (group) => { instrumentation.disable() }) + test('reports the deduplicated job instead of the skipped one', async ({ assert }) => { + const { instrumentation } = await setupWithWrappers() + + const jobData: JobData = { id: 'new-id', name: 'DedupJob', payload: {}, attempts: 0 } + const message: JobDispatchMessage = { jobs: [jobData], queue: 'default' } + await dispatchChannel.tracePromise(async () => { + message.dedupOutcome = 'skipped' + message.dedupJobId = 'existing-id' + }, message) + + const spans = getFinishedSpans() + assert.equal(spans[0].attributes['messaging.message.id'], 'existing-id') + assert.equal(spans[0].attributes['messaging.job.dedup_outcome'], 'skipped') + + instrumentation.disable() + }) + test('includes delay_ms and batch_count attributes', async ({ assert }) => { const { instrumentation } = await setupWithWrappers() @@ -211,6 +231,29 @@ test.group('QueueInstrumentation | execute via executionWrapper', (group) => { }) group.each.setup(() => resetSpans()) + test('does not start a CONSUMER span after disable()', async ({ assert }) => { + const { instrumentation, executionWrapper } = await setupWithWrappers() + instrumentation.disable() + + const job = makeJob({ id: 'disabled-1' }) + const message: JobExecuteMessage = { job, queue: 'default', status: 'completed' } + let activeSpan: unknown + const result = await executeChannel.tracePromise(async () => { + return executionWrapper( + async () => { + activeSpan = trace.getActiveSpan() + return 'done' + }, + job, + 'default' + ) + }, message) + + assert.equal(result, 'done') + assert.isUndefined(activeSpan) + assert.lengthOf(getFinishedSpans(), 0) + }) + test('creates CONSUMER span with semconv attributes', async ({ assert }) => { const { instrumentation, executionWrapper } = await setupWithWrappers() diff --git a/yarn.lock b/yarn.lock index b693bdf..5d05224 100644 --- a/yarn.lock +++ b/yarn.lock @@ -47,11 +47,11 @@ __metadata: "@japa/file-system": "npm:^3.0.0" "@japa/runner": "npm:^5.4.0" "@lukeed/ms": "npm:^2.0.2" - "@opentelemetry/api": "npm:^1.9.1" - "@opentelemetry/context-async-hooks": "npm:^2.12.0" - "@opentelemetry/core": "npm:^2.12.0" - "@opentelemetry/instrumentation": "npm:^0.223.0" - "@opentelemetry/sdk-trace-base": "npm:^2.12.0" + "@opentelemetry/api": "npm:^1.10.0" + "@opentelemetry/context-async-hooks": "npm:^3.1.1" + "@opentelemetry/core": "npm:^3.1.1" + "@opentelemetry/instrumentation": "npm:^0.301.1" + "@opentelemetry/sdk-trace": "npm:^3.1.1" "@poppinss/ts-exec": "npm:^1.5.1" "@poppinss/utils": "npm:^7.1.2" "@release-it/conventional-changelog": "npm:^12.0.2" @@ -76,9 +76,9 @@ __metadata: tsup: "npm:^9.0.0" typescript: "npm:^6.0.3" peerDependencies: - "@opentelemetry/api": ^1.9.0 - "@opentelemetry/core": ^1.30.0 || ^2.0.0 - "@opentelemetry/instrumentation": ">=0.200.0 <0.300.0" + "@opentelemetry/api": ^1.10.0 + "@opentelemetry/core": ^3.0.0 + "@opentelemetry/instrumentation": ">=0.300.0 <0.400.0" ioredis: ^5.0.0 || ^6.0.0 knex: ^3.0.0 kysely: ^0.29.0 @@ -927,98 +927,67 @@ __metadata: languageName: node linkType: hard -"@opentelemetry/api-logs@npm:0.223.0": - version: 0.223.0 - resolution: "@opentelemetry/api-logs@npm:0.223.0" - dependencies: - "@opentelemetry/api": "npm:^1.3.0" - checksum: 10c0/3be606243df08c7356d584d48194464050f3c379d63a92cc059de4f45dd5e81bbd7810f0a95419bed71fe4b790d2e495663eaf563f2e67bde30b381182add32e +"@opentelemetry/api@npm:^1.10.0": + version: 1.10.0 + resolution: "@opentelemetry/api@npm:1.10.0" + checksum: 10c0/c666f89d882e2e3ea3780b4e419de3ba7e00a80bf334269c3c8507a5e2b4e453e2bc8368f6230f894d789b206c14dde313ea6f51fa7bc629f0eff8480b5d356b languageName: node linkType: hard -"@opentelemetry/api@npm:^1.3.0": - version: 1.9.0 - resolution: "@opentelemetry/api@npm:1.9.0" - checksum: 10c0/9aae2fe6e8a3a3eeb6c1fdef78e1939cf05a0f37f8a4fae4d6bf2e09eb1e06f966ece85805626e01ba5fab48072b94f19b835449e58b6d26720ee19a58298add - languageName: node - linkType: hard - -"@opentelemetry/api@npm:^1.9.1": - version: 1.9.1 - resolution: "@opentelemetry/api@npm:1.9.1" - checksum: 10c0/c608485fc8b5a91e1f7e05e843b45b509307456b31cd2ad365933d90813e40ebfedf179f1451c762037e82d7c76aa8500e95d2da3609f640a1206cde5322cd14 - languageName: node - linkType: hard - -"@opentelemetry/context-async-hooks@npm:^2.12.0": - version: 2.12.0 - resolution: "@opentelemetry/context-async-hooks@npm:2.12.0" +"@opentelemetry/context-async-hooks@npm:^3.1.1": + version: 3.1.1 + resolution: "@opentelemetry/context-async-hooks@npm:3.1.1" peerDependencies: - "@opentelemetry/api": ">=1.0.0 <1.10.0" - checksum: 10c0/fc4185cc643562527532cc5de985edc963a02a68aa1cf6ec517da5602cb51620ea9ffd2201f0fb75295eb9eb7d281a517598a081eeda79a905245a110f21282c + "@opentelemetry/api": ">=1.0.0 <1.11.0" + checksum: 10c0/70d26beee812852b6edc51282fe8053ca7190157987ccf0487020f53984ecc33ddda2561ae21d9ed02a966316f7c7c98d9fa901de65cc138abd89d9177fdf16f languageName: node linkType: hard -"@opentelemetry/core@npm:2.12.0, @opentelemetry/core@npm:^2.12.0": - version: 2.12.0 - resolution: "@opentelemetry/core@npm:2.12.0" +"@opentelemetry/core@npm:3.1.1, @opentelemetry/core@npm:^3.1.1": + version: 3.1.1 + resolution: "@opentelemetry/core@npm:3.1.1" dependencies: "@opentelemetry/semantic-conventions": "npm:^1.29.0" peerDependencies: - "@opentelemetry/api": ">=1.0.0 <1.10.0" - checksum: 10c0/e0d0338d314ca827b4395acf8c7eb8f88a58fd63f1effc6068ba4dd8e3443e02e5796333ae6723fb926a7e6c18dd6d2102f5cc7e531b185618562dadbf6cad71 + "@opentelemetry/api": ">=1.10.0 <1.11.0" + checksum: 10c0/4776fb7f6dd08e96c1069f8b08e329ae854353ab617759539ca4b5bf925a1cce77b7c6f744f14565e75297cb73117a111c5d96b6489e64e789502032f041fceb languageName: node linkType: hard -"@opentelemetry/instrumentation@npm:^0.223.0": - version: 0.223.0 - resolution: "@opentelemetry/instrumentation@npm:0.223.0" +"@opentelemetry/instrumentation@npm:^0.301.1": + version: 0.301.1 + resolution: "@opentelemetry/instrumentation@npm:0.301.1" dependencies: - "@opentelemetry/api-logs": "npm:0.223.0" import-in-the-middle: "npm:^3.0.0" require-in-the-middle: "npm:^8.0.0" peerDependencies: - "@opentelemetry/api": ^1.3.0 - checksum: 10c0/91d376a6625ed005655b27492922068f0d2fbe04372002407a6498d96e953b0b4669a97b9946cc49286a28082195b6db849ad35583b9772bc3a359b4e672756f + "@opentelemetry/api": ">=1.10.0 <1.11.0" + checksum: 10c0/3fa4a7a3fc8b77addcdf7374372175d46219aeba390bcd560f541920d52cbedda63a2d2cc85a909a97e191ced90212c6a10295eb2b921635cff3937bd28d2429 languageName: node linkType: hard -"@opentelemetry/resources@npm:2.12.0": - version: 2.12.0 - resolution: "@opentelemetry/resources@npm:2.12.0" - dependencies: - "@opentelemetry/core": "npm:2.12.0" - "@opentelemetry/semantic-conventions": "npm:^1.29.0" - peerDependencies: - "@opentelemetry/api": ">=1.3.0 <1.10.0" - checksum: 10c0/ef3f8470b9ada8406fa4e4a0f002ca31537c1b7fe61f06102fb9023c97a56b0d978c687dd392539f1e005c20f0467f6a1c66f9c43e8d31c2900023164529f2bc - languageName: node - linkType: hard - -"@opentelemetry/sdk-trace-base@npm:^2.12.0": - version: 2.12.0 - resolution: "@opentelemetry/sdk-trace-base@npm:2.12.0" +"@opentelemetry/resources@npm:3.1.1": + version: 3.1.1 + resolution: "@opentelemetry/resources@npm:3.1.1" dependencies: - "@opentelemetry/core": "npm:2.12.0" - "@opentelemetry/resources": "npm:2.12.0" - "@opentelemetry/sdk-trace": "npm:2.12.0" + "@opentelemetry/core": "npm:3.1.1" "@opentelemetry/semantic-conventions": "npm:^1.29.0" peerDependencies: - "@opentelemetry/api": ">=1.3.0 <1.10.0" - checksum: 10c0/cd1de7e881f60a98047d662915b74b7b39cde69a51e9fc146fb4f02af4df4441a785af57e7ad267ea22bbc6ac19c5497684b51d7407573c2aef51f9e402aea4b + "@opentelemetry/api": ">=1.10.0 <1.11.0" + checksum: 10c0/4ff4e28966b7ae33cae40e7d52b502277dd8391ecd985ee5479aff9855e7e08589018f571dabf8699ba0eb2e5b2ac55877059f3d45571d851658f7bf8dc4f53e languageName: node linkType: hard -"@opentelemetry/sdk-trace@npm:2.12.0": - version: 2.12.0 - resolution: "@opentelemetry/sdk-trace@npm:2.12.0" +"@opentelemetry/sdk-trace@npm:^3.1.1": + version: 3.1.1 + resolution: "@opentelemetry/sdk-trace@npm:3.1.1" dependencies: - "@opentelemetry/core": "npm:2.12.0" - "@opentelemetry/resources": "npm:2.12.0" + "@opentelemetry/core": "npm:3.1.1" + "@opentelemetry/resources": "npm:3.1.1" "@opentelemetry/semantic-conventions": "npm:^1.29.0" peerDependencies: - "@opentelemetry/api": ">=1.3.0 <1.10.0" - checksum: 10c0/bc36d6c3c0a007e0e1f928a7e3c9e6e6d130e157efab8da36a07a624000c2fd25309c0a3d42d4e58f6b30ae233cddfbc6270c2a8566c9f611b9d964f5b7ea8a6 + "@opentelemetry/api": ">=1.10.0 <1.11.0" + checksum: 10c0/9373d5fe28fb8fe8d40d4b37a67033b9976d8c1d732fb05885250ec716df1e47e2e0285ee4fdfe4c587a13d5c50af192d197a0ffb7ecdfa96905899d804ae6e1 languageName: node linkType: hard