tutorial

How to Monitor VerneMQ Cluster Health, Subscription Count, and Message Drop Rate with Vigilmon

VerneMQ cluster splits, high message drop rates, and retained message store failures can silently break MQTT pipelines. Learn to monitor VerneMQ node health, subscription counts, Prometheus metrics, and consumer liveness with Vigilmon.

VerneMQ is a high-performance MQTT broker designed for horizontal scaling — but a cluster partition, a node going offline, or a surge in message drop rates can silently break device communication without alerting your team. Subscriptions stay open, clients keep connecting, but messages stop flowing.

Vigilmon gives you external visibility into VerneMQ health through HTTP probe monitoring of its built-in HTTP API and Prometheus endpoint, plus heartbeat monitoring for downstream consumer applications. This tutorial covers both.

Why VerneMQ Needs External Monitoring

VerneMQ ships with a CLI (vmq-admin) and a Prometheus metrics endpoint, but these require active polling by a human or a monitoring stack you already maintain. External monitoring with Vigilmon adds:

  • Proactive alerting when cluster nodes become unreachable or leave the ring
  • Message drop rate detection before devices report delivery failures
  • Subscription count drift — catches mass client disconnects that precede fleet outages
  • Retained message store health — detects corruption or size breaches in the retained message backend
  • Multi-region probing from outside your network to catch infrastructure-layer failures

Step 1: Enable VerneMQ HTTP API and Prometheus Endpoint

VerneMQ exposes a built-in HTTP API on port 8888 and a Prometheus metrics endpoint. Enable them in your vernemq.conf:

# vernemq.conf
listener.http.default = 0.0.0.0:8888
listener.http.default.http_modules = vmq_status_http

Also enable the Prometheus plugin:

vmq-admin plugin enable --name vmq_prometheus
# Restart or reload config
vmq-admin cluster show

Verify both endpoints are available:

# Node status API
curl http://localhost:8888/status.json

# Prometheus metrics
curl http://localhost:8888/metrics

Step 2: Build a VerneMQ Health Endpoint

Expose a composite health endpoint that aggregates cluster and message metrics from the VerneMQ HTTP API and Prometheus scrape.

Node.js Health Sidecar

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

const app = express();

const VERNEMQ_API = process.env.VERNEMQ_API_URL || 'http://localhost:8888';
const MIN_NODES = parseInt(process.env.VERNEMQ_MIN_NODES || '1');
const DROP_RATE_THRESHOLD = parseInt(process.env.VERNEMQ_DROP_THRESHOLD || '50');

app.get('/health/vernemq', async (req, res) => {
  try {
    const statusRes = await axios.get(`${VERNEMQ_API}/status.json`, { timeout: 5000 });
    const status = statusRes.data;

    const nodes = status.nodes || [];
    const runningNodes = nodes.filter(n => n.status === 'running');

    if (runningNodes.length < MIN_NODES) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'insufficient_cluster_nodes',
        running: runningNodes.length,
        required: MIN_NODES,
        nodes: nodes.map(n => ({ name: n.node, status: n.status })),
      });
    }

    const downNodes = nodes.filter(n => n.status !== 'running');
    if (downNodes.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'cluster_nodes_down',
        nodes: downNodes.map(n => ({ name: n.node, status: n.status })),
      });
    }

    return res.status(200).json({
      status: 'ok',
      total_nodes: nodes.length,
      running_nodes: runningNodes.length,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

// Parse Prometheus text format for specific metrics
function parsePrometheusMetric(text, metricName) {
  const lines = text.split('\n');
  for (const line of lines) {
    if (line.startsWith(metricName) && !line.startsWith('#')) {
      const parts = line.split(' ');
      return parseFloat(parts[parts.length - 1]);
    }
  }
  return null;
}

app.get('/health/vernemq/metrics', async (req, res) => {
  try {
    const metricsRes = await axios.get(`${VERNEMQ_API}/metrics`, { timeout: 5000 });
    const text = metricsRes.data;

    const subscriptions = parsePrometheusMetric(text, 'vmq_router_subscriptions');
    const msgDropped = parsePrometheusMetric(text, 'vmq_queue_message_drop');
    const msgInFlight = parsePrometheusMetric(text, 'vmq_queue_in_flight_messages');
    const connections = parsePrometheusMetric(text, 'vmq_sessions_count');
    const bytesReceived = parsePrometheusMetric(text, 'vmq_bytes_received');
    const bytesSent = parsePrometheusMetric(text, 'vmq_bytes_sent');

    if (msgDropped !== null && msgDropped > DROP_RATE_THRESHOLD) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'high_message_drop_rate',
        dropped: msgDropped,
        threshold: DROP_RATE_THRESHOLD,
      });
    }

    return res.status(200).json({
      status: 'ok',
      subscriptions,
      connections,
      msg_dropped: msgDropped,
      msg_in_flight: msgInFlight,
      bytes_received: bytesReceived,
      bytes_sent: bytesSent,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3007);

Python Health Sidecar

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

app = Flask(__name__)

VERNEMQ_API = os.environ.get('VERNEMQ_API_URL', 'http://localhost:8888')
MIN_NODES = int(os.environ.get('VERNEMQ_MIN_NODES', 1))
DROP_THRESHOLD = int(os.environ.get('VERNEMQ_DROP_THRESHOLD', 50))

def parse_metric(text, name):
    for line in text.splitlines():
        if line.startswith(name) and not line.startswith('#'):
            try:
                return float(line.split()[-1])
            except ValueError:
                pass
    return None

@app.route('/health/vernemq')
def vernemq_health():
    try:
        status = requests.get(f'{VERNEMQ_API}/status.json', timeout=5).json()
        nodes = status.get('nodes', [])
        running = [n for n in nodes if n.get('status') == 'running']
        if len(running) < MIN_NODES:
            return jsonify(status='degraded', reason='insufficient_nodes',
                           running=len(running), required=MIN_NODES), 503
        down = [n['node'] for n in nodes if n.get('status') != 'running']
        if down:
            return jsonify(status='degraded', reason='nodes_down', nodes=down), 503
        return jsonify(status='ok', nodes=len(nodes), running=len(running))
    except Exception as e:
        return jsonify(status='down', error=str(e)), 503

@app.route('/health/vernemq/metrics')
def vernemq_metrics():
    try:
        text = requests.get(f'{VERNEMQ_API}/metrics', timeout=5).text
        dropped = parse_metric(text, 'vmq_queue_message_drop')
        if dropped and dropped > DROP_THRESHOLD:
            return jsonify(status='degraded', reason='high_drop_rate',
                           dropped=dropped, threshold=DROP_THRESHOLD), 503
        return jsonify(
            status='ok',
            subscriptions=parse_metric(text, 'vmq_router_subscriptions'),
            connections=parse_metric(text, 'vmq_sessions_count'),
            msg_dropped=dropped,
            msg_in_flight=parse_metric(text, 'vmq_queue_in_flight_messages'),
        )
    except Exception as e:
        return jsonify(status='down', error=str(e)), 503

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

Step 3: Monitor Retained Message Store Health

VerneMQ stores retained messages in a LevelDB-based backend (plumtree / leveled). A corrupted or excessively large retained store can slow down subscriber reconnects and cause cluster sync issues.

Add a retained message health check:

app.get('/health/vernemq/retained', async (req, res) => {
  try {
    const metricsRes = await axios.get(`${VERNEMQ_API}/metrics`, { timeout: 5000 });
    const text = metricsRes.data;

    const retainedCount = parsePrometheusMetric(text, 'vmq_retain_messages');
    const retainedBytes = parsePrometheusMetric(text, 'vmq_retain_memory');
    const MAX_RETAINED = parseInt(process.env.MAX_RETAINED_MESSAGES || '500000');
    const MAX_RETAINED_BYTES = parseInt(process.env.MAX_RETAINED_BYTES || '524288000'); // 500MB

    if (retainedCount !== null && retainedCount > MAX_RETAINED) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'retained_store_oversized',
        retained_count: retainedCount,
        threshold: MAX_RETAINED,
      });
    }

    if (retainedBytes !== null && retainedBytes > MAX_RETAINED_BYTES) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'retained_store_memory_exceeded',
        retained_bytes: retainedBytes,
        threshold_bytes: MAX_RETAINED_BYTES,
      });
    }

    return res.status(200).json({
      status: 'ok',
      retained_messages: retainedCount,
      retained_bytes: retainedBytes,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

Step 4: Configure Vigilmon HTTP Monitors

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

Add monitors for additional endpoints:

| Monitor URL | Purpose | Interval | Priority | |---|---|---|---| | /health/vernemq | Cluster node health | 1 min | P1 | | /health/vernemq/metrics | Subscriptions, connections, drop rate | 1 min | P2 | | /health/vernemq/retained | Retained message store size | 5 min | P3 |

For multi-node clusters, you may want individual node health monitors too — point each at a per-node health sidecar or use Vigilmon's multi-location probing to verify cluster-wide availability.

Step 5: Heartbeat Monitoring for VerneMQ Consumers

VerneMQ often sits in front of data consumers — telemetry ingestors, alerting pipelines, command dispatchers. These can stall while the broker remains healthy.

Set Up the Heartbeat Monitor

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Name: vernemq-telemetry-ingestor
  3. Expected interval: 5 minutes
  4. Grace period: 10 minutes
  5. Save — copy the heartbeat URL

Wire Into a Node.js MQTT Consumer

// consumer/telemetry.js
const mqtt = require('mqtt');
const axios = require('axios');

const client = mqtt.connect(process.env.VERNEMQ_BROKER_URL, {
  clientId: `telemetry-ingestor-${process.pid}`,
  clean: true,
  keepalive: 60,
});

const HEARTBEAT_URL = process.env.VIGILMON_HEARTBEAT_URL;
let lastProcessedAt = null;

client.on('connect', () => {
  client.subscribe('sensors/+/data', { qos: 1 });
});

client.on('message', async (topic, payload) => {
  try {
    await ingestTelemetry(topic, payload);
    lastProcessedAt = Date.now();
  } catch (err) {
    console.error('Ingest error:', err.message);
  }
});

// Heartbeat: send to Vigilmon every 60s if processing was recent
setInterval(async () => {
  if (lastProcessedAt && Date.now() - lastProcessedAt < 300_000) {
    await axios.get(HEARTBEAT_URL).catch(() => {});
  }
}, 60_000);

Wire Into a Python MQTT Consumer

import paho.mqtt.client as mqtt
import requests, os, time, threading

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

def on_message(client, userdata, msg):
    global last_processed
    try:
        ingest_telemetry(msg.topic, msg.payload)
        last_processed = time.time()
    except Exception as e:
        print(f'Ingest error: {e}')

def heartbeat_loop():
    while True:
        time.sleep(60)
        if time.time() - last_processed < 300:
            try:
                requests.get(HEARTBEAT_URL, timeout=5)
            except Exception:
                pass

threading.Thread(target=heartbeat_loop, daemon=True).start()

client = mqtt.Client(client_id=f'telemetry-ingestor-{os.getpid()}')
client.on_message = on_message
client.connect(os.environ['VERNEMQ_HOST'], int(os.environ.get('VERNEMQ_PORT', 1883)))
client.subscribe('sensors/+/data', qos=1)
client.loop_forever()

Step 6: Alert Routing

VerneMQ cluster failures cascade: a node leaving the ring causes subscription rebalancing, which causes brief client disconnects, which can cause a message drop spike. Alert on broker issues before the cascade reaches consumers.

Recommended Vigilmon alert routing:

| Monitor | Alert Channel | Priority | |---|---|---| | Cluster node health | Slack + PagerDuty | P1 | | Message drop rate | Slack | P2 | | Retained store size | Email | P3 | | Heartbeat: telemetry ingestor | Slack + email | P2 |

Set response time thresholds:

  • Alert at 2000ms for cluster health (HTTP API slowness precedes node pressure)
  • Alert at 500ms for Prometheus metrics endpoint (should always respond fast)

For clusters with rolling restarts or planned maintenance, use Vigilmon's maintenance windows to suppress expected alerts without disabling monitors.

Summary

VerneMQ failures are invisible from the client side — devices keep connecting while messages drop and cluster state diverges. External monitoring with Vigilmon gives you end-to-end visibility:

| Monitor Type | What It Covers | |---|---| | HTTP: /health/vernemq | Cluster node status, ring membership | | HTTP: /health/vernemq/metrics | Subscription count, connections, message drop rate | | HTTP: /health/vernemq/retained | Retained message store size and memory | | Heartbeat | Consumer liveness, data processing health |

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

Monitor your app with Vigilmon

Free plan — 5 monitors, no credit card required. Up and running in 60 seconds.

Start free →
← Back to all articles