How to Monitor EMQX Broker Health, Message Throughput, and Cluster Connectivity with Vigilmon

EMQX handles millions of concurrent MQTT connections for IoT fleets, connected vehicles, and real-time telemetry pipelines. When a broker node fails, a rule engine route breaks, or message throughput drops, connected devices keep publishing — but data stops flowing. The dashboard looks fine; your pipeline is silent.

Vigilmon gives you external visibility into EMQX health through HTTP probe monitoring of the EMQX Dashboard API and heartbeat monitoring for downstream data pipeline applications. This tutorial covers both.

Why EMQX Needs External Monitoring

EMQX ships with a management dashboard and REST API — but these only alert you if someone is watching. External monitoring with Vigilmon adds:

• Proactive alerting when broker nodes leave the cluster or become unreachable
• Throughput anomaly detection — catches drops in message-in/out rates before devices report issues
• Rule engine execution monitoring — detects failed data forwarding to databases, Kafka, or webhooks
• Subscriber count drift — catches mass disconnects that precede device fleet outages
• Multi-region probing from outside your network to catch infrastructure-layer failures

Step 1: Enable the EMQX REST API

EMQX exposes a Management API on port 18083 (EMQX 5.x) or port 8081 (EMQX 4.x). Enable it in your configuration:

# EMQX 5.x — verify the dashboard is running
curl http://localhost:18083/api/v5/status

Create a dedicated API key for monitoring:

1. Open the EMQX Dashboard at http://localhost:18083
2. Go to System → API Keys → Create
3. Name: vigilmon-monitoring
4. Permissions: Read-only
5. Copy the API key and secret

Test access:

curl -u <api-key>:<api-secret> http://localhost:18083/api/v5/nodes

Step 2: Build an EMQX Health Endpoint

Expose a composite health endpoint from your application tier that aggregates broker metrics from the EMQX REST API.

Node.js Health Sidecar

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

const app = express();

const EMQX_API = process.env.EMQX_API_URL || 'http://localhost:18083/api/v5';
const EMQX_KEY = process.env.EMQX_API_KEY;
const EMQX_SECRET = process.env.EMQX_API_SECRET;
const MIN_NODES = parseInt(process.env.EMQX_MIN_NODES || '1');
const MSG_DROP_THRESHOLD = parseInt(process.env.EMQX_MSG_DROP_THRESHOLD || '100');

const auth = { auth: { username: EMQX_KEY, password: EMQX_SECRET } };

app.get('/health/emqx', async (req, res) => {
  try {
    const nodesRes = await axios.get(`${EMQX_API}/nodes`, auth);
    const nodes = nodesRes.data;

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

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

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

app.get('/health/emqx/metrics', async (req, res) => {
  try {
    const statsRes = await axios.get(`${EMQX_API}/stats`, auth);
    const stats = statsRes.data;

    const connections = stats['connections.count'] || 0;
    const subscriptions = stats['subscriptions.count'] || 0;

    // Fetch message metrics
    const metricsRes = await axios.get(`${EMQX_API}/metrics`, auth);
    const metrics = metricsRes.data;

    const msgDropped = metrics.find(m => m.key === 'messages.dropped')?.value || 0;
    if (msgDropped > MSG_DROP_THRESHOLD) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'high_message_drop_rate',
        dropped: msgDropped,
        threshold: MSG_DROP_THRESHOLD,
      });
    }

    return res.status(200).json({
      status: 'ok',
      connections,
      subscriptions,
      msg_dropped: msgDropped,
      msg_sent: metrics.find(m => m.key === 'messages.sent')?.value || 0,
      msg_received: metrics.find(m => m.key === 'messages.received')?.value || 0,
    });
  } catch (err) {
    return res.status(503).json({ status: 'down', error: err.message });
  }
});

app.listen(3006);

Python Health Sidecar

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

app = Flask(__name__)

EMQX_API = os.environ.get('EMQX_API_URL', 'http://localhost:18083/api/v5')
AUTH = (os.environ['EMQX_API_KEY'], os.environ['EMQX_API_SECRET'])
MIN_NODES = int(os.environ.get('EMQX_MIN_NODES', 1))
DROP_THRESHOLD = int(os.environ.get('EMQX_MSG_DROP_THRESHOLD', 100))

@app.route('/health/emqx')
def emqx_health():
    try:
        nodes = requests.get(f'{EMQX_API}/nodes', auth=AUTH, timeout=5).json()
        running = [n for n in nodes if n.get('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 not n.get('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/emqx/metrics')
def emqx_metrics():
    try:
        stats = requests.get(f'{EMQX_API}/stats', auth=AUTH, timeout=5).json()
        metrics_raw = requests.get(f'{EMQX_API}/metrics', auth=AUTH, timeout=5).json()
        metrics = {m['key']: m['value'] for m in metrics_raw}

        dropped = metrics.get('messages.dropped', 0)
        if dropped > DROP_THRESHOLD:
            return jsonify(status='degraded', reason='high_drop_rate', dropped=dropped), 503

        return jsonify(
            status='ok',
            connections=stats.get('connections.count', 0),
            subscriptions=stats.get('subscriptions.count', 0),
            msg_dropped=dropped,
            msg_sent=metrics.get('messages.sent', 0),
            msg_received=metrics.get('messages.received', 0),
        )
    except Exception as e:
        return jsonify(status='down', error=str(e)), 503

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

Step 3: Monitor Rule Engine Execution

EMQX rule engine routes messages to databases, Kafka topics, HTTP webhooks, and other bridges. A broken rule engine action silently drops data without affecting MQTT connectivity.

Extend your health sidecar to check rule engine execution statistics:

app.get('/health/emqx/rules', async (req, res) => {
  try {
    const rulesRes = await axios.get(`${EMQX_API}/rules`, auth);
    const rules = rulesRes.data;

    const brokenRules = rules.filter(r => r.enabled && r.status?.status === 'disconnected');
    if (brokenRules.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'rule_engine_disconnected',
        rules: brokenRules.map(r => ({ id: r.id, name: r.name, status: r.status?.status })),
      });
    }

    // Check for high failure rates in rule metrics
    const metrics_r = await axios.get(`${EMQX_API}/rules/metrics`, auth);
    const metrics = metrics_r.data;
    const highFailure = metrics.filter(m => {
      const total = (m.matched || 0);
      const failed = (m.failed || 0);
      return total > 10 && (failed / total) > 0.1;
    });

    if (highFailure.length > 0) {
      return res.status(503).json({
        status: 'degraded',
        reason: 'rule_engine_high_failure_rate',
        rules: highFailure.map(m => ({ id: m.id, matched: m.matched, failed: m.failed })),
      });
    }

    return res.status(200).json({
      status: 'ok',
      total_rules: rules.length,
      enabled_rules: rules.filter(r => r.enabled).length,
    });
  } 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. Set the URL: https://your-app.example.com/health/emqx
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 for P1 incidents
7. Save

Add monitors for each additional endpoint:

| Monitor URL | Purpose | Interval | Priority | |---|---|---|---| | /health/emqx | Cluster node health | 1 min | P1 | | /health/emqx/metrics | Connection count, message drop rate | 1 min | P2 | | /health/emqx/rules | Rule engine execution health | 2 min | P2 |

Step 5: Heartbeat Monitoring for EMQX Data Pipeline Consumers

EMQX routes messages to downstream consumers — time-series databases, analytics pipelines, webhook forwarders. These consumers can stall silently while the broker looks healthy.

Set Up the Heartbeat Monitor

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

Wire Into Your Pipeline Consumer

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

const client = mqtt.connect(process.env.EMQX_BROKER_URL);
const HEARTBEAT_URL = process.env.VIGILMON_HEARTBEAT_URL;

let msgCount = 0;
const HEARTBEAT_EVERY = 200;

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

client.on('message', async (topic, payload) => {
  try {
    await storeTelemetry(topic, payload);
    msgCount++;

    if (msgCount % HEARTBEAT_EVERY === 0) {
      await axios.get(HEARTBEAT_URL).catch(() => {});
    }
  } catch (err) {
    console.error('Pipeline error:', err.message);
  }
});

For sparse IoT topics where messages arrive infrequently, use a time-based heartbeat:

// Ping Vigilmon every 60 seconds if last message was recent
setInterval(async () => {
  const secondsSinceLast = (Date.now() - lastMsgAt) / 1000;
  if (secondsSinceLast < 300) { // active within last 5 minutes
    await axios.get(HEARTBEAT_URL).catch(() => {});
  }
}, 60_000);

Step 6: Alert Routing for IoT Scale

At IoT scale, brief subscriber count drops and message rate fluctuations are normal — you want to alert on sustained anomalies, not transient spikes. Configure alert routing in Vigilmon:

| Monitor | Alert Channel | Priority | |---|---|---| | Cluster node health | Slack + PagerDuty | P1 | | Message drop rate | Slack | P2 | | Rule engine health | Slack | P2 | | Heartbeat: telemetry pipeline | Slack + email | P2 | | Heartbeat: command dispatcher | PagerDuty | P1 |

Set response time thresholds:

• Alert at 2000ms for cluster health (EMQX API slowness precedes node pressure)
• Alert at 1000ms for metrics endpoint (should respond instantly)

For command-and-control topics where latency matters (vehicle fleets, industrial equipment), use a tighter heartbeat interval (2 minutes) and a shorter grace period (5 minutes) for the command dispatcher consumer.

Summary

EMQX failures are silent from the device perspective — connected clients keep publishing while data stops flowing downstream. External monitoring with Vigilmon gives you end-to-end visibility:

| Monitor Type | What It Covers | |---|---| | HTTP: /health/emqx | Cluster node health, running node count | | HTTP: /health/emqx/metrics | Connections, subscriptions, message drop rate | | HTTP: /health/emqx/rules | Rule engine execution and failure rates | | Heartbeat | Pipeline consumer liveness, data processing health |

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