Locust ile MQTT Yük Testi

Locust, canlı web dashboard'u, dağıtık worker'ları ve sıradan kod olarak yazılan senaryolarıyla bir Python yük testi framework'üdür. Varsayılan olarak HTTP konuşur, ancak User sınıfı herhangi bir protokolü sürebilir — dolayısıyla Locust ile MQTT yük testi, paho-mqtt'yi bir kullanıcı sınıfına sarmalamak ve her connect ile publish işlemini Locust istatistiklerine raporlamaktan ibarettir.

Bu rehber eksiksiz bir locustfile.py sunuyor: yeniden kullanılabilir bir MQTT temel kullanıcısı, telemetri publish eden sensör kullanıcıları, uçtan uca gecikmeyi ölçen bir dashboard kullanıcısı ve bunu tarayıcıdan, CI'da headless olarak ve birden fazla makineye dağıtarak çalıştırma komutları.

MQTT için neden Locust?

Bir "kullanıcının" cihaz gibi davranması gerektiğinde — bağlanmak, bir zamanlamaya göre publish etmek, mesajlara tepki vermek, yeniden bağlanmak — ve bu mantığı Python'da yazmak istediğinizde Locust doğru seçimdir. Hızlıca bir throughput tavanı bulmak için CLI benchmark'ı daha basittir; bkz. MQTT broker yük testi.

Entegrasyon nasıl çalışır

Her Locust kullanıcısı hafif bir greenlet'tir (Locust gevent üzerinde çalışır). MQTT kullanıcımız:

1. Locust kullanıcısı başına bir paho-mqtt istemcisi oluşturur; yani 1.000 kullanıcı 1.000 MQTT bağlantısı demektir.
2. connect() çağrısından CONNACK'e kadar geçen bağlantı süresini ölçer.
3. QoS 1 için publish() çağrısından PUBACK'e kadar geçen publish süresini ölçer (QoS 0 yazıldığı anda raporlanır).
4. Her sonucu environment.events.request.fire(...) ile raporlar; bu, Locust'un HTTP istemcisinin dahili olarak yaptığının aynısıdır — böylece dashboard, CSV dışa aktarımı ve hata sekmesi değişiklik gerektirmeden çalışır.

locustfile

import json, os, random, threading, time, uuid
import paho.mqtt.client as mqtt
from locust import User, between, constant, task
from locust.exception import StopUser

HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", 1883))
PREFIX = "testmqtt/locust"

class MqttUser(User):
    abstract = True
    subscriptions: list[tuple[str, int]] = []

    def on_start(self):
        self.client_id = f"locust-{uuid.uuid4().hex[:10]}"
        self.pending, self.lock = {}, threading.Lock()
        self.connected, self.connect_rc = threading.Event(), None
        c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=self.client_id)
        if os.environ.get("MQTT_USER"):
            c.username_pw_set(os.environ["MQTT_USER"], os.environ.get("MQTT_PASS"))
        c.on_connect, c.on_publish, c.on_message = self._on_connect, self._on_publish, self.on_message
        self.client = c
        start = time.perf_counter()
        try:
            c.connect(HOST, PORT, keepalive=60)
            c.loop_start()
            if not self.connected.wait(10):
                raise TimeoutError("no CONNACK within 10 s")
            if self.connect_rc.is_failure:
                raise ConnectionError(f"CONNACK: {self.connect_rc}")
            self._report("CONNECT", "connect", start, 0)
        except Exception as e:
            self._report("CONNECT", "connect", start, 0, e)
            c.loop_stop()
            raise StopUser()

    def on_stop(self):
        with self.lock:
            leftovers, self.pending = list(self.pending.values()), {}
        for start, name, size in leftovers:
            self._report("PUBLISH", name, start, size, TimeoutError("no PUBACK before stop"))
        self.client.disconnect()
        self.client.loop_stop()

    # --- reporting -------------------------------------------------------
    def _report(self, kind, name, start, size, exc=None):
        self.environment.events.request.fire(
            request_type=kind, name=name, response_time=(time.perf_counter() - start) * 1000,
            response_length=size, exception=exc, context={"client_id": self.client_id})

    # --- paho callbacks (run in the network loop) -----------------------
    def _on_connect(self, client, userdata, flags, reason_code, properties):
        self.connect_rc = reason_code
        for topic, qos in self.subscriptions:          # also restores them after a reconnect
            client.subscribe(topic, qos)
        self.connected.set()

    def _on_publish(self, client, userdata, mid, reason_code, properties):
        with self.lock:
            item = self.pending.pop(mid, None)
        if item:
            start, name, size = item
            exc = Exception(f"PUBACK: {reason_code}") if reason_code.is_failure else None
            self._report("PUBLISH", name, start, size, exc)

    def on_message(self, client, userdata, msg):
        pass

    # --- API for tasks ---------------------------------------------------
    def publish(self, topic, payload, qos=1, name=None):
        name, start = name or topic, time.perf_counter()
        with self.lock:
            info = self.client.publish(topic, payload, qos=qos)
            if info.rc == mqtt.MQTT_ERR_SUCCESS and qos > 0:
                self.pending[info.mid] = (start, name, len(payload))
                return
        exc = None if info.rc == mqtt.MQTT_ERR_SUCCESS else Exception(mqtt.error_string(info.rc))
        self._report("PUBLISH", name, start, len(payload), exc)

class Sensor(MqttUser):
    weight = 20
    wait_time = between(4, 6)                          # one reading every ~5 s

    @task(10)
    def telemetry(self):
        payload = json.dumps({"ts": time.time(), "temp": round(random.gauss(21, 0.5), 2)})
        self.publish(f"{PREFIX}/{self.client_id}/telemetry", payload, qos=1, name="telemetry")

    @task(1)
    def heartbeat(self):
        self.publish(f"{PREFIX}/{self.client_id}/status", "ok", qos=0, name="status")

class Dashboard(MqttUser):
    weight = 1
    wait_time = constant(1)
    subscriptions = [(f"{PREFIX}/+/telemetry", 1)]

    def on_message(self, client, userdata, msg):
        sent = json.loads(msg.payload)["ts"]
        self.environment.events.request.fire(
            request_type="RECEIVE", name="telemetry end-to-end",
            response_time=(time.time() - sent) * 1000, response_length=len(msg.payload),
            exception=None, context={})

    @task
    def idle(self):
        pass                                           # messages arrive via on_message

Önemli birkaç ayrıntı:

• name="telemetry" tüm sensörlerin publish işlemlerini tek bir satırda toplar. Bu olmadan her istemcinin benzersiz topic'i istatistik tablosunda ayrı bir satır olur ve dashboard okunamaz hale gelir.
• publish() ve pending etrafındaki kilit, çok hızlı gelen bir PUBACK'in mesaj ID'si kaydedilmeden önce işlenmesi şeklindeki yarış durumunu önler.
• Ağırlıklar — weight = 20 ve weight = 1 ile 210 kullanıcılı bir test yaklaşık 200 sensör ve 10 dashboard başlatır. Her dashboard tüm sensörlerin telemetrisini aldığı için dashboard'lar fan-out'u da test eder.
• Uçtan uca gecikme iki istemcideki time.time() değerlerini karşılaştırır. Tek makinede bu doğrudur; makineler arasında saatlerin NTP ile senkronize olması gerekir, aksi halde sayılar saat sapmasını da içerir.

Testi çalıştırmak

pip install locust "paho-mqtt>=2"
export MQTT_HOST=your-broker-host    # your own broker instance, not the shared endpoint
export MQTT_USER=username MQTT_PASS=password
ulimit -n 8192                         # one socket per user
locust -f locustfile.py

http://localhost:8089 adresini açın, kullanıcı sayısını ve spawn rate'i (saniyede başlatılan kullanıcı) girin ve başlatın. Statistics sekmesi her ad için bir satır gösterir — connect, telemetry, status, telemetry end-to-end — istek hızı, hatalar ve yüzdeliklerle birlikte. Charts saniyedeki istekleri, yanıt sürelerini ve kullanıcı sayısını canlı çizer; Failures ise The client is not currently connected. veya no CONNACK within 10 s gibi hataları gruplar.

CI veya uzun, gözetimsiz çalıştırmalar için headless modu kullanın ve raporları saklayın:

locust -f locustfile.py --headless \
  --users 2000 --spawn-rate 50 --run-time 15m \
  --csv results/mqtt --html results/mqtt-report.html

--spawn-rate sizin bağlantı hızınızdır. Gerçek filonuzun makul olarak üretebileceği bir değerde tutun; saniyede 50 ile 2.000 kullanıcı 40 saniyede ayağa kalkar. Çok yüksek bir spawn rate, yük testini bir yeniden bağlanma fırtınası testine dönüştürür — faydalıdır ama başka bir sorunun cevabıdır.

Ölçeği büyütmeden önce birkaç kullanıcıyla çalıştırın ve payload'ların ve topic adlarının doğru olduğunu teyit etmek için topic'leri çevrimiçi MQTT istemcisinde (testmqtt/locust/#) izleyin. Bu tür smoke testler paylaşımlı broker.connectmq.com uç noktasında sorunsuz çalışır, ancak bu uç nokta trafiğe hız sınırı uygular; bu yüzden asıl yük testi için MQTT_HOST değerini kendi broker örneğinize ayarlayın.

Ölçeklendirme: worker'lar ve süreçler

Bir Locust süreci bir CPU çekirdeği kullanır. Yük üreteci doyduğunda Locust CPU usage above 90%! This may constrain your throughput and may even give inconsistent response time measurements! mesajını loglar — bu noktadan sonra yanıt süreleri broker'ı değil dizüstü bilgisayarınızı yansıtır. Süreç ekleyin:

# all cores on one machine (Linux/macOS)
locust -f locustfile.py --processes -1

# or a master plus workers on several machines
locust -f locustfile.py --master
locust -f locustfile.py --worker --master-host 10.0.0.5

Worker'lar aynı locustfile'ı paylaşır; master istatistikleri tek bir dashboard'da birleştirir. On binlerce bağlantının ötesine geçiyorsanız worker'ları farklı kaynak IP'lere dağıtın — tek bir host'taki geçici port ve dosya tanımlayıcı limitleri sık karşılaşılan sahte bir tavandır.

MQTT için Locust sonuçlarını okumak

PUBLISH telemetry RPS değerini, RECEIVE değerinin dashboard sayısına bölümüyle karşılaştırın: alım publish'in gerisinde kalıyorsa mesajlar kuyrukta birikiyor veya düşürülüyordur. QoS paket akışını, dolayısıyla sayıları değiştirir — MQTT QoS seviyeleri QoS 1 gecikmesinin neden tam bir gidiş-dönüş içerdiğini açıklıyor. Yüzdelikleri yapılandırılmış bir şekilde kapasite kararına dönüştürmek için MQTT benchmark sonuçlarını yorumlama yazısına bakın.

Eklemeye değer varyasyonlar

• TLS — TLS el sıkışmalarını bağlantı süresine dahil etmek için connect() öncesinde c.tls_set() çağırın ve MQTT_PORT=8883 ayarlayın.
• MQTT 5 — MQTT 5 broker'larını test etmek için mqtt.Client'a protocol=mqtt.MQTTv5 verin; PUBACK reason code'ları bu durumda dashboard'da hata olarak görünür.
• Yük şekilleri — bir LoadTestShape sınıfı yükü basamaklar halinde artırabilir veya otomatik olarak bir sabah zirvesini simüle edebilir.
• Komutlar — aşağı yönlü yolu da yük altında test etmek için testmqtt/locust/<id>/cmd topic'ine publish eden bir backend kullanıcısı ekleyin ve sensörlerin onay vermesini sağlayın.

Locust'un MQTT'deki gücü, bir kullanıcının sadece Python olmasıdır: gerçekçi cihaz davranışı, tüm ekibin okuyabileceği bir dashboard ve dizüstü bilgisayarda, CI'da veya bir worker filosunda çalışan aynı dosya. Küçük başlayın, topic'leri doğrulayın, ardından yükü artırın ve sınırların nerede olduğunu yüzdeliklerin söylemesine izin verin.

Sıkça sorulan sorular