BullMQ Programmatic Observability: OpenTelemetry, Prometheus Metrics, and Infrastructure-as-Code Monitoring

Most BullMQ monitoring guides stop at "enable metrics middleware and add a Grafana dashboard." But production queue systems demand more: distributed traces that follow a job from producer through Redis into the worker, custom Prometheus metrics with job-level granularity, dashboards defined as version-controlled JSON, and alerting rules that detect real incidents without noise.

This post walks through five concrete, deployable pieces of infrastructure that transform your BullMQ deployment from monitored to truly observable. Each section includes complete TypeScript and YAML examples you can adapt to your own stack.

Section 1: End-to-End Distributed Tracing with OpenTelemetry

Standard queue metrics tell you how many jobs are slow. OpenTelemetry traces tell you which job is slow and why. With distributed tracing, a single trace can span from an HTTP request handler that enqueues a job, through Redis, into the worker, and across every sub-operation inside the processor — database query, API call, file write — all in one waterfall view.

Setting Up the OpenTelemetry SDK

Start by initializing the Node.js OpenTelemetry SDK with the BullMQ instrumentation package. This file must be imported before any other module:

// tracing.ts — MUST be imported before any other module
import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http';
import { PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics';
import { Resource } from '@opentelemetry/resources';
import { SEMRESATTRS_SERVICE_NAME } from '@opentelemetry/semantic-conventions';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { BullMQInstrumentation } from '@appsignal/opentelemetry-instrumentation-bullmq';

const resource = new Resource({
  [SEMRESATTRS_SERVICE_NAME]: 'bullmq-platform',
  'service.version': '1.0.0',
  'deployment.environment': process.env.NODE_ENV ?? 'production',
});

const traceExporter = new OTLPTraceExporter({
  url: process.env.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT
    ?? 'http://localhost:4318/v1/traces',
});

const metricExporter = new OTLPMetricExporter({
  url: process.env.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT
    ?? 'http://localhost:4318/v1/metrics',
});

const sdk = new NodeSDK({
  resource,
  traceExporter,
  metricReader: new PeriodicExportingMetricReader({
    exporter: metricExporter,
    exportIntervalMillis: 10_000,
  }),
  instrumentations: [
    getNodeAutoInstrumentations(),
    new BullMQInstrumentation({
      useProducerSpanAsConsumerParent: true,
      emitCreateSpansForBulk: true,
    }),
  ],
});

sdk.start();

process.on('SIGTERM', () => sdk.shutdown().catch(console.error));
process.on('SIGINT', () => sdk.shutdown().catch(console.error));

Two configuration options are particularly important:

• useProducerSpanAsConsumerParent: true — keeps the produce-span and consume-span in the same trace. When false, the consumer starts a new trace with a link, which breaks waterfall views in Jaeger or Grafana Tempo.
• emitCreateSpansForBulk: true — generates individual create spans for each job in an addBulk() call, enabling per-job creation tracing.

Enabling BullMQ's Built-in OpenTelemetry Metrics

BullMQ ships its own OTel metric instrumentations via the bullmq-otel module. Wire it alongside the trace instrumentation:

import { MeterProvider, PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics';
import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http';
import { metrics } from '@opentelemetry/api';

const meterProvider = new MeterProvider({
  readers: [
    new PeriodicExportingMetricReader({
      exporter: new OTLPMetricExporter({
        url: 'http://localhost:4318/v1/metrics',
      }),
      exportIntervalMillis: 10_000,
    }),
  ],
});

metrics.setGlobalMeterProvider(meterProvider);

// Now create BullMQ instances with telemetry enabled
import { BullMQOtel } from 'bullmq-otel';
import { Queue, Worker } from 'bullmq';
import IORedis from 'ioredis';

const connection = new IORedis({
  host: 'localhost',
  port: 6379,
  maxRetriesPerRequest: null,
});

const telemetry = new BullMQOtel({
  tracerName: 'queue-platform',
  meterName: 'queue-platform',
  version: '1.0.0',
  enableMetrics: true,
});

const queue = new Queue('email-notifications', { connection, telemetry });
const worker = new Worker('email-notifications', async (job) => {
  // job processing logic — child spans auto-propagate
  const currentSpan = opentelemetry.trace.getActiveSpan();
  currentSpan?.setAttribute('job.id', job.id);
  currentSpan?.setAttribute('job.attempts', job.attemptsMade);
  // ... actual work
}, { connection, telemetry });

Once wired, BullMQ automatically exports:

• Counters: bullmq.jobs.completed, bullmq.jobs.failed, bullmq.jobs.delayed, bullmq.jobs.retried, bullmq.jobs.waiting
• Histograms: bullmq.job.duration (ms)
• Gauges: bullmq.queue.jobs with a state dimension (waiting, active, completed, failed, delayed)

All metrics carry the bullmq.queue.name attribute and job-level metadata.

Adding Custom Child Spans Inside Job Processors

For complex jobs with multiple sub-operations, create child spans to isolate where time is being spent:

import { trace, context, SpanStatusCode } from '@opentelemetry/api';
import { Job } from 'bullmq';

const tracer = trace.getTracer('email-worker');

async function processEmailJob(job: Job) {
  return tracer.startActiveSpan('email.send', async (span) => {
    span.setAttribute('job.id', job.id);
    span.setAttribute('recipient.domain', job.data.email.split('@')[1]);

    try {
      // Child span: template rendering
      await tracer.startActiveSpan('render-template', async (childSpan) => {
        const html = renderTemplate(job.data.template, job.data.variables);
        childSpan.end();
      });

      // Child span: SMTP delivery
      await tracer.startActiveSpan('smtp-delivery', async (childSpan) => {
        childSpan.setAttribute('smtp.host', process.env.SMTP_HOST);
        await sendEmail({ to: job.data.email, html });
        childSpan.end();
      });

      span.setStatus({ code: SpanStatusCode.OK });
    } catch (err) {
      span.setStatus({
        code: SpanStatusCode.ERROR,
        message: (err as Error).message,
      });
      span.recordException(err as Error);
      throw err;
    } finally {
      span.end();
    }
  });
}

This produces a trace waterfall where the email.send span contains one render-template and one smtp-delivery child span. If SMTP delivery is slow, you know instantly.

Section 2: Building a Custom Prometheus Metrics Exporter

BullMQ's built-in exportPrometheusMetrics() exposes per-queue job counts by state, but it won't give you processing latency histograms, error-type breakdowns, or throughput rates. For those, you need a custom Prometheus exporter.

Defining the Metrics Registry

Using the prom-client library, define the metrics that matter for queue operations:

// metrics.ts
import { Registry, Counter, Gauge, Histogram, collectDefaultMetrics } from 'prom-client';

const register = new Registry();
collectDefaultMetrics({ register });

// Queue depth by status
export const queueDepthByStatus = new Gauge({
  name: 'bullmq_queue_depth',
  help: 'Number of jobs in a queue, by status',
  labelNames: ['queue', 'status'],
  registers: [register],
});

// Processing latency histogram
export const processingLatency = new Histogram({
  name: 'bullmq_processing_latency_seconds',
  help: 'Job processing duration in seconds',
  labelNames: ['queue', 'job_name'],
  buckets: [0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 120, 300],
  registers: [register],
});

// End-to-end latency (add -> completed)
export const endToEndLatency = new Histogram({
  name: 'bullmq_end_to_end_latency_seconds',
  help: 'Time from job creation to completion in seconds',
  labelNames: ['queue', 'job_name'],
  buckets: [0.1, 0.5, 1, 5, 10, 30, 60, 300, 600, 1800],
  registers: [register],
});

// Jobs processed by result
export const jobsProcessed = new Counter({
  name: 'bullmq_jobs_processed_total',
  help: 'Total number of jobs processed, by queue and result',
  labelNames: ['queue', 'job_name', 'result'],
  registers: [register],
});

// Failures by error type
export const failuresByError = new Counter({
  name: 'bullmq_failures_by_error_total',
  help: 'Job failures categorized by error name',
  labelNames: ['queue', 'job_name', 'error_type'],
  registers: [register],
});

// Active worker count
export const workerCount = new Gauge({
  name: 'bullmq_workers_active',
  help: 'Number of active workers per queue',
  labelNames: ['queue'],
  registers: [register],
});

// Job throughput rate
export const jobThroughput = new Gauge({
  name: 'bullmq_throughput_per_second',
  help: 'Jobs processed per second, per queue',
  labelNames: ['queue'],
  registers: [register],
});

// Stalled job count
export const stalledJobsCount = new Gauge({
  name: 'bullmq_stalled_jobs',
  help: 'Number of stalled jobs currently detected per queue',
  labelNames: ['queue'],
  registers: [register],
});

export { register };

The Event-Driven Collector Engine

The collector listens to BullMQ queue events (active, completed, failed, stalled) to capture timing data, and polls queue state on a configurable interval for depth and throughput:

// collector.ts
import { Queue, QueueEvents } from 'bullmq';
import IORedis from 'ioredis';
import {
  queueDepthByStatus,
  processingLatency,
  endToEndLatency,
  jobsProcessed,
  failuresByError,
  workerCount,
  jobThroughput,
  stalledJobsCount,
} from './metrics';

interface CollectorConfig {
  queues: string[];
  redisConnection: IORedis.Redis;
  pollIntervalMs?: number;
}

interface JobTimestamps {
  [jobId: string]: {
    queuedAt: number;
    startedAt: number;
    jobName: string;
  };
}

export class BullMQExporterCollector {
  private queues: string[];
  private connection: IORedis.Redis;
  private pollIntervalMs: number;
  private timestamps: JobTimestamps = {};
  private pollTimer?: NodeJS.Timeout;
  private previousCompletedCount: Map<string, number> = new Map();

  constructor(config: CollectorConfig) {
    this.queues = config.queues;
    this.connection = config.redisConnection;
    this.pollIntervalMs = config.pollIntervalMs ?? 15_000;
  }

  async start(): Promise<void> {
    for (const queueName of this.queues) {
      await this.attachQueueListeners(queueName);
    }
    this.pollTimer = setInterval(
      () => this.pollAllQueues(),
      this.pollIntervalMs,
    );
    await this.pollAllQueues();
  }

  private async attachQueueListeners(queueName: string): Promise<void> {
    const queueEvents = new QueueEvents(queueName, {
      connection: this.connection,
    });

    queueEvents.on('active', ({ jobId }) => {
      this.timestamps[jobId] = {
        ...this.timestamps[jobId],
        startedAt: Date.now(),
      };
    });

    queueEvents.on('completed', ({ jobId }) => {
      const ts = this.timestamps[jobId];
      if (ts) {
        const durationSec = (Date.now() - ts.startedAt) / 1000;
        const e2eSec = (Date.now() - ts.queuedAt) / 1000;
        processingLatency.observe(
          { queue: queueName, job_name: ts.jobName },
          durationSec,
        );
        endToEndLatency.observe(
          { queue: queueName, job_name: ts.jobName },
          e2eSec,
        );
        jobsProcessed.inc({
          queue: queueName,
          job_name: ts.jobName,
          result: 'completed',
        });
        delete this.timestamps[jobId];
      }
    });

    queueEvents.on('failed', ({ jobId, failedReason }) => {
      const ts = this.timestamps[jobId];
      if (ts) {
        const durationSec = (Date.now() - ts.startedAt) / 1000;
        processingLatency.observe(
          { queue: queueName, job_name: ts.jobName },
          durationSec,
        );
        jobsProcessed.inc({
          queue: queueName,
          job_name: ts.jobName,
          result: 'failed',
        });

        const errorType = failedReason.match(/^(\w+)/)?.[1] ?? 'UnknownError';
        failuresByError.inc({
          queue: queueName,
          job_name: ts.jobName,
          error_type: errorType,
        });
        delete this.timestamps[jobId];
      }
    });

    queueEvents.on('stalled', ({ jobId }) => {
      stalledJobsCount.inc({ queue: queueName });
    });
  }

  private async pollAllQueues(): Promise<void> {
    for (const queueName of this.queues) {
      await this.pollQueue(queueName);
    }
  }

  private async pollQueue(queueName: string): Promise<void> {
    try {
      const queue = new Queue(queueName, { connection: this.connection });

      const jobCounts = await queue.getJobCounts(
        'waiting',
        'active',
        'completed',
        'failed',
        'delayed',
        'paused',
      );

      for (const [status, count] of Object.entries(jobCounts)) {
        queueDepthByStatus.set({ queue: queueName, status }, count);
      }

      const workers = await queue.getWorkers();
      workerCount.set({ queue: queueName }, workers.length);

      const currentCompleted = jobCounts.completed ?? 0;
      const previousCompleted =
        this.previousCompletedCount.get(queueName) ?? currentCompleted;
      const delta = currentCompleted - previousCompleted;
      const throughput = delta / (this.pollIntervalMs / 1000);
      jobThroughput.set({ queue: queueName }, Math.max(0, throughput));
      this.previousCompletedCount.set(queueName, currentCompleted);

      await queue.close();
    } catch (err) {
      console.error(`Failed to poll queue ${queueName}:`, err);
    }
  }

  stop(): void {
    if (this.pollTimer) clearInterval(this.pollTimer);
  }
}

Exposing the /metrics Endpoint

Wire the collector into an Express server that Prometheus can scrape:

// server.ts
import express from 'express';
import IORedis from 'ioredis';
import { BullMQExporterCollector } from './collector';
import { register } from './metrics';

const app = express();
const connection = new IORedis({
  host: process.env.REDIS_HOST ?? 'localhost',
  port: Number(process.env.REDIS_PORT ?? 6379),
  maxRetriesPerRequest: null,
});

const QUEUES = (
  process.env.BULLMQ_QUEUES ??
  'email-notifications,order-processing,report-generation'
)
  .split(',')
  .map((s) => s.trim());

const collector = new BullMQExporterCollector({
  queues: QUEUES,
  redisConnection: connection,
  pollIntervalMs: 15_000,
});

app.get('/metrics', async (_req, res) => {
  try {
    res.set('Content-Type', register.contentType);
    res.send(await register.metrics());
  } catch (err) {
    res.status(500).send(
      err instanceof Error ? err.message : 'Metrics error',
    );
  }
});

app.get('/health', (_req, res) => {
  res.json({ status: 'ok', queues: QUEUES });
});

const PORT = Number(process.env.METRICS_PORT ?? 9090);
app.listen(PORT, async () => {
  await collector.start();
  console.log(`BullMQ metrics exporter listening on :${PORT}`);
  console.log(`Monitoring queues: ${QUEUES.join(', ')}`);
});

process.on('SIGTERM', () => {
  collector.stop();
  process.exit(0);
});

Deploy as a standalone Docker container and add a Prometheus scrape target:

# prometheus.yml snippet
scrape_configs:
  - job_name: 'bullmq-exporter'
    static_configs:
      - targets: ['bullmq-exporter:9090']
    metrics_path: '/metrics'
    scrape_interval: 15s

Section 3: Grafana Dashboard as Code (JSON Provisioning)

Click-ops in the Grafana UI is fine for exploration, but production dashboards should be version-controlled, peer-reviewed, and CI/CD-deployed alongside the exporter.

Below is a complete dashboard JSON with 9 panels covering every dimension of queue health. The dashboard uses a $queue template variable populated from Prometheus label values, so you can filter by individual queue or view an aggregate.

Key panels:

• Queue Depth by Status — bar gauge with color thresholds (green < 1K, orange < 5K, red beyond)
• Processing Latency P99 / P95 / P50 — time series showing latency percentiles over the last hour
• Latency Heatmap — shows the distribution of processing latencies, with log-scale Y-axis to highlight slow outliers
• Status Breakdown — pie chart showing the proportion of waiting, active, failed, delayed, and completed jobs
• Throughput — jobs per second as a sparkline
• Failure Rate by Error Type — stacked bar chart grouped by error type (ValidationError, TimeoutError, etc.)
• Active Workers — stat panel with threshold coloring (red = 0, yellow = 1, green = 2+)
• Stalled Jobs — stat panel that turns red the instant any stalled job is detected
• Queue Depth Sparklines — area-fill time series for waiting, failed, and active counts over time

Here's the template variable configuration that drives all panels:

{
  "templating": {
    "list": [
      {
        "name": "queue",
        "type": "query",
        "query": "label_values(bullmq_queue_depth, queue)",
        "refresh": 1,
        "includeAll": true,
        "allValue": ".*",
        "multi": true,
        "current": { "selected": true, "text": "All", "value": ["$__all"] }
      }
    ]
  },
  "time": { "from": "now-1h", "to": "now" }
}

Deploy via Grafana provisioning:

# /etc/grafana/provisioning/dashboards/bullmq.yaml
apiVersion: 1
providers:
  - name: 'BullMQ'
    orgId: 1
    folder: 'Queues'
    type: file
    disableDeletion: false
    updateIntervalSeconds: 30
    options:
      path: /var/lib/grafana/dashboards/bullmq

Or push via the Grafana HTTP API as part of your CI/CD pipeline.

Section 4: Synthetic Health Checks for Queue Infrastructure

A dedicated health-check service validates Redis connectivity, queue responsiveness, and worker liveness — suitable for Kubernetes liveness/readiness probes and external uptime monitoring.

Health Check Interface and Checks

Define a standard health check interface, then implement individual checks:

// health.ts
export interface HealthCheckResult {
  name: string;
  healthy: boolean;
  latencyMs: number;
  detail?: Record<string, unknown>;
  error?: string;
}

export interface HealthSummary {
  healthy: boolean;
  checks: HealthCheckResult[];
  timestamp: string;
}

// checks.ts
import { Redis } from 'ioredis';
import { Queue } from 'bullmq';
import { performance } from 'perf_hooks';
import { HealthCheckResult } from './health';

export async function checkRedisPing(
  connection: Redis,
): Promise<HealthCheckResult> {
  const start = performance.now();
  try {
    const result = await connection.ping();
    return {
      name: 'redis_ping',
      healthy: result === 'PONG',
      latencyMs: Math.round(performance.now() - start),
      detail: { response: result },
    };
  } catch (err) {
    return {
      name: 'redis_ping',
      healthy: false,
      latencyMs: Math.round(performance.now() - start),
      error: (err as Error).message,
    };
  }
}

export async function checkQueueHealth(
  queueName: string,
  connection: Redis,
): Promise<HealthCheckResult> {
  const start = performance.now();
  try {
    const queue = new Queue(queueName, { connection });
    const jobCounts = await queue.getJobCounts();
    const activeJobs = await queue.getActive(0, 100);
    await queue.close();

    // Detect stalled jobs: active jobs older than 5 minutes
    const stalled = activeJobs.filter(
      (job) => job.timestamp && Date.now() - job.timestamp > 300_000,
    );

    return {
      name: `queue_${queueName}`,
      healthy: stalled.length === 0,
      latencyMs: Math.round(performance.now() - start),
      detail: {
        waiting: jobCounts.waiting,
        active: jobCounts.active,
        failed: jobCounts.failed,
        stalled: stalled.length,
      },
    };
  } catch (err) {
    return {
      name: `queue_${queueName}`,
      healthy: false,
      latencyMs: Math.round(performance.now() - start),
      error: (err as Error).message,
    };
  }
}

export async function checkWorkerLiveness(
  queueName: string,
  connection: Redis,
): Promise<HealthCheckResult> {
  const start = performance.now();
  try {
    const queue = new Queue(queueName, { connection });
    const workers = await queue.getWorkers();
    await queue.close();

    return {
      name: `workers_${queueName}`,
      healthy: workers.length > 0,
      latencyMs: Math.round(performance.now() - start),
      detail: { workerCount: workers.length },
    };
  } catch (err) {
    return {
      name: `workers_${queueName}`,
      healthy: false,
      latencyMs: Math.round(performance.now() - start),
      error: (err as Error).message,
    };
  }
}

export async function checkRedisMemory(
  connection: Redis,
): Promise<HealthCheckResult> {
  const start = performance.now();
  try {
    const info = await connection.info('memory');
    const used = info.match(/used_memory_human:([^\r\n]+)/)?.[1]?.trim();
    const peak = info.match(/used_memory_peak_human:([^\r\n]+)/)?.[1]?.trim();
    const max = info.match(/maxmemory_human:([^\r\n]+)/)?.[1]?.trim() ?? 'unlimited';

    return {
      name: 'redis_memory',
      healthy: true,
      latencyMs: Math.round(performance.now() - start),
      detail: { usedMemory: used, peakMemory: peak, maxMemory: max },
    };
  } catch (err) {
    return {
      name: 'redis_memory',
      healthy: false,
      latencyMs: Math.round(performance.now() - start),
      error: (err as Error).message,
    };
  }
}

Combined Aggregator and Express Endpoints

Aggregate all checks into a single health summary, and expose three HTTP endpoints:

// aggregator.ts
import { Redis } from 'ioredis';
import {
  checkRedisPing,
  checkQueueHealth,
  checkWorkerLiveness,
  checkRedisMemory,
} from './checks';
import { HealthCheckResult, HealthSummary } from './health';

export class QueueHealthAggregator {
  private connection: Redis;
  private queues: string[];

  constructor(connection: Redis, queues: string[]) {
    this.connection = connection;
    this.queues = queues;
  }

  async runAllChecks(): Promise<HealthSummary> {
    const checks: HealthCheckResult[] = [];

    checks.push(await checkRedisPing(this.connection));
    checks.push(await checkRedisMemory(this.connection));

    for (const queueName of this.queues) {
      checks.push(await checkQueueHealth(queueName, this.connection));
      checks.push(await checkWorkerLiveness(queueName, this.connection));
    }

    return {
      healthy: checks.every((c) => c.healthy),
      checks,
      timestamp: new Date().toISOString(),
    };
  }
}

// health-server.ts
import express, { Request, Response } from 'express';
import IORedis from 'ioredis';
import { QueueHealthAggregator } from './aggregator';

const connection = new IORedis({
  host: process.env.REDIS_HOST ?? 'localhost',
  port: Number(process.env.REDIS_PORT ?? 6379),
  maxRetriesPerRequest: null,
  enableReadyCheck: true,
});

const QUEUES = (
  process.env.BULLMQ_QUEUES ?? 'default'
).split(',').map((s) => s.trim());

const aggregator = new QueueHealthAggregator(connection, QUEUES);
const app = express();

// Liveness — is the process alive?
app.get('/health/live', (_req: Request, res: Response) => {
  res.json({ status: 'alive', uptime: process.uptime() });
});

// Readiness — can it do work?
app.get('/health/ready', async (_req: Request, res: Response) => {
  try {
    await connection.ping();
    res.json({ status: 'ready' });
  } catch {
    res.status(503).json({ status: 'not ready' });
  }
});

// Detailed — comprehensive diagnostics
app.get('/health/detailed', async (_req: Request, res: Response) => {
  const summary = await aggregator.runAllChecks();
  res.status(summary.healthy ? 200 : 503).json(summary);
});

const PORT = Number(process.env.HEALTH_PORT ?? 8080);
app.listen(PORT, () => {
  console.log(`Health server listening on :${PORT}`);
});

Kubernetes Probe Configuration

Use the endpoints as native Kubernetes probes:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: bullmq-health
spec:
  template:
    spec:
      containers:
        - name: health
          image: myapp/bullmq-health:latest
          ports:
            - containerPort: 8080
          livenessProbe:
            httpGet:
              path: /health/live
              port: 8080
            initialDelaySeconds: 10
            periodSeconds: 15
            timeoutSeconds: 5
            failureThreshold: 3
          readinessProbe:
            httpGet:
              path: /health/ready
              port: 8080
            initialDelaySeconds: 5
            periodSeconds: 10
            timeoutSeconds: 3
            failureThreshold: 2

Section 5: Prometheus Alerting Rules as Code

With metrics flowing into Prometheus, define recording and alerting rules that detect real queue incidents. Version-control these YAML files and deploy them alongside your Prometheus configuration.

Recording Rules

Pre-compute commonly queried metrics to make alert expressions simpler and more efficient:

# prometheus/rules/bullmq_recording.yml
groups:
  - name: bullmq_recording_rules
    interval: 30s
    rules:
      # Job failure rate over 5 minutes
      - record: job:bullmq_failure_rate_5m
        expr: |
          sum(rate(bullmq_failures_by_error_total[5m])) by (queue)
          /
          (sum(rate(bullmq_jobs_processed_total[5m])) by (queue) > 0 or vector(1))

      # P99, P95, P50 processing latency
      - record: job:bullmq_processing_p99:5m
        expr: |
          histogram_quantile(0.99,
            sum(rate(bullmq_processing_latency_seconds_bucket[5m])) by (le, queue)
          )

      - record: job:bullmq_processing_p95:5m
        expr: |
          histogram_quantile(0.95,
            sum(rate(bullmq_processing_latency_seconds_bucket[5m])) by (le, queue)
          )

      # Queue backlog ratio (waiting / completed baseline)
      - record: job:bullmq_backlog_ratio:5m
        expr: |
          bullmq_queue_depth{status="waiting"}
          / on(queue)
          (label_replace(
            bullmq_queue_depth{status="completed"} > 0 or vector(1),
            "queue", "$1", "queue", "(.*)"
          ))

      # Stalled job rate
      - record: job:bullmq_stalled_rate:5m
        expr: |
          sum(rate(bullmq_stalled_jobs[5m])) by (queue)

Alerting Rules

Define alerts with clear thresholds, severity levels, and runbook links:

# prometheus/rules/bullmq_alerts.yml
groups:
  - name: bullmq_queue_alerts
    interval: 30s
    rules:
      - alert: BullMQHighFailureRate
        expr: job:bullmq_failure_rate_5m > 0.05
        for: 5m
        labels:
          severity: critical
          team: platform
        annotations:
          summary: "BullMQ high failure rate on {{ $labels.queue }}"
          description: |
            Queue {{ $labels.queue }} has a failure rate of {{ $value | humanizePercentage }}
            over the last 5 minutes. This exceeds the 5% threshold.
          runbook_url: "https://wiki.example.com/runbooks/bullmq-high-failure-rate"

      - alert: BullMQHighProcessingLatency
        expr: job:bullmq_processing_p99:5m > 30
        for: 3m
        labels:
          severity: warning
          team: platform
        annotations:
          summary: "BullMQ high P99 latency on {{ $labels.queue }}"
          description: |
            Queue {{ $labels.queue }} P99 processing latency is {{ $value | humanizeDuration }}.
            Threshold is 30 seconds.

      - alert: BullMQCriticalProcessingLatency
        expr: job:bullmq_processing_p99:5m > 120
        for: 2m
        labels:
          severity: critical
          team: platform
        annotations:
          summary: "BullMQ critical P99 latency on {{ $labels.queue }}"
          description: |
            Queue {{ $labels.queue }} P99 processing latency is {{ $value | humanizeDuration }}.
            Threshold is 120 seconds. Immediate investigation required.

      - alert: BullMQQueueBacklog
        expr: bullmq_queue_depth{status="waiting"} > 10000
        for: 10m
        labels:
          severity: warning
          team: platform
        annotations:
          summary: "BullMQ backlog exceeded on {{ $labels.queue }}"
          description: |
            Queue {{ $labels.queue }} has {{ $value }} waiting jobs (threshold: 10,000).
            Consider scaling up workers.

      - alert: BullMQStalledJobs
        expr: sum(bullmq_stalled_jobs) > 0
        for: 2m
        labels:
          severity: warning
          team: platform
        annotations:
          summary: "BullMQ stalled jobs detected"
          description: |
            {{ $value }} stalled job(s) detected. Workers may have crashed.

      - alert: BullMQNoActiveWorkers
        expr: sum(bullmq_workers_active) == 0
        for: 1m
        labels:
          severity: critical
          team: platform
        annotations:
          summary: "No active BullMQ workers"
          description: |
            Zero workers are currently active. No jobs will be processed.

      - alert: BullMQThroughputDrop
        expr: |
          (avg(bullmq_throughput_per_second) by (queue))
          /
          (avg(bullmq_throughput_per_second offset 1h) by (queue) > 0)
          < 0.5
        for: 10m
        labels:
          severity: warning
          team: platform
        annotations:
          summary: "BullMQ throughput drop on {{ $labels.queue }}"
          description: |
            Queue {{ $labels.queue }} throughput has dropped by >50% compared to 1 hour ago.

Validate your rules with promtool before deploying:

promtool check rules prometheus/rules/bullmq_alerts.yml

How Queue Hub Complements Custom Monitoring

Building and maintaining the infrastructure described above — the standalone Prometheus exporter, custom OTel instrumentation, Grafana provisioning files, synthetic health endpoints, and alerting rules — requires significant engineering effort.

Queue Hub offers a managed alternative that addresses many of the same observability needs without the operational burden:

• Multi-backend queue overview dashboard — immediately see job counts, processing rates, and worker health across all queues without writing a single metric query
• Live worker view — real-time visibility into active worker processes, their concurrency, and the jobs they're handling (similar to queue.getWorkers() in code, but with a polished UI)
• Job detail view — inspect individual job payloads, attempts, timestamps, and error stacks, filling the same role as per-job traces for debugging
• Redis support across deployment models — local, TLS, Valkey, AWS ElastiCache, and agent tunneling for private Redis instances
• Multi-org and team management — share queue visibility across teams without building your own RBAC layer on top of Prometheus and Grafana

Queue Hub is not a replacement for Prometheus-based alerting — you still want recording rules and alert thresholds for automated incident response. But for day-to-day queue introspection, debugging, and team-wide observability, Queue Hub provides the visibility layer that would otherwise require maintaining the exporter, dashboard JSON, and health check service described in this post.

Conclusion

Building production-grade queue observability for BullMQ means going beyond built-in metrics middleware. The five pieces covered in this post — OpenTelemetry tracing, a custom Prometheus exporter, Grafana dashboards as code, synthetic health checks, and Prometheus alerting rules — form a complete programmatic observability stack that version-controls, deploys, and scales with your infrastructure.

Start with OpenTelemetry instrumentation to get distributed traces across the producer → queue → worker pipeline. Layer in the custom Prometheus exporter for queue-depth histograms, failure-type counters, and throughput gauges. Define your Grafana dashboard as JSON to bake it into your CI/CD pipeline. Add synthetic health checks for Kubernetes-native liveness and readiness probing. Finally, codify your SLIs into Prometheus recording and alerting rules.

Each component is independently useful. Together they eliminate the blind spots that plague queue systems in production.