MQTT-Soak-Test: Speicherlecks und schleichende Verschlechterung finden

Ein Soak-Test (Dauertest) betreibt ein MQTT-System lange unter normaler Last – meist 24 bis 72 Stunden –, um Probleme zu finden, die ein zehnminütiger Benchmark nie sieht: Speicher, der um ein paar Megabyte pro Stunde wächst, Dateideskriptoren, die nie freigegeben werden, Queues für Sessions, die nie zurückkommen, und Latenz, die langsam steigt, bis Geräte in Timeouts laufen.

Dieser Leitfaden zeigt, was du messen solltest, liefert ein Python-Skript, das alles als CSV aufzeichnet, und erklärt, wie du das Ergebnis liest, damit aus „sieht gut aus" eine Zahl wird.

Soak-Test vs. Lasttest

Wenn du die Kapazität noch nicht gemessen hast, fang mit MQTT-Broker-Lasttests an. Ein Soak-Test setzt voraus, dass das System die Last jetzt bewältigt, und fragt, ob das an Tag drei immer noch so ist.

Was sich langsam verschlechtert

Die meisten Langzeitfehler folgen einer Handvoll Muster:

• Speicherlecks im Client – Callbacks, die an eine Liste anhängen, eine wachsende In-flight-Queue, weil der Client schneller publiziert als Acknowledgements eintreffen, oder bei jedem Reconnect ein neues Client-Objekt, ohne das alte zu stoppen. Siehe Speicherlecks in MQTT-Clients.
• Unbegrenzte Broker-Queues – persistente Sessions (clean_session=False / Session Expiry) von Geräten, die nie wiederkommen, sammeln QoS-1- und QoS-2-Nachrichten, bis dem Broker der Speicher ausgeht. Das klassische Ergebnis beschreibt MQTT-Broker-Abstürze durch OOM.
• Wachsende Retained Messages – Topics mit eindeutigen IDs im Pfad (.../session-8f3a/state), die mit retain=true publiziert werden, sammeln sich für immer an.
• Lecks bei Dateideskriptoren und Sockets – Reconnect-Pfade, die den alten Socket nicht schließen. Der Prozess scheitert irgendwann mit Too many open files.
• Zeitgesteuerte Ereignisse – Log-Rotation, Zertifikatserneuerung, Token-Ablauf, tägliche Backups oder eine Platte, die sich mit Persistenzdateien füllt. Sie treten nur auf, wenn der Test lange genug läuft.

Die Metriken, die du aufzeichnen solltest

Zeichne Client- und Broker-Werte auf derselben Zeitachse auf, einmal pro Minute:

Ein Python-Skript für den Soak-Test

Dieses Skript (paho-mqtt 2.x und psutil) publiziert in festem Takt eine durchnummerierte Nachricht, abonniert sein eigenes Topic und schreibt pro Minute eine CSV-Zeile mit Durchsatz, Lücken, Duplikaten, Latenz-Perzentilen, eigenem Speicher und offenen Dateideskriptoren. Es abonniert in on_connect neu und übersteht so Reconnects, statt unbemerkt taub zu werden.

import csv, json, os, threading, time, uuid
import paho.mqtt.client as mqtt
import psutil

HOST, PORT = os.environ.get("MQTT_HOST", "broker.connectmq.com"), 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/soak/{RUN}/data"
RATE = 20            # Nachrichten pro Sekunde
DURATION_H = 72      # Gesamtlaufzeit
WINDOW_S = 60        # eine CSV-Zeile pro Minute

lock = threading.Lock()
st = {"recv": 0, "gaps": 0, "dups": 0, "last": -1, "lat": [], "disc": 0}

def on_connect(client, userdata, flags, reason_code, properties):
    if client is sub:
        client.subscribe(TOPIC, qos=1)

def on_disconnect(client, userdata, flags, reason_code, properties):
    with lock:
        st["disc"] += 1

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    with lock:
        st["recv"] += 1
        st["lat"].append((time.monotonic() - data["t"]) * 1000)
        if data["seq"] <= st["last"]:
            st["dups"] += 1
        else:
            st["gaps"] += data["seq"] - st["last"] - 1
            st["last"] = data["seq"]

def make_client(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"soak-{RUN}-{name}")
    c.username_pw_set("username", "password")
    c.on_connect, c.on_disconnect = on_connect, on_disconnect
    c.reconnect_delay_set(min_delay=1, max_delay=30)
    return c

sub = make_client("sub")
sub.on_message = on_message
sub.connect(HOST, PORT, keepalive=30)
sub.loop_start()
pub = make_client("pub")
pub.connect(HOST, PORT, keepalive=30)
pub.loop_start()
time.sleep(2)

proc = psutil.Process()
pct = lambda xs, q: xs[max(0, int(len(xs) * q) - 1)] if xs else 0.0

with open(f"soak-{RUN}.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["time", "sent", "recv", "gaps", "dups", "disconnects",
                "p50_ms", "p99_ms", "rss_mb", "fds"])
    end = time.monotonic() + DURATION_H * 3600
    next_row = time.monotonic() + WINDOW_S
    seq = 0
    while time.monotonic() < end:
        pub.publish(TOPIC, json.dumps({"seq": seq, "t": time.monotonic()}), qos=1)
        seq += 1
        time.sleep(1 / RATE)
        if time.monotonic() >= next_row:
            with lock:
                lat, st["lat"] = sorted(st["lat"]), []
                counts = [st["recv"], st["gaps"], st["dups"], st["disc"]]
            w.writerow([time.strftime("%Y-%m-%dT%H:%M:%S"), seq, *counts,
                        round(pct(lat, 0.50), 1), round(pct(lat, 0.99), 1),
                        round(proc.memory_info().rss / 1e6, 1), proc.num_fds()])
            f.flush()
            next_row += WINDOW_S

Starte es mit nohup python soak.py & (oder in tmux) auf einer Maschine, die drei Tage durchläuft. Die CSV wird jede Minute geschrieben, du kannst sie also schon während des Tests auswerten. num_fds() gibt es unter Linux und macOS; unter Windows nimmst du proc.num_handles(). Ein kurzer Probelauf gegen den geteilten Endpunkt broker.connectmq.com ist in Ordnung, für den echten mehrtägigen Lauf setzt du MQTT_HOST aber auf deine eigene Broker-Instanz: Der geteilte Endpunkt ist für schnelle Tests gedacht, hat ein Rate-Limit und speichert keine persistenten Sessions – die unten beschriebenen Session-Fehlerbilder siehst du also nur auf einem eigenen Broker.

Auch die Broker-Seite beobachten

Die Client-Zahlen sagen dir, dass sich etwas verschlechtert hat; die Broker-Zahlen sagen dir, wo. Wenn dein Broker $SYS-Topics veröffentlicht (Mosquitto tut das), abonniere die Zähler, die Wachstum zeigen:

mosquitto_sub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" -v \
  -t '$SYS/broker/clients/connected' \
  -t '$SYS/broker/messages/stored' \
  -t '$SYS/broker/retained messages/count' \
  -t '$SYS/broker/heap/current'

Läuft der Broker in Docker auf einem Host, den du kontrollierst, miss Speicher und CPU einmal pro Minute:

while true; do
  docker stats --no-stream --format "$(date -u +%FT%TZ),{{.Name}},{{.MemUsage}},{{.CPUPerc}}" mosquitto
  sleep 60
done >> broker-stats.csv

Für alles, was länger als ein Test läuft, exportierst du dieselben Werte in eine Zeitreihendatenbank – Monitoring mit Prometheus und Grafana und $SYS-Topic-Monitoring zeigen das Setup.

Aus der CSV ein Urteil machen

Eine CSV mit 4.000 Zeilen mit bloßem Auge zu prüfen funktioniert nicht. Lege stattdessen eine Trendlinie hindurch. Überspringe die erste Stunde (Caches, JIT, Connection-Pools wärmen sich auf) und berechne die Steigung des Speichers über die Zeit mit der Standardbibliothek:

import csv, statistics, sys

rows = list(csv.DictReader(open(sys.argv[1])))[60:]   # 1 h Aufwärmphase überspringen
hours = [i / 60 for i in range(len(rows))]
for col in ("rss_mb", "fds", "p99_ms"):
    ys = [float(r[col]) for r in rows]
    slope, _ = statistics.linear_regression(hours, ys)
    print(f"{col:8} start {ys[0]:8.1f}  end {ys[-1]:8.1f}  trend {slope:+.3f}/hour")

last = rows[-1]
print(f"sent {last['sent']}  recv {last['recv']}  gaps {last['gaps']}  "
      f"dups {last['dups']}  disconnects {last['disconnects']}")

So interpretierst du das Ergebnis:

• RSS-Trend nahe null mit Sägezahn in den Rohdaten – normale Garbage Collection. Eine stetig positive Steigung bei konstanter Last ist ein Leck; multipliziere sie mit der Laufzeit deines Deployments, um zu sehen, ob es relevant ist (5 MB/Stunde sind 3,6 GB pro Monat).
• Steigende Dateideskriptoren – irgendetwas öffnet Sockets oder Dateien, ohne sie zu schließen, typischerweise im Reconnect-Pfad.
• p99 steigt, p50 bleibt flach – wachsende Queues, längere GC-Pausen oder ein Broker, der gespeicherte Nachrichten anhäuft.
• Lücken, die zu Disconnects passen – bei QoS 0 erwartbar, bei QoS 1 mit persistenter Session ein Bug. Lücken ohne Disconnect deuten darauf hin, dass der Broker Nachrichten verwirft (Queue-Limits) oder ein Subscriber nicht hinterherkommt.
• messages/stored des Brokers steigt – meist persistente Sessions von Clients, die verschwunden sind. Setze ein Session Expiry (MQTT 5) oder räume veraltete Client-IDs auf.

Den Soak-Test realistisch machen

Ein konstanter Strom zwischen zwei Clients findet Client-Lecks. Um Broker-Lecks zu finden, muss der Traffic die Codepfade durchlaufen, die Speicher allozieren:

1. Verbindungswechsel – ein Teil der Clients trennt sich alle paar Minuten und verbindet sich neu, manche sauber, andere durch Abreißen des Sockets, damit das Last Will ausgelöst wird.
2. Unterschiedliche Sessions – mische Clean Sessions mit persistenten und nimm ein paar persistente Clients dazu, die stundenlang offline gehen.
3. Unterschiedliche Topics – publiziere auf Tausende verschiedene Topics statt auf eines; nimm Retained Messages und Wildcard-Subscriber dazu.
4. Echte Payload-Größen – erzeuge sie mit einem MQTT-Testdatengenerator statt mit festen Strings.
5. Netzwerkfehler – baue ein paar kurze Ausfälle ein, um die Wiederherstellung zu prüfen; MQTT-Reconnect-Tests zeigt, wie.

Bevor du den dreitägigen Lauf startest, prüfe Topic und Zugangsdaten mit dem Online-MQTT-Client und vergewissere dich, dass Nachrichten ankommen.

Soak-Traffic im Browser beobachten

Ein Soak-Test ist billig: eine kleine Maschine, ein Skript und Geduld. Lass ihn vor jedem größeren Release und nach jedem Broker- oder Bibliotheks-Upgrade laufen, bewahre die CSVs auf und vergleiche die Trendlinien – schleichende Verschlechterung bleibt nur unsichtbar, wenn niemand sie misst.

Häufig gestellte Fragen