Chaos testing com MQTT: partições de rede, perda de pacotes e reinícios do broker

Chaos testing com MQTT significa quebrar a rede e o broker de propósito (latência, perda de pacotes, partições, reinícios) enquanto você mede se os clientes se recuperam sozinhos. A maioria dos bugs de MQTT que chega à produção é de recuperação: um cliente que nunca percebe uma conexão morta, uma assinatura que não é restaurada, uma mensagem QoS 1 perdida porque a sessão não era persistente. Este guia mostra as ferramentas (tc netem, iptables, Toxiproxy, docker) e um script sonda que transforma cada experimento em números.

Primeiro defina o que significa "recuperado"

Um experimento de chaos sem critério de aprovação é só uma queda. Anote o estado estável ao qual o sistema deve voltar:

Dois mecanismos fazem a maior parte do trabalho: o intervalo de keep-alive, que limita quanto tempo uma conexão morta passa despercebida, e as sessões persistentes, que permitem ao broker guardar mensagens QoS 1 de um subscriber enquanto ele está fora.

A sonda: medir perdas, duplicatas e reconexões

Este script com paho-mqtt 2.x roda um publisher e um subscriber com uma sessão persistente MQTT 5. Ele publica uma mensagem numerada a cada 100 ms com QoS 1 e, a cada cinco segundos, informa quantas mensagens confirmadas ainda não chegaram (missing), quantas chegaram duas vezes (dups), o número de reconexões e a pior latência.

import json, os, threading, time, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

HOST = os.getenv("MQTT_HOST", "localhost")
PORT = int(os.getenv("MQTT_PORT", "1883"))
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/chaos/{RUN}/seq"
lock = threading.Lock()
seen, sent = set(), []            # sent: (seq, MQTTMessageInfo)
stats = {"dups": 0, "reconnects": 0, "max_ms": 0.0}

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"[{userdata['role']}] connected: {reason_code}, session present={flags.session_present}")
    if userdata["ever"]:
        stats["reconnects"] += 1
    userdata["ever"] = True
    if userdata["role"] == "sub":
        client.subscribe(TOPIC, qos=1)

def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f"[{userdata['role']}] disconnected: {reason_code}")

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    with lock:
        if data["seq"] in seen:
            stats["dups"] += 1
        seen.add(data["seq"])
        stats["max_ms"] = max(stats["max_ms"], (time.time() - data["t"]) * 1000)

def make(role, session_expiry):
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = session_expiry
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"chaos-{role}-{RUN}",
                    protocol=mqtt.MQTTv5, userdata={"role": role, "ever": False})
    c.on_connect, c.on_disconnect, c.on_message = on_connect, on_disconnect, on_message
    c.reconnect_delay_set(min_delay=1, max_delay=16)
    c.connect_async(HOST, PORT, keepalive=10, properties=props)
    c.loop_start()
    return c

sub = make("sub", session_expiry=600)   # session survives outages up to 10 minutes
pub = make("pub", session_expiry=0)
time.sleep(2)

seq, next_report = 0, time.time() + 5
while True:
    info = pub.publish(TOPIC, json.dumps({"seq": seq, "t": time.time()}), qos=1)
    sent.append((seq, info))
    seq += 1
    if time.time() >= next_report:
        acked = {s for s, i in sent if i.is_published()}
        with lock:
            print(f"{time.strftime('%H:%M:%S')} sent={seq} acked={len(acked)} "
                  f"missing={len(acked - seen)} dups={stats['dups']} "
                  f"reconnects={stats['reconnects']} max_latency={stats['max_ms']:.0f}ms")
            stats["max_ms"] = 0.0
        next_report += 5
    time.sleep(0.1)

Enquanto a rede está fora, o paho guarda as mensagens QoS 1 na fila em memória e as envia depois de reconectar, então sent continua crescendo enquanto acked para. Durante uma queda, missing pode subir brevemente; após a recuperação ele precisa voltar a 0. Um valor diferente de zero permanente significa que o broker confirmou mensagens que depois se perderam, normalmente por uma sessão que não sobreviveu.

Experimento 1: latência e perda de pacotes com tc netem

No Linux, o tc netem degrada uma interface. Rode-o dentro de uma VM ou de um container com --cap-add NET_ADMIN para não deixar sua estação de trabalho inteira lenta.

IFACE=eth0
sudo tc qdisc add dev $IFACE root netem delay 300ms 100ms loss 5%     # bad cellular
sudo tc qdisc change dev $IFACE root netem delay 2000ms loss 30%      # nearly unusable
sudo tc qdisc del dev $IFACE root                                     # back to normal

O esperado: com 300 ms e 5 % de perda, as retransmissões do TCP escondem a perda; a latência sobe, mas não há reconexões e missing fica em 0. Com 2 s e 30 % de perda, o PINGRESP pode chegar tarde demais e os clientes começam a reconectar; verifique se eles se estabilizam quando a regra é removida. Para perfis realistas de rede celular, veja testar MQTT em condições de rede celular.

Experimento 2: partições de rede com iptables

Uma rede pode falhar de duas formas bem diferentes, e os clientes muitas vezes só tratam uma delas:

# Blackhole: packets silently vanish. Only keep-alive can detect this.
sudo iptables -A OUTPUT -p tcp --dport 1883 -j DROP
# ...wait 60 s, then heal:
sudo iptables -D OUTPUT -p tcp --dport 1883 -j DROP

# Hard failure: the client gets a TCP reset immediately.
sudo iptables -A OUTPUT -p tcp --dport 1883 -j REJECT --reject-with tcp-reset
sudo iptables -D OUTPUT -p tcp --dport 1883 -j REJECT --reject-with tcp-reset

Com keep-alive de 10 segundos, a sonda deve registrar disconnected cerca de 10–20 segundos depois do início do blackhole, tentar de novo com backoff enquanto ele durar e reconectar com session present=True quando a rede voltar. Do lado do broker, outros clientes devem receber o Last Will da sonda após cerca de 1,5 × keep-alive. Um cliente que continua "conectado" por minutos durante um blackhole está com keep-alive desativado ou alto demais. Testar QoS com corte de rede aprofunda o que cada nível de QoS garante nesse caso.

Experimento 3: falhas por conexão com Toxiproxy

O Toxiproxy é um proxy TCP cujo comportamento você altera por uma API HTTP. Funciona também em macOS e Windows, afeta só as conexões que passam por ele e é fácil de roteirizar no CI.

docker run -d --name toxiproxy -p 8474:8474 -p 21883:21883 ghcr.io/shopify/toxiproxy
curl -s -X POST localhost:8474/proxies \
  -d '{"name":"mqtt","listen":"0.0.0.0:21883","upstream":"broker.connectmq.com:1883"}'

# Point the probe at the proxy
MQTT_HOST=localhost MQTT_PORT=21883 python chaos_probe.py

Depois injete falhas a partir de um segundo terminal:

API=localhost:8474/proxies/mqtt
# 1.5 s latency with 500 ms jitter
curl -s -X POST $API/toxics -d '{"name":"lag","type":"latency","attributes":{"latency":1500,"jitter":500}}'
curl -s -X DELETE $API/toxics/lag
# Blackhole: data stops, connection stays open until the toxic is removed
curl -s -X POST $API/toxics -d '{"name":"cut","type":"timeout","attributes":{"timeout":0}}'
curl -s -X DELETE $API/toxics/cut
# Connection reset by peer
curl -s -X POST $API/toxics -d '{"name":"rst","type":"reset_peer","attributes":{"timeout":0}}'
curl -s -X DELETE $API/toxics/rst
# Take the whole "network" down and up again
curl -s -X POST $API -d '{"enabled":false}'; sleep 30; curl -s -X POST $API -d '{"enabled":true}'

O toxic timeout com 0 é o equivalente no Toxiproxy a um DROP do iptables; reset_peer corresponde ao REJECT. Testar MQTT com Toxiproxy cobre limites de banda, slicing e integração com frameworks de teste.

Experimento 4: reinícios e quedas do broker

Experimentos de reinício precisam de um broker que você mesmo roda. Um container local do Mosquitto basta para encontrar a maioria dos bugs do lado do cliente:

docker run -d --name chaos-broker -p 1883:1883 eclipse-mosquitto:2 mosquitto -c /mosquitto-no-auth.conf
MQTT_HOST=localhost MQTT_PORT=1883 python chaos_probe.py     # in another terminal

docker restart chaos-broker                                   # graceful: SIGTERM, then start
docker kill --signal=KILL chaos-broker && docker start chaos-broker   # crash

O que você deve observar:

• Sem persistência (a configuração padrão acima), o broker esquece todas as sessões. A sonda reconecta com session present=False, assina de novo, e mensagens publicadas durante o reinício podem ficar missing para sempre.
• Com persistence true e um volume para o diretório de dados, sessões e mensagens enfileiradas sobrevivem a um reinício ordenado. Depois de um KILL, tudo desde o último autosave ainda pode se perder; confira se isso bate com o que seu design promete.
• Os clientes não podem reconectar todos juntos. Com centenas de clientes, um reinício vira um pico de conexões. Backoff com jitter espalha esse pico; veja tempestades de reconexão MQTT.

Em brokers em cluster, troque o reinício por parar um nó de cada vez e verifique se os clientes fazem failover. Testes de reconexão MQTT traz mais cenários do lado do cliente.

Um plano de experimentos repetível

Rode o plano antes de cada release importante do cliente ou do broker, guarde a saída da sonda como artefato e adicione uma linha nova sempre que um incidente real ensinar um novo modo de falha.

Veja as mensagens da sonda chegarem no navegador

Chaos testing com MQTT é, acima de tudo, sobre recuperação: detectar a falha rápido, reconectar com suavidade e não perder nada que você prometeu entregar. Com netem, iptables, Toxiproxy e um broker que pode ser reiniciado, você reproduz quase qualquer queda em um notebook, e a sonda diz em números se seus clientes passam.

Perguntas frequentes