Pruebas de carga MQTT con Locust

Locust es un framework de pruebas de carga en Python con dashboard web en directo, workers distribuidos y escenarios escritos como código normal. Por defecto habla HTTP, pero su clase User puede manejar cualquier protocolo, así que hacer pruebas de carga MQTT con Locust consiste en envolver paho-mqtt en una clase de usuario y notificar cada connect y cada publish a las estadísticas de Locust.

Esta guía te da un locustfile.py completo: un usuario MQTT base reutilizable, usuarios sensor que publican telemetría, un usuario dashboard que mide la latencia extremo a extremo y los comandos para ejecutarlo desde el navegador, en modo headless en CI y repartido entre varias máquinas.

¿Por qué Locust para MQTT?

Locust es la opción adecuada cuando un "usuario" debe comportarse como un dispositivo: conectarse, publicar según un calendario, reaccionar a mensajes, reconectarse, y quieres escribir esa lógica en Python. Para obtener rápido un techo de throughput, un benchmark de CLI es más sencillo; consulta load testing de un broker MQTT.

Cómo funciona la integración

Cada usuario de Locust es un greenlet ligero (Locust funciona sobre gevent). Nuestro usuario MQTT:

1. Crea un cliente paho-mqtt por usuario de Locust, así que 1.000 usuarios son 1.000 conexiones MQTT.
2. Mide el tiempo de conexión desde connect() hasta el CONNACK.
3. Mide el tiempo de publish desde publish() hasta el PUBACK en QoS 1 (QoS 0 se notifica en cuanto se escribe).
4. Notifica cada resultado con environment.events.request.fire(...), que es exactamente lo que hace internamente el cliente HTTP de Locust; por eso el dashboard, la exportación CSV y la pestaña de fallos funcionan sin cambios.

El 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

Algunos detalles importantes:

• name="telemetry" agrupa el publish de todos los sensores en una sola fila. Sin él, el topic único de cada cliente se convierte en una línea propia de la tabla de estadísticas y el dashboard se vuelve ilegible.
• El lock alrededor de publish() y de pending evita una condición de carrera en la que un PUBACK muy rápido se procesa antes de haber registrado el ID del mensaje.
• Pesos: con weight = 20 y weight = 1, una prueba con 210 usuarios arranca unos 200 sensores y 10 dashboards. Cada dashboard recibe la telemetría de todos los sensores, así que los dashboards también prueban el fan-out.
• La latencia extremo a extremo compara time.time() en dos clientes. Es precisa en una sola máquina; entre máquinas, los relojes deben estar sincronizados con NTP o los números incluirán el desfase de reloj.

Ejecutar la prueba

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

Abre http://localhost:8089, introduce el número de usuarios y el spawn rate (usuarios arrancados por segundo) y empieza. La pestaña Statistics muestra una fila por nombre (connect, telemetry, status, telemetry end-to-end) con tasa de peticiones, fallos y percentiles. Charts dibuja en directo las peticiones por segundo, los tiempos de respuesta y el número de usuarios; Failures agrupa errores como The client is not currently connected. o no CONNACK within 10 s.

Para CI o ejecuciones largas sin supervisión, usa el modo headless y guarda los informes:

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

--spawn-rate es tu ritmo de conexiones. Mantenlo en un valor que tu flota real podría producir; 2.000 usuarios a 50 por segundo se completan en 40 segundos. Un spawn rate muy alto convierte una prueba de carga en una prueba de tormenta de reconexiones: útil, pero es otra pregunta.

Antes de escalar, ejecuta con unos pocos usuarios y mira los topics en el cliente MQTT online (testmqtt/locust/#) para confirmar que los payloads y los nombres de topic son correctos. Ese tipo de smoke test funciona bien en el endpoint compartido broker.connectmq.com, pero limita la tasa de tráfico, así que define MQTT_HOST con tu propia instancia de broker para la prueba de carga real.

Escalar: workers y procesos

Un proceso de Locust usa un núcleo de CPU. Cuando el generador de carga se satura, Locust registra CPU usage above 90%! This may constrain your throughput and may even give inconsistent response time measurements!; a partir de ahí, los tiempos de respuesta reflejan tu portátil, no el broker. Añade procesos:

# 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

Los workers comparten el mismo locustfile; el master agrega las estadísticas en un único dashboard. Reparte los workers entre varias IP de origen si pasas de decenas de miles de conexiones: los límites de puertos efímeros y de descriptores de archivo de un solo host son un falso techo habitual.

Interpretar los resultados de Locust para MQTT

Compara las RPS de PUBLISH telemetry con las de RECEIVE divididas entre el número de dashboards: si la recepción se queda por detrás de la publicación, los mensajes se están encolando o descartando. El QoS cambia el flujo de paquetes y por tanto los números; niveles de QoS en MQTT explica por qué la latencia en QoS 1 incluye un viaje de ida y vuelta completo. Para convertir percentiles en una decisión de capacidad de forma estructurada, consulta interpretar resultados de benchmarks MQTT.

Variantes que merece la pena añadir

• TLS: llama a c.tls_set() antes de connect() y define MQTT_PORT=8883 para incluir los handshakes TLS en el tiempo de conexión.
• MQTT 5: pasa protocol=mqtt.MQTTv5 a mqtt.Client para probar brokers MQTT 5; los reason codes del PUBACK aparecerán entonces como fallos en el dashboard.
• Formas de carga: una clase LoadTestShape puede subir por escalones o simular automáticamente un pico matinal.
• Comandos: añade un usuario backend que publique en testmqtt/locust/<id>/cmd y haz que los sensores confirmen, para probar también la carga del canal de bajada.

La gran ventaja de Locust para MQTT es que un usuario es solo Python: comportamiento realista de dispositivos, un dashboard que todo tu equipo puede leer y el mismo archivo funcionando en un portátil, en CI o en una flota de workers. Empieza con poco, verifica los topics y después sube la carga y deja que los percentiles te digan dónde están los límites.

Preguntas frecuentes