Prueba de resistencia (soak test) MQTT: detecta fugas de memoria y degradación lenta

Un soak test (prueba de resistencia) mantiene un sistema MQTT con carga normal durante mucho tiempo — normalmente de 24 a 72 horas — para detectar problemas que un benchmark de diez minutos nunca ve: memoria que crece unos megabytes por hora, descriptores de archivo que nunca se liberan, colas de sesiones que nunca vuelven y una latencia que sube poco a poco hasta que los dispositivos empiezan a dar timeouts.

Esta guía explica qué medir, te da un script Python que lo registra todo en CSV y muestra cómo leer el resultado para que "parece que va bien" se convierta en un número.

Soak test frente a load test

Si todavía no has medido la capacidad, empieza por load testing de un broker MQTT. Un soak test da por hecho que el sistema soporta la carga hoy y pregunta si la seguirá soportando el tercer día.

Qué se degrada lentamente

La mayoría de los fallos a largo plazo encajan en unos pocos patrones:

• Fugas de memoria en el cliente — callbacks que añaden elementos a una lista, una cola in-flight que crece porque el cliente publica más rápido de lo que llegan los acknowledgements, o un objeto cliente nuevo en cada reconexión sin detener el anterior. Consulta fugas de memoria en clientes MQTT.
• Colas del broker sin límite — las sesiones persistentes (clean_session=False / session expiry) de dispositivos que nunca se reconectan siguen acumulando mensajes QoS 1 y 2 hasta que el broker se queda sin memoria. El resultado clásico se describe en caídas del broker MQTT por OOM.
• Crecimiento de mensajes retenidos — topics con IDs únicos en la ruta (.../session-8f3a/state) publicados con retain=true se acumulan para siempre.
• Fugas de descriptores de archivo y sockets — rutas de reconexión que no cierran el socket anterior. El proceso acaba fallando con Too many open files.
• Eventos temporales — rotación de logs, renovación de certificados, expiración de tokens, copias de seguridad diarias o un disco que se llena con archivos de persistencia. Solo ocurren si la prueba dura lo suficiente para atravesarlos.

Las métricas que hay que registrar

Registra los datos del cliente y del broker en la misma línea temporal, una vez por minuto:

Un script Python para el soak test

Este script (paho-mqtt 2.x y psutil) publica un mensaje numerado a un ritmo fijo, se suscribe a su propio topic y escribe una fila CSV por minuto con throughput, huecos, duplicados, percentiles de latencia, su propia memoria y los descriptores abiertos. Se vuelve a suscribir en on_connect, así sobrevive a las reconexiones en lugar de quedarse sordo sin 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            # mensajes por segundo
DURATION_H = 72      # duración total
WINDOW_S = 60        # una fila 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

Ejecútalo con nohup python soak.py & (o dentro de tmux) en una máquina que vaya a seguir encendida tres días. El CSV se vuelca cada minuto, así que puedes revisarlo mientras la prueba sigue en marcha. num_fds() está disponible en Linux y macOS; en Windows usa proc.num_handles(). Una prueba corta contra el endpoint compartido broker.connectmq.com está bien, pero para la ejecución real de varios días define MQTT_HOST con tu propia instancia de broker: el endpoint compartido está pensado para pruebas rápidas, tiene límite de tasa y no guarda sesiones persistentes, así que los fallos relacionados con sesiones que se describen abajo solo aparecen en un broker propio.

Vigila también el lado del broker

Los números del cliente te dicen que algo se ha degradado; los del broker te dicen dónde. Si tu broker publica topics $SYS (Mosquitto lo hace), suscríbete a los contadores que revelan crecimiento:

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'

Si el broker corre en Docker en un host que controlas, toma una muestra de memoria y CPU una 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 todo lo que vaya a funcionar más tiempo que una prueba, exporta esos mismos valores a una base de datos de series temporales — monitorización con Prometheus y Grafana y monitorización de topics $SYS explican la configuración.

Convertir el CSV en un veredicto

Mirar a ojo un CSV de 4.000 filas no funciona. Ajusta una línea de tendencia. Descarta la primera hora (cachés, JIT, pools de conexiones calentándose) y calcula la pendiente de la memoria en el tiempo con la librería estándar:

import csv, statistics, sys

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

Cómo interpretarlo:

• Tendencia de RSS cercana a cero con diente de sierra en los datos brutos — recolección de basura normal. Una pendiente positiva constante con carga constante es una fuga; multiplícala por el tiempo que estará en marcha tu despliegue para ver si importa (5 MB/hora son 3,6 GB al mes).
• Descriptores de archivo en aumento — algo abre sockets o archivos sin cerrarlos, normalmente en la ruta de reconexión.
• p99 sube mientras p50 se mantiene — colas que crecen, pausas de recolección de basura más largas o un broker que acumula mensajes almacenados.
• Huecos que coinciden con desconexiones — esperable en QoS 0, un bug en QoS 1 con sesión persistente. Huecos sin ninguna desconexión apuntan a que el broker descarta mensajes (límites de cola) o a un suscriptor que no da abasto.
• messages/stored del broker subiendo — normalmente sesiones persistentes de clientes que ya no existen. Configura un session expiry (MQTT 5) o limpia los client ID obsoletos.

Haz que el soak test sea realista

Un flujo constante entre dos clientes encuentra fugas del cliente. Para encontrar fugas del broker, el tráfico tiene que recorrer las rutas de código que reservan memoria:

1. Rotación de conexiones — una parte de los clientes se desconecta y reconecta cada pocos minutos, unos limpiamente y otros cortando el socket para que se dispare el Last Will.
2. Variedad de sesiones — mezcla sesiones limpias con persistentes e incluye algunos clientes persistentes que pasan horas offline.
3. Variedad de topics — publica en miles de topics distintos, no en uno; incluye mensajes retenidos y suscriptores con wildcards.
4. Tamaños de payload reales — genéralos con un generador de datos de prueba MQTT en lugar de cadenas fijas.
5. Fallos de red — provoca algunos cortes breves para confirmar la recuperación; pruebas de reconexión MQTT explica cómo.

Antes de lanzar la ejecución de tres días, comprueba el topic y las credenciales con el cliente MQTT online y confirma que los mensajes llegan.

Mira el tráfico del soak test en tu navegador

Un soak test es barato: una máquina pequeña, un script y paciencia. Ejecútalo antes de cada versión importante y después de cada actualización del broker o de la librería, guarda los CSV y compara las líneas de tendencia — la degradación lenta solo es invisible cuando nadie la mide.

Preguntas frecuentes