MQTT için Kaos Testi: Ağ Bölünmeleri, Paket Kaybı ve Broker Yeniden Başlatmaları

MQTT için kaos testi, ağı ve broker'ı bilerek bozmak (gecikme, paket kaybı, ağ bölünmeleri, yeniden başlatmalar) ve bu sırada istemcilerinizin kendi kendine toparlanıp toparlanmadığını ölçmek demektir. Production'a ulaşan MQTT hatalarının çoğu toparlanma hatalarıdır: ölü bir bağlantıyı hiç fark etmeyen bir istemci, geri yüklenmeyen bir abonelik, oturum kalıcı olmadığı için kaybolan bir QoS 1 mesajı. Bu rehber araçları (tc netem, iptables, Toxiproxy, docker) ve her deneyi sayılara döken bir sonda betiğini gösteriyor.

Önce "toparlandı"nın ne demek olduğunu tanımlayın

Başarı kriteri olmayan bir kaos deneyi sadece bir kesintidir. Sisteminizin geri dönmesi gereken kararlı durumu yazın:

İşin çoğunu iki mekanizma yapar: ölü bir bağlantının ne kadar süre fark edilmeden kalabileceğini sınırlayan keep-alive aralığı ve subscriber uzaktayken broker'ın QoS 1 mesajlarını onun için saklamasını sağlayan kalıcı oturumlar.

Sonda: kayıpları, tekrarları ve yeniden bağlanmaları ölçmek

Bu paho-mqtt 2.x betiği, kalıcı bir MQTT 5 oturumuyla bir publisher ve bir subscriber çalıştırır. Her 100 ms'de QoS 1 ile numaralı bir mesaj yayınlar ve beş saniyede bir, onaylanmış mesajlardan kaçının henüz ulaşmadığını (missing), kaçının iki kez geldiğini (dups), yeniden bağlanma sayısını ve en kötü gecikmeyi raporlar.

import json, os, threading, time, uuid
import paho.mqtt.client as mqtt
from paho.mqtt.packettypes import PacketTypes
from paho.mqtt.properties import Properties

HOST = os.getenv("MQTT_HOST", "localhost")
PORT = int(os.getenv("MQTT_PORT", "1883"))
RUN = uuid.uuid4().hex[:8]
TOPIC = f"testmqtt/chaos/{RUN}/seq"
lock = threading.Lock()
seen, sent = set(), []            # sent: (seq, MQTTMessageInfo)
stats = {"dups": 0, "reconnects": 0, "max_ms": 0.0}

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"[{userdata['role']}] connected: {reason_code}, session present={flags.session_present}")
    if userdata["ever"]:
        stats["reconnects"] += 1
    userdata["ever"] = True
    if userdata["role"] == "sub":
        client.subscribe(TOPIC, qos=1)

def on_disconnect(client, userdata, flags, reason_code, properties):
    print(f"[{userdata['role']}] disconnected: {reason_code}")

def on_message(client, userdata, msg):
    data = json.loads(msg.payload)
    with lock:
        if data["seq"] in seen:
            stats["dups"] += 1
        seen.add(data["seq"])
        stats["max_ms"] = max(stats["max_ms"], (time.time() - data["t"]) * 1000)

def make(role, session_expiry):
    props = Properties(PacketTypes.CONNECT)
    props.SessionExpiryInterval = session_expiry
    c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id=f"chaos-{role}-{RUN}",
                    protocol=mqtt.MQTTv5, userdata={"role": role, "ever": False})
    c.on_connect, c.on_disconnect, c.on_message = on_connect, on_disconnect, on_message
    c.reconnect_delay_set(min_delay=1, max_delay=16)
    c.connect_async(HOST, PORT, keepalive=10, properties=props)
    c.loop_start()
    return c

sub = make("sub", session_expiry=600)   # session survives outages up to 10 minutes
pub = make("pub", session_expiry=0)
time.sleep(2)

seq, next_report = 0, time.time() + 5
while True:
    info = pub.publish(TOPIC, json.dumps({"seq": seq, "t": time.time()}), qos=1)
    sent.append((seq, info))
    seq += 1
    if time.time() >= next_report:
        acked = {s for s, i in sent if i.is_published()}
        with lock:
            print(f"{time.strftime('%H:%M:%S')} sent={seq} acked={len(acked)} "
                  f"missing={len(acked - seen)} dups={stats['dups']} "
                  f"reconnects={stats['reconnects']} max_latency={stats['max_ms']:.0f}ms")
            stats["max_ms"] = 0.0
        next_report += 5
    time.sleep(0.1)

Ağ kapalıyken paho QoS 1 mesajlarını bellekteki kuyruğunda tutar ve yeniden bağlandıktan sonra gönderir; bu yüzden acked dururken sent artmaya devam eder. Kesinti sırasında missing kısa süreliğine yükselebilir; toparlanmadan sonra 0'a dönmelidir. Kalıcı olarak sıfırdan farklı bir değer, broker'ın onayladığı mesajların daha sonra kaybolduğu anlamına gelir; genellikle ayakta kalamayan bir oturum yüzünden.

Deney 1: tc netem ile gecikme ve paket kaybı

Linux'ta tc netem bir ağ arayüzünü kötüleştirir. Tüm iş istasyonunuzu yavaşlatmamak için bunu bir VM içinde ya da --cap-add NET_ADMIN ile başlatılmış bir container'da çalıştırın.

IFACE=eth0
sudo tc qdisc add dev $IFACE root netem delay 300ms 100ms loss 5%     # bad cellular
sudo tc qdisc change dev $IFACE root netem delay 2000ms loss 30%      # nearly unusable
sudo tc qdisc del dev $IFACE root                                     # back to normal

Beklenen: 300 ms ve %5 kayıpta TCP yeniden iletimleri kaybı gizler; gecikme artar ama yeniden bağlanma olmaz ve missing 0'da kalır. 2 sn ve %30 kayıpta PINGRESP çok geç gelebilir ve istemciler yeniden bağlanmaya başlar; kural kaldırıldığında durumun oturduğunu kontrol edin. Gerçekçi hücresel ağ profilleri için MQTT'yi hücresel ağ koşullarında test etmek makalesine bakın.

Deney 2: iptables ile ağ bölünmeleri

Bir ağ birbirinden çok farklı iki şekilde arızalanabilir ve istemciler çoğu zaman bunlardan yalnızca birini doğru ele alır:

# Blackhole: packets silently vanish. Only keep-alive can detect this.
sudo iptables -A OUTPUT -p tcp --dport 1883 -j DROP
# ...wait 60 s, then heal:
sudo iptables -D OUTPUT -p tcp --dport 1883 -j DROP

# Hard failure: the client gets a TCP reset immediately.
sudo iptables -A OUTPUT -p tcp --dport 1883 -j REJECT --reject-with tcp-reset
sudo iptables -D OUTPUT -p tcp --dport 1883 -j REJECT --reject-with tcp-reset

10 saniyelik keep-alive ile sonda, blackhole başladıktan yaklaşık 10–20 saniye sonra disconnected yazmalı, kesinti sürdükçe backoff ile tekrar denemeli ve ağ düzelince session present=True ile yeniden bağlanmalıdır. Broker tarafında diğer istemciler, yaklaşık 1,5 × keep-alive sonra sondanın Last Will mesajını almalıdır. Bir blackhole sırasında dakikalarca "bağlı" kalan bir istemcinin keep-alive'ı kapalıdır ya da çok yüksek ayarlanmıştır. Ağ kesintisiyle QoS testi, her QoS seviyesinin burada neyi garanti ettiğini daha ayrıntılı ele alır.

Deney 3: Toxiproxy ile bağlantı bazlı arızalar

Toxiproxy, davranışını bir HTTP API'si üzerinden değiştirdiğiniz bir TCP proxy'sidir. macOS ve Windows'ta da çalışır, yalnızca kendisinden geçen bağlantıları etkiler ve CI'da kolayca betiğe dökülebilir.

docker run -d --name toxiproxy -p 8474:8474 -p 21883:21883 ghcr.io/shopify/toxiproxy
curl -s -X POST localhost:8474/proxies \
  -d '{"name":"mqtt","listen":"0.0.0.0:21883","upstream":"broker.connectmq.com:1883"}'

# Point the probe at the proxy
MQTT_HOST=localhost MQTT_PORT=21883 python chaos_probe.py

Ardından ikinci bir terminalden arıza enjekte edin:

API=localhost:8474/proxies/mqtt
# 1.5 s latency with 500 ms jitter
curl -s -X POST $API/toxics -d '{"name":"lag","type":"latency","attributes":{"latency":1500,"jitter":500}}'
curl -s -X DELETE $API/toxics/lag
# Blackhole: data stops, connection stays open until the toxic is removed
curl -s -X POST $API/toxics -d '{"name":"cut","type":"timeout","attributes":{"timeout":0}}'
curl -s -X DELETE $API/toxics/cut
# Connection reset by peer
curl -s -X POST $API/toxics -d '{"name":"rst","type":"reset_peer","attributes":{"timeout":0}}'
curl -s -X DELETE $API/toxics/rst
# Take the whole "network" down and up again
curl -s -X POST $API -d '{"enabled":false}'; sleep 30; curl -s -X POST $API -d '{"enabled":true}'

0 değerli timeout toxic'i, iptables DROP'un Toxiproxy'deki karşılığıdır; reset_peer ise REJECT'e karşılık gelir. MQTT'yi Toxiproxy ile test etmek bant genişliği sınırlarını, slicing'i ve test framework'leriyle entegrasyonu anlatır.

Deney 4: broker yeniden başlatmaları ve çökmeleri

Yeniden başlatma deneyleri, kendi çalıştırdığınız bir broker gerektirir. İstemci tarafındaki hataların çoğunu bulmak için yerel bir Mosquitto container'ı yeterlidir:

docker run -d --name chaos-broker -p 1883:1883 eclipse-mosquitto:2 mosquitto -c /mosquitto-no-auth.conf
MQTT_HOST=localhost MQTT_PORT=1883 python chaos_probe.py     # in another terminal

docker restart chaos-broker                                   # graceful: SIGTERM, then start
docker kill --signal=KILL chaos-broker && docker start chaos-broker   # crash

Gözlemlemeniz gerekenler:

• Kalıcılık olmadan (yukarıdaki varsayılan yapılandırma) broker tüm oturumları unutur. Sonda session present=False ile yeniden bağlanır, yeniden abone olur ve yeniden başlatma sırasında yayınlanan mesajlar kalıcı olarak missing kalabilir.
• persistence true ve veri dizini için bir volume ile oturumlar ve kuyruktaki mesajlar düzgün bir yeniden başlatmadan sağ çıkar. Bir KILL sonrasında ise son autosave'den bu yana olan her şey yine kaybolabilir; bunun tasarımınızın vaat ettiği şeyle örtüşüp örtüşmediğini kontrol edin.
• İstemciler aynı anda yeniden bağlanmamalıdır. Yüzlerce istemciyle bir yeniden başlatma, bir bağlantı patlamasına dönüşür. Jitter'lı backoff bunu zamana yayar; MQTT yeniden bağlanma fırtınaları makalesine bakın.

Cluster'lı broker'larda yeniden başlatma yerine node'ları tek tek durdurun ve istemcilerin failover yaptığını doğrulayın. MQTT yeniden bağlanma testleri istemci tarafında daha fazla senaryo içerir.

Tekrarlanabilir bir deney planı

Planı her büyük istemci veya broker sürümünden önce çalıştırın, sonda çıktısını bir artifact olarak saklayın ve gerçek bir olay size yeni bir arıza türü öğrettiğinde tabloya yeni bir satır ekleyin.

Sonda mesajlarının gelişini tarayıcıda izleyin

MQTT için kaos testi esas olarak toparlanmayla ilgilidir: arızayı hızla fark etmek, nazikçe yeniden bağlanmak ve teslim etmeyi vaat ettiğiniz hiçbir şeyi kaybetmemek. netem, iptables, Toxiproxy ve yeniden başlatılabilen bir broker ile neredeyse her kesintiyi bir dizüstü bilgisayarda yeniden üretebilirsiniz; sonda da istemcilerinizin geçip geçmediğini size sayılarla söyler.

Sıkça sorulan sorular