k6 ve xk6-mqtt ile MQTT Yük Testi

k6, xk6-mqtt eklentisi sayesinde bir MQTT broker'ına yük testi yapabilir: testi MQTT.js tarzı bir Client ile JavaScript'te yazarsınız, k6 onu yüzlerce ya da binlerce sanal kullanıcıyla çalıştırır ve threshold'lar sonucu CI'da çalıştırabileceğiniz bir geçti/kaldı kontrolüne dönüştürür. Yük testlerini kod olarak, uygulamanızla aynı repository'de sürümlemek istediğinizde çok uygun bir seçimdir.

Bu rehberde MQTT destekli bir k6 binary'si edinmeyi, ilk script'lerin çoğunu tökezleten event loop modelini, publisher, subscriber ve uçtan uca gecikme metriği içeren bir yük script'ini ve broker yavaşladığında çalıştırmayı başarısız kılan threshold'ları ele alıyoruz.

MQTT destekli k6 edinin

MQTT, k6'nın çekirdek binary'sinin bir parçası değildir. Grafana tarafından sürdürülen ve k6/x/mqtt olarak import edilen xk6-mqtt eklentisinden gelir. Kullanmanın iki yolu vardır:

Otomatik eklenti çözümleme (k6 v1.2.0 ve sonrası). Modülü script'inizde import edin ve her zamanki gibi çalıştırın; k6 import'u algılar ve eklentiyi hazırlar:

k6 version          # v1.2.0 veya daha yenisi gerekir
k6 run mqtt-smoke.js

xk6 ile özel binary. İnternet erişimi olmayan CI ortamlarında ya da eklentinin tam sürümünü sabitlemek istediğinizde işe yarar:

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

Önce bir smoke test

Bağlanan, subscribe olan, bir mesaj publish eden ve mesaj geri geldiğinde bağlantıyı kesen tek bir sanal kullanıcıyla başlayın. Bu başarısız olursa yük testi yalnızca daha hızlı başarısız olur.

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, "k6'dan merhaba", { qos: 1 });
  });

  client.on("message", (t, payload) => {
    console.log(`${t} üzerinden alındı: ${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

Çalıştırırken online MQTT client üzerinde testmqtt/k6/# topic'ine abone kalın: mesajın geldiğini görürsünüz; bu, topic'i ve kimlik bilgilerini k6'dan bağımsız olarak doğrular. Paylaşımlı broker.connectmq.com endpoint'i bu smoke test için yeterlidir; aşağıdaki yük script'ini MQTT_URL değerini ve kimlik bilgilerini ayarlayarak kendi özel broker instance'ınıza karşı çalıştırın.

Event loop modeli

HTTP testlerinden farklı olan kısım budur. xk6-mqtt ile bir iterasyon, default fonksiyon döndüğünde bitmez. VU, bağlantı kapanana kadar — client.end() çağırdığınız için ya da broker bağlantıyı kopardığı için — event loop'unu çalıştırmaya, yani connect, message ve timer callback'lerini işlemeye devam eder.

Script'leriniz için sonuçları:

• Bir iterasyon = bir MQTT oturumu. Bir oturumun ne kadar süreceğine karar verin ve onu setTimeout(() => client.end(), …) ile açıkça sonlandırın.
• Sabit hızda publish etmek için connect handler'ı içinde setInterval kullanın. sleep() event loop'u bloklar.
• error olayını her zaman ele alın ve orada client.end() çağırın; aksi halde hata alan bir VU, senaryonun graceful stop süresine kadar takılı kalabilir.
• Her bağlantıya benzersiz bir client_id verin (ya da broker'ın atanması için boş bırakın). Tekrarlanan ID'ler rastgele kopmalar gibi görünen session takeover durumlarına yol açar.

Uçtan uca gecikme ölçen bir yük script'i

Yerleşik metrikler bağlantıları, publish'leri ve alınan mesajları sayar ve bağlantı süresini ölçer, ama bir mesajın publisher'dan subscriber'a ne kadar sürede ulaştığını söylemez. Bunu kendiniz ekleyin: payload'a Date.now() koyun ve farkı özel bir Trend içinde kaydedin. Publisher ve subscriber aynı VU'da çalıştığı için aynı saati paylaşırlar.

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;   // bir iterasyon = 30 sn'lik bir oturum
const INTERVAL_MS = 100;    // VU başına saniyede 10 mesaj

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 });
}

100 VU × saniyede 10 mesaj ile hedef saniyede 1.000 publish edilen ve 1.000 teslim edilen mesajdır. Payload yaklaşık 250 bayttır; pad değerini cihazlarınıza göre ayarlayın ve hattaki gerçek boyutu paket boyutu hesaplayıcı ile kontrol edin.

gracefulRampDown, oturum süresinin biraz üzerinde tutulur; böylece VU'lar ramp-down sırasında akışın ortasında kesilmek yerine mevcut oturumlarını tamamlar. Aksi halde bu, kayıp mesajlar olarak görünürdü.

Çalıştırmayı başarısız kılan threshold'lar

Threshold'lar bir koşul ihlal edildiğinde k6'nın sıfırdan farklı bir çıkış koduyla sonlanmasını sağlar; bir pipeline'da tam da istediğiniz budur:

Yukarıdaki değerler örnektir. Bunları kendi baseline'ınızdan belirleyin: testi broker boştayken çalıştırın, p95/p99 değerlerini not edin ve biraz pay bırakın. Test sonu özetindeki yerleşik mqtt_messages_sent ve mqtt_messages_received sayaçlarını da karşılaştırın — QoS 1 ve topic başına bir subscriber ile birbirine yakın olmalıdırlar; büyüyen bir fark, mesajların kuyrukta beklediği ya da kaybolduğu anlamına gelir. Daha derin analiz için benchmark sonuçlarını yorumlama yazısına bakın.

Executor seçimi

Bir iterasyon bir MQTT oturumu olduğu için executor, kaç bağlantının açık olacağını ve yeni bağlantıların ne sıklıkla geleceğini belirler:

Bir yeniden bağlanma fırtınası testi için SESSION_MS değerini birkaç saniyeye düşürün ve constant-arrival-rate kullanın: her iterasyon tam bir CONNECT, SUBSCRIBE ve DISCONNECT döngüsü yapar ve mqtt_connect_duration, geliş hızı arttıkça bağlantı süresinin nasıl kötüleştiğini gösterir. Nelere dikkat etmeniz gerektiği için yeniden bağlanma fırtınaları yazısına bakın.

TLS ve WebSocket'ler

Aynı script ile diğer transport'ları test etmek için URL şemasını değiştirin:

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, k6'nın standart TLS yapılandırmasını kullanır (örneğin client sertifikaları için tlsAuth). Aynı senaryoyu TCP ve TLS üzerinden karşılaştırmak, handshake'in size maliyetini görmenin en hızlı yoludur — bkz. MQTT TLS performans testi.

k6'yı CI'da çalıştırma

1. mqtt-smoke.js dosyasını her pull request'te hızlı bir kontrol olarak tutun, mqtt-load.js dosyasını ise her gece veya release öncesinde çalıştırın.
2. Kimlik bilgilerini -e MQTT_USER=… -e MQTT_PASS=… ya da ortam değişkenleri üzerinden secret olarak geçirin; asla koda gömmeyin.
3. Bir geçmiş tutmak için sonuçları --summary-export=summary.json veya --out json=results.json gibi bir output ile dışa aktarın.
4. Yük üreticiyi cihazlarınızın broker endpoint'iyle aynı bölgede çalıştırın ve CPU'sunu izleyin; doymuş bir üretici gecikmeyi şişirir.

Eksiksiz bir workflow için GitHub Actions'ta MQTT testleri yazısına bakın.

k6 trafiğini online client'ta izleyin

xk6-mqtt ile k6, bir MQTT yük testini net geçti/kaldı kriterlerine sahip kısa ve gözden geçirilebilir bir script'e dönüştürür. Event loop modelini doğru kurun, uçtan uca gecikmeyi kendiniz ölçün ve bir broker değişikliği işleri yavaşlattığında bunu size threshold'ların söylemesine izin verin.

Sıkça sorulan sorular