MQTT Soak Testi: Bellek Sızıntılarını ve Yavaş Bozulmayı Bulun

Soak testi (dayanıklılık testi), bir MQTT sistemini uzun süre — genellikle 24 ila 72 saat — normal yük altında çalıştırarak on dakikalık bir benchmark'ın asla göremeyeceği sorunları yakalar: saatte birkaç megabayt büyüyen bellek, hiç serbest bırakılmayan dosya tanımlayıcıları, bir daha geri gelmeyen oturumlar için biriken kuyruklar ve cihazlar timeout almaya başlayana kadar yavaşça yükselen gecikme.

Bu rehber neleri ölçmeniz gerektiğini gösterir, her şeyi CSV'ye kaydeden bir Python betiği sunar ve "iyi görünüyor" ifadesini bir sayıya dönüştürmek için sonucu nasıl okuyacağınızı açıklar.

Soak testi ve yük testi

Kapasiteyi henüz ölçmediyseniz MQTT broker yük testi ile başlayın. Soak testi, sistemin yükü şu anda kaldırdığını varsayar ve üçüncü gün de kaldırıp kaldırmayacağını sorar.

Yavaşça bozulan şeyler

Uzun süreli arızaların çoğu birkaç kalıba uyar:

• İstemci bellek sızıntıları — bir listeye sürekli ekleme yapan callback'ler, istemci acknowledgement'lar gelmeden daha hızlı publish ettiği için büyüyen in-flight kuyruğu ya da eskisi durdurulmadan her yeniden bağlanmada oluşturulan yeni bir istemci nesnesi. Bkz. MQTT istemci bellek sızıntıları.
• Sınırsız broker kuyrukları — bir daha bağlanmayan cihazların kalıcı oturumları (clean_session=False / session expiry), broker'ın belleği bitene kadar QoS 1 ve 2 mesajlarını biriktirmeye devam eder. Klasik sonuç MQTT broker çökmeleri ve OOM yazısında anlatılıyor.
• Retained mesaj artışı — yolunda benzersiz ID'ler bulunan (.../session-8f3a/state) ve retain=true ile publish edilen topic'ler sonsuza kadar birikir.
• Dosya tanımlayıcısı ve soket sızıntıları — eski soketi kapatmayan yeniden bağlanma yolları. Süreç sonunda Too many open files hatasıyla çöker.
• Zamana bağlı olaylar — log rotasyonu, sertifika yenileme, token süresinin dolması, günlük yedeklemeler veya kalıcılık dosyalarıyla dolan bir disk. Bunlar yalnızca test bu anlara denk gelecek kadar uzun sürerse ortaya çıkar.

Kaydedilecek metrikler

İstemci ve broker tarafındaki değerleri aynı zaman çizgisinde, dakikada bir kaydedin:

Bir Python soak testi betiği

Bu betik (paho-mqtt 2.x ve psutil) sabit bir hızda sıra numaralı mesaj publish eder, kendi topic'ine subscribe olur ve her dakika throughput, boşluklar, kopyalar, gecikme yüzdelikleri, kendi belleği ve açık dosya tanımlayıcılarıyla bir CSV satırı yazar. on_connect içinde yeniden subscribe olduğu için yeniden bağlanmalardan sonra sessizce sağırlaşmak yerine çalışmaya devam eder.

import csv, json, os, threading, time, uuid
import paho.mqtt.client as mqtt
import psutil

HOST, PORT = os.environ.get("MQTT_HOST", "broker.connectmq.com"), 1883
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/soak/{RUN}/data"
RATE = 20            # saniyedeki mesaj sayısı
DURATION_H = 72      # toplam çalışma süresi
WINDOW_S = 60        # dakikada bir CSV satırı

lock = threading.Lock()
st = {"recv": 0, "gaps": 0, "dups": 0, "last": -1, "lat": [], "disc": 0}

def on_connect(client, userdata, flags, reason_code, properties):
    if client is sub:
        client.subscribe(TOPIC, qos=1)

def on_disconnect(client, userdata, flags, reason_code, properties):
    with lock:
        st["disc"] += 1

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    with lock:
        st["recv"] += 1
        st["lat"].append((time.monotonic() - data["t"]) * 1000)
        if data["seq"] <= st["last"]:
            st["dups"] += 1
        else:
            st["gaps"] += data["seq"] - st["last"] - 1
            st["last"] = data["seq"]

def make_client(name):
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"soak-{RUN}-{name}")
    c.username_pw_set("username", "password")
    c.on_connect, c.on_disconnect = on_connect, on_disconnect
    c.reconnect_delay_set(min_delay=1, max_delay=30)
    return c

sub = make_client("sub")
sub.on_message = on_message
sub.connect(HOST, PORT, keepalive=30)
sub.loop_start()
pub = make_client("pub")
pub.connect(HOST, PORT, keepalive=30)
pub.loop_start()
time.sleep(2)

proc = psutil.Process()
pct = lambda xs, q: xs[max(0, int(len(xs) * q) - 1)] if xs else 0.0

with open(f"soak-{RUN}.csv", "w", newline="") as f:
    w = csv.writer(f)
    w.writerow(["time", "sent", "recv", "gaps", "dups", "disconnects",
                "p50_ms", "p99_ms", "rss_mb", "fds"])
    end = time.monotonic() + DURATION_H * 3600
    next_row = time.monotonic() + WINDOW_S
    seq = 0
    while time.monotonic() < end:
        pub.publish(TOPIC, json.dumps({"seq": seq, "t": time.monotonic()}), qos=1)
        seq += 1
        time.sleep(1 / RATE)
        if time.monotonic() >= next_row:
            with lock:
                lat, st["lat"] = sorted(st["lat"]), []
                counts = [st["recv"], st["gaps"], st["dups"], st["disc"]]
            w.writerow([time.strftime("%Y-%m-%dT%H:%M:%S"), seq, *counts,
                        round(pct(lat, 0.50), 1), round(pct(lat, 0.99), 1),
                        round(proc.memory_info().rss / 1e6, 1), proc.num_fds()])
            f.flush()
            next_row += WINDOW_S

Betiği üç gün boyunca açık kalacak bir makinede nohup python soak.py & ile (veya tmux içinde) çalıştırın. CSV her dakika diske yazılır, böylece test sürerken dosyayı inceleyebilirsiniz. num_fds() Linux ve macOS'ta vardır; Windows'ta bunun yerine proc.num_handles() kullanın. Paylaşımlı broker.connectmq.com uç noktasına karşı kısa bir deneme yapmak sorun değildir; ancak gerçek çok günlük çalıştırma için MQTT_HOST değerini kendi broker örneğinize ayarlayın: paylaşımlı uç nokta hızlı testler için tasarlanmıştır, hız sınırı uygular ve kalıcı oturum tutmaz; bu yüzden aşağıda anlatılan oturumla ilgili arızaları yalnızca kendi broker'ınızda gözlemleyebilirsiniz.

Broker tarafını da izleyin

İstemci sayıları bir şeyin bozulduğunu söyler; broker sayıları nerede bozulduğunu söyler. Broker'ınız $SYS topic'leri yayınlıyorsa (Mosquitto yayınlar), büyümeyi gösteren sayaçlara subscribe olun:

mosquitto_sub -h broker.connectmq.com -p 1883 -u "$MQTT_USER" -P "$MQTT_PASS" -v \
  -t '$SYS/broker/clients/connected' \
  -t '$SYS/broker/messages/stored' \
  -t '$SYS/broker/retained messages/count' \
  -t '$SYS/broker/heap/current'

Broker, kontrolünüzdeki bir host üzerinde Docker'da çalışıyorsa belleğini ve CPU'sunu dakikada bir örnekleyin:

while true; do
  docker stats --no-stream --format "$(date -u +%FT%TZ),{{.Name}},{{.MemUsage}},{{.CPUPerc}}" mosquitto
  sleep 60
done >> broker-stats.csv

Bir testten daha uzun süre çalışacak her şey için aynı değerleri bir zaman serisi veritabanına aktarın — kurulum Prometheus ve Grafana ile izleme ve $SYS topic izleme yazılarında anlatılıyor.

CSV'yi bir karara dönüştürmek

4.000 satırlık bir CSV'ye gözle bakmak işe yaramaz. Bunun yerine bir eğilim çizgisi uydurun. İlk saati atlayın (önbellekler, JIT, ısınan bağlantı havuzları) ve standart kütüphaneyle belleğin zaman içindeki eğimini hesaplayın:

import csv, statistics, sys

rows = list(csv.DictReader(open(sys.argv[1])))[60:]   # 1 saatlik ısınmayı atla
hours = [i / 60 for i in range(len(rows))]
for col in ("rss_mb", "fds", "p99_ms"):
    ys = [float(r[col]) for r in rows]
    slope, _ = statistics.linear_regression(hours, ys)
    print(f"{col:8} start {ys[0]:8.1f}  end {ys[-1]:8.1f}  trend {slope:+.3f}/hour")

last = rows[-1]
print(f"sent {last['sent']}  recv {last['recv']}  gaps {last['gaps']}  "
      f"dups {last['dups']}  disconnects {last['disconnects']}")

Nasıl yorumlanır:

• Sıfıra yakın RSS eğilimi ve ham veride testere dişi — normal çöp toplama. Sabit yükte sürekli pozitif bir eğim sızıntıdır; önemli olup olmadığını görmek için dağıtımınızın çalışma süresiyle çarpın (saatte 5 MB, ayda 3,6 GB demektir).
• Artan dosya tanımlayıcıları — bir şey soket veya dosya açıp kapatmıyor, genellikle yeniden bağlanma yolunda.
• p50 sabitken p99'un yükselmesi — büyüyen kuyruklar, uzayan çöp toplama duraklamaları veya saklanan mesajları biriktiren bir broker.
• Kopmalarla örtüşen boşluklar — QoS 0'da beklenir, kalıcı oturumlu QoS 1'de bir hatadır. Hiç kopma olmadan oluşan boşluklar, broker'ın mesaj düşürdüğüne (kuyruk sınırları) veya yetişemeyen bir subscriber'a işaret eder.
• Broker'da messages/stored değerinin artması — genellikle artık var olmayan istemcilerin kalıcı oturumları. Bir session expiry (MQTT 5) ayarlayın veya eski client ID'leri temizleyin.

Soak testini gerçekçi hale getirin

İki istemci arasındaki sabit bir akış istemci sızıntılarını bulur. Broker sızıntılarını bulmak için trafiğin bellek ayıran kod yollarından geçmesi gerekir:

1. Bağlantı değişimi — istemcilerin bir kısmı birkaç dakikada bir bağlantıyı kesip yeniden bağlanır; bazıları düzgünce, bazıları da Last Will tetiklensin diye soketi koparak.
2. Oturum çeşitliliği — temiz oturumları kalıcı olanlarla karıştırın ve saatlerce çevrimdışı kalan birkaç kalıcı istemci ekleyin.
3. Topic çeşitliliği — tek bir topic'e değil, binlerce farklı topic'e publish edin; retained mesajlar ve wildcard subscriber'lar ekleyin.
4. Gerçek payload boyutları — sabit string'ler yerine bir MQTT test verisi üreticisi ile oluşturun.
5. Ağ arızaları — toparlanmayı doğrulamak için birkaç kısa kesinti yaratın; MQTT yeniden bağlanma testi nasıl yapılacağını gösteriyor.

Üç günlük çalıştırmayı başlatmadan önce topic'i ve kimlik bilgilerini çevrimiçi MQTT istemcisi ile kontrol edin ve mesajların ulaştığını doğrulayın.

Soak trafiğini tarayıcınızda izleyin

Soak testi ucuzdur: küçük bir makine, bir betik ve sabır. Her büyük sürümden önce ve her broker ya da kütüphane güncellemesinden sonra çalıştırın, CSV'leri saklayın ve eğilim çizgilerini karşılaştırın — yavaş bozulma ancak kimse ölçmediğinde görünmez kalır.

Sıkça sorulan sorular