From beed3925eca0606293fa053297ab3465720655d0 Mon Sep 17 00:00:00 2001 From: Victor Carvalho Date: Tue, 20 Oct 2026 22:11:28 -0300 Subject: [PATCH] fix: give each BullMQ stalled check a trace of its own (v0.1.3) BullMQ's Worker starts its stalled-check loop inside the startStalledCheckTimer span, so every moveStalledJobsToWait (every 30 s per queue) and its redis call joined that one trace for the life of the process. In Backbone production each such trace reached Tempo's 5 MB per-trace limit after about a day and a half, and Tempo then refused its spans (trace_too_large). The BullMQ adapter now starts moveStalledJobsToWait on the root context unless BullMQ passes an explicit parent. Every other operation, and a job's propagated parent, is unchanged. Co-Authored-By: Claude Opus 5.5 --- README.md | 4 +- dist/bullmq-loops.d.ts | 6 ++ dist/bullmq-loops.js | 25 ++++++++ dist/sdk-boot.js | 3 +- package.json | 2 +- src/__tests__/bullmq-loops.test.ts | 97 ++++++++++++++++++++++++++++++ src/bullmq-loops.ts | 30 +++++++++ src/sdk-boot.ts | 3 +- 8 files changed, 166 insertions(+), 4 deletions(-) create mode 100644 dist/bullmq-loops.d.ts create mode 100644 dist/bullmq-loops.js create mode 100644 src/__tests__/bullmq-loops.test.ts create mode 100644 src/bullmq-loops.ts diff --git a/README.md b/README.md index becddb1..51ff8eb 100644 --- a/README.md +++ b/README.md @@ -90,7 +90,9 @@ Environment: `OTEL_EXPORTER_OTLP_ENDPOINT` (on/off), `OTEL_SDK_DISABLED=true` (k `bullmq` (`^6.3.8`) is an optional peer dependency, used only by the `./bullmq` entry, so an app without it typechecks the main entry even with `skipLibCheck` off. `bullmqTelemetry()` returns -the adapter once telemetry has started, and `undefined` before. +the adapter once telemetry has started, and `undefined` before. Each `moveStalledJobsToWait` (a +Worker's stalled check, every 30 s per queue) starts a trace of its own; BullMQ would otherwise +nest every check for the life of the process under one `startStalledCheckTimer` trace. ## Logging diff --git a/dist/bullmq-loops.d.ts b/dist/bullmq-loops.d.ts new file mode 100644 index 0000000..d575fe0 --- /dev/null +++ b/dist/bullmq-loops.d.ts @@ -0,0 +1,6 @@ +import type { Telemetry } from "bullmq"; +/** + * Wraps a BullMQ adapter so each iteration of a lifelong loop is a trace of its own. A span given + * an explicit context (a job's propagated parent) keeps it. + */ +export declare function rootLoopSpans(telemetry: Telemetry): Telemetry; diff --git a/dist/bullmq-loops.js b/dist/bullmq-loops.js new file mode 100644 index 0000000..82c065c --- /dev/null +++ b/dist/bullmq-loops.js @@ -0,0 +1,25 @@ +import { ROOT_CONTEXT } from "@opentelemetry/api"; +/** + * BullMQ operations a Worker repeats for its whole life. `startStalledCheckTimer` starts the + * stalled-check loop inside its own span and returns, so without this every + * `moveStalledJobsToWait` (every 30 s, per queue) would join that one trace, which would grow + * until Tempo refuses it (`trace_too_large`). + */ +const LOOP_OPERATIONS = ["moveStalledJobsToWait"]; +function isLoopSpan(name) { + return LOOP_OPERATIONS.some((op) => name === op || name.startsWith(`${op} `)); +} +/** + * Wraps a BullMQ adapter so each iteration of a lifelong loop is a trace of its own. A span given + * an explicit context (a job's propagated parent) keeps it. + */ +export function rootLoopSpans(telemetry) { + const { tracer } = telemetry; + return { + tracer: { + startSpan: (name, options, context) => tracer.startSpan(name, options, context ?? (isLoopSpan(name) ? ROOT_CONTEXT : undefined)), + }, + contextManager: telemetry.contextManager, + meter: telemetry.meter, + }; +} diff --git a/dist/sdk-boot.js b/dist/sdk-boot.js index b522e03..fc88b85 100644 --- a/dist/sdk-boot.js +++ b/dist/sdk-boot.js @@ -11,6 +11,7 @@ import { logs } from "@opentelemetry/api-logs"; import { resourceFromAttributes } from "@opentelemetry/resources"; import { ATTR_SERVICE_INSTANCE_ID, ATTR_SERVICE_NAME, ATTR_SERVICE_VERSION, } from "@opentelemetry/semantic-conventions"; import { BullMQOtel } from "bullmq-otel"; +import { rootLoopSpans } from "./bullmq-loops.js"; import { buildInstrumentations } from "./instrumentations.js"; import { metricViews } from "./metric-views.js"; import { RedactingSpanExporter } from "./redacting-exporter.js"; @@ -74,7 +75,7 @@ export async function bootSdk(config, options) { // After the records the sink has already queued for emission. return Promise.resolve().then(() => sdk.shutdown()); }, - bullmq: new BullMQOtel({ tracerName: logScope }), + bullmq: rootLoopSpans(new BullMQOtel({ tracerName: logScope })), }; if (exportLogs) setLogLineSink(logLineSink(logs.getLogger(logScope))); diff --git a/package.json b/package.json index a7e7453..3dfc03e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@codelittinc/otel-node", - "version": "0.1.2", + "version": "0.1.3", "description": "In-process OpenTelemetry for Node services: SDK boot, layer-1 PII scrub, OTLP logs.", "license": "UNLICENSED", "repository": "github:codelittinc/otel-node", diff --git a/src/__tests__/bullmq-loops.test.ts b/src/__tests__/bullmq-loops.test.ts new file mode 100644 index 0000000..b5a0e34 --- /dev/null +++ b/src/__tests__/bullmq-loops.test.ts @@ -0,0 +1,97 @@ +import { afterAll, beforeEach, describe, expect, it } from "vitest"; +import { context, propagation, trace as otelTrace } from "@opentelemetry/api"; +import { AsyncLocalStorageContextManager } from "@opentelemetry/context-async-hooks"; +import { W3CTraceContextPropagator } from "@opentelemetry/core"; +import { + BasicTracerProvider, + InMemorySpanExporter, + SimpleSpanProcessor, +} from "@opentelemetry/sdk-trace-base"; +import { SpanKind, trace } from "bullmq"; +import { BullMQOtel } from "bullmq-otel"; +import { rootLoopSpans } from "../bullmq-loops.js"; + +const exporter = new InMemorySpanExporter(); +const provider = new BasicTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }); +context.setGlobalContextManager(new AsyncLocalStorageContextManager().enable()); +otelTrace.setGlobalTracerProvider(provider); +// NodeSDK installs this in an app; BullMQ carries a job's parent through it. +propagation.setGlobalPropagator(new W3CTraceContextPropagator()); +const telemetry = rootLoopSpans(new BullMQOtel({ tracerName: "bullmq-loops-test" })); + +afterAll(async () => { + await provider.shutdown(); + otelTrace.disable(); + propagation.disable(); + context.disable(); +}); + +beforeEach(() => exporter.reset()); + +const span = (name: string) => exporter.getFinishedSpans().filter((s) => s.name === name); + +/** BullMQ 6's Worker.startStalledCheckTimer: the loop starts inside the span and outlives it. */ +async function stalledCheckTimer(queue: string, iterations: number): Promise { + let loop: Promise = Promise.resolve(); + await trace(telemetry, SpanKind.INTERNAL, queue, "startStalledCheckTimer", queue, async () => { + loop = (async () => { + for (let i = 0; i < iterations; i++) { + await new Promise((resolve) => setTimeout(resolve, 1)); + await trace(telemetry, SpanKind.INTERNAL, queue, "moveStalledJobsToWait", queue, async () => + otelTrace.getTracer("redis").startSpan("evalsha").end(), + ); + } + })(); + }); + await loop; +} + +describe("rootLoopSpans", () => { + // WHY: Backbone's worker grew one trace per queue for its whole uptime, every check a child of + // startStalledCheckTimer, until Tempo refused it (trace_too_large) a day and a half in. + it("gives each stalled check a trace of its own, with its redis call inside it", async () => { + await stalledCheckTimer("q", 3); + const [timer] = span("startStalledCheckTimer q"); + const checks = span("moveStalledJobsToWait q"); + const traceIds = new Set(checks.map((s) => s.spanContext().traceId)); + + expect(checks).toHaveLength(3); + expect(traceIds.size).toBe(3); + expect(traceIds.has(timer!.spanContext().traceId)).toBe(false); + for (const check of checks) expect(check.parentSpanContext).toBeUndefined(); + for (const call of span("evalsha")) { + expect(traceIds.has(call.spanContext().traceId)).toBe(true); + expect(call.parentSpanContext?.spanId).toBeDefined(); + } + }); + + it("leaves every other operation under the active span", async () => { + const parent = otelTrace.getTracer("app").startSpan("request"); + await context.with(otelTrace.setSpan(context.active(), parent), () => + trace(telemetry, SpanKind.PRODUCER, "q", "add", "q.job", async () => undefined), + ); + parent.end(); + const [add] = span("add q.job"); + expect(add!.spanContext().traceId).toBe(parent.spanContext().traceId); + expect(add!.parentSpanContext?.spanId).toBe(parent.spanContext().spanId); + }); + + it("keeps a job's propagated parent", async () => { + let metadata = ""; + await trace(telemetry, SpanKind.PRODUCER, "q", "add", "q.job", async (_span, dst) => { + metadata = dst!; + }); + await trace( + telemetry, + SpanKind.CONSUMER, + "q", + "process", + "q", + async () => undefined, + metadata, + ); + const [add] = span("add q.job"); + const [processed] = span("process q"); + expect(processed!.spanContext().traceId).toBe(add!.spanContext().traceId); + }); +}); diff --git a/src/bullmq-loops.ts b/src/bullmq-loops.ts new file mode 100644 index 0000000..5f75a90 --- /dev/null +++ b/src/bullmq-loops.ts @@ -0,0 +1,30 @@ +import { ROOT_CONTEXT } from "@opentelemetry/api"; +import type { Telemetry } from "bullmq"; + +/** + * BullMQ operations a Worker repeats for its whole life. `startStalledCheckTimer` starts the + * stalled-check loop inside its own span and returns, so without this every + * `moveStalledJobsToWait` (every 30 s, per queue) would join that one trace, which would grow + * until Tempo refuses it (`trace_too_large`). + */ +const LOOP_OPERATIONS = ["moveStalledJobsToWait"]; + +function isLoopSpan(name: string): boolean { + return LOOP_OPERATIONS.some((op) => name === op || name.startsWith(`${op} `)); +} + +/** + * Wraps a BullMQ adapter so each iteration of a lifelong loop is a trace of its own. A span given + * an explicit context (a job's propagated parent) keeps it. + */ +export function rootLoopSpans(telemetry: Telemetry): Telemetry { + const { tracer } = telemetry; + return { + tracer: { + startSpan: (name, options, context) => + tracer.startSpan(name, options, context ?? (isLoopSpan(name) ? ROOT_CONTEXT : undefined)), + }, + contextManager: telemetry.contextManager, + meter: telemetry.meter, + }; +} diff --git a/src/sdk-boot.ts b/src/sdk-boot.ts index 28db98c..83d5297 100644 --- a/src/sdk-boot.ts +++ b/src/sdk-boot.ts @@ -18,6 +18,7 @@ import { BullMQOtel } from "bullmq-otel"; import type { Telemetry } from "bullmq"; import type { Instrumentation } from "@opentelemetry/instrumentation"; import type { TelemetryConfig } from "./config.js"; +import { rootLoopSpans } from "./bullmq-loops.js"; import { buildInstrumentations } from "./instrumentations.js"; import { metricViews } from "./metric-views.js"; import type { MetricViewOptions } from "./options.js"; @@ -101,7 +102,7 @@ export async function bootSdk(config: TelemetryConfig, options: BootOptions): Pr // After the records the sink has already queued for emission. return Promise.resolve().then(() => sdk.shutdown()); }, - bullmq: new BullMQOtel({ tracerName: logScope }), + bullmq: rootLoopSpans(new BullMQOtel({ tracerName: logScope })), }; if (exportLogs) setLogLineSink(logLineSink(logs.getLogger(logScope))); if (hook) await acknowledgedWithin(hook.waitForAllMessagesAcknowledged, HOOK_ACK_TIMEOUT_MS);