How to Monitor RabbitMQ Queue Depth, Consumer Lag, and Cluster Health with Vigilmon

RabbitMQ powers millions of async workflows — but a stalled consumer, a memory watermark breach, or a cluster partition can bring your message pipeline to a halt without emitting a single HTTP error. Queue depth climbs silently, consumers fall behind, and downstream services get nothing while your broker dashboard collects dust.

Vigilmon gives you external visibility into RabbitMQ health through HTTP probe monitoring of the Management API and heartbeat monitoring for your consumer applications. This tutorial covers both.

Why RabbitMQ Needs External Monitoring

The RabbitMQ Management Plugin ships with a dashboard and REST API — but that only helps if someone is watching. External monitoring with Vigilmon adds:

• Proactive alerting when queue depth breaches thresholds or the cluster loses nodes
• Memory watermark detection before the broker starts blocking publishers
• Consumer liveness monitoring via heartbeats sent from each consumer after successful message processing
• Exchange throughput visibility to catch routing failures and misconfigured bindings
• Multi-region probing from outside your VPC to catch network-layer failures

Step 1: Enable and Secure the Management API

RabbitMQ's Management Plugin exposes a rich HTTP API at port 15672. Enable it if you haven't already:

rabbitmq-plugins enable rabbitmq_management

Create a dedicated monitoring user with read-only access:

rabbitmqctl add_user vigilmon <strong-password>
rabbitmqctl set_user_tags vigilmon monitoring
rabbitmqctl set_permissions -p / vigilmon "" "" ".*"

The monitoring tag grants read access to the management API without permission to publish or consume messages.

Verify the API is accessible:

curl -u vigilmon:<password> http://localhost:15672/api/healthchecks/node
# {"status":"ok"}

Step 2: Build a Composite Health Endpoint

Rather than pointing Vigilmon directly at the RabbitMQ Management API (which may not be externally accessible), expose a thin health endpoint from your application tier that aggregates the metrics you care about.

Node.js Health Sidecar

// health/rabbitmq.js
const express = require('express');
const axios = require('axios');

const app = express();

const RABBITMQ_API = process.env.RABBITMQ_API_URL || 'http://localhost:15672/api';
const RABBITMQ_USER = process.env.RABBITMQ_USER || 'vigilmon';
const RABBITMQ_PASS = process.env.RABBITMQ_PASS;
const QUEUE_DEPTH_THRESHOLD = parseInt(process.env.QUEUE_DEPTH_THRESHOLD || '10000');
const CONSUMER_LAG_THRESHOLD = parseInt(process.env.CONSUMER_LAG_THRESHOLD || '5000');

const auth = { auth: { username: RABBITMQ_USER, password: RABBITMQ_PASS } };

app.get('/health/rabbitmq', async (req, res) => {
  try {
    // Node health check
    const nodeRes = await axios.get(`${RABBITMQ_API}/healthchecks/node`, auth);
    if (nodeRes.data.status !== 'ok') {
      return res.status(503).json({ status: 'down', reason: 'node_unhealthy', detail: nodeRes.data });
    }

    // Cluster overview — check all nodes are running
    const overviewRes = await axios.get(`${RABBITMQ_API}/nodes`, auth);
    const nodes = overviewRes.data;
    const downNodes = nodes.filter(n => !n.running);
    if (downNodes.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'cluster_nodes_down',
        nodes: downNodes.map(n => n.name),
      });
    }

    // Memory watermark — if any node is blocking publishers
    const memBlocked = nodes.filter(n => n.mem_alarm);
    if (memBlocked.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'memory_watermark_breached',
        nodes: memBlocked.map(n => n.name),
      });
    }

    // Disk alarm check
    const diskAlarm = nodes.filter(n => n.disk_free_alarm);
    if (diskAlarm.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'disk_free_alarm',
        nodes: diskAlarm.map(n => n.name),
      });
    }

    return res.status(200).json({
      status: 'ok',
      cluster_nodes: nodes.length,
      running_nodes: nodes.filter(n => n.running).length,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.get('/health/rabbitmq/queues', async (req, res) => {
  try {
    const queuesRes = await axios.get(`${RABBITMQ_API}/queues`, auth);
    const queues = queuesRes.data;

    const problematic = queues.filter(q => {
      const depth = q.messages || 0;
      const consumerCount = q.consumers || 0;
      return depth > QUEUE_DEPTH_THRESHOLD || (depth > CONSUMER_LAG_THRESHOLD && consumerCount === 0);
    });

    if (problematic.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'queue_depth_exceeded',
        queues: problematic.map(q => ({
          name: q.name,
          vhost: q.vhost,
          depth: q.messages,
          consumers: q.consumers,
        })),
      });
    }

    return res.status(200).json({
      status: 'ok',
      total_queues: queues.length,
      total_messages: queues.reduce((sum, q) => sum + (q.messages || 0), 0),
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3005);

Python Health Sidecar

# health/rabbitmq_health.py
from flask import Flask, jsonify
import requests, os

app = Flask(__name__)

RABBITMQ_API = os.environ.get('RABBITMQ_API_URL', 'http://localhost:15672/api')
AUTH = (os.environ['RABBITMQ_USER'], os.environ['RABBITMQ_PASS'])
DEPTH_THRESHOLD = int(os.environ.get('QUEUE_DEPTH_THRESHOLD', 10000))

@app.route('/health/rabbitmq')
def rabbitmq_health():
    try:
        node_r = requests.get(f'{RABBITMQ_API}/healthchecks/node', auth=AUTH, timeout=5)
        if node_r.json().get('status') != 'ok':
            return jsonify(status='down', reason='node_unhealthy'), 503

        nodes_r = requests.get(f'{RABBITMQ_API}/nodes', auth=AUTH, timeout=5)
        nodes = nodes_r.json()
        down = [n['name'] for n in nodes if not n.get('running')]
        if down:
            return jsonify(status='degraded', reason='cluster_nodes_down', nodes=down), 503

        mem_alarm = [n['name'] for n in nodes if n.get('mem_alarm')]
        if mem_alarm:
            return jsonify(status='degraded', reason='memory_watermark', nodes=mem_alarm), 503

        return jsonify(status='ok', nodes=len(nodes))
    except Exception as e:
        return jsonify(status='down', error=str(e)), 503

@app.route('/health/rabbitmq/queues')
def queue_health():
    try:
        r = requests.get(f'{RABBITMQ_API}/queues', auth=AUTH, timeout=5)
        queues = r.json()
        deep = [
            {'name': q['name'], 'depth': q.get('messages', 0), 'consumers': q.get('consumers', 0)}
            for q in queues if q.get('messages', 0) > DEPTH_THRESHOLD
        ]
        if deep:
            return jsonify(status='degraded', reason='queue_depth_exceeded', queues=deep), 503
        total = sum(q.get('messages', 0) for q in queues)
        return jsonify(status='ok', total_queues=len(queues), total_messages=total)
    except Exception as e:
        return jsonify(status='down', error=str(e)), 503

if __name__ == '__main__':
    app.run(port=3005)

Step 3: Configure Vigilmon HTTP Monitors for RabbitMQ

1. Log in to vigilmon.online and go to Monitors → New Monitor
2. Choose HTTP / HTTPS
3. Set the URL to your cluster health endpoint: https://your-app.example.com/health/rabbitmq
4. Set check interval: 1 minute
5. Under Expected response:
   • Status code: 200
   • Response body contains: "status":"ok"
   • Response time threshold: 3000ms
6. Assign your Slack or PagerDuty alert channel
7. Save

Add a second monitor for queue depth:

• URL: https://your-app.example.com/health/rabbitmq/queues
• Expected: 200, body contains "status":"ok"
• Interval: 1 minute
• Separate alert channel (queue depth alerts are usually P2 vs P1 for broker failures)

Vigilmon's multi-region consensus prevents false alerts from transient Management API latency.

Step 4: Monitor Exchange Throughput

Exchange-level metrics reveal routing failures — messages being dropped at the exchange because no bindings match, or publish rates spiking beyond consumer capacity. Extend the health sidecar:

app.get('/health/rabbitmq/exchanges', async (req, res) => {
  try {
    const res2 = await axios.get(`${RABBITMQ_API}/exchanges`, auth);
    const exchanges = res2.data.filter(e => e.name !== '' && !e.name.startsWith('amq.'));

    const throughputRes = await axios.get(`${RABBITMQ_API}/overview`, auth);
    const stats = throughputRes.data.message_stats || {};

    return res.status(200).json({
      status: 'ok',
      publish_rate: stats.publish_details?.rate || 0,
      deliver_rate: stats.deliver_get_details?.rate || 0,
      ack_rate: stats.ack_details?.rate || 0,
      exchanges: exchanges.length,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

Set up a Vigilmon monitor on this endpoint with a response body check for "status":"ok". You can also use the numeric fields in your alerting logic — if publish_rate is high but ack_rate is near zero, consumers are falling behind.

Step 5: Heartbeat Monitoring for RabbitMQ Consumers

Queue depth and cluster health tell you about the broker — but consumers can silently stall (stuck in a retry loop, blocked on a downstream service, paused after connection failure) while broker metrics look healthy.

Vigilmon heartbeat monitors catch silent consumer stalls: your consumer pings Vigilmon after each successful batch.

Set Up the Heartbeat Monitor

1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
2. Name: rabbitmq-order-consumer
3. Expected interval: 5 minutes
4. Grace period: 10 minutes
5. Save — copy the heartbeat URL, e.g. https://vigilmon.online/heartbeat/abc123xyz

Wire Into Your Consumer (Node.js / amqplib)

// consumer.js
const amqp = require('amqplib');
const axios = require('axios');

const HEARTBEAT_URL = process.env.VIGILMON_HEARTBEAT_URL;
const HEARTBEAT_EVERY_N = 50;

let processed = 0;

async function startConsumer() {
  const conn = await amqp.connect(process.env.RABBITMQ_URL);
  const channel = await conn.createChannel();
  await channel.assertQueue('orders', { durable: true });
  channel.prefetch(10);

  channel.consume('orders', async (msg) => {
    if (!msg) return;
    try {
      await processMessage(msg.content);
      channel.ack(msg);
      processed++;

      if (processed % HEARTBEAT_EVERY_N === 0) {
        await axios.get(HEARTBEAT_URL).catch(() => {});
      }
    } catch (err) {
      channel.nack(msg, false, true);
    }
  });
}

startConsumer();

Wire Into Your Consumer (Python / pika)

import pika, os, requests, time

HEARTBEAT_URL = os.environ['VIGILMON_HEARTBEAT_URL']
last_heartbeat = time.time()

def on_message(channel, method, properties, body):
    global last_heartbeat
    try:
        process_message(body)
        channel.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        channel.basic_nack(delivery_tag=method.delivery_tag, requeue=True)
        return

    if time.time() - last_heartbeat > 60:
        try:
            requests.get(HEARTBEAT_URL, timeout=5)
        except Exception:
            pass
        last_heartbeat = time.time()

params = pika.URLParameters(os.environ['RABBITMQ_URL'])
connection = pika.BlockingConnection(params)
channel = connection.channel()
channel.queue_declare(queue='orders', durable=True)
channel.basic_qos(prefetch_count=10)
channel.basic_consume(queue='orders', on_message_callback=on_message)
channel.start_consuming()

Step 6: Alert Routing and Priority Tiers

RabbitMQ failures cascade: a memory watermark blocks publishers, which causes queue depth to stop growing (masking the problem), while consumers continue draining an increasingly stale backlog.

Recommended alert routing in Vigilmon:

| Monitor | Alert Channel | Priority | |---|---|---| | Cluster health /health/rabbitmq | Slack + PagerDuty | P1 | | Queue depth /health/rabbitmq/queues | Slack | P2 | | Exchange throughput /health/rabbitmq/exchanges | Slack | P2 | | Heartbeat: order consumer | Slack + email | P2 | | Heartbeat: analytics consumer | Email | P3 |

Set response time thresholds for early warning:

• Alert at 2000ms for cluster health (Management API slowness precedes broker pressure)
• Alert at 500ms for queue depth (fast check, should always respond quickly)

Summary

RabbitMQ failures are quiet — queues fill up, consumers stall, and memory watermarks block publishers without generating HTTP errors. External monitoring with Vigilmon gives you end-to-end visibility:

| Monitor Type | What It Covers | |---|---| | HTTP: /health/rabbitmq | Cluster node health, memory watermarks, disk alarms | | HTTP: /health/rabbitmq/queues | Queue depth, consumer count, lag detection | | HTTP: /health/rabbitmq/exchanges | Publish and delivery throughput rates | | Heartbeat | Consumer application liveness, processing health |

Get started free at vigilmon.online — your first RabbitMQ monitor is running in under two minutes.