Teste de carga MQTT com Locust

O Locust é um framework de testes de carga em Python com dashboard web ao vivo, workers distribuídos e cenários escritos como código comum. Por padrão ele fala HTTP, mas a classe User pode controlar qualquer protocolo — então fazer teste de carga MQTT com Locust se resume a encapsular o paho-mqtt numa classe de usuário e reportar cada connect e publish para as estatísticas do Locust.

Este guia entrega um locustfile.py completo: um usuário MQTT base reutilizável, usuários sensor que publicam telemetria, um usuário dashboard que mede a latência ponta a ponta e os comandos para rodar tudo pelo navegador, em modo headless no CI e distribuído entre várias máquinas.

Por que Locust para MQTT?

O Locust é a escolha certa quando um "usuário" precisa se comportar como um dispositivo: conectar, publicar num cronograma, reagir a mensagens, reconectar — e você quer escrever essa lógica em Python. Para descobrir rapidamente o teto de throughput, um benchmark de CLI é mais simples; veja load testing de broker MQTT.

Como a integração funciona

Cada usuário do Locust é um greenlet leve (o Locust roda sobre gevent). Nosso usuário MQTT:

1. Cria um cliente paho-mqtt por usuário do Locust, então 1.000 usuários significam 1.000 conexões MQTT.
2. Mede o tempo de conexão de connect() até o CONNACK.
3. Mede o tempo de publish de publish() até o PUBACK em QoS 1 (QoS 0 é reportado assim que é escrito).
4. Reporta cada resultado com environment.events.request.fire(...), exatamente o que o cliente HTTP do Locust faz internamente — por isso dashboard, exportação CSV e aba de falhas funcionam sem mudanças.

O 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

Alguns detalhes que importam:

• name="telemetry" agrupa o publish de todos os sensores numa linha só. Sem isso, o topic único de cada cliente vira uma linha própria na tabela de estatísticas e o dashboard fica ilegível.
• O lock em volta de publish() e pending evita uma condição de corrida em que um PUBACK muito rápido é processado antes de o ID da mensagem ser registrado.
• Pesos — com weight = 20 e weight = 1, um teste com 210 usuários inicia cerca de 200 sensores e 10 dashboards. Cada dashboard recebe a telemetria de todos os sensores, então os dashboards também testam fan-out.
• A latência ponta a ponta compara time.time() em dois clientes. Isso é preciso numa única máquina; entre máquinas, os relógios precisam estar sincronizados com NTP, ou os números incluem a diferença de relógio.

Rodando o teste

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

Abra http://localhost:8089, informe o número de usuários e o spawn rate (usuários iniciados por segundo) e comece. A aba Statistics mostra uma linha por nome — connect, telemetry, status, telemetry end-to-end — com taxa de requisições, falhas e percentis. Charts plota ao vivo requisições por segundo, tempos de resposta e número de usuários; Failures agrupa erros como The client is not currently connected. ou no CONNACK within 10 s.

Para CI ou execuções longas sem supervisão, use o modo headless e guarde os relatórios:

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

--spawn-rate é a sua taxa de conexões. Mantenha um valor que sua frota real poderia produzir; 2.000 usuários a 50 por segundo sobem em 40 segundos. Um spawn rate muito alto transforma o teste de carga num teste de tempestade de reconexões — útil, mas é outra pergunta.

Antes de escalar, rode com poucos usuários e acompanhe os topics no cliente MQTT online (testmqtt/locust/#) para confirmar que payloads e nomes de topic estão corretos. Esse tipo de smoke test funciona bem no endpoint compartilhado broker.connectmq.com, mas ele limita a taxa de tráfego, então defina MQTT_HOST com a sua própria instância de broker para o teste de carga de verdade.

Escalando: workers e processos

Um processo do Locust usa um núcleo de CPU. Quando o gerador de carga satura, o Locust registra CPU usage above 90%! This may constrain your throughput and may even give inconsistent response time measurements! — a partir daí, os tempos de resposta refletem o seu notebook, não o broker. Adicione processos:

# 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

Os workers compartilham o mesmo locustfile; o master agrega as estatísticas num único dashboard. Distribua os workers entre IPs de origem diferentes se passar de dezenas de milhares de conexões — limites de portas efêmeras e de descritores de arquivo num único host são um falso teto comum.

Lendo os resultados do Locust para MQTT

Compare o RPS de PUBLISH telemetry com o de RECEIVE dividido pelo número de dashboards: se a recepção fica para trás da publicação, as mensagens estão enfileirando ou sendo descartadas. O QoS muda o fluxo de pacotes e, portanto, os números — níveis de QoS no MQTT explica por que a latência em QoS 1 inclui uma ida e volta completa. Para transformar percentis numa decisão de capacidade de forma estruturada, veja como interpretar resultados de benchmark MQTT.

Variações que valem a pena

• TLS — chame c.tls_set() antes de connect() e defina MQTT_PORT=8883 para incluir os handshakes TLS no tempo de conexão.
• MQTT 5 — passe protocol=mqtt.MQTTv5 para mqtt.Client para testar brokers MQTT 5; os reason codes do PUBACK passam a aparecer como falhas no dashboard.
• Formatos de carga — uma classe LoadTestShape pode subir em degraus ou simular automaticamente um pico matinal.
• Comandos — adicione um usuário backend que publica em testmqtt/locust/<id>/cmd e faça os sensores confirmarem, para testar a carga no caminho de descida também.

A força do Locust para MQTT é que um usuário é só Python: comportamento realista de dispositivo, um dashboard que o time inteiro consegue ler e o mesmo arquivo rodando num notebook, no CI ou numa frota de workers. Comece pequeno, verifique os topics, depois aumente a carga e deixe os percentis mostrarem onde estão os limites.

Perguntas frequentes