Tests de charge MQTT avec Locust

Locust est un framework Python de tests de charge avec un dashboard web en direct, des workers distribués et des scénarios écrits en code ordinaire. Il parle HTTP par défaut, mais sa classe User peut piloter n'importe quel protocole : faire des tests de charge MQTT avec Locust revient donc à encapsuler paho-mqtt dans une classe utilisateur et à signaler chaque connect et chaque publish aux statistiques de Locust.

Ce guide fournit un locustfile.py complet : un utilisateur MQTT de base réutilisable, des utilisateurs capteurs qui publient de la télémétrie, un utilisateur dashboard qui mesure la latence de bout en bout, et les commandes pour l'exécuter depuis le navigateur, en headless dans la CI et réparti sur plusieurs machines.

Pourquoi Locust pour MQTT ?

Locust est le bon choix lorsqu'un « utilisateur » doit se comporter comme un appareil : se connecter, publier selon un calendrier, réagir aux messages, se reconnecter — et que vous voulez écrire cette logique en Python. Pour obtenir rapidement un plafond de débit, un benchmark en ligne de commande est plus simple ; voir le test de charge d'un broker MQTT.

Comment fonctionne l'intégration

Chaque utilisateur Locust est un greenlet léger (Locust repose sur gevent). Notre utilisateur MQTT :

1. Crée un client paho-mqtt par utilisateur Locust : 1 000 utilisateurs font donc 1 000 connexions MQTT.
2. Mesure le temps de connexion de connect() jusqu'au CONNACK.
3. Mesure le temps de publish de publish() jusqu'au PUBACK en QoS 1 (QoS 0 est signalé dès l'écriture).
4. Signale chaque résultat avec environment.events.request.fire(...), exactement comme le fait en interne le client HTTP de Locust : dashboard, export CSV et onglet des erreurs fonctionnent donc sans modification.

Le locustfile

import json, os, random, threading, time, uuid
import paho.mqtt.client as mqtt
from locust import User, between, constant, task
from locust.exception import StopUser

HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", 1883))
PREFIX = "testmqtt/locust"

class MqttUser(User):
    abstract = True
    subscriptions: list[tuple[str, int]] = []

    def on_start(self):
        self.client_id = f"locust-{uuid.uuid4().hex[:10]}"
        self.pending, self.lock = {}, threading.Lock()
        self.connected, self.connect_rc = threading.Event(), None
        c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=self.client_id)
        if os.environ.get("MQTT_USER"):
            c.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
        c.on_connect, c.on_publish, c.on_message = self._on_connect, self._on_publish, self.on_message
        self.client = c
        start = time.perf_counter()
        try:
            c.connect(HOST, PORT, keepalive=60)
            c.loop_start()
            if not self.connected.wait(10):
                raise TimeoutError("no CONNACK within 10 s")
            if self.connect_rc.is_failure:
                raise ConnectionError(f"CONNACK: {self.connect_rc}")
            self._report("CONNECT", "connect", start, 0)
        except Exception as e:
            self._report("CONNECT", "connect", start, 0, e)
            c.loop_stop()
            raise StopUser()

    def on_stop(self):
        with self.lock:
            leftovers, self.pending = list(self.pending.values()), {}
        for start, name, size in leftovers:
            self._report("PUBLISH", name, start, size, TimeoutError("no PUBACK before stop"))
        self.client.disconnect()
        self.client.loop_stop()

    # --- reporting -------------------------------------------------------
    def _report(self, kind, name, start, size, exc=None):
        self.environment.events.request.fire(
            request_type=kind, name=name, response_time=(time.perf_counter() - start) * 1000,
            response_length=size, exception=exc, context={"client_id": self.client_id})

    # --- paho callbacks (run in the network loop) -----------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        self.connect_rc = reason_code
        for topic, qos in self.subscriptions:          # also restores them after a reconnect
            client.subscribe(topic, qos)
        self.connected.set()

    def _on_publish(self, client, userdata, mid, reason_code, properties):
        with self.lock:
            item = self.pending.pop(mid, None)
        if item:
            start, name, size = item
            exc = Exception(f"PUBACK: {reason_code}") if reason_code.is_failure else None
            self._report("PUBLISH", name, start, size, exc)

    def on_message(self, client, userdata, msg):
        pass

    # --- API for tasks ---------------------------------------------------
    def publish(self, topic, payload, qos=1, name=None):
        name, start = name or topic, time.perf_counter()
        with self.lock:
            info = self.client.publish(topic, payload, qos=qos)
            if info.rc == mqtt.MQTT_ERR_SUCCESS and qos > 0:
                self.pending[info.mid] = (start, name, len(payload))
                return
        exc = None if info.rc == mqtt.MQTT_ERR_SUCCESS else Exception(mqtt.error_string(info.rc))
        self._report("PUBLISH", name, start, len(payload), exc)

class Sensor(MqttUser):
    weight = 20
    wait_time = between(4, 6)                          # one reading every ~5 s

    @task(10)
    def telemetry(self):
        payload = json.dumps({"ts": time.time(), "temp": round(random.gauss(21, 0.5), 2)})
        self.publish(f"{PREFIX}/{self.client_id}/telemetry", payload, qos=1, name="telemetry")

    @task(1)
    def heartbeat(self):
        self.publish(f"{PREFIX}/{self.client_id}/status", "ok", qos=0, name="status")

class Dashboard(MqttUser):
    weight = 1
    wait_time = constant(1)
    subscriptions = [(f"{PREFIX}/+/telemetry", 1)]

    def on_message(self, client, userdata, msg):
        sent = json.loads(msg.payload)["ts"]
        self.environment.events.request.fire(
            request_type="RECEIVE", name="telemetry end-to-end",
            response_time=(time.time() - sent) * 1000, response_length=len(msg.payload),
            exception=None, context={})

    @task
    def idle(self):
        pass                                           # messages arrive via on_message

Quelques détails qui comptent :

• name="telemetry" regroupe les publish de tous les capteurs sur une seule ligne. Sans cela, le topic unique de chaque client devient une ligne à part dans le tableau des statistiques et le dashboard devient illisible.
• Le verrou autour de publish() et de pending évite une situation de concurrence où un PUBACK très rapide est traité avant l'enregistrement de l'identifiant du message.
• Les poids — avec weight = 20 et weight = 1, un test à 210 utilisateurs démarre environ 200 capteurs et 10 dashboards. Chaque dashboard reçoit la télémétrie de tous les capteurs : les dashboards testent donc aussi le fan-out.
• La latence de bout en bout compare time.time() sur deux clients. C'est exact sur une seule machine ; entre plusieurs machines, les horloges doivent être synchronisées par NTP, sinon les chiffres incluent le décalage d'horloge.

Lancer le test

pip install locust "paho-mqtt>=2"
export MQTT_HOST=your-broker-host    # your own broker instance, not the shared endpoint
export MQTT_USER=username MQTT_PASS=password
ulimit -n 8192                         # one socket per user
locust -f locustfile.py

Ouvrez http://localhost:8089, saisissez le nombre d'utilisateurs et le spawn rate (utilisateurs démarrés par seconde), puis lancez. L'onglet Statistics affiche une ligne par nom — connect, telemetry, status, telemetry end-to-end — avec le débit de requêtes, les échecs et les percentiles. Charts trace en direct les requêtes par seconde, les temps de réponse et le nombre d'utilisateurs ; Failures regroupe les erreurs comme The client is not currently connected. ou no CONNACK within 10 s.

Pour la CI ou de longues exécutions sans surveillance, utilisez le mode headless et conservez les rapports :

locust -f locustfile.py --headless \
  --users 2000 --spawn-rate 50 --run-time 15m \
  --csv results/mqtt --html results/mqtt-report.html

--spawn-rate correspond à votre cadence de connexions. Gardez une valeur que votre vraie flotte pourrait plausiblement produire ; 2 000 utilisateurs à 50 par seconde sont lancés en 40 secondes. Un spawn rate très élevé transforme le test de charge en test de tempête de reconnexions — utile, mais c'est une autre question.

Avant de monter en charge, lancez quelques utilisateurs et observez les topics dans le client MQTT en ligne (testmqtt/locust/#) pour vérifier que les payloads et les noms de topics sont corrects. Ce type de smoke test fonctionne très bien sur le point d'accès partagé broker.connectmq.com, mais celui-ci limite le débit : définissez MQTT_HOST sur votre propre instance de broker pour le vrai test de charge.

Passer à l'échelle : workers et processus

Un processus Locust utilise un cœur de CPU. Lorsque le générateur de charge sature, Locust journalise CPU usage above 90%! This may constrain your throughput and may even give inconsistent response time measurements! — à partir de là, les temps de réponse reflètent votre portable, pas le broker. Ajoutez des processus :

# all cores on one machine (Linux/macOS)
locust -f locustfile.py --processes -1

# or a master plus workers on several machines
locust -f locustfile.py --master
locust -f locustfile.py --worker --master-host 10.0.0.5

Les workers partagent le même locustfile ; le master agrège les statistiques dans un seul dashboard. Répartissez les workers sur plusieurs IP sources si vous dépassez quelques dizaines de milliers de connexions : les limites de ports éphémères et de descripteurs de fichiers d'un hôte unique sont un faux plafond fréquent.

Lire les résultats Locust pour MQTT

Comparez le RPS de PUBLISH telemetry avec celui de RECEIVE divisé par le nombre de dashboards : si la réception prend du retard sur la publication, les messages s'accumulent ou sont perdus. La QoS change le flux de paquets et donc les chiffres — les niveaux de QoS MQTT expliquent pourquoi la latence en QoS 1 inclut un aller-retour complet. Pour transformer des percentiles en décision de capacité de façon structurée, voir interpréter les résultats de benchmark MQTT.

Variantes utiles

• TLS — appelez c.tls_set() avant connect() et définissez MQTT_PORT=8883 pour inclure les handshakes TLS dans le temps de connexion.
• MQTT 5 — passez protocol=mqtt.MQTTv5 à mqtt.Client pour tester des brokers MQTT 5 ; les reason codes du PUBACK apparaissent alors comme des échecs dans le dashboard.
• Profils de charge — une classe LoadTestShape peut monter par paliers ou simuler automatiquement un pic matinal.
• Commandes — ajoutez un utilisateur backend qui publie sur testmqtt/locust/<id>/cmd et faites acquitter les capteurs, pour tester aussi la charge sur le chemin descendant.

La force de Locust pour MQTT, c'est qu'un utilisateur n'est que du Python : un comportement d'appareil réaliste, un dashboard lisible par toute l'équipe et le même fichier qui tourne sur un portable, dans la CI ou sur une flotte de workers. Commencez petit, vérifiez les topics, puis montez en charge et laissez les percentiles vous montrer où sont les limites.

Questions fréquentes