Teste de carga MQTT com k6 e xk6-mqtt

O k6 consegue fazer teste de carga em um broker MQTT por meio da extensão xk6-mqtt: você escreve o teste em JavaScript com um Client no estilo MQTT.js, o k6 o executa com centenas ou milhares de usuários virtuais e os thresholds transformam o resultado em um check de aprovado/reprovado que você roda no CI. É uma ótima escolha quando você quer testes de carga como código, versionados junto com a aplicação.

Este guia mostra como obter um binário do k6 com suporte a MQTT, o modelo de event loop que derruba a maioria dos primeiros scripts, um script de carga com publishers, subscribers e uma métrica de latência ponta a ponta, e thresholds que reprovam a execução quando o broker fica lento.

Obter o k6 com suporte a MQTT

O MQTT não faz parte do binário principal do k6. Ele vem da extensão xk6-mqtt, mantida pela Grafana e importada como k6/x/mqtt. Há duas formas de usá-la:

Resolução automática de extensões (k6 v1.2.0 em diante). Importe o módulo no script e rode normalmente; o k6 detecta o import e provisiona a extensão:

k6 version          # precisa da v1.2.0 ou mais recente
k6 run mqtt-smoke.js

Binário próprio com xk6. Útil em CI sem acesso à internet ou quando você quer fixar a versão exata da extensão:

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

Primeiro, um smoke test

Comece com um único usuário virtual que conecta, faz subscribe, publica uma mensagem e desconecta quando ela volta. Se isso falhar, um teste de carga só vai falhar mais 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, "olá do k6", { qos: 1 });
  });

  client.on("message", (t, payload) => {
    console.log(`recebido em ${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

Deixe o cliente MQTT online inscrito em testmqtt/k6/# enquanto roda: você verá a mensagem chegar, o que confirma topic e credenciais independentemente do k6. O endpoint compartilhado broker.connectmq.com serve para este smoke test; rode o script de carga abaixo contra a sua própria instância privada de broker definindo MQTT_URL e as credenciais.

O modelo de event loop

Esta é a parte que difere dos testes HTTP. Com o xk6-mqtt, uma iteração não termina quando a função default retorna. O VU continua executando o event loop — tratando os callbacks de connect, message e dos timers — até a conexão fechar, seja porque você chamou client.end(), seja porque o broker a derrubou.

Consequências para os seus scripts:

• Uma iteração = uma sessão MQTT. Decida quanto tempo dura uma sessão e encerre-a explicitamente com setTimeout(() => client.end(), …).
• Use setInterval dentro do handler connect para publicar em ritmo fixo. sleep() bloquearia o event loop.
• Sempre trate error e chame client.end() ali; senão, um VU com falha pode ficar travado até o graceful stop do cenário.
• Dê a cada conexão um client_id único (ou deixe vazio para o broker atribuir um). Duplicatas causam session takeovers que parecem desconexões aleatórias.

Um script de carga com latência ponta a ponta

As métricas embutidas contam conexões, publicações e mensagens recebidas e medem o tempo de conexão, mas não dizem quanto tempo uma mensagem levou do publisher ao subscriber. Adicione isso você mesmo: coloque Date.now() no payload e registre a diferença em uma Trend personalizada. Publisher e subscriber rodam no mesmo VU, então compartilham o mesmo relógio.

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;   // uma iteração = uma sessão de 30 s
const INTERVAL_MS = 100;    // 10 mensagens 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 });
}

Com 100 VUs × 10 mensagens por segundo, a meta é 1.000 mensagens publicadas e 1.000 entregues por segundo. O payload tem cerca de 250 bytes; ajuste pad para ficar igual aos seus dispositivos e confira o tamanho real na rede com a calculadora de tamanho de pacote.

gracefulRampDown fica um pouco acima da duração da sessão para que os VUs terminem a sessão atual durante o ramp-down em vez de serem cortados no meio, o que apareceria como mensagens perdidas.

Thresholds que reprovam a execução

Os thresholds fazem o k6 sair com código diferente de zero quando uma condição é violada — exatamente o que você quer em um pipeline:

Os valores acima são exemplos. Defina-os a partir da sua própria linha de base: rode o teste com o broker ocioso, anote p95/p99 e deixe uma margem. Compare também os contadores embutidos mqtt_messages_sent e mqtt_messages_received no resumo final — com QoS 1 e um subscriber por topic, eles devem ficar próximos; uma diferença crescente significa mensagens enfileiradas ou perdidas. Para uma análise mais profunda, veja interpretando resultados de benchmark.

Escolhendo um executor

Como uma iteração é uma sessão MQTT, o executor decide quantas conexões ficam abertas e com que frequência chegam novas:

Para um teste de tempestade de reconexões, reduza SESSION_MS para poucos segundos e use constant-arrival-rate: cada iteração passa a fazer um ciclo completo de CONNECT, SUBSCRIBE e DISCONNECT, e mqtt_connect_duration mostra como o tempo de conexão piora conforme a taxa de chegada sobe. Veja tempestades de reconexão para saber o que observar.

TLS e WebSockets

Troque o esquema da URL para testar outros transportes com o mesmo 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

O TLS usa a configuração TLS padrão do k6 (por exemplo tlsAuth para certificados de cliente). Comparar o mesmo cenário em TCP e TLS é o jeito mais rápido de ver quanto o handshake custa — veja teste de desempenho TLS no MQTT.

Rodando o k6 no CI

1. Mantenha o mqtt-smoke.js como verificação rápida em todo pull request e rode o mqtt-load.js toda noite ou antes de releases.
2. Passe as credenciais como secrets via -e MQTT_USER=… -e MQTT_PASS=… ou variáveis de ambiente; nunca as deixe fixas no código.
3. Exporte resultados com --summary-export=summary.json ou uma saída como --out json=results.json para manter um histórico.
4. Rode o gerador de carga na mesma região do endpoint do broker dos seus dispositivos e acompanhe a CPU; um gerador saturado infla a latência.

Veja testes MQTT no GitHub Actions para um workflow completo.

Veja o tráfego do k6 no cliente online

O k6 com xk6-mqtt transforma um teste de carga MQTT em um script curto e revisável, com critérios claros de aprovação. Acerte o modelo de event loop, meça você mesmo a latência ponta a ponta e deixe os thresholds avisarem quando uma mudança no broker deixou tudo mais lento.

Perguntas frequentes