MQTT-Lasttests mit k6 und xk6-mqtt

k6 kann einen MQTT-Broker über die Extension xk6-mqtt unter Last testen: Du schreibst den Test in JavaScript mit einem Client im Stil von MQTT.js, k6 führt ihn mit Hunderten oder Tausenden virtueller User aus, und Thresholds machen aus dem Ergebnis einen Pass/Fail-Check für die CI. Das passt gut, wenn du Lasttests als Code willst, versioniert direkt neben deiner Anwendung.

Diese Anleitung zeigt, wie du ein k6-Binary mit MQTT-Support bekommst, erklärt das Event-Loop-Modell, an dem die meisten ersten Skripte scheitern, liefert ein Lastskript mit Publishern, Subscribern und End-to-End-Latenzmetrik und setzt Thresholds, die den Lauf scheitern lassen, wenn der Broker langsam wird.

k6 mit MQTT-Support besorgen

MQTT ist nicht Teil des k6-Kern-Binarys. Es kommt aus der von Grafana gepflegten Extension xk6-mqtt, importiert als k6/x/mqtt. Es gibt zwei Wege:

Automatische Extension-Auflösung (ab k6 v1.2.0). Importiere das Modul im Skript und starte es wie gewohnt; k6 erkennt den Import und stellt die Extension bereit:

k6 version          # benötigt v1.2.0 oder neuer
k6 run mqtt-smoke.js

Eigenes Binary mit xk6. Praktisch in abgeschotteten CI-Umgebungen oder wenn du die exakte Extension-Version festnageln willst:

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

Zuerst ein Smoke-Test

Starte mit einem einzigen virtuellen User, der sich verbindet, subscribt, eine Nachricht publisht und sich trennt, sobald sie zurückkommt. Scheitert das, scheitert ein Lasttest nur schneller.

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, "hallo von k6", { qos: 1 });
  });

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

Lass den Online-MQTT-Client währenddessen auf testmqtt/k6/# subscribed: Du siehst die Nachricht ankommen und hast Topic und Zugangsdaten unabhängig von k6 bestätigt. Für diesen Smoke-Test reicht der geteilte Endpoint broker.connectmq.com; das Lastskript unten fährst du gegen deine eigene private Broker-Instanz, indem du MQTT_URL und die Zugangsdaten setzt.

Das Event-Loop-Modell

Das ist der Teil, der sich von HTTP-Tests unterscheidet. Mit xk6-mqtt endet eine Iteration nicht, wenn die Default-Funktion zurückkehrt. Der VU führt seine Event-Loop weiter aus — er verarbeitet connect-, message- und Timer-Callbacks —, bis die Verbindung geschlossen wird, entweder weil du client.end() aufrufst oder weil der Broker sie trennt.

Folgen für deine Skripte:

• Eine Iteration = eine MQTT-Session. Lege fest, wie lange eine Session dauert, und beende sie explizit mit setTimeout(() => client.end(), …).
• Nutze setInterval im connect-Handler, um mit fester Rate zu publishen. sleep() würde die Event-Loop blockieren.
• Behandle immer error und rufe dort client.end() auf, sonst hängt ein fehlerhafter VU bis zum Graceful Stop des Szenarios.
• Gib jeder Verbindung eine eindeutige client_id (oder lass sie leer, damit der Broker eine vergibt). Duplikate verursachen Session-Takeovers, die wie zufällige Verbindungsabbrüche aussehen.

Ein Lastskript mit End-to-End-Latenz

Die eingebauten Metriken zählen Verbindungen, Publishes und empfangene Nachrichten und messen die Verbindungsdauer, verraten aber nicht, wie lange eine Nachricht vom Publisher zum Subscriber gebraucht hat. Das ergänzt du selbst: Schreibe Date.now() in den Payload und erfasse die Differenz in einer eigenen Trend. Publisher und Subscriber laufen im selben VU und teilen sich daher eine Uhr.

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;   // eine Iteration = eine Session von 30 s
const INTERVAL_MS = 100;    // 10 Nachrichten pro Sekunde und 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 });
}

Bei 100 VUs × 10 Nachrichten pro Sekunde liegt das Ziel bei 1.000 publizierten und 1.000 zugestellten Nachrichten pro Sekunde. Der Payload hat etwa 250 Byte; passe pad an deine Geräte an und prüfe die tatsächliche Größe auf der Leitung mit dem Packet-Size-Rechner.

gracefulRampDown liegt etwas über der Session-Dauer, damit VUs ihre laufende Session im Ramp-down beenden können, statt mittendrin abgeschnitten zu werden — das würde als fehlende Nachrichten auftauchen.

Thresholds, die den Lauf scheitern lassen

Thresholds sorgen dafür, dass k6 mit einem Exit-Code ungleich null endet, wenn eine Bedingung verletzt wird — genau das willst du in einer Pipeline:

Die Werte sind Beispiele. Leite sie aus deiner eigenen Baseline ab: Führe den Test ohne Last auf dem Broker aus, notiere p95/p99 und plane etwas Spielraum ein. Vergleiche außerdem die eingebauten Zähler mqtt_messages_sent und mqtt_messages_received in der Zusammenfassung am Ende — bei QoS 1 und einem Subscriber pro Topic sollten sie nah beieinander liegen; eine wachsende Lücke bedeutet gepufferte oder verlorene Nachrichten. Für eine tiefere Analyse siehe Benchmark-Ergebnisse interpretieren.

Den passenden Executor wählen

Da eine Iteration eine MQTT-Session ist, bestimmt der Executor, wie viele Verbindungen offen sind und wie oft neue hinzukommen:

Für einen Reconnect-Sturm-Test kürzt du SESSION_MS auf wenige Sekunden und nutzt constant-arrival-rate: Jede Iteration durchläuft dann einen vollständigen CONNECT-, SUBSCRIBE- und DISCONNECT-Zyklus, und mqtt_connect_duration zeigt, wie sich die Verbindungszeit mit steigender Ankunftsrate verschlechtert. Worauf du achten solltest, steht unter Reconnect-Stürme.

TLS und WebSockets

Ändere das URL-Schema, um mit demselben Skript andere Transporte zu testen:

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 nutzt die Standard-TLS-Konfiguration von k6 (zum Beispiel tlsAuth für Client-Zertifikate). Dasselbe Szenario über TCP und TLS zu vergleichen, ist der schnellste Weg zu sehen, was dich der Handshake kostet — siehe MQTT-TLS-Performancetests.

k6 in der CI ausführen

1. Nutze mqtt-smoke.js als schnellen Check bei jedem Pull Request und führe mqtt-load.js nächtlich oder vor Releases aus.
2. Übergib Zugangsdaten als Secrets mit -e MQTT_USER=… -e MQTT_PASS=… oder über Umgebungsvariablen; nie hartkodieren.
3. Exportiere Ergebnisse mit --summary-export=summary.json oder einem Output wie --out json=results.json, um eine Historie zu behalten.
4. Lass den Lastgenerator in derselben Region wie den Broker-Endpoint deiner Geräte laufen und beobachte seine CPU; ein ausgelasteter Generator bläht die Latenz auf.

Einen kompletten Workflow findest du unter MQTT-Tests in GitHub Actions.

k6-Traffic im Online-Client ansehen

k6 mit xk6-mqtt macht aus einem MQTT-Lasttest ein kurzes, reviewbares Skript mit klaren Pass/Fail-Kriterien. Versteh das Event-Loop-Modell, miss die End-to-End-Latenz selbst und lass dich von Thresholds warnen, wenn eine Broker-Änderung alles langsamer gemacht hat.

Häufig gestellte Fragen