BullMQ

Trace BullMQ 5 queues, flows and workers with @unhandled/instrumentation-bullmq, including job sampling through the AdonisJS agent.

@unhandled/instrumentation-bullmq traces BullMQ 5: jobs added through queues and flow producers, and jobs run by workers. Dispatch and execution spans feed the jobs views. It needs bullmq 5.0.0 or later.

Install

Terminal

npm install @unhandled/instrumentation-bullmq

Register with the AdonisJS agent

Add the integration to config/unhandled.ts. Nothing is traced until it is listed.

import env from '#start/env'
import { defineConfig, Sample } from '@unhandled/agent-adonisjs'
import { bullmq } from '@unhandled/instrumentation-bullmq/integration'

export default defineConfig({
  dsn: env.get('UNHANDLED_DSN'),
  integrations: [
    bullmq({
      useProducerSpanAsConsumerParent: true,
      sampling: (job) => (job.name === 'SyncInventory' ? Sample.rate(0.1) : Sample.default()),
    }),
  ],
})

By default the integration uses the agent's sampling callback for every worker execution. bullmq({ sampling }) uses a different callback for BullMQ jobs only, and bullmq({ enabled: false }) turns the integration off. The callback runs before the worker body, so a dropped job creates no spans at all. See job sampling for the rules.

Register standalone

Without the agent, create and enable the instrumentation before bullmq is imported:

import { BullMQInstrumentation } from '@unhandled/instrumentation-bullmq'

new BullMQInstrumentation({ requireParentSpanForPublish: true }).enable()

Both require and import are hooked; ESM needs the import-in-the-middle loader registered before bullmq loads. When a bundler or load order defeats the hooks, patch the module yourself with instrumentation.patch(bullmq) and undo it with instrumentation.unpatch(bullmq). Standalone use does no job sampling.

Options

Option	Default	Effect
`emitCreateSpansForBulk`	`true`	One `Job.addJob` span per job in `Queue.addBulk`.
`emitCreateSpansForFlow`	`true`	One `Job.addJob` span per job in `FlowProducer.add`, and per flow in `FlowProducer.addBulk`.
`requireParentSpanForPublish`	`false`	Producer spans are created only when a span is already active.
`useProducerSpanAsConsumerParent`	`false`	The worker span joins the producer's trace as its child instead of starting a new trace with a link.
`enabled`	`true`	Integration only. `false` registers nothing.
`sampling`	agent callback	Integration only. Sampling callback for BullMQ jobs.

Spans

Operation	Kind	Name
`Queue.add`	PRODUCER	`Queue.add {queue}`
`Queue.addBulk`	INTERNAL	`Queue.addBulk {queue}`
Each job of a bulk or flow	PRODUCER	`Job.addJob {queue}`
`FlowProducer.add`	INTERNAL	`FlowProducer.add {queue}`
`FlowProducer.addBulk`	INTERNAL	`FlowProducer.addBulk`
Worker job execution	CONSUMER	`Worker.run {queue}`

The producer's trace context is written to the job's opts.telemetry.metadata, the field BullMQ itself uses for propagation. Jobs added with telemetry.omitContext: true carry no context. By default the worker span starts a new trace with a link to the producer span; with useProducerSpanAsConsumerParent it becomes the producer span's child.

A failed execution records the exception and sets ERROR status. The rescheduling errors DelayedError, WaitingError, WaitingChildrenError and RateLimitError are not failures.

Attributes

Every span has messaging.system=bullmq and messaging.operation.type: send on Queue.add and the bulk and flow parents, create on Job.addJob, process on Worker.run. All spans except FlowProducer.addBulk carry messaging.destination.name.

Attribute	Spans
`messaging.job.name`, `messaging.bullmq.job.name`	Single-job spans
`messaging.message.id`	Single-job spans
`messaging.bullmq.job.timestamp`, `.delay`, `.attempts`, `.repeatJobKey`	Single-job spans
`messaging.batch.message_count`, `messaging.bullmq.job.bulk.names`	Bulk parents
`messaging.message.retry.count`, `messaging.job.queue_time_ms`	`Worker.run`
`messaging.bullmq.job.processedOn`, `.finishedOn`, `.failedReason`	`Worker.run`; `failedReason` only after a failure
`messaging.bullmq.worker.concurrency`, `.lockDuration`, `.lockRenewTime`, `.rateLimiter.max`, `.rateLimiter.duration`	`Worker.run`, when set on the worker

messaging.job.queue_time_ms is processedOn - timestamp - delay, so a delayed job's wait does not count as queue time.

Events on Worker.run

Event	Attributes
`bullmq.job.lock_extended`	`messaging.message.id`, `messaging.bullmq.lock.duration`
`bullmq.job.removed`	`messaging.message.id`
`bullmq.job.retry`	`messaging.message.id`; for automatic retries also `messaging.message.retry.count` and `messaging.bullmq.retry.delay`

The attribute and event names are exported as BULLMQ_ATTR and BULLMQ_EVENTS.