Chaos-Tests für MQTT: Netzwerkpartitionen, Paketverlust und Broker-Neustarts

Chaos-Tests für MQTT heißt: Netzwerk und Broker absichtlich stören – Latenz, Paketverlust, Partitionen, Neustarts – und dabei messen, ob sich deine Clients selbst erholen. Die meisten MQTT-Bugs, die in Produktion landen, sind Recovery-Bugs: ein Client, der eine tote Verbindung nie bemerkt, ein Abonnement, das nicht wiederhergestellt wird, eine QoS-1-Nachricht, die verloren geht, weil die Session nicht persistent war. Dieser Guide zeigt die Werkzeuge (tc netem, iptables, Toxiproxy, docker) und ein Probe-Skript, das jedes Experiment in Zahlen übersetzt.

Zuerst festlegen, was „erholt“ bedeutet

Ein Chaos-Experiment ohne Erfolgskriterium ist einfach ein Ausfall. Schreib den Steady State auf, zu dem dein System zurückkehren muss:

Zwei Mechanismen leisten die Hauptarbeit: das Keep-Alive-Intervall, das begrenzt, wie lange eine tote Verbindung unbemerkt bleibt, und persistente Sessions, mit denen der Broker QoS-1-Nachrichten für einen abwesenden Subscriber aufbewahrt.

Das Probe: Verluste, Duplikate und Reconnects messen

Dieses paho-mqtt-2.x-Skript betreibt einen Publisher und einen Subscriber mit persistenter MQTT-5-Session. Es veröffentlicht alle 100 ms eine nummerierte Nachricht mit QoS 1 und meldet alle fünf Sekunden, wie viele bestätigte Nachrichten noch nicht angekommen sind (missing), wie viele doppelt kamen (dups), die Anzahl der Reconnects und die schlechteste Latenz.

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)

Solange das Netz weg ist, hält paho QoS-1-Nachrichten in seiner In-Memory-Queue und sendet sie nach dem Reconnect; sent wächst also weiter, während acked pausiert. Während eines Ausfalls kann missing kurz steigen; nach der Erholung muss der Wert auf 0 zurückgehen. Ein dauerhaft von null verschiedener Wert heißt, dass der Broker Nachrichten bestätigt hat, die später verloren gingen – meist wegen einer Session, die nicht überlebt hat.

Experiment 1: Latenz und Paketverlust mit tc netem

Unter Linux verschlechtert tc netem ein Interface. Führe es in einer VM oder in einem Container mit --cap-add NET_ADMIN aus, damit nicht deine ganze Workstation langsam wird.

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

Erwartet: Bei 300 ms und 5 % Verlust kaschieren TCP-Retransmits den Verlust – die Latenz steigt, aber es gibt keine Reconnects und missing bleibt bei 0. Bei 2 s und 30 % Verlust kommt das PINGRESP eventuell zu spät, und Clients beginnen, sich neu zu verbinden; prüfe, ob sie sich nach Entfernen der Regel stabilisieren. Realistische Mobilfunkprofile findest du unter MQTT unter Mobilfunkbedingungen testen.

Experiment 2: Netzwerkpartitionen mit iptables

Ein Netzwerk kann auf zwei sehr unterschiedliche Arten ausfallen, und Clients beherrschen oft nur eine davon:

# 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

Mit 10 Sekunden Keep-Alive sollte das Probe etwa 10–20 Sekunden nach Beginn des Blackholes disconnected protokollieren, währenddessen mit Backoff neu versuchen und nach der Heilung mit session present=True wieder verbunden sein. Auf Broker-Seite sollten andere Clients nach etwa 1,5 × Keep-Alive den Last Will des Probes erhalten. Ein Client, der während eines Blackholes minutenlang „verbunden“ bleibt, hat Keep-Alive deaktiviert oder viel zu hoch eingestellt. QoS mit Netzwerkunterbrechung testen geht tiefer darauf ein, was jedes QoS-Level hier garantiert.

Experiment 3: Fehler pro Verbindung mit Toxiproxy

Toxiproxy ist ein TCP-Proxy, dessen Verhalten du über eine HTTP-API änderst. Er läuft auch unter macOS und Windows, betrifft nur Verbindungen, die durch ihn laufen, und lässt sich in CI leicht skripten.

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

Dann injizierst du Fehler aus einem zweiten 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}'

Der timeout-Toxic mit 0 entspricht in Toxiproxy einem iptables-DROP; reset_peer entspricht REJECT. MQTT mit Toxiproxy testen behandelt Bandbreitenlimits, Slicing und die Integration in Test-Frameworks.

Experiment 4: Broker-Neustarts und -Abstürze

Neustart-Experimente brauchen einen Broker, den du selbst betreibst. Ein lokaler Mosquitto-Container reicht, um die meisten clientseitigen Bugs zu finden:

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

Was du beobachten solltest:

• Ohne Persistenz (die Standardkonfiguration oben) vergisst der Broker alle Sessions. Das Probe verbindet sich mit session present=False neu, abonniert erneut, und während des Neustarts veröffentlichte Nachrichten können dauerhaft missing sein.
• Mit persistence true und einem Volume für das Datenverzeichnis überleben Sessions und gepufferte Nachrichten einen sauberen Neustart. Nach einem KILL kann alles seit dem letzten Autosave trotzdem verloren gehen – prüfe, ob das zu den Zusagen deines Designs passt.
• Clients dürfen sich nicht im Gleichschritt neu verbinden. Bei Hunderten Clients wird ein Neustart zur Verbindungsspitze. Backoff mit Jitter verteilt sie; siehe MQTT-Reconnect-Stürme.

Bei geclusterten Brokern ersetzt du den Neustart durch das Stoppen jeweils eines Knotens und prüfst, ob die Clients per Failover wechseln. MQTT-Reconnect-Tests enthält weitere clientseitige Szenarien.

Ein wiederholbarer Experimentplan

Führe den Plan vor jedem größeren Client- oder Broker-Release aus, bewahre die Probe-Ausgabe als Artefakt auf und ergänze eine neue Zeile, wann immer dir ein echter Vorfall einen neuen Fehlerfall zeigt.

Probe-Nachrichten live im Browser verfolgen

Bei Chaos-Tests für MQTT geht es vor allem um Erholung: den Fehler schnell erkennen, behutsam neu verbinden und nichts verlieren, dessen Zustellung du zugesagt hast. Mit netem, iptables, Toxiproxy und einem neustartbaren Broker reproduzierst du fast jeden Ausfall auf einem Laptop, und das Probe sagt dir in Zahlen, ob deine Clients bestehen.

Häufig gestellte Fragen