Health check d'un broker MQTT : scripts de surveillance de disponibilité

Un health check de broker MQTT doit prouver la seule chose dont vos appareils dépendent : un client peut se connecter, s'abonner, publier et recevoir son propre message en quelques secondes. Tout ce qui en fait moins — un ping, un port ouvert, un processus en cours d'exécution — peut rester au vert pendant que le broker refuse silencieusement les connexions ou cesse de router.

Voici trois scripts prêts à l'emploi (Bash, Python, Node.js) qui effectuent cet aller-retour et renvoient des codes de sortie compatibles avec vos outils de surveillance, ainsi que des exemples pour cron, les sondes Kubernetes et l'alerting.

Ce qu'un health check doit vérifier

C'est la dernière ligne qu'il faut automatiser. Chaque exécution doit utiliser un client ID unique (deux vérifications avec le même ID se déconnectent mutuellement — voir bonnes pratiques pour le client ID MQTT) et un topic unique, afin qu'un message retardé d'une exécution précédente ne puisse jamais faire paraître saine l'exécution en cours.

Si vous cherchez un diagnostic ponctuel plus large plutôt qu'une sonde récurrente, commencez par vérifier l'état d'un broker MQTT ou explorez le broker de façon interactive dans le client MQTT en ligne.

Des codes de sortie compris par les outils de surveillance

Tous les scripts de cet article suivent la convention des plugins Nagios, comprise par Icinga, Zabbix, Checkmk, Sensu et la plupart des wrappers cron :

Kubernetes, systemd et cron ne distinguent que zéro et non zéro : les mêmes scripts fonctionnent donc partout.

Bash : mosquitto_sub + mosquitto_pub

La vérification la plus rapide utilise les clients en ligne de commande de Mosquitto (apt install mosquitto-clients ou brew install mosquitto). Un subscriber en arrière-plan attend exactement un message avec un timeout, et le publisher envoie un jeton aléatoire sur le même 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

Points essentiels :

• -C 1 fait quitter mosquitto_sub après un message ; -W le fait abandonner après le timeout au lieu de rester bloqué indéfiniment.
• Le script compare le payload au jeton au lieu de se fier au seul code de sortie : un message égaré sur le topic ne peut donc pas produire un faux OK.
• Le texte d'erreur des clients (par exemple Connection error: Connection Refused: not authorised.) se retrouve dans la ligne de sortie, exactement ce que vous voulez voir dans une alerte. Recherchez n'importe quel reason code avec l'outil de reason codes MQTT.
• /etc/ssl/certs est le répertoire des CA sous Debian, Ubuntu et Alpine. Sur d'autres systèmes, passez --cafile avec votre bundle de CA.

Python : aller-retour avec seuil de latence

La version Python (paho-mqtt 2.x) mesure le temps d'aller-retour et renvoie WARNING lorsqu'il dépasse un seuil. Chaque étape a une échéance, si bien que le script ne bloque jamais un agent de surveillance :

#!/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()

Le texte après | correspond aux données de performance Nagios : les outils qui les tracent afficheront gratuitement le temps d'aller-retour. Distinguer les échecs de CONNACK, de SUBACK et de livraison compte à 3 heures du matin : « subscribe refused » oriente vers une modification d'ACL, « no CONNACK » vers un broker surchargé ou à moitié mort, « not delivered » vers le routage ou un bridge. Si des abonnements sont refusés de façon inattendue, ACL MQTT refusée en silence explique les causes habituelles.

Node.js : MQTT.js 5

La même logique avec MQTT.js 5 et son API à base de promesses. Un unique timer global garantit que le processus se termine même si un callback ne se déclenche jamais :

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}`);
  }
});

Définir reconnectPeriod: 0 est important : par défaut, MQTT.js se reconnecte indéfiniment, ce qui est parfait pour une application et désastreux pour une sonde qui doit échouer vite. De mauvais identifiants apparaissent sous la forme Connection refused: Not authorized ou Connection refused: Bad username or password, selon la façon dont le broker les signale.

Planifier la vérification

Cron avec 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)"

La plupart des agents de surveillance (Nagios, Icinga, Checkmk, Zabbix avec external checks) peuvent exécuter directement ces scripts comme plugins. Les services d'uptime proposant un type de moniteur MQTT conviennent aussi, mais un script vous donne la maîtrise des topics, du QoS et des seuils. Pour un tableau de bord complet, combinez la sonde avec les métriques du broker comme décrit dans surveillance de la disponibilité MQTT et monitoring avec Prometheus et Grafana.

Sondes Kubernetes

Pour un pod Mosquitto auto-hébergé, une sonde exec qui effectue un aller-retour local est bien plus utile qu'une sonde tcpSocket. L'image officielle eclipse-mosquitto contient déjà les clients en ligne de commande :

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

Si votre listener exige une authentification, ajoutez -u/-P depuis un Secret monté en variables d'environnement, et autorisez l'utilisateur de la sonde dans vos ACL. Gardez timeoutSeconds supérieur au timeout du script lui-même, et réglez failureThreshold pour qu'une seule vérification lente pendant une pause de garbage collection ne redémarre pas le broker. Plus de détails de déploiement dans Mosquitto sur Kubernetes.

Surveiller un broker managé

Avec un broker managé, vous ne contrôlez pas le processus, mais vous restez responsable du côté client du contrat : votre nom d'hôte, vos identifiants, vos ACL et le chemin réseau depuis vos appareils. Une vérification externe par aller-retour détecte les identifiants expirés, les modifications d'ACL accidentelles et les problèmes réseau de votre côté que la surveillance du fournisseur ne peut pas voir.

Tester la vérification elle-même

Un health check qui ne se déclenche jamais est pire que pas de health check du tout. Avant de vous y fier, provoquez chaque mode d'échec une fois :

1. Lancez-le normalement et confirmez le code de sortie 0 : ./mqtt-health.sh; echo $?.
2. Utilisez un mauvais mot de passe et confirmez 2 avec un message « not authorised ».
3. Pointez-le vers un port fermé (MQTT_PORT=1999) et confirmez qu'il échoue dans le délai imparti au lieu de rester bloqué.
4. Définissez MQTT_WARN_MS=1 pour le script Python ou Node et confirmez que le chemin WARNING renvoie 1.
5. Observez testmqtt/health/# dans le client en ligne pendant la vérification pour voir arriver le jeton.

Voir les messages du health check en direct

Un bon health check MQTT est petit, rapide et honnête : une connexion, un abonnement, un message, un timeout strict et un code de sortie clair. Gardez le script dans votre dépôt à côté de la configuration du broker, exécutez-le depuis l'intérieur et l'extérieur de votre réseau, et vous serez averti des problèmes du broker avant vos appareils.

Questions fréquentes