Telemetry integration for BullMQ.

BullMQ emits various telemetry events that you can attach handlers to for metrics, logging, and monitoring.

Event Names

All events are prefixed with [:bullmq, ...].

Job Events

• [:bullmq, :job, :add] - A job was added to the queue

  • Measurements: %{queue_time: native_time} (time to add to Redis)
  • Metadata: %{queue: name, job_id: id, job_name: name}

• [:bullmq, :job, :start] - A job started processing

  • Measurements: %{system_time: native_time}
  • Metadata: %{queue: name, job_id: id, job_name: name, worker: pid}

• [:bullmq, :job, :complete] - A job completed successfully

  • Measurements: %{duration: native_time}
  • Metadata: %{queue: name, job_id: id, job_name: name, worker: pid}

• [:bullmq, :job, :fail] - A job failed

  • Measurements: %{duration: native_time}
  • Metadata: %{queue: name, job_id: id, job_name: name, worker: pid, error: term}

• [:bullmq, :job, :retry] - A job is being retried

  • Measurements: %{attempt: integer, delay: ms}
  • Metadata: %{queue: name, job_id: id, job_name: name}

• [:bullmq, :job, :progress] - Job progress updated

  • Measurements: %{progress: 0..100}
  • Metadata: %{queue: name, job_id: id}

Worker Events

• [:bullmq, :worker, :start] - Worker started

  • Measurements: %{concurrency: integer}
  • Metadata: %{queue: name, worker: pid}

• [:bullmq, :worker, :stop] - Worker stopped

  • Measurements: %{uptime: native_time}
  • Metadata: %{queue: name, worker: pid}

• [:bullmq, :worker, :stalled_check] - Stalled job check executed

  • Measurements: %{recovered: integer, failed: integer}
  • Metadata: %{queue: name}

Queue Events

• [:bullmq, :queue, :pause] - Queue was paused
• [:bullmq, :queue, :resume] - Queue was resumed
• [:bullmq, :queue, :drain] - Queue was drained

Rate Limiting Events

• [:bullmq, :rate_limit, :hit] - Rate limit was hit
  • Measurements: %{delay: ms}
  • Metadata: %{queue: name}

Example Setup

# In your application.ex
:telemetry.attach_many(
  "bullmq-logger",
  [
    [:bullmq, :job, :complete],
    [:bullmq, :job, :fail],
    [:bullmq, :rate_limit, :hit]
  ],
  &MyApp.Telemetry.handle_event/4,
  nil
)

Prometheus/StatsD Integration

defmodule MyApp.Telemetry do
  def handle_event([:bullmq, :job, :complete], measurements, metadata, _config) do
    :prometheus_histogram.observe(
      :job_duration_seconds,
      [metadata.queue],
      measurements.duration / 1_000_000_000
    )
  end

  def handle_event([:bullmq, :job, :fail], _measurements, metadata, _config) do
    :prometheus_counter.inc(:job_failures_total, [metadata.queue])
  end
end