How to Monitor Mosquitto Broker Uptime, Client Count, and TLS Expiry with Vigilmon

Mosquitto is the reference MQTT broker — lightweight, fast, and deployed in millions of IoT systems from Raspberry Pis to industrial edge nodes. But when Mosquitto goes down, connected clients don't get an HTTP error — they lose connection silently and start retrying, often indefinitely, while your data pipeline receives nothing.

Vigilmon gives you external visibility into Mosquitto uptime through TCP port probing, HTTP health sidecars, and heartbeat monitoring for consumer applications. This tutorial covers all three.

Why Mosquitto Needs External Monitoring

Mosquitto is intentionally minimal — no built-in HTTP API, no dashboard, no management plane. This simplicity is its strength, but it means you have no built-in alerting. External monitoring with Vigilmon adds:

• Broker uptime monitoring via TCP port probe (port 1883 / 8883) — alerts within seconds when Mosquitto stops accepting connections
• Connected client count tracking via the $SYS topic published by Mosquitto itself
• Message throughput monitoring — catch rate drops before devices report delivery failures
• TLS certificate expiry alerting — catch expiring certificates before clients fail to handshake
• Consumer heartbeat monitoring — detect silent stalls in applications that subscribe to Mosquitto

Step 1: Enable the Mosquitto $SYS Topic

Mosquitto publishes broker statistics to a set of topics under the $SYS hierarchy. Enable them in mosquitto.conf:

# mosquitto.conf
sys_interval 10

# Allow local stats subscription (lock down in production)
listener 1883 0.0.0.0

# For TLS
listener 8883 0.0.0.0
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key
cafile /etc/mosquitto/certs/ca.crt

Key $SYS topics published by Mosquitto:

| Topic | Description | |---|---| | $SYS/broker/uptime | Broker uptime string | | $SYS/broker/clients/connected | Current connected client count | | $SYS/broker/clients/total | Total client connections since start | | $SYS/broker/messages/sent | Total messages sent | | $SYS/broker/messages/received | Total messages received | | $SYS/broker/subscriptions/count | Current active subscription count | | $SYS/broker/bytes/sent | Total bytes sent | | $SYS/broker/bytes/received | Total bytes received | | $SYS/broker/publish/messages/dropped | Messages dropped (queue overflow) |

Step 2: Build a Mosquitto Health Endpoint

Since Mosquitto has no HTTP API, you need a small health sidecar that subscribes to $SYS topics and exposes the metrics over HTTP.

Node.js Health Sidecar

// health/mosquitto.js
const express = require('express');
const mqtt = require('mqtt');

const app = express();

const BROKER_URL = process.env.MOSQUITTO_BROKER_URL || 'mqtt://localhost:1883';
const CLIENT_COUNT_MIN = parseInt(process.env.MIN_CLIENTS || '0');
const DROP_THRESHOLD = parseInt(process.env.DROP_THRESHOLD || '10');

const metrics = {
  uptime: null,
  clients_connected: null,
  clients_total: null,
  messages_sent: null,
  messages_received: null,
  messages_dropped: null,
  subscriptions: null,
  bytes_sent: null,
  bytes_received: null,
  last_updated: null,
  broker_reachable: false,
};

const client = mqtt.connect(BROKER_URL, {
  clientId: `health-sidecar-${process.pid}`,
  clean: true,
  reconnectPeriod: 5000,
  connectTimeout: 10000,
});

client.on('connect', () => {
  metrics.broker_reachable = true;
  client.subscribe('$SYS/broker/#');
});

client.on('error', (err) => {
  metrics.broker_reachable = false;
  console.error('Mosquitto connection error:', err.message);
});

client.on('close', () => {
  metrics.broker_reachable = false;
});

client.on('message', (topic, payload) => {
  const value = payload.toString();
  metrics.last_updated = Date.now();

  switch (topic) {
    case '$SYS/broker/uptime':
      metrics.uptime = value;
      break;
    case '$SYS/broker/clients/connected':
      metrics.clients_connected = parseInt(value);
      break;
    case '$SYS/broker/clients/total':
      metrics.clients_total = parseInt(value);
      break;
    case '$SYS/broker/messages/sent':
      metrics.messages_sent = parseInt(value);
      break;
    case '$SYS/broker/messages/received':
      metrics.messages_received = parseInt(value);
      break;
    case '$SYS/broker/publish/messages/dropped':
      metrics.messages_dropped = parseInt(value);
      break;
    case '$SYS/broker/subscriptions/count':
      metrics.subscriptions = parseInt(value);
      break;
    case '$SYS/broker/bytes/sent':
      metrics.bytes_sent = parseInt(value);
      break;
    case '$SYS/broker/bytes/received':
      metrics.bytes_received = parseInt(value);
      break;
  }
});

app.get('/health/mosquitto', (req, res) => {
  const staleSec = metrics.last_updated
    ? (Date.now() - metrics.last_updated) / 1000
    : null;

  if (!metrics.broker_reachable) {
    return res.status(503).json({ status: 'down', reason: 'broker_unreachable' });
  }

  if (staleSec !== null && staleSec > 30) {
    return res.status(503).json({
      status: 'degraded',
      reason: 'sys_topics_stale',
      last_updated_secs_ago: staleSec,
    });
  }

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

  return res.status(200).json({
    status: 'ok',
    uptime: metrics.uptime,
    clients_connected: metrics.clients_connected,
    subscriptions: metrics.subscriptions,
    messages_sent: metrics.messages_sent,
    messages_received: metrics.messages_received,
    messages_dropped: metrics.messages_dropped,
  });
});

app.listen(3008);

Python Health Sidecar

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

app = Flask(__name__)

BROKER_HOST = os.environ.get('MOSQUITTO_HOST', 'localhost')
BROKER_PORT = int(os.environ.get('MOSQUITTO_PORT', 1883))
DROP_THRESHOLD = int(os.environ.get('DROP_THRESHOLD', 10))

state = {
    'reachable': False,
    'last_updated': None,
    'clients_connected': None,
    'messages_dropped': None,
    'subscriptions': None,
    'uptime': None,
    'messages_sent': None,
    'messages_received': None,
}

def on_connect(client, userdata, flags, rc):
    state['reachable'] = rc == 0
    if rc == 0:
        client.subscribe('$SYS/broker/#')

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

def on_message(client, userdata, msg):
    state['last_updated'] = time.time()
    val = msg.payload.decode()
    topic_map = {
        '$SYS/broker/uptime': 'uptime',
        '$SYS/broker/clients/connected': 'clients_connected',
        '$SYS/broker/publish/messages/dropped': 'messages_dropped',
        '$SYS/broker/subscriptions/count': 'subscriptions',
        '$SYS/broker/messages/sent': 'messages_sent',
        '$SYS/broker/messages/received': 'messages_received',
    }
    key = topic_map.get(msg.topic)
    if key:
        try:
            state[key] = int(val) if key != 'uptime' else val
        except ValueError:
            state[key] = val

mqttc = mqtt.Client(client_id=f'health-sidecar-{os.getpid()}')
mqttc.on_connect = on_connect
mqttc.on_disconnect = on_disconnect
mqttc.on_message = on_message
mqttc.connect_async(BROKER_HOST, BROKER_PORT)
threading.Thread(target=mqttc.loop_forever, daemon=True).start()

@app.route('/health/mosquitto')
def health():
    if not state['reachable']:
        return jsonify(status='down', reason='broker_unreachable'), 503

    stale = (time.time() - state['last_updated']) if state['last_updated'] else None
    if stale and stale > 30:
        return jsonify(status='degraded', reason='sys_topics_stale', stale_secs=stale), 503

    dropped = state.get('messages_dropped') or 0
    if dropped > DROP_THRESHOLD:
        return jsonify(status='degraded', reason='high_drop_rate',
                       dropped=dropped, threshold=DROP_THRESHOLD), 503

    return jsonify(
        status='ok',
        uptime=state['uptime'],
        clients_connected=state['clients_connected'],
        subscriptions=state['subscriptions'],
        messages_sent=state['messages_sent'],
        messages_received=state['messages_received'],
        messages_dropped=dropped,
    )

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

Step 3: Monitor TLS Certificate Expiry

For Mosquitto instances with TLS (port 8883), certificate expiry is a common operational failure — the broker starts refusing handshakes and devices can't reconnect after their session expires.

Expose Certificate Expiry from the Health Sidecar

const tls = require('tls');

app.get('/health/mosquitto/tls', (req, res) => {
  const options = {
    host: process.env.MOSQUITTO_TLS_HOST || 'localhost',
    port: parseInt(process.env.MOSQUITTO_TLS_PORT || '8883'),
    servername: process.env.MOSQUITTO_TLS_HOST || 'localhost',
    rejectUnauthorized: false, // We want to check even near-expired certs
  };

  const socket = tls.connect(options, () => {
    const cert = socket.getPeerCertificate();
    socket.destroy();

    if (!cert || !cert.valid_to) {
      return res.status(503).json({ status: 'degraded', reason: 'no_certificate_returned' });
    }

    const expiresAt = new Date(cert.valid_to);
    const daysRemaining = Math.floor((expiresAt - Date.now()) / (1000 * 60 * 60 * 24));
    const WARNING_DAYS = parseInt(process.env.TLS_WARNING_DAYS || '30');

    if (daysRemaining <= 0) {
      return res.status(503).json({ status: 'down', reason: 'tls_certificate_expired', days_remaining: 0 });
    }

    if (daysRemaining <= WARNING_DAYS) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'tls_certificate_expiring_soon',
        days_remaining: daysRemaining,
        expires_at: expiresAt.toISOString(),
      });
    }

    return res.status(200).json({
      status: 'ok',
      days_remaining: daysRemaining,
      expires_at: expiresAt.toISOString(),
      subject: cert.subject?.CN,
    });
  });

  socket.on('error', (err) => {
    return res.status(503).json({ status: 'down', reason: 'tls_connection_failed', error: err.message });
  });

  socket.setTimeout(5000, () => {
    socket.destroy();
    return res.status(503).json({ status: 'down', reason: 'tls_connection_timeout' });
  });
});

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/mosquitto
4. Check interval: 1 minute
5. Under Expected response:
   • Status code: 200
   • Response body contains: "status":"ok"
   • Response time threshold: 3000ms
6. Assign your PagerDuty alert channel
7. Save

Add monitors for TLS and message rates:

| Monitor URL | Purpose | Interval | Priority | |---|---|---|---| | /health/mosquitto | Broker uptime, client count, drop rate | 1 min | P1 | | /health/mosquitto/tls | TLS certificate expiry | 30 min | P2 |

For the TLS monitor, set a response body check for "status":"ok" — any other response (including 503 with days_remaining near zero) triggers your alert.

Alternatively, Vigilmon has a built-in SSL/TLS expiry monitor type that checks certificates directly without requiring a health sidecar. You can configure this in Monitors → New Monitor → SSL Certificate and point it at your Mosquitto TLS listener.

Step 5: Heartbeat Monitoring for Mosquitto Consumers

Mosquitto consumers — telemetry ingestors, event processors, alerting subscribers — can stall silently while the broker stays healthy. Heartbeat monitoring catches consumer stalls independently of broker health.

Set Up the Heartbeat Monitor

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

Wire Into a Node.js Consumer

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

const broker = mqtt.connect(process.env.MOSQUITTO_BROKER_URL);
const HEARTBEAT_URL = process.env.VIGILMON_HEARTBEAT_URL;

let lastMsgAt = null;

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

broker.on('message', async (topic, payload) => {
  try {
    await processTelemetry(topic, payload);
    lastMsgAt = Date.now();
  } catch (err) {
    console.error('Processing error:', err.message);
  }
});

// Heartbeat: ping Vigilmon every 60s if messages were recent
setInterval(async () => {
  if (lastMsgAt && Date.now() - lastMsgAt < 300_000) {
    await axios.get(HEARTBEAT_URL).catch(() => {});
  }
}, 60_000);

Wire Into a Python Consumer

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

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

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

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

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

client = mqtt.Client()
client.on_message = on_message
client.connect(os.environ['MOSQUITTO_HOST'], int(os.environ.get('MOSQUITTO_PORT', 1883)))
client.subscribe('sensors/+/telemetry', qos=1)
client.loop_forever()

Step 6: Alert Routing

Mosquitto failures tend to be hard binary failures rather than gradual degradation. A broker restart disconnects all clients simultaneously, causing a reconnect storm that can appear as a message drop spike before clients re-establish sessions.

Recommended Vigilmon alert routing:

| Monitor | Alert Channel | Priority | |---|---|---| | Broker uptime / client count | Slack + PagerDuty | P1 | | TLS certificate expiry | Slack + email | P2 | | Heartbeat: telemetry consumer | Slack + email | P2 |

Response time thresholds:

• Alert at 3000ms for the health sidecar (MQTT $SYS subscription should always be fast)
• Alert at 5000ms for the TLS check (TLS handshake has inherent overhead)

For Mosquitto instances that serve IoT devices on intermittent connections (battery-powered sensors, mobile assets), expect normal fluctuations in clients_connected — focus your alerting on sustained drops below your minimum expected client count, not individual connection changes.

Summary

Mosquitto's simplicity is its greatest strength — and its monitoring blind spot. Without an HTTP API or built-in alerting, you need external monitoring to catch broker failures, TLS expiry, and consumer stalls before they impact your device fleet.

| Monitor Type | What It Covers | |---|---| | HTTP: /health/mosquitto | Broker reachability, client count, message drop rate | | HTTP: /health/mosquitto/tls | TLS certificate expiry and handshake health | | Heartbeat | Consumer liveness, telemetry processing health |

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