Test d'endurance MQTT (soak test) : détecter fuites mémoire et dégradation lente

Un soak test (test d'endurance) fait tourner un système MQTT sous charge normale pendant longtemps — généralement de 24 à 72 heures — pour attraper les problèmes qu'un benchmark de dix minutes ne voit jamais : une mémoire qui grossit de quelques mégaoctets par heure, des descripteurs de fichiers jamais libérés, des files pour des sessions qui ne reviennent jamais et une latence qui grimpe lentement jusqu'à ce que les appareils commencent à expirer.

Ce guide explique quoi mesurer, fournit un script Python qui enregistre tout en CSV et montre comment lire le résultat pour que « ça a l'air d'aller » devienne un chiffre.

Soak test ou test de charge

Si vous n'avez pas encore mesuré la capacité, commencez par le test de charge d'un broker MQTT. Un soak test part du principe que le système tient la charge aujourd'hui et vérifie qu'il la tiendra encore le troisième jour.

Ce qui se dégrade lentement

La plupart des défaillances de longue durée se rangent dans quelques catégories :

• Fuites mémoire côté client — des callbacks qui ajoutent à une liste, une file in-flight qui grossit parce que le client publie plus vite que les acquittements n'arrivent, ou un nouvel objet client créé à chaque reconnexion sans arrêter l'ancien. Voir fuites mémoire des clients MQTT.
• Files du broker non bornées — les sessions persistantes (clean_session=False / session expiry) d'appareils qui ne se reconnectent jamais continuent d'accumuler des messages QoS 1 et 2 jusqu'à ce que le broker manque de mémoire. Le résultat classique est décrit dans plantages du broker MQTT par OOM.
• Croissance des messages retenus — des topics contenant des identifiants uniques (.../session-8f3a/state) publiés avec retain=true s'accumulent indéfiniment.
• Fuites de descripteurs de fichiers et de sockets — des chemins de reconnexion qui ne ferment pas l'ancien socket. Le processus finit par échouer avec Too many open files.
• Événements liés au temps — rotation des logs, renouvellement des certificats, expiration des jetons, sauvegardes quotidiennes ou un disque qui se remplit de fichiers de persistance. Ils ne surviennent que si le test dure assez longtemps pour les croiser.

Les métriques à enregistrer

Enregistrez les valeurs côté client et côté broker sur la même échelle de temps, une fois par minute :

Un script Python de soak test

Ce script (paho-mqtt 2.x et psutil) publie à cadence fixe un message numéroté, s'abonne à son propre topic et écrit une ligne CSV par minute avec le débit, les trous, les doublons, les percentiles de latence, sa propre mémoire et ses descripteurs ouverts. Il se réabonne dans on_connect : il survit donc aux reconnexions au lieu de devenir sourd sans prévenir.

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            # messages par seconde
DURATION_H = 72      # durée totale
WINDOW_S = 60        # une ligne CSV par 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

Lancez-le avec nohup python soak.py & (ou dans tmux) sur une machine qui restera allumée trois jours. Le CSV est écrit chaque minute : vous pouvez donc l'examiner pendant que le test tourne. num_fds() existe sous Linux et macOS ; sous Windows, utilisez proc.num_handles(). Un court essai sur le point d'accès partagé broker.connectmq.com convient, mais pour l'exécution réelle de plusieurs jours, définissez MQTT_HOST sur votre propre instance de broker : le point d'accès partagé est prévu pour les tests rapides, limite le débit et ne conserve pas de sessions persistantes, si bien que les défaillances liées aux sessions décrites plus bas n'apparaissent que sur votre propre broker.

Surveillez aussi le côté broker

Les chiffres du client vous disent qu'*il y a une dégradation ; ceux du broker vous disent *où. Si votre broker publie des topics $SYS (c'est le cas de Mosquitto), abonnez-vous aux compteurs qui révèlent une croissance :

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'

Si le broker tourne dans Docker sur un hôte que vous contrôlez, relevez sa mémoire et son CPU une fois par minute :

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

Pour tout ce qui tourne plus longtemps qu'un test, exportez ces mêmes valeurs vers une base de séries temporelles — supervision avec Prometheus et Grafana et supervision des topics $SYS détaillent la mise en place.

Transformer le CSV en verdict

Parcourir à l'œil un CSV de 4 000 lignes ne marche pas. Tracez plutôt une droite de tendance. Ignorez la première heure (caches, JIT, pools de connexions qui chauffent) et calculez la pente de la mémoire dans le temps avec la bibliothèque standard :

import csv, statistics, sys

rows = list(csv.DictReader(open(sys.argv[1])))[60:]   # ignore 1 h de montée en température
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']}")

Comment l'interpréter :

• Tendance RSS proche de zéro avec des dents de scie dans les données brutes — garbage collector normal. Une pente positive régulière à charge constante est une fuite ; multipliez-la par la durée de fonctionnement de votre déploiement pour juger de sa gravité (5 Mo/heure font 3,6 Go par mois).
• Descripteurs de fichiers en hausse — quelque chose ouvre des sockets ou des fichiers sans les fermer, généralement dans le chemin de reconnexion.
• p99 qui monte alors que p50 reste plat — des files qui grossissent, des pauses de GC plus longues ou un broker qui accumule des messages stockés.
• Des trous qui coïncident avec des déconnexions — normal en QoS 0, un bug en QoS 1 avec une session persistante. Des trous sans aucune déconnexion signalent un broker qui jette des messages (limites de file) ou un subscriber qui ne suit pas.
• messages/stored du broker en hausse — généralement des sessions persistantes de clients disparus. Définissez un session expiry (MQTT 5) ou nettoyez les client ID obsolètes.

Rendre le soak test réaliste

Un flux constant entre deux clients trouve les fuites du client. Pour trouver celles du broker, le trafic doit emprunter les chemins de code qui allouent de la mémoire :

1. Rotation des connexions — une partie des clients se déconnecte et se reconnecte toutes les quelques minutes, certains proprement, d'autres en coupant le socket pour déclencher le Last Will.
2. Diversité des sessions — mélangez sessions propres et persistantes, et incluez quelques clients persistants qui restent hors ligne pendant des heures.
3. Diversité des topics — publiez sur des milliers de topics différents, pas sur un seul ; incluez des messages retenus et des subscribers avec wildcards.
4. Tailles de payload réelles — générez-les avec un générateur de données de test MQTT plutôt qu'avec des chaînes fixes.
5. Pannes réseau — provoquez quelques coupures courtes pour vérifier la reprise ; tester la reconnexion MQTT explique comment.

Avant de lancer l'exécution de trois jours, vérifiez le topic et les identifiants avec le client MQTT en ligne et confirmez que les messages arrivent.

Observez le trafic du soak test dans votre navigateur

Un soak test ne coûte presque rien : une petite machine, un script et de la patience. Lancez-le avant chaque version majeure et après chaque mise à jour du broker ou de la bibliothèque, conservez les CSV et comparez les droites de tendance — la dégradation lente n'est invisible que lorsque personne ne la mesure.

Questions fréquentes