Toxiproxy ile MQTT için Kötü Ağ Koşullarını Simüle Etmek

Toxiproxy, Shopify'ın geliştirdiği küçük bir TCP proxy'sidir. MQTT client'ınız ile broker arasına girer ve bağlantıyı bilerek bozmanızı sağlar: gecikme ve jitter eklemek, bant genişliğini kısmak, trafiği dondurmak ya da soketi sıfırlamak. Her şey bir HTTP API ile yönetilir; böylece testleriniz hataları koddan açıp kapatabilir.

Bu rehber Toxiproxy'yi bir MQTT broker'ının önüne kurar, MQTT için önemli olan toxic'leri anlatır ve bunları keep-alive, yeniden bağlanma ve QoS davranışı için otomatik testlere dönüştürür.

Neden kablo çekmek yerine proxy

Wi-Fi'ı kapatmak veya broker'ı öldürmek kaba araçlardır. Zamanlaması zordur, CI'da tekrarlanması zordur ve yalnızca tek bir hatayı simüle eder: "her şey gitti". Gerçek ağlar daha sinsi şekillerde bozulur: iki saniye gecikme ekleyen bir mobil hat, boşta kalan bir akışı sessizce düşüren bir NAT cihazı, saniyede 5 KB taşıyabilen tıkanmış bir uplink. Yeniden bağlanma döngülerindeki, inflight pencerelerindeki ve timeout'lardaki hataları ortaya çıkaran koşullar tam olarak bunlardır.

Toxiproxy bu hataların her birini tek bir TCP bağlantısı üzerinde, isimli ve geri alınabilir bir "toxic" olarak sunar. Broker ve diğer client'larınız etkilenmez; test de toxic aktifken kodunuzun tam olarak ne yaptığını doğrulayabilir.

Toxiproxy'yi broker'ın önünde çalıştırmak

En hızlı başlangıç resmi container'dır. 8474 portu kontrol API'sidir; oluşturduğunuz her proxy kendi dinleme portunu alır.

docker run -d --name toxiproxy \
  -p 8474:8474 -p 21883:21883 \
  ghcr.io/shopify/toxiproxy

21883 portunu dinleyen ve trafiği broker'a ileten bir proxy oluşturun:

curl -s -X POST localhost:8474/proxies -d '{
  "name": "mqtt",
  "listen": "0.0.0.0:21883",
  "upstream": "broker.connectmq.com:1883"
}'

Herhangi bir MQTT client'ını broker yerine localhost:21883 adresine yönlendirin ve trafiğin hâlâ aktığını doğrulayın:

mosquitto_sub -h localhost -p 21883 -t 'testmqtt/toxiproxy/#' -v &
mosquitto_pub -h localhost -p 21883 -t 'testmqtt/toxiproxy/hello' -m 'through the proxy'

Aynı topic'i doğrudan broker'a bağlı online MQTT client üzerinden izleyebilirsiniz: proxy üzerinden yayınlanan mesajlar orada da görünür; böylece bir toxic'in mesajları ne zaman geciktirdiğini veya engellediğini kolayca görürsünüz.

Bir test ortamı için proxy'yi bir yapılandırma dosyasında tanımlayın; container başlar başlamaz hazır olsun:

services:
  toxiproxy:
    image: ghcr.io/shopify/toxiproxy
    command: ["-host=0.0.0.0", "-config=/config/toxiproxy.json"]
    volumes:
      - ./toxiproxy.json:/config/toxiproxy.json:ro
    ports:
      - "8474:8474"
      - "21883:21883"

[
  { "name": "mqtt", "listen": "0.0.0.0:21883", "upstream": "broker.connectmq.com:1883", "enabled": true }
]

Broker aynı Compose dosyasında çalışıyorsa (örneğin Testcontainers ile bir Mosquitto container'ı ya da mosquitto adlı bir servis), upstream olarak mosquitto:1883 verin.

MQTT için önemli toxic'ler

Toxic'ler proxy başına ve yön başına eklenir. downstream broker → client yönüdür (CONNACK, PUBACK, teslim edilen mesajlar), upstream ise client → broker (CONNECT, PUBLISH, PINGREQ). Varsayılan yön downstream'dir.

# 800 ms ± 200 ms on everything the broker sends back
curl -s -X POST localhost:8474/proxies/mqtt/toxics -d '{
  "name": "slow", "type": "latency", "stream": "downstream",
  "attributes": { "latency": 800, "jitter": 200 }
}'

# Throttle the client's uplink to 4 KB/s
curl -s -X POST localhost:8474/proxies/mqtt/toxics -d '{
  "name": "thin", "type": "bandwidth", "stream": "upstream",
  "attributes": { "rate": 4 }
}'

curl -s localhost:8474/proxies/mqtt/toxics       # list active toxics
curl -s -X DELETE localhost:8474/proxies/mqtt/toxics/slow
curl -s -X POST localhost:8474/reset              # remove all toxics, re-enable proxies

MQTT'ye özgü birkaç not:

• latency her veri parçasını geciktirir. Downstream'de uygulandığında her gidiş-dönüşü uzatır: CONNACK, SUBACK ve PUBACK geç gelir. Fazla sıkı timeout'ları bulmak için kullanın; örneğin 1 saniyelik bir connect timeout, gecikme bunu aştığı anda başarısız olur.
• bandwidth (rate, KB/s cinsinden), hattın taşıyabileceğinden hızlı publish ettiğinizde ne olduğunu gösterir. QoS 1'de onaylanmamış mesajlar client'ın inflight penceresinde birikir; bkz. MQTT akış kontrolü.
• timeout, timeout: 0 ile tüm veriyi durdurur ama soketi açık tutar. İki taraf da TCP hatası almaz; bu klasik half-open bağlantıdır ve yalnızca MQTT keep alive bunu fark eder.
• reset_peer, timeout ms sonra bağlantıyı TCP RST ile kapatır; client tarafında ECONNRESET görürsünüz.
• slicer veriyi aralarında gecikme olan küçük segmentlere böler. Doğru yazılmış bir client MQTT paketlerini birden fazla okumada birleştirir; naif bir gömülü parser bunu yapamayabilir.
• limit_data, bytes byte sonra bağlantıyı kapatır; büyük bir PUBLISH'i ortasından kesmek için kullanışlıdır.

Bağlantıyı tamamen kesmek için proxy'yi devre dışı bırakın: curl -X POST localhost:8474/proxies/mqtt -d '{"enabled": false}'. Mevcut bağlantılar kapanır, yeniden etkinleştirene kadar yeni bağlantılar reddedilir.

Otomasyon: Python ile keep-alive testi

Kontrol API'si düz HTTP olduğundan herhangi bir client kütüphanesine gerek yoktur. Bu test (paho-mqtt 2.x) proxy üzerinden 5 saniyelik keep-alive ile bağlanır, hattı dondurur ve client'ın bunu keep-alive penceresi içinde fark ettiğini doğrular.

import json, threading, time, urllib.request
import paho.mqtt.client as mqtt

API = "http://localhost:8474"

def toxiproxy(method, path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(API + path, data=data, method=method)
    with urllib.request.urlopen(req) as resp:
        return resp.read()

def test_client_detects_half_open_connection():
    toxiproxy("POST", "/reset")
    connected, disconnected = threading.Event(), threading.Event()

    client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="toxi-keepalive-test")
    client.on_connect = lambda c, u, f, rc, p: connected.set()
    client.on_disconnect = lambda c, u, f, rc, p: disconnected.set()
    client.connect("localhost", 21883, keepalive=5)
    client.loop_start()
    try:
        assert connected.wait(5), "no CONNACK through the proxy"

        # Bağlantıyı kapatmadan iki yönde de dondur
        for stream in ("upstream", "downstream"):
            toxiproxy("POST", "/proxies/mqtt/toxics", {
                "name": f"freeze_{stream}", "type": "timeout",
                "stream": stream, "attributes": {"timeout": 0},
            })
        started = time.monotonic()

        assert disconnected.wait(15), "client never noticed the dead link"
        print(f"dead link detected after {time.monotonic() - started:.1f} s")
    finally:
        toxiproxy("POST", "/reset")
        client.loop_stop()
        client.disconnect()

pytest -s test_keepalive.py ile çalıştırın. Dondurmanın ping döngüsünün hangi noktasına denk geldiğine bağlı olarak tespit yaklaşık bir ile iki keep-alive aralığı sürmelidir. Çok daha uzun sürüyorsa veya hiç gerçekleşmiyorsa client'ınız keep-alive'ı uygulamıyor demektir; bu, gerçek cihazları saatlerce hiçbir şeye "bağlı" bırakacak bir hatadır. Daha düzenli bir fixture yapısı için MQTT için pytest fixture'ları yazısına bakın.

Aynı kalıp başka senaryoları da kapsar:

1. Yeniden bağlanma: proxy'yi devre dışı bırakın, bekleyin, tekrar etkinleştirin ve client'ın yeniden bağlanıp tekrar subscribe olduğunu doğrulayın (ayrıntılar MQTT yeniden bağlanma testleri yazısında).
2. QoS 1 yeniden gönderimi: upstream'e limit_data ekleyin, büyük bir QoS 1 mesajı publish edin, toxic'i kaldırın ve subscriber'ın mesajı tasarımınızın beklediği gibi tam bir kez veya en az bir kez aldığını kontrol edin.
3. Timeout bütçesi: downstream'e 3 saniye gecikme ekleyin ve connect ya da request-response timeout'unuzun takılıp kalmak yerine anlaşılır bir hatayla temiz biçimde başarısız olduğunu doğrulayın.

Toxiproxy üzerinden TLS ve WebSocket ile MQTT

Toxiproxy ham byte ilettiği için TLS (8883) ve WebSocket (8083, /mqtt path'i ile 8084) değişiklik olmadan çalışır: upstream'i broker.connectmq.com:8883 olan bir proxy oluşturun, handshake doğrudan geçer.

Çözmeniz gereken tek şey hostname doğrulamasıdır. Client localhost'a bağlanır ama sertifikada broker'ın adı yazar. Seçenekler:

• Client'ta broker hostname'ini koruyun ve test runner'da /etc/hosts ya da Docker'ın extra_hosts ayarıyla proxy'ye yönlendirin.
• TLS server name değerini açıkça verin: MQTT.js'te servername seçeneği; Python'da soketi server_hostname broker adı olacak şekilde sarmalayın.
• Ağ hatası testlerinde düz TCP kullanın, TLS'i ayrı bir TLS bağlantı testine bırakın.

Production ayarlarıyla da çalışan bir testte proxy'yi çalıştırmak için sertifika doğrulamasını asla kapatmayın; bu alışkanlık genellikle gerçek koda sızar.

Sık yapılan hatalar

• Gecikmeyi yalnızca upstream'e eklemek. İlginç timeout'ların çoğu broker'ın yanıtını bekler. İki yönü de test edin ya da önce downstream'i.
• timeout: 0 ile proxy'yi devre dışı bırakmayı karıştırmak. Devre dışı bırakmak soketleri kapatır (client anında disconnect görür); timeout: 0 ise onları dondurur (sizi yalnızca keep-alive kurtarır).
• Yalnızca QoS 0 ile test etmek. QoS 0, onaylanmamış mesajlara ne olduğunu gizler. Kritik senaryoları QoS 1 ile tekrarlayın.
• Public bir broker üzerinden ölçüm yapmak. Paylaşımlı sandbox'lar kendi dalgalanmalarını ekler. Assert edeceğiniz her şey için kontrolünüzdeki bir broker kullanın.

Proxy üzerinden gelen mesajları tarayıcıda izleyin

Toxiproxy, "sahada bazen bağlantı kopuyor" durumunu her commit'te çalıştırabileceğiniz bir teste dönüştürür. Bir proxy, dondurulmuş hatla bir keep-alive testi ve bir yeniden bağlanma testiyle başlayın; bunlar geçtikten sonra cihazlarınızın gerçekte kullandığı ağlara benzeyen gecikme ve bant genişliği profilleri ekleyin; örneğin yavaş ve kararsız hücresel ağlarda MQTT testi yazısındaki profiller.

Sıkça sorulan sorular