Soak test MQTT: encontre vazamentos de memória e degradação lenta

Um soak test (teste de resistência) mantém um sistema MQTT sob carga normal por muito tempo — geralmente de 24 a 72 horas — para pegar problemas que um benchmark de dez minutos nunca vê: memória que cresce alguns megabytes por hora, descritores de arquivo que nunca são liberados, filas de sessões que nunca voltam e uma latência que sobe devagar até os dispositivos começarem a dar timeout.

Este guia mostra o que medir, entrega um script Python que grava tudo em CSV e explica como ler o resultado para que "parece estar tudo bem" vire um número.

Soak test vs load test

Se você ainda não mediu a capacidade, comece por load testing de broker MQTT. Um soak test parte do princípio de que o sistema aguenta a carga agora e pergunta se ainda vai aguentar no terceiro dia.

O que se degrada devagar

A maioria das falhas de longo prazo cai em poucos padrões:

• Vazamentos de memória no cliente — callbacks que vão adicionando itens a uma lista, uma fila in-flight que cresce porque o cliente publica mais rápido do que os acknowledgements chegam, ou um novo objeto cliente criado a cada reconexão sem parar o antigo. Veja vazamentos de memória em clientes MQTT.
• Filas do broker sem limite — sessões persistentes (clean_session=False / session expiry) de dispositivos que nunca reconectam continuam acumulando mensagens QoS 1 e 2 até o broker ficar sem memória. O resultado clássico está em broker MQTT caindo por OOM.
• Crescimento de mensagens retidas — topics com IDs únicos no caminho (.../session-8f3a/state) publicados com retain=true se acumulam para sempre.
• Vazamento de descritores de arquivo e sockets — caminhos de reconexão que não fecham o socket antigo. O processo acaba falhando com Too many open files.
• Eventos baseados em tempo — rotação de logs, renovação de certificados, expiração de tokens, backups diários ou um disco enchendo com arquivos de persistência. Eles só acontecem se o teste durar o bastante para atravessá-los.

As métricas a registrar

Registre números do cliente e do broker na mesma linha do tempo, uma vez por minuto:

Um script Python para soak test

Este script (paho-mqtt 2.x e psutil) publica uma mensagem numerada em ritmo fixo, assina o próprio topic e grava uma linha CSV por minuto com throughput, lacunas, duplicatas, percentis de latência, a própria memória e os descritores abertos. Ele assina de novo no on_connect, então sobrevive a reconexões em vez de ficar surdo sem avisar.

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            # mensagens por segundo
DURATION_H = 72      # duração total
WINDOW_S = 60        # uma linha CSV por minuto

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

Rode com nohup python soak.py & (ou dentro do tmux) em uma máquina que vai ficar ligada por três dias. O CSV é gravado a cada minuto, então você pode olhar o arquivo com o teste ainda rodando. num_fds() existe no Linux e no macOS; no Windows use proc.num_handles(). Um teste curto contra o endpoint compartilhado broker.connectmq.com não tem problema, mas para a execução real de vários dias defina MQTT_HOST com a sua própria instância de broker: o endpoint compartilhado é feito para testes rápidos, tem rate limit e não guarda sessões persistentes, então as falhas ligadas a sessões descritas abaixo só aparecem num broker seu.

Observe também o lado do broker

Os números do cliente dizem que algo degradou; os do broker dizem onde. Se o seu broker publica topics $SYS (o Mosquitto publica), assine os contadores que revelam crescimento:

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'

Se o broker roda em Docker num host que você controla, colete memória e CPU uma vez por minuto:

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

Para qualquer coisa que rode por mais tempo que um teste, exporte esses valores para um banco de séries temporais — monitoramento com Prometheus e Grafana e monitoramento de topics $SYS mostram a configuração.

Transformando o CSV em um veredito

Olhar um CSV de 4.000 linhas no olho não funciona. Ajuste uma linha de tendência. Pule a primeira hora (caches, JIT, pools de conexão aquecendo) e calcule a inclinação da memória ao longo do tempo com a biblioteca padrão:

import csv, statistics, sys

rows = list(csv.DictReader(open(sys.argv[1])))[60:]   # pula 1 h de aquecimento
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']}")

Como interpretar:

• Tendência de RSS perto de zero com dente de serra nos dados brutos — coleta de lixo normal. Uma inclinação positiva constante com carga constante é vazamento; multiplique pelo tempo de atividade do seu deploy para ver se importa (5 MB/hora são 3,6 GB por mês).
• Descritores de arquivo subindo — algo abre sockets ou arquivos sem fechar, normalmente no caminho de reconexão.
• p99 subindo enquanto p50 fica estável — filas crescendo, pausas de coleta de lixo mais longas ou um broker acumulando mensagens armazenadas.
• Lacunas que batem com desconexões — esperado em QoS 0, um bug em QoS 1 com sessão persistente. Lacunas sem nenhuma desconexão indicam que o broker está descartando mensagens (limites de fila) ou um subscriber que não dá conta.
• messages/stored do broker subindo — geralmente sessões persistentes de clientes que sumiram. Defina um session expiry (MQTT 5) ou limpe client IDs antigos.

Deixe o soak test realista

Um fluxo constante entre dois clientes encontra vazamentos no cliente. Para achar vazamentos no broker, o tráfego precisa passar pelos caminhos de código que alocam memória:

1. Rotatividade de conexões — uma parte dos clientes desconecta e reconecta a cada poucos minutos, alguns de forma limpa e outros derrubando o socket para que o Last Will dispare.
2. Variedade de sessões — misture sessões limpas e persistentes, e inclua alguns clientes persistentes que ficam offline por horas.
3. Variedade de topics — publique em milhares de topics distintos, não em um só; inclua mensagens retidas e subscribers com wildcards.
4. Tamanhos de payload reais — gere-os com um gerador de dados de teste MQTT em vez de strings fixas.
5. Falhas de rede — provoque algumas quedas curtas para confirmar a recuperação; testes de reconexão MQTT mostra como.

Antes de iniciar a execução de três dias, confira o topic e as credenciais com o cliente MQTT online e confirme que as mensagens chegam.

Acompanhe o tráfego do soak test no navegador

Um soak test é barato: uma máquina pequena, um script e paciência. Rode antes de cada release importante e depois de cada atualização de broker ou biblioteca, guarde os CSVs e compare as linhas de tendência — degradação lenta só é invisível quando ninguém mede.

Perguntas frequentes