Health check de un broker MQTT: scripts para monitorizar la disponibilidad

Un health check de un broker MQTT debe demostrar lo único de lo que dependen tus dispositivos: un cliente puede conectarse, suscribirse, publicar y recibir su propio mensaje en pocos segundos. Cualquier cosa menos —un ping, un puerto abierto, un proceso en ejecución— puede seguir en verde mientras el broker rechaza logins en silencio o deja de enrutar.

A continuación tienes tres scripts listos para usar (Bash, Python, Node.js) que hacen ese round trip y devuelven códigos de salida aptos para monitorización, además de ejemplos para cron, probes de Kubernetes y alertas.

Qué debe verificar un health check

La última fila es la que hay que automatizar. Cada ejecución debe usar un client ID único (dos comprobaciones con el mismo ID se desconectan entre sí; consulta buenas prácticas para el client ID de MQTT) y un topic único, para que un mensaje retrasado de una ejecución anterior nunca haga parecer sana la actual.

Si buscas un diagnóstico puntual más amplio en lugar de un probe recurrente, empieza por cómo comprobar el estado de un broker MQTT o prueba el broker de forma interactiva en el cliente MQTT online.

Códigos de salida que entienden las herramientas de monitorización

Todos los scripts de este artículo usan la convención de plugins de Nagios, que entienden Icinga, Zabbix, Checkmk, Sensu y la mayoría de wrappers de cron:

Kubernetes, systemd y cron solo distinguen cero de distinto de cero, así que los mismos scripts funcionan en todas partes.

Bash: mosquitto_sub + mosquitto_pub

La comprobación más rápida usa los clientes de línea de comandos de Mosquitto (apt install mosquitto-clients o brew install mosquitto). Un suscriptor en segundo plano espera exactamente un mensaje con un timeout, y el publicador envía un token aleatorio al mismo 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

Detalles clave:

• -C 1 hace que mosquitto_sub termine tras un mensaje; -W hace que se rinda tras el timeout en lugar de quedarse colgado para siempre.
• El script compara el payload con el token en lugar de fiarse solo del código de salida, así un mensaje perdido en el topic no puede producir un falso OK.
• El texto de error de los clientes (por ejemplo Connection error: Connection Refused: not authorised.) acaba en la línea de salida, que es justo lo que quieres ver en una alerta. Busca cualquier reason code en la herramienta de reason codes de MQTT.
• /etc/ssl/certs es el directorio de CA en Debian, Ubuntu y Alpine. En otros sistemas pasa --cafile con tu bundle de CA.

Python: round trip con umbral de latencia

La versión en Python (paho-mqtt 2.x) mide el tiempo de round trip y devuelve WARNING cuando es más lento que un umbral. Cada paso tiene un límite de tiempo, así que el script nunca bloquea a un agente de monitorización:

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

El texto tras | son datos de rendimiento de Nagios, así que las herramientas que los grafican dibujarán el tiempo de round trip sin coste adicional. Separar los fallos de CONNACK, SUBACK y entrega importa a las 3 de la madrugada: "subscribe refused" apunta a un cambio de ACL, "no CONNACK" a un broker sobrecargado o medio muerto y "not delivered" al enrutamiento o a un bridge. Si las suscripciones se rechazan de forma inesperada, ACL de MQTT denegada en silencio explica las causas habituales.

Node.js: MQTT.js 5

La misma lógica con MQTT.js 5 y su API de promesas. Un único temporizador global garantiza que el proceso termina aunque un callback no se dispare nunca:

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

Configurar reconnectPeriod: 0 es importante: por defecto MQTT.js se reconecta indefinidamente, lo cual es estupendo para aplicaciones y terrible para un probe que debe fallar rápido. Las credenciales incorrectas aparecen como Connection refused: Not authorized o Connection refused: Bad username or password, según cómo las notifique el broker.

Ejecutar la comprobación de forma programada

Cron con alertas

# 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 mayoría de agentes de monitorización (Nagios, Icinga, Checkmk, Zabbix con external checks) pueden ejecutar cualquiera de estos scripts directamente como plugin. Los servicios de uptime con un tipo de monitor MQTT también sirven, pero un script te da control sobre los topics, el QoS y los umbrales. Para un dashboard completo, combina el probe con las métricas del broker como se describe en monitorización de uptime de MQTT y monitorización con Prometheus y Grafana.

Probes de Kubernetes

Para un pod de Mosquitto autoalojado, un probe exec que hace un round trip local es mucho más útil que un probe tcpSocket. La imagen oficial eclipse-mosquitto ya incluye los clientes de línea de comandos:

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 tu listener exige autenticación, añade -u/-P desde un Secret montado como variables de entorno y permite al usuario del probe en tus ACL. Mantén timeoutSeconds por encima del timeout del propio script y ajusta failureThreshold para que una sola comprobación lenta durante una pausa de garbage collection no reinicie el broker. Tienes más detalles de despliegue en Mosquitto en Kubernetes.

Monitorizar un broker gestionado

Con un broker gestionado no controlas el proceso, pero sigues siendo responsable del lado cliente del contrato: tu hostname, tus credenciales, tus ACL y la ruta de red desde tus dispositivos. Una comprobación externa de round trip detecta credenciales caducadas, cambios accidentales de ACL y problemas de red en tu lado que la monitorización del propio proveedor no puede ver.

Probar la propia comprobación

Un health check que nunca salta es peor que no tener ninguno. Antes de confiar en él, fuerza cada modo de fallo una vez:

1. Ejecútalo normalmente y confirma el código de salida 0: ./mqtt-health.sh; echo $?.
2. Usa una contraseña incorrecta y confirma 2 con un mensaje "not authorised".
3. Apúntalo a un puerto cerrado (MQTT_PORT=1999) y confirma que falla dentro del timeout en lugar de quedarse colgado.
4. Pon MQTT_WARN_MS=1 en el script de Python o de Node y confirma que la ruta WARNING devuelve 1.
5. Observa testmqtt/health/# en el cliente online mientras se ejecuta la comprobación para ver llegar el token.

Ver en directo los mensajes del health check

Un buen health check de MQTT es pequeño, rápido y honesto: una conexión, una suscripción, un mensaje, un timeout estricto y un código de salida claro. Guarda el script en tu repositorio junto a la configuración del broker, ejecútalo tanto desde dentro como desde fuera de tu red y te enterarás de los problemas del broker antes que tus dispositivos.

Preguntas frecuentes