Chaos testing MQTT : partitions réseau, perte de paquets et redémarrages du broker

Le chaos testing MQTT consiste à casser volontairement le réseau et le broker (latence, perte de paquets, partitions, redémarrages) tout en mesurant si vos clients récupèrent d'eux-mêmes. La plupart des bugs MQTT qui atteignent la production sont des bugs de reprise : un client qui ne remarque jamais une connexion morte, un abonnement qui n'est pas restauré, un message QoS 1 perdu parce que la session n'était pas persistante. Ce guide présente les outils (tc netem, iptables, Toxiproxy, docker) et un script sonde qui transforme chaque expérience en chiffres.

Définissez d'abord ce que « récupéré » veut dire

Une expérience de chaos sans critère de réussite n'est qu'une panne. Notez l'état stable vers lequel votre système doit revenir :

Deux mécanismes font l'essentiel du travail : l'intervalle de keep-alive, qui borne la durée pendant laquelle une connexion morte passe inaperçue, et les sessions persistantes, qui permettent au broker de conserver les messages QoS 1 d'un abonné absent.

La sonde : mesurer pertes, doublons et reconnexions

Ce script paho-mqtt 2.x exécute un publisher et un abonné avec une session MQTT 5 persistante. Il publie un message numéroté toutes les 100 ms en QoS 1 et indique toutes les cinq secondes combien de messages acquittés ne sont pas encore arrivés (missing), combien sont arrivés deux fois (dups), le nombre de reconnexions et la pire latence.

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)

Pendant que le réseau est coupé, paho garde les messages QoS 1 dans sa file en mémoire et les envoie après la reconnexion : sent continue donc de croître pendant que acked marque une pause. Pendant une panne, missing peut augmenter brièvement ; après la reprise, il doit revenir à 0. Une valeur non nulle permanente signifie que le broker a acquitté des messages perdus ensuite, généralement à cause d'une session qui n'a pas survécu.

Expérience 1 : latence et perte de paquets avec tc netem

Sous Linux, tc netem dégrade une interface. Lancez-le dans une VM ou un conteneur avec --cap-add NET_ADMIN pour ne pas ralentir tout votre poste de travail.

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

Attendu : à 300 ms et 5 % de pertes, les retransmissions TCP masquent les pertes ; la latence augmente, mais il n'y a pas de reconnexion et missing reste à 0. À 2 s et 30 % de pertes, le PINGRESP peut arriver trop tard et les clients commencent à se reconnecter ; vérifiez qu'ils se stabilisent une fois la règle retirée. Pour des profils cellulaires réalistes, voir tester MQTT dans des conditions de réseau cellulaire.

Expérience 2 : partitions réseau avec iptables

Un réseau peut tomber en panne de deux façons très différentes, et les clients n'en gèrent souvent qu'une :

# 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

Avec un keep-alive de 10 secondes, la sonde devrait journaliser disconnected environ 10 à 20 secondes après le début du trou noir, réessayer avec backoff tant qu'il dure, puis se reconnecter avec session present=True une fois le réseau rétabli. Côté broker, les autres clients devraient recevoir le Last Will de la sonde après environ 1,5 × keep-alive. Un client qui reste « connecté » pendant des minutes dans un trou noir a un keep-alive désactivé ou beaucoup trop élevé. Tester la QoS avec une coupure réseau détaille ce que chaque niveau de QoS garantit dans ce cas.

Expérience 3 : pannes par connexion avec Toxiproxy

Toxiproxy est un proxy TCP dont vous modifiez le comportement via une API HTTP. Il fonctionne aussi sous macOS et Windows, n'affecte que les connexions qui le traversent et se scripte facilement 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

Injectez ensuite des pannes depuis un second 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}'

Le toxic timeout à 0 est l'équivalent Toxiproxy d'un DROP iptables ; reset_peer correspond à REJECT. Tester MQTT avec Toxiproxy couvre les limites de bande passante, le slicing et l'intégration aux frameworks de test.

Expérience 4 : redémarrages et crashs du broker

Les expériences de redémarrage nécessitent un broker que vous exploitez vous-même. Un conteneur Mosquitto local suffit pour trouver la plupart des bugs côté client :

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

Ce que vous devriez observer :

• Sans persistance (la configuration par défaut ci-dessus), le broker oublie toutes les sessions. La sonde se reconnecte avec session present=False, se réabonne, et les messages publiés pendant le redémarrage peuvent rester définitivement missing.
• Avec persistence true et un volume pour le répertoire de données, les sessions et les messages en file survivent à un redémarrage propre. Après un KILL, tout ce qui suit le dernier autosave peut malgré tout être perdu ; vérifiez que cela correspond à ce que promet votre conception.
• Les clients ne doivent pas se reconnecter tous en même temps. Avec des centaines de clients, un redémarrage devient un pic de connexions. Un backoff avec jitter l'étale ; voir les tempêtes de reconnexion MQTT.

Pour les brokers en cluster, remplacez le redémarrage par l'arrêt d'un nœud à la fois et vérifiez que les clients basculent. Les tests de reconnexion MQTT proposent d'autres scénarios côté client.

Un plan d'expériences reproductible

Exécutez ce plan avant chaque release majeure du client ou du broker, conservez la sortie de la sonde comme artefact et ajoutez une ligne chaque fois qu'un incident réel vous révèle un nouveau mode de défaillance.

Voir arriver les messages de la sonde dans le navigateur

Le chaos testing MQTT porte avant tout sur la reprise : détecter la panne vite, se reconnecter en douceur et ne rien perdre de ce que vous avez promis de livrer. Avec netem, iptables, Toxiproxy et un broker redémarrable, vous reproduisez presque n'importe quelle panne sur un portable, et la sonde vous dit en chiffres si vos clients passent le test.

Questions fréquentes