tutorial

How to Monitor rumqttd Broker Health, MQTT Client Connections, and Message Throughput with Vigilmon

rumqttd is a high-performance Rust MQTT broker used in embedded IoT fleets. When the broker crashes, clients disconnect and data stops. Learn to monitor rumqttd process health, active connections, publish rates, QoS delivery, and TLS certificate expiry with Vigilmon.

rumqttd is an open-source, high-performance MQTT broker written in Rust and developed by Bytebeam as part of the rumq project. Unlike JVM-based brokers such as HiveMQ or Erlang-based VerneMQ, rumqttd achieves very low memory overhead and high connection density on commodity hardware — making it ideal for edge IoT gateways and embedded device fleets. It passes all MQTT v3.1.1 and v5 conformance tests and supports TLS, WebSocket, and multiple listener configurations.

When rumqttd crashes, every connected IoT device loses its MQTT session simultaneously. A silent process restart loop, a TLS certificate that expires at 2 AM, or a retained message store that fills up will all cause device data to silently stop flowing. Vigilmon gives you external probing of rumqttd health endpoints and heartbeat monitoring so you catch broker failures before your devices do.

Why rumqttd Needs External Monitoring

rumqttd ships with logging but no built-in alerting or external health API. External monitoring with Vigilmon adds:

  • Proactive alerting when the rumqttd process crashes or becomes unresponsive on its MQTT port
  • Connection count drift detection — a >20% drop from baseline signals mass device disconnection before support tickets arrive
  • Message publish rate monitoring — catches publish rate drops in production IoT deployments where devices should always be publishing
  • QoS delivery tracking — unacknowledged QoS 1/2 messages accumulate silently when clients disconnect under load
  • TLS certificate expiry alerts — rumqttd rejects device connections when its TLS certificate expires; 30-day advance warning prevents fleet outages
  • Retained message store monitoring — an unbounded retained message store can exhaust memory and crash the broker

Step 1: Build a rumqttd Health Sidecar

rumqttd does not expose an HTTP health API natively. Add a lightweight health sidecar that probes the broker's MQTT port, tracks connection metrics, and exposes an HTTP endpoint for Vigilmon to poll.

Node.js Health Sidecar

// health/rumqttd.js
const express = require('express');
const mqtt = require('mqtt');
const tls = require('tls');
const fs = require('fs');

const app = express();

const MQTT_HOST = process.env.RUMQTTD_HOST || 'localhost';
const MQTT_PORT = parseInt(process.env.RUMQTTD_PORT || '1883');
const MQTT_TLS_PORT = parseInt(process.env.RUMQTTD_TLS_PORT || '8883');
const TLS_CERT_PATH = process.env.RUMQTTD_TLS_CERT;
const TLS_WARNING_DAYS = parseInt(process.env.TLS_WARNING_DAYS || '30');
const CONN_BASELINE = parseInt(process.env.RUMQTTD_CONN_BASELINE || '0');
const CONN_DROP_THRESHOLD = parseFloat(process.env.RUMQTTD_CONN_DROP_THRESHOLD || '0.2');

let activeConnections = 0;
let peakConnections = 0;
let publishCount = 0;
let qos1PubackCount = 0;
let lastWindowPublish = 0;
let lastCheck = Date.now();

// Track our own MQTT monitoring connection
const monitorClient = mqtt.connect(`mqtt://${MQTT_HOST}:${MQTT_PORT}`, {
  clientId: `vigilmon-health-${Math.random().toString(16).slice(2)}`,
  clean: true,
  connectTimeout: 5000,
  reconnectPeriod: 10000,
});

let brokerReachable = false;
let lastConnectError = null;

monitorClient.on('connect', () => {
  brokerReachable = true;
  lastConnectError = null;
  activeConnections++;
  if (activeConnections > peakConnections) peakConnections = activeConnections;
  monitorClient.subscribe('$SYS/#', { qos: 0 });
});

monitorClient.on('error', (err) => {
  brokerReachable = false;
  lastConnectError = err.message;
});

monitorClient.on('offline', () => {
  brokerReachable = false;
});

// rumqttd publishes stats to $SYS topics (when enabled)
monitorClient.on('message', (topic, payload) => {
  const val = parseInt(payload.toString()) || 0;
  if (topic === '$SYS/broker/clients/connected') {
    activeConnections = val;
    if (val > peakConnections) peakConnections = val;
  }
  if (topic === '$SYS/broker/messages/publish/received') {
    publishCount = val;
  }
  if (topic === '$SYS/broker/messages/publish/sent') {
    qos1PubackCount = val;
  }
});

function checkTlsCertExpiry() {
  if (!TLS_CERT_PATH || !fs.existsSync(TLS_CERT_PATH)) return null;
  try {
    const certPem = fs.readFileSync(TLS_CERT_PATH, 'utf8');
    const cert = new tls.TLSSocket(null);
    // Parse expiry from PEM via openssl-style extraction
    const match = certPem.match(/Not After\s*:\s*(.+)/);
    if (match) {
      const expiry = new Date(match[1]);
      const daysLeft = Math.floor((expiry - Date.now()) / (1000 * 60 * 60 * 24));
      return { expiry: expiry.toISOString(), daysLeft };
    }
    return null;
  } catch {
    return null;
  }
}

app.get('/health/rumqttd', (req, res) => {
  if (!brokerReachable) {
    return res.status(503).json({
      status: 'down',
      reason: 'broker_unreachable',
      error: lastConnectError,
      host: MQTT_HOST,
      port: MQTT_PORT,
    });
  }

  // Check connection count drop from baseline
  if (CONN_BASELINE > 0 && activeConnections < CONN_BASELINE * (1 - CONN_DROP_THRESHOLD)) {
    return res.status(503).json({
      status: 'degraded',
      reason: 'connection_count_drop',
      active: activeConnections,
      baseline: CONN_BASELINE,
      threshold_pct: CONN_DROP_THRESHOLD * 100,
    });
  }

  // Check TLS cert expiry
  const tlsInfo = checkTlsCertExpiry();
  if (tlsInfo && tlsInfo.daysLeft < TLS_WARNING_DAYS) {
    return res.status(503).json({
      status: 'degraded',
      reason: 'tls_cert_expiring_soon',
      expiry: tlsInfo.expiry,
      days_left: tlsInfo.daysLeft,
      warning_threshold: TLS_WARNING_DAYS,
    });
  }

  return res.status(200).json({
    status: 'ok',
    broker: { host: MQTT_HOST, port: MQTT_PORT },
    connections: {
      active: activeConnections,
      peak: peakConnections,
      baseline: CONN_BASELINE,
    },
    messages: {
      publish_received: publishCount,
      qos1_puback_sent: qos1PubackCount,
    },
    tls: tlsInfo,
  });
});

app.get('/health/rumqttd/publish-rate', (req, res) => {
  const now = Date.now();
  const elapsed = (now - lastCheck) / 1000;
  const rate = elapsed > 0 ? (publishCount - lastWindowPublish) / elapsed : 0;
  lastWindowPublish = publishCount;
  lastCheck = now;

  return res.status(200).json({
    status: 'ok',
    publish_rate_per_sec: Math.round(rate * 10) / 10,
    total_publish_received: publishCount,
    window_seconds: Math.round(elapsed),
  });
});

app.listen(3007, () => console.log('rumqttd health sidecar listening on :3007'));

Python Health Sidecar

# health/rumqttd_health.py
import os, time, threading
from flask import Flask, jsonify
import paho.mqtt.client as mqtt

app = Flask(__name__)

MQTT_HOST = os.environ.get('RUMQTTD_HOST', 'localhost')
MQTT_PORT = int(os.environ.get('RUMQTTD_PORT', 1883))
CONN_BASELINE = int(os.environ.get('RUMQTTD_CONN_BASELINE', 0))
CONN_DROP_PCT = float(os.environ.get('RUMQTTD_CONN_DROP_THRESHOLD', 0.2))

state = {
    'reachable': False,
    'active_connections': 0,
    'peak_connections': 0,
    'publish_received': 0,
    'error': None,
}

def on_connect(client, userdata, flags, rc):
    if rc == 0:
        state['reachable'] = True
        state['error'] = None
        client.subscribe('$SYS/#')
    else:
        state['reachable'] = False
        state['error'] = f'Connect failed rc={rc}'

def on_disconnect(client, userdata, rc):
    state['reachable'] = False

def on_message(client, userdata, msg):
    val = int(msg.payload.decode() or 0)
    if msg.topic == '$SYS/broker/clients/connected':
        state['active_connections'] = val
        if val > state['peak_connections']:
            state['peak_connections'] = val
    elif msg.topic == '$SYS/broker/messages/publish/received':
        state['publish_received'] = val

def on_error(client, userdata, err):
    state['reachable'] = False
    state['error'] = str(err)

client = mqtt.Client(client_id=f'vigilmon-health', clean_session=True)
client.on_connect = on_connect
client.on_disconnect = on_disconnect
client.on_message = on_message
client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=30)
client.loop_start()

@app.route('/health/rumqttd')
def rumqttd_health():
    if not state['reachable']:
        return jsonify(
            status='down',
            reason='broker_unreachable',
            error=state['error'],
            host=MQTT_HOST,
            port=MQTT_PORT,
        ), 503

    active = state['active_connections']
    if CONN_BASELINE > 0 and active < CONN_BASELINE * (1 - CONN_DROP_PCT):
        return jsonify(
            status='degraded',
            reason='connection_count_drop',
            active=active,
            baseline=CONN_BASELINE,
            threshold_pct=CONN_DROP_PCT * 100,
        ), 503

    return jsonify(
        status='ok',
        connections=dict(active=active, peak=state['peak_connections'], baseline=CONN_BASELINE),
        messages=dict(publish_received=state['publish_received']),
    )

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

Step 2: Monitor Process Health via TCP Port Check

Before HTTP-level probes, verify that the MQTT listener port is reachable at all. A TCP check catches the case where the rumqttd process has crashed but the health sidecar hasn't detected it yet.

  1. Log in to vigilmon.online and go to Monitors → New Monitor
  2. Choose TCP Port as the type
  3. Enter your broker hostname and port 1883 (or 8883 for TLS)
  4. Check interval: 1 minute
  5. Save

If TLS is enabled, add a separate TCP check on port 8883 and on port 8083 for WebSocket.

| TCP Monitor | Port | What It Catches | |---|---|---| | MQTT plaintext | 1883 | rumqttd process crash, port binding failure | | MQTT TLS | 8883 | TLS listener crash or certificate binding failure | | MQTT WebSocket | 8083 | WebSocket listener crash (if enabled) |

Step 3: Configure HTTP Monitors for Broker Metrics

  1. In Vigilmon, go to Monitors → New Monitor → HTTP / HTTPS
  2. URL: https://your-server.example.com/health/rumqttd
  3. Check interval: 1 minute
  4. Under Expected response:
    • Status code: 200
    • Response body contains: "status":"ok"
    • Response time threshold: 2000ms
  5. Assign your primary alert channel
  6. Save

Add a second monitor for publish rate:

| Monitor URL | Purpose | Interval | |---|---|---| | /health/rumqttd | Process, connections, TLS cert | 1 min | | /health/rumqttd/publish-rate | Message publish rate | 2 min |

Step 4: Heartbeat Monitoring for MQTT Subscribers

Your rumqttd subscribers (data pipeline consumers, telemetry processors) can silently stall even when the broker is healthy. Heartbeat monitoring catches subscriber liveness.

  1. In Vigilmon, go to Monitors → New Monitor → Heartbeat
  2. Name: rumqttd-telemetry-subscriber
  3. Expected interval: 5 minutes
  4. Grace period: 10 minutes
  5. Copy the heartbeat URL

Wire the heartbeat into your subscriber:

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

const client = mqtt.connect(`mqtt://${process.env.RUMQTTD_HOST}:1883`);
const HB_URL = process.env.VIGILMON_HEARTBEAT_URL;
let msgCount = 0;

client.on('connect', () => client.subscribe('devices/+/telemetry'));

client.on('message', async (topic, payload) => {
  await processMessage(topic, payload);
  msgCount++;
  if (msgCount % 100 === 0) {
    await axios.get(HB_URL).catch(() => {});
  }
});

// Time-based fallback for low-volume topics
setInterval(async () => {
  if (msgCount > 0) await axios.get(HB_URL).catch(() => {});
}, 60_000);

Step 5: Alert Routing

| Monitor | Alert Channel | Priority | Condition | |---|---|---|---| | TCP: MQTT port 1883 | Slack + PagerDuty | P1 | Port unreachable — broker crash | | HTTP: /health/rumqttd | Slack + PagerDuty | P1 | Broker down or connections dropped >20% | | HTTP: /health/rumqttd (TLS warning) | Email | P2 | TLS cert expiring within 30 days | | HTTP: /health/rumqttd/publish-rate | Slack | P2 | Publish rate drops from baseline | | Heartbeat: subscriber | Slack + email | P2 | Subscriber stalled |

For production IoT fleets, set the TCP check grace period to 0 — a broker crash needs an immediate page, not a delayed alert. For TLS certificate expiry, configure a secondary email-only alert so on-call staff see it without waking the whole team.

Summary

rumqttd failures are invisible at the device level — connected clients keep trying to publish while the broker is gone. External monitoring with Vigilmon gives you end-to-end visibility across process health, connection counts, message throughput, QoS delivery, and TLS certificate validity.

| Monitor Type | What It Covers | |---|---| | TCP: port 1883/8883/8083 | Process crash, port binding failure | | HTTP: /health/rumqttd | Connections, baseline drop, TLS expiry | | HTTP: /health/rumqttd/publish-rate | Message throughput, publish rate baseline | | Heartbeat | Subscriber liveness, pipeline stall detection |

Get started free at vigilmon.online — your first rumqttd 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