Chaos testing en MQTT: particiones de red, pérdida de paquetes y reinicios del broker

El chaos testing en MQTT consiste en romper a propósito la red y el broker (latencia, pérdida de paquetes, particiones, reinicios) mientras mides si tus clientes se recuperan solos. La mayoría de los bugs MQTT que llegan a producción son bugs de recuperación: un cliente que nunca detecta una conexión muerta, una suscripción que no se restaura, un mensaje QoS 1 perdido porque la sesión no era persistente. Esta guía muestra las herramientas (tc netem, iptables, Toxiproxy, docker) y un script sonda que convierte cada experimento en números.

Primero define qué significa "recuperado"

Un experimento de chaos sin criterio de éxito es simplemente una caída. Anota el estado estable al que tu sistema debe volver:

Dos mecanismos hacen la mayor parte del trabajo: el intervalo de keep-alive, que limita cuánto tiempo pasa desapercibida una conexión muerta, y las sesiones persistentes, que permiten al broker guardar los mensajes QoS 1 de un suscriptor mientras está ausente.

La sonda: medir pérdidas, duplicados y reconexiones

Este script con paho-mqtt 2.x ejecuta un publicador y un suscriptor con una sesión persistente de MQTT 5. Publica un mensaje numerado cada 100 ms con QoS 1 y cada cinco segundos informa de cuántos mensajes confirmados aún no han llegado (missing), cuántos llegaron dos veces (dups), el número de reconexiones y la peor latencia.

import json, os, threading, time, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

HOST = os.getenv("MQTT_HOST", "localhost")
PORT = int(os.getenv("MQTT_PORT", "1883"))
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/chaos/{RUN}/seq"
lock = threading.Lock()
seen, sent = set(), []            # sent: (seq, MQTTMessageInfo)
stats = {"dups": 0, "reconnects": 0, "max_ms": 0.0}

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"[{userdata['role']}] connected: {reason_code}, session present={flags.session_present}")
    if userdata["ever"]:
        stats["reconnects"] += 1
    userdata["ever"] = True
    if userdata["role"] == "sub":
        client.subscribe(TOPIC, qos=1)

def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f"[{userdata['role']}] disconnected: {reason_code}")

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    with lock:
        if data["seq"] in seen:
            stats["dups"] += 1
        seen.add(data["seq"])
        stats["max_ms"] = max(stats["max_ms"], (time.time() - data["t"]) * 1000)

def make(role, session_expiry):
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = session_expiry
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"chaos-{role}-{RUN}",
                    protocol=mqtt.MQTTv5, userdata={"role": role, "ever": False})
    c.on_connect, c.on_disconnect, c.on_message = on_connect, on_disconnect, on_message
    c.reconnect_delay_set(min_delay=1, max_delay=16)
    c.connect_async(HOST, PORT, keepalive=10, properties=props)
    c.loop_start()
    return c

sub = make("sub", session_expiry=600)   # session survives outages up to 10 minutes
pub = make("pub", session_expiry=0)
time.sleep(2)

seq, next_report = 0, time.time() + 5
while True:
    info = pub.publish(TOPIC, json.dumps({"seq": seq, "t": time.time()}), qos=1)
    sent.append((seq, info))
    seq += 1
    if time.time() >= next_report:
        acked = {s for s, i in sent if i.is_published()}
        with lock:
            print(f"{time.strftime('%H:%M:%S')} sent={seq} acked={len(acked)} "
                  f"missing={len(acked - seen)} dups={stats['dups']} "
                  f"reconnects={stats['reconnects']} max_latency={stats['max_ms']:.0f}ms")
            stats["max_ms"] = 0.0
        next_report += 5
    time.sleep(0.1)

Mientras la red está caída, paho guarda los mensajes QoS 1 en su cola en memoria y los envía tras reconectar, así que sent sigue creciendo mientras acked se detiene. Durante una caída missing puede subir brevemente; tras la recuperación debe volver a 0. Un valor distinto de cero permanente significa que el broker confirmó mensajes que después se perdieron, normalmente por una sesión que no sobrevivió.

Experimento 1: latencia y pérdida de paquetes con tc netem

En Linux, tc netem degrada una interfaz. Ejecútalo dentro de una VM o de un contenedor con --cap-add NET_ADMIN para no ralentizar toda tu estación de trabajo.

IFACE=eth0
sudo tc qdisc add dev $IFACE root netem delay 300ms 100ms loss 5%     # bad cellular
sudo tc qdisc change dev $IFACE root netem delay 2000ms loss 30%      # nearly unusable
sudo tc qdisc del dev $IFACE root                                     # back to normal

Lo esperado: con 300 ms y un 5 % de pérdida, las retransmisiones de TCP ocultan la pérdida; la latencia sube, pero no hay reconexiones y missing se mantiene en 0. Con 2 s y un 30 % de pérdida, el PINGRESP puede llegar demasiado tarde y los clientes empiezan a reconectarse; comprueba que se estabilizan al quitar la regla. Para perfiles realistas de red móvil, consulta probar MQTT en condiciones de red móvil.

Experimento 2: particiones de red con iptables

Una red puede fallar de dos formas muy distintas, y los clientes a menudo solo gestionan una de ellas:

# Blackhole: packets silently vanish. Only keep-alive can detect this.
sudo iptables -A OUTPUT -p tcp --dport 1883 -j DROP
# ...wait 60 s, then heal:
sudo iptables -D OUTPUT -p tcp --dport 1883 -j DROP

# Hard failure: the client gets a TCP reset immediately.
sudo iptables -A OUTPUT -p tcp --dport 1883 -j REJECT --reject-with tcp-reset
sudo iptables -D OUTPUT -p tcp --dport 1883 -j REJECT --reject-with tcp-reset

Con un keep-alive de 10 segundos, la sonda debería registrar disconnected unos 10–20 segundos después de empezar el agujero negro, reintentar con backoff mientras dure y reconectar con session present=True cuando se resuelva. En el lado del broker, otros clientes deberían recibir el Last Will de la sonda tras aproximadamente 1,5 × keep-alive. Un cliente que sigue "conectado" durante minutos en un agujero negro tiene el keep-alive desactivado o configurado demasiado alto. Probar QoS con un corte de red profundiza en lo que garantiza aquí cada nivel de QoS.

Experimento 3: fallos por conexión con Toxiproxy

Toxiproxy es un proxy TCP cuyo comportamiento cambias mediante una API HTTP. Funciona también en macOS y Windows, solo afecta a las conexiones que pasan por él y es fácil de automatizar en CI.

docker run -d --name toxiproxy -p 8474:8474 -p 21883:21883 ghcr.io/shopify/toxiproxy
curl -s -X POST localhost:8474/proxies \
  -d '{"name":"mqtt","listen":"0.0.0.0:21883","upstream":"broker.connectmq.com:1883"}'

# Point the probe at the proxy
MQTT_HOST=localhost MQTT_PORT=21883 python chaos_probe.py

Después inyecta fallos desde una segunda terminal:

API=localhost:8474/proxies/mqtt
# 1.5 s latency with 500 ms jitter
curl -s -X POST $API/toxics -d '{"name":"lag","type":"latency","attributes":{"latency":1500,"jitter":500}}'
curl -s -X DELETE $API/toxics/lag
# Blackhole: data stops, connection stays open until the toxic is removed
curl -s -X POST $API/toxics -d '{"name":"cut","type":"timeout","attributes":{"timeout":0}}'
curl -s -X DELETE $API/toxics/cut
# Connection reset by peer
curl -s -X POST $API/toxics -d '{"name":"rst","type":"reset_peer","attributes":{"timeout":0}}'
curl -s -X DELETE $API/toxics/rst
# Take the whole "network" down and up again
curl -s -X POST $API -d '{"enabled":false}'; sleep 30; curl -s -X POST $API -d '{"enabled":true}'

El toxic timeout con 0 es el equivalente en Toxiproxy de un DROP de iptables; reset_peer equivale a REJECT. Probar MQTT con Toxiproxy cubre límites de ancho de banda, slicing e integración con frameworks de test.

Experimento 4: reinicios y caídas del broker

Los experimentos de reinicio necesitan un broker que gestiones tú. Un contenedor local de Mosquitto basta para encontrar la mayoría de los bugs del lado del cliente:

docker run -d --name chaos-broker -p 1883:1883 eclipse-mosquitto:2 mosquitto -c /mosquitto-no-auth.conf
MQTT_HOST=localhost MQTT_PORT=1883 python chaos_probe.py     # in another terminal

docker restart chaos-broker                                   # graceful: SIGTERM, then start
docker kill --signal=KILL chaos-broker && docker start chaos-broker   # crash

Lo que deberías observar:

• Sin persistencia (la configuración por defecto de arriba), el broker olvida todas las sesiones. La sonda se reconecta con session present=False, se vuelve a suscribir, y los mensajes publicados durante el reinicio pueden quedar missing para siempre.
• Con persistence true y un volumen para el directorio de datos, las sesiones y los mensajes en cola sobreviven a un reinicio ordenado. Tras un KILL, todo lo posterior al último autosave puede perderse igualmente; comprueba si eso coincide con lo que promete tu diseño.
• Los clientes no deben reconectarse a la vez. Con cientos de clientes, un reinicio se convierte en un pico de conexiones. El backoff con jitter lo reparte; consulta tormentas de reconexión MQTT.

En brokers en clúster, sustituye el reinicio por detener un nodo cada vez y verifica que los clientes hacen failover. Pruebas de reconexión MQTT tiene más escenarios del lado del cliente.

Un plan de experimentos repetible

Ejecuta el plan antes de cada versión importante del cliente o del broker, guarda la salida de la sonda como artefacto y añade una fila nueva cada vez que un incidente real te enseñe un modo de fallo nuevo.

Mira llegar los mensajes de la sonda en tu navegador

El chaos testing en MQTT trata sobre todo de la recuperación: detectar el fallo rápido, reconectar con suavidad y no perder nada de lo que prometiste entregar. Con netem, iptables, Toxiproxy y un broker que puedas reiniciar, puedes reproducir casi cualquier caída en un portátil, y la sonda te dice en números si tus clientes superan la prueba.

Preguntas frecuentes