BullMQ

@monocle.sh/instrumentation-bullmq traces job dispatch and processing for BullMQ queues, flows, and workers. Jobs then show up in the Jobs dashboard with throughput, failures, and queue delays.

Installation

npm install @monocle.sh/instrumentation-bullmq

Peer dependency: bullmq >= 5.0.0

With the AdonisJS agent

Install this integration separately and add bullmq() to config/monocle.ts. Installing the package alone does not activate it.

import { defineConfig } from '@monocle.sh/adonisjs-agent'
import { bullmq } from '@monocle.sh/instrumentation-bullmq/integration'
import env from '#start/env'

export default defineConfig({
  dsn: env.get('MONOCLE_DSN'),
  integrations: [bullmq()],
})

The integration inherits the agent's sampling callback. Pass sampling to bullmq() to override it for BullMQ only. Omit the integration or use bullmq({ enabled: false }) to disable it. See Job Sampling for sampling rules.

Without the AdonisJS agent

Create an instance of BullMQInstrumentation and call enable(). This must happen before you import or create any BullMQ Queue, Worker, or FlowProducer instances.

import { BullMQInstrumentation } from '@monocle.sh/instrumentation-bullmq'

const instrumentation = new BullMQInstrumentation()
instrumentation.enable()

Once enabled, all Queue.add, Queue.addBulk, FlowProducer.add, FlowProducer.addBulk, and Worker processing calls are automatically instrumented.

import { Queue, Worker } from 'bullmq'

const queue = new Queue('emails', { connection })

// This creates a PRODUCER span
await queue.add('welcome-email', { userId: '123' })

// This creates a CONSUMER span for each processed job
const worker = new Worker(
  'emails',
  async (job) => {
    await sendEmail(job.data.userId)
  },
  { connection },
)

Configuration

const instrumentation = new BullMQInstrumentation({
  // Create individual spans for each job in addBulk (default: true)
  emitCreateSpansForBulk: true,

  // Create individual spans for each job in FlowProducer.add (default: true)
  emitCreateSpansForFlow: true,

  // Only create producer spans when a parent span exists (default: false)
  requireParentSpanForPublish: false,

  // Use producer span as consumer parent instead of linking (default: false)
  useProducerSpanAsConsumerParent: false,
})

Span linking vs parenting

By default, consumer spans are linked to the producer span. This means they appear as separate traces connected by a link. Set useProducerSpanAsConsumerParent: true to make the consumer span a child of the producer, keeping them in the same trace.

Spans

Producer spans

Queue.add creates a PRODUCER span named Queue.add {queueName}.

Queue.addBulk creates an INTERNAL parent span named Queue.addBulk {queueName}, with individual PRODUCER child spans for each job (named Job.addJob {queueName}).

FlowProducer.add creates an INTERNAL parent span named FlowProducer.add {queueName}, with child spans for each job in the flow tree.

FlowProducer.addBulk creates an INTERNAL parent span named FlowProducer.addBulk, with child spans per flow.

Consumer spans

Worker processing creates a CONSUMER span named Worker.run {queueName} for each job processed. The span captures job metadata, queue time, and any errors thrown during processing.

Lifecycle events

Lock extensions, job removals, and retries are recorded as events on the active consumer span (not separate spans).

BullMQ-specific attributes

In addition to the standard job attributes, the BullMQ instrumentation emits these extra attributes: