MQTT-Broker Health Check: Skripte für das Uptime-Monitoring

Ein Health Check für deinen MQTT-Broker sollte genau das beweisen, wovon deine Geräte abhängen: Ein Client kann sich verbinden, abonnieren, publishen und seine eigene Nachricht innerhalb weniger Sekunden empfangen. Alles darunter – ein Ping, ein offener Port, ein laufender Prozess – kann grün bleiben, während der Broker still Logins ablehnt oder nichts mehr routet.

Unten findest du drei einsatzbereite Skripte (Bash, Python, Node.js), die diesen Round Trip ausführen und Monitoring-taugliche Exit-Codes liefern, plus Beispiele für Cron, Kubernetes-Probes und Alerting.

Was ein Health Check prüfen muss

Die letzte Zeile ist die, die du automatisieren solltest. Jeder Lauf sollte eine eindeutige Client-ID verwenden (zwei Checks mit derselben ID werfen sich gegenseitig raus – siehe Best Practices für MQTT-Client-IDs) und ein eindeutiges Topic, damit eine verspätete Nachricht aus einem früheren Lauf den aktuellen Lauf nie gesund aussehen lässt.

Wenn du statt eines wiederkehrenden Probes eine umfassendere Einmal-Diagnose willst, fang mit MQTT-Broker-Status prüfen an oder teste den Broker interaktiv im Online-MQTT-Client.

Exit-Codes, die Monitoring-Tools verstehen

Alle Skripte in diesem Artikel folgen der Nagios-Plugin-Konvention, die Icinga, Zabbix, Checkmk, Sensu und die meisten Cron-Wrapper verstehen:

Kubernetes, systemd und Cron unterscheiden nur null und ungleich null, deshalb funktionieren dieselben Skripte überall.

Bash: mosquitto_sub + mosquitto_pub

Der schnellste Check nutzt die Kommandozeilen-Clients von Mosquitto (apt install mosquitto-clients oder brew install mosquitto). Ein Subscriber im Hintergrund wartet mit Timeout auf genau eine Nachricht, und der Publisher schickt ein zufälliges Token an dasselbe Topic:

#!/usr/bin/env bash
# Round-trip health check. Exit: 0 OK, 2 CRITICAL, 3 UNKNOWN.
HOST="${MQTT_HOST:-broker.connectmq.com}"
PORT="${MQTT_PORT:-8883}"
TIMEOUT="${MQTT_TIMEOUT:-5}"
TOPIC="testmqtt/health/$(hostname)-$$"
TOKEN="ok-$(date +%s)-$RANDOM"

command -v mosquitto_pub >/dev/null || { echo "UNKNOWN - mosquitto clients not installed"; exit 3; }

OPTS=(-h "$HOST" -p "$PORT")
[ "$PORT" = "8883" ] && OPTS+=(--capath /etc/ssl/certs)
[ -n "$MQTT_USER" ] && OPTS+=(-u "$MQTT_USER" -P "$MQTT_PASS")

OUT=$(mktemp)
mosquitto_sub "${OPTS[@]}" -i "health-sub-$$" -t "$TOPIC" -q 1 -C 1 -W "$TIMEOUT" >"$OUT" 2>&1 &
SUB_PID=$!
sleep 1   # give the subscriber time to receive its SUBACK

if ! ERR=$(mosquitto_pub "${OPTS[@]}" -i "health-pub-$$" -t "$TOPIC" -q 1 -m "$TOKEN" 2>&1); then
  kill "$SUB_PID" 2>/dev/null
  echo "CRITICAL - publish failed: $ERR"; rm -f "$OUT"; exit 2
fi

wait "$SUB_PID"
RECEIVED=$(cat "$OUT"); rm -f "$OUT"

if [ "$RECEIVED" = "$TOKEN" ]; then
  echo "OK - round trip via $HOST:$PORT"; exit 0
fi
echo "CRITICAL - no message within ${TIMEOUT}s (${RECEIVED:-no output})"; exit 2

Wichtige Details:

• -C 1 lässt mosquitto_sub nach einer Nachricht beenden; -W lässt es nach dem Timeout aufgeben, statt ewig zu hängen.
• Das Skript vergleicht die Payload mit dem Token, statt sich nur auf den Exit-Status zu verlassen – so kann eine verirrte Nachricht auf dem Topic kein falsches OK erzeugen.
• Fehlertexte der Clients (zum Beispiel Connection error: Connection Refused: not authorised.) landen in der Ausgabezeile – genau das, was du in einem Alert sehen willst. Jeden Reason Code kannst du im MQTT-Reason-Code-Tool nachschlagen.
• /etc/ssl/certs ist das CA-Verzeichnis unter Debian, Ubuntu und Alpine. Auf anderen Systemen übergib --cafile mit deinem CA-Bundle.

Python: Round Trip mit Latenzschwelle

Die Python-Version (paho-mqtt 2.x) misst die Round-Trip-Zeit und liefert WARNING, wenn sie langsamer als eine Schwelle ist. Jeder Schritt hat eine Deadline, sodass das Skript nie einen Monitoring-Agenten blockiert:

#!/usr/bin/env python3
"""MQTT round-trip health check. Exit: 0 OK, 1 WARNING, 2 CRITICAL."""
import os, sys, threading, time, uuid
import paho.mqtt.client as mqtt

HOST = os.getenv("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.getenv("MQTT_PORT", "8883"))
TIMEOUT = float(os.getenv("MQTT_TIMEOUT", "5"))
WARN_MS = float(os.getenv("MQTT_WARN_MS", "500"))

topic = f"testmqtt/health/{uuid.uuid4().hex[:12]}"
token = uuid.uuid4().hex
connected, subscribed, received = threading.Event(), threading.Event(), threading.Event()
state = {"connack": None, "suback": None, "t0": 0.0, "rtt": 0.0}


def on_connect(client, userdata, flags, reason_code, properties):
    state["connack"] = reason_code
    connected.set()
    if not reason_code.is_failure:
        client.subscribe(topic, qos=1)


def on_subscribe(client, userdata, mid, reason_codes, properties):
    state["suback"] = reason_codes[0]
    subscribed.set()
    if not reason_codes[0].is_failure:
        state["t0"] = time.perf_counter()
        client.publish(topic, token, qos=1)


def on_message(client, userdata, msg):
    if msg.payload.decode() == token:
        state["rtt"] = (time.perf_counter() - state["t0"]) * 1000
        received.set()


def finish(code, text):
    print(text)
    sys.exit(code)


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"health-{uuid.uuid4().hex[:8]}")
client.on_connect, client.on_subscribe, client.on_message = on_connect, on_subscribe, on_message
if os.getenv("MQTT_USER"):
    client.username_pw_set(os.environ["MQTT_USER"], os.getenv("MQTT_PASS"))
if PORT == 8883:
    client.tls_set()  # system CA bundle

try:
    client.connect(HOST, PORT, keepalive=30)
except Exception as exc:  # DNS, refused, TLS errors, timeouts
    finish(2, f"CRITICAL - cannot connect to {HOST}:{PORT}: {exc}")

client.loop_start()
try:
    if not connected.wait(TIMEOUT):
        finish(2, "CRITICAL - no CONNACK")
    if state["connack"].is_failure:
        finish(2, f"CRITICAL - connection refused: {state['connack']}")
    if not subscribed.wait(TIMEOUT):
        finish(2, "CRITICAL - no SUBACK")
    if state["suback"].is_failure:
        finish(2, f"CRITICAL - subscribe refused: {state['suback']}")
    if not received.wait(TIMEOUT):
        finish(2, f"CRITICAL - message not delivered within {TIMEOUT:.0f}s")
    rtt = state["rtt"]
    level, code = ("WARNING", 1) if rtt > WARN_MS else ("OK", 0)
    finish(code, f"{level} - round trip {rtt:.0f} ms | rtt={rtt:.1f}ms;{WARN_MS:.0f}")
finally:
    client.disconnect()
    client.loop_stop()

Der Text nach | sind Nagios-Performance-Daten – Tools, die sie grafisch darstellen, zeichnen die Round-Trip-Zeit also gratis mit. CONNACK-, SUBACK- und Zustellungsfehler zu trennen, zahlt sich um 3 Uhr nachts aus: „subscribe refused" deutet auf eine ACL-Änderung, „no CONNACK" auf einen überlasteten oder halbtoten Broker, „not delivered" auf Routing oder Bridging. Wenn Subscriptions unerwartet abgelehnt werden, erklärt MQTT-ACL lehnt still ab die üblichen Ursachen.

Node.js: MQTT.js 5

Dieselbe Logik mit MQTT.js 5 und seiner Promise-API. Ein einziger globaler Timer garantiert, dass der Prozess endet, auch wenn ein Callback nie feuert:

import mqtt from 'mqtt';
import { randomUUID } from 'node:crypto';

const url = process.env.MQTT_URL ?? 'mqtts://broker.connectmq.com:8883';
const timeoutMs = Number(process.env.MQTT_TIMEOUT_MS ?? 5000);
const warnMs = Number(process.env.MQTT_WARN_MS ?? 500);
const topic = `testmqtt/health/${randomUUID().slice(0, 12)}`;
const token = randomUUID();

function finish(code, text) {
  console.log(text);
  process.exit(code);
}
setTimeout(() => finish(2, `CRITICAL - no round trip within ${timeoutMs} ms`), timeoutMs);

const client = mqtt.connect(url, {
  clientId: `health-${randomUUID().slice(0, 8)}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  connectTimeout: timeoutMs,
  reconnectPeriod: 0, // a health check must not retry silently
});

client.on('error', (err) => finish(2, `CRITICAL - ${err.message}`));

client.on('connect', async () => {
  try {
    const granted = await client.subscribeAsync(topic, { qos: 1 });
    if (!granted[0] || granted[0].qos > 2) finish(2, 'CRITICAL - subscribe refused');

    let t0 = 0;
    client.on('message', (_topic, payload) => {
      if (payload.toString() !== token) return;
      const ms = Math.round(performance.now() - t0);
      const [code, level] = ms > warnMs ? [1, 'WARNING'] : [0, 'OK'];
      client.end(true, () => finish(code, `${level} - round trip ${ms} ms | rtt=${ms}ms;${warnMs}`));
    });
    t0 = performance.now();
    await client.publishAsync(topic, token, { qos: 1 });
  } catch (err) {
    finish(2, `CRITICAL - ${err.message}`);
  }
});

reconnectPeriod: 0 ist wichtig: Standardmäßig verbindet sich MQTT.js endlos neu – großartig für Anwendungen, fatal für einen Probe, der schnell fehlschlagen soll. Falsche Zugangsdaten erscheinen als Connection refused: Not authorized oder Connection refused: Bad username or password, je nachdem, wie der Broker sie meldet.

Den Check zeitgesteuert ausführen

Cron mit Alerting

# every minute; alert only on non-zero exit
* * * * * /opt/checks/mqtt-health.sh >/tmp/mqtt-health.log 2>&1 || /opt/checks/notify.sh "$(cat /tmp/mqtt-health.log)"

Die meisten Monitoring-Agenten (Nagios, Icinga, Checkmk, Zabbix mit External Checks) können jedes dieser Skripte direkt als Plugin ausführen. Uptime-Dienste mit einem MQTT-Monitortyp funktionieren auch, aber ein Skript gibt dir die Kontrolle über Topics, QoS und Schwellenwerte. Für ein vollständiges Dashboard kombinierst du den Probe mit Broker-Metriken, wie in MQTT-Uptime-Monitoring und Monitoring mit Prometheus und Grafana beschrieben.

Kubernetes-Probes

Für einen selbst betriebenen Mosquitto-Pod ist ein exec-Probe mit lokalem Round Trip deutlich nützlicher als ein tcpSocket-Probe. Das offizielle eclipse-mosquitto-Image enthält die Kommandozeilen-Clients bereits:

containers:
  - name: mosquitto
    image: eclipse-mosquitto:2
    ports:
      - containerPort: 1883
    readinessProbe:
      tcpSocket:
        port: 1883
      periodSeconds: 5
    livenessProbe:
      exec:
        command:
          - sh
          - -c
          - >
            mosquitto_sub -h 127.0.0.1 -i k8s-probe-sub -t testmqtt/health/k8s -C 1 -W 5 &
            sleep 1;
            mosquitto_pub -h 127.0.0.1 -i k8s-probe-pub -t testmqtt/health/k8s -m ok;
            wait $!
      initialDelaySeconds: 10
      periodSeconds: 30
      timeoutSeconds: 8
      failureThreshold: 3

Wenn dein Listener Authentifizierung verlangt, ergänze -u/-P aus einem Secret, das als Umgebungsvariablen eingebunden ist, und erlaube den Probe-User in deinen ACLs. Halte timeoutSeconds größer als den Timeout des Skripts und setze failureThreshold so, dass ein einzelner langsamer Check während einer Garbage-Collection-Pause den Broker nicht neu startet. Mehr Details zum Deployment findest du in Mosquitto auf Kubernetes.

Einen Managed Broker überwachen

Bei einem Managed Broker kontrollierst du den Prozess nicht, aber die Client-Seite des Vertrags gehört weiterhin dir: dein Hostname, deine Zugangsdaten, deine ACLs und der Netzwerkpfad von deinen Geräten. Ein externer Round-Trip-Check erkennt abgelaufene Zugangsdaten, versehentliche ACL-Änderungen und Netzwerkprobleme auf deiner Seite, die das Monitoring des Anbieters nicht sehen kann.

Den Check selbst testen

Ein Health Check, der nie anschlägt, ist schlimmer als keiner. Bevor du dich darauf verlässt, provoziere jeden Fehlerfall einmal:

1. Führe ihn normal aus und bestätige Exit-Code 0: ./mqtt-health.sh; echo $?.
2. Verwende ein falsches Passwort und bestätige 2 mit einer „not authorised"-Meldung.
3. Richte ihn auf einen geschlossenen Port (MQTT_PORT=1999) und bestätige, dass er innerhalb des Timeouts fehlschlägt, statt zu hängen.
4. Setze MQTT_WARN_MS=1 beim Python- oder Node-Skript und bestätige, dass der WARNING-Pfad 1 liefert.
5. Beobachte testmqtt/health/# im Online-Client, während der Check läuft, und sieh zu, wie das Token ankommt.

Health-Check-Nachrichten live ansehen

Ein guter MQTT Health Check ist klein, schnell und ehrlich: eine Verbindung, eine Subscription, eine Nachricht, ein strikter Timeout und ein klarer Exit-Code. Leg das Skript neben deiner Broker-Konfiguration ins Repository, führe es innerhalb und außerhalb deines Netzwerks aus, und du erfährst von Broker-Problemen, bevor deine Geräte es tun.

Häufig gestellte Fragen