Pruebas de carga MQTT con k6 y xk6-mqtt

k6 puede hacer pruebas de carga a un broker MQTT mediante la extensión xk6-mqtt: escribes la prueba en JavaScript con un Client al estilo MQTT.js, k6 la ejecuta con cientos o miles de usuarios virtuales y los thresholds convierten el resultado en un check de éxito/fallo que puedes ejecutar en CI. Encaja muy bien cuando quieres pruebas de carga como código, versionadas junto a tu aplicación.

Esta guía explica cómo conseguir un binario de k6 con soporte MQTT, el modelo de event loop que hace tropezar a la mayoría de los primeros scripts, un script de carga con publishers y subscribers y una métrica de latencia extremo a extremo, y thresholds que hacen fallar la ejecución cuando el broker se vuelve lento.

Conseguir k6 con soporte MQTT

MQTT no forma parte del binario base de k6. Viene de la extensión xk6-mqtt mantenida por Grafana, que se importa como k6/x/mqtt. Hay dos formas de usarla:

Resolución automática de extensiones (k6 v1.2.0 y posteriores). Importa el módulo en tu script y ejecútalo como siempre; k6 detecta el import y aprovisiona la extensión:

k6 version          # necesita v1.2.0 o superior
k6 run mqtt-smoke.js

Binario propio con xk6. Útil en CI sin acceso a internet o cuando quieres fijar la versión exacta de la extensión:

go install go.k6.io/xk6/cmd/xk6@latest
xk6 build --with github.com/grafana/xk6-mqtt
./k6 run mqtt-smoke.js

Primero, un smoke test

Empieza con un único usuario virtual que se conecta, se suscribe, publica un mensaje y se desconecta cuando lo recibe. Si esto falla, una prueba de carga solo fallará más rápido.

import { Client } from "k6/x/mqtt";

const BROKER = __ENV.MQTT_URL || "mqtt://broker.connectmq.com:1883";

export default function () {
  const client = new Client({
    username: __ENV.MQTT_USER,
    password: __ENV.MQTT_PASS,
  });
  const topic = "testmqtt/k6/smoke";

  client.on("connect", () => {
    client.subscribe(topic, { qos: 1 });
    client.publish(topic, "hola desde k6", { qos: 1 });
  });

  client.on("message", (t, payload) => {
    console.log(`recibido en ${t}: ${new TextDecoder().decode(payload)}`);
    client.end();
  });

  client.on("error", (err) => {
    console.error(`${err.method}: ${err.message}`);
    client.end();
  });

  client.connect(BROKER);
}

MQTT_USER=username MQTT_PASS=password k6 run mqtt-smoke.js

Mantén el cliente MQTT online suscrito a testmqtt/k6/# mientras lo ejecutas: verás llegar el mensaje, lo que confirma el topic y las credenciales de forma independiente de k6. El endpoint compartido broker.connectmq.com sirve para este smoke test; ejecuta el script de carga de abajo contra tu propia instancia privada de broker definiendo MQTT_URL y las credenciales.

El modelo de event loop

Esta es la parte que difiere de las pruebas HTTP. Con xk6-mqtt, una iteración no termina cuando la función default retorna. El VU sigue ejecutando su event loop — gestionando los callbacks de connect, message y los timers — hasta que la conexión se cierra, ya sea porque llamas a client.end() o porque el broker la corta.

Consecuencias para tus scripts:

• Una iteración = una sesión MQTT. Decide cuánto dura una sesión y termínala explícitamente con setTimeout(() => client.end(), …).
• Usa setInterval dentro del handler connect para publicar a un ritmo fijo. sleep() bloquearía el event loop.
• Gestiona siempre error y llama ahí a client.end(), o un VU que falla puede quedarse colgado hasta el graceful stop del escenario.
• Da a cada conexión un client_id único (o déjalo vacío para que el broker asigne uno). Los duplicados provocan session takeovers que parecen desconexiones aleatorias.

Un script de carga con latencia extremo a extremo

Las métricas integradas cuentan conexiones, publicaciones y mensajes recibidos, y miden el tiempo de conexión, pero no dicen cuánto tardó un mensaje del publisher al subscriber. Añádelo tú: mete Date.now() en el payload y registra la diferencia en una Trend personalizada. Publisher y subscriber se ejecutan en el mismo VU, así que comparten el mismo reloj.

import { Client } from "k6/x/mqtt";
import { Trend, Counter } from "k6/metrics";
import exec from "k6/execution";

const BROKER = __ENV.MQTT_URL || "mqtt://broker.connectmq.com:1883";
const SESSION_MS = 30000;   // una iteración = una sesión de 30 s
const INTERVAL_MS = 100;    // 10 mensajes por segundo por VU

const e2eLatency = new Trend("mqtt_e2e_latency", true);
const mqttErrors = new Counter("mqtt_errors");

export const options = {
  scenarios: {
    telemetry: {
      executor: "ramping-vus",
      startVUs: 0,
      stages: [
        { duration: "30s", target: 100 },
        { duration: "3m", target: 100 },
        { duration: "15s", target: 0 },
      ],
      gracefulRampDown: "35s",
    },
  },
  thresholds: {
    mqtt_e2e_latency: ["p(95)<100", "p(99)<250"],
    mqtt_connect_duration: ["p(95)<1000"],
    mqtt_errors: ["count<1"],
  },
};

export default function () {
  const vu = exec.vu.idInTest;
  const topic = `testmqtt/k6/${vu}/data`;
  const client = new Client({
    client_id: `k6-${vu}-${exec.vu.iterationInScenario}`,
    username: __ENV.MQTT_USER,
    password: __ENV.MQTT_PASS,
  });

  client.on("connect", () => {
    client.subscribe(topic, { qos: 1 });
    let seq = 0;
    const timer = setInterval(() => {
      const body = JSON.stringify({ seq: seq++, t: Date.now(), pad: "x".repeat(200) });
      client.publish(topic, body, { qos: 1 });
    }, INTERVAL_MS);
    setTimeout(() => {
      clearInterval(timer);
      client.end();
    }, SESSION_MS);
  });

  client.on("message", (t, payload) => {
    const msg = JSON.parse(new TextDecoder().decode(payload));
    e2eLatency.add(Date.now() - msg.t);
  });

  client.on("error", (err) => {
    mqttErrors.add(1, { method: err.method });
    client.end();
  });

  client.connect(BROKER, { keepalive: 30 });
}

Con 100 VUs × 10 mensajes por segundo, el objetivo es 1.000 mensajes publicados y 1.000 entregados por segundo. El payload ocupa unos 250 bytes; ajusta pad para que coincida con tus dispositivos y comprueba el tamaño real en la red con la calculadora de tamaño de paquetes.

gracefulRampDown está un poco por encima de la duración de la sesión para que los VUs terminen su sesión actual durante el ramp-down en lugar de cortarse a mitad, lo que aparecería como mensajes perdidos.

Thresholds que hacen fallar la ejecución

Los thresholds hacen que k6 termine con un código distinto de cero cuando se incumple una condición, justo lo que quieres en un pipeline:

Los valores anteriores son ejemplos. Fíjalos a partir de tu propia línea base: ejecuta la prueba con el broker en reposo, anota el p95/p99 y deja algo de margen. Compara también los contadores integrados mqtt_messages_sent y mqtt_messages_received en el resumen final: con QoS 1 y un subscriber por topic deberían estar cerca; una diferencia creciente significa que hay mensajes encolados o perdidos. Para un análisis más profundo, consulta interpretar resultados de benchmarks.

Elegir un executor

Como una iteración es una sesión MQTT, el executor decide cuántas conexiones hay abiertas y con qué frecuencia llegan nuevas:

Para una prueba de tormenta de reconexiones, reduce SESSION_MS a unos pocos segundos y usa constant-arrival-rate: cada iteración realiza entonces un ciclo completo de CONNECT, SUBSCRIBE y DISCONNECT, y mqtt_connect_duration muestra cómo se degrada el tiempo de conexión a medida que sube la tasa de llegada. Consulta tormentas de reconexión para saber qué vigilar.

TLS y WebSockets

Cambia el esquema de la URL para probar otros transportes con el mismo script:

k6 run -e MQTT_URL=mqtts://broker.connectmq.com:8883 mqtt-load.js
k6 run -e MQTT_URL=wss://broker.connectmq.com:8084/mqtt mqtt-load.js

TLS usa la configuración TLS estándar de k6 (por ejemplo tlsAuth para certificados de cliente). Comparar el mismo escenario sobre TCP y TLS es la forma más rápida de ver cuánto te cuesta el handshake; consulta pruebas de rendimiento TLS en MQTT.

Ejecutar k6 en CI

1. Mantén mqtt-smoke.js como control rápido en cada pull request y ejecuta mqtt-load.js cada noche o antes de cada release.
2. Pasa las credenciales como secretos con -e MQTT_USER=… -e MQTT_PASS=… o variables de entorno; nunca las escribas en el código.
3. Exporta resultados con --summary-export=summary.json o una salida como --out json=results.json para guardar un histórico.
4. Ejecuta el generador de carga en la misma región que el endpoint del broker de tus dispositivos y vigila su CPU; un generador saturado infla la latencia.

Consulta pruebas MQTT en GitHub Actions para un workflow completo.

Observa el tráfico de k6 en el cliente online

k6 con xk6-mqtt convierte una prueba de carga MQTT en un script corto y revisable con criterios claros de éxito y fallo. Entiende bien el modelo de event loop, mide tú mismo la latencia extremo a extremo y deja que los thresholds te avisen cuando un cambio en el broker empeore las cosas.

Preguntas frecuentes