fix: give each BullMQ stalled check a trace of its own (v0.1.3)

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
 

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",

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<void> {
+  let loop: Promise<void> = 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);
+  });
+});

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,
+  };
+}

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);