$SYS Topic'leri ile MQTT Broker İzleme

Çoğu MQTT broker'ı kendi durumunu yine MQTT üzerinden bildirir. $SYS/# topic'ine abone olduğunuzda broker kaç istemcinin bağlı olduğunu, kaç mesaj ve kaç byte aldığını ve gönderdiğini, kaç abonelik ve retained mesaj tuttuğunu ve ne kadar süredir çalıştığını yayınlar. Bu da $SYS'i bir broker'ı izlemenin en hızlı yolu yapar. Exporter'a ya da agent'a gerek yoktur, bir MQTT istemcisi yeterlidir.

Bu rehberde $SYS'in yaygın broker'larda neleri sunduğunu, ona doğru şekilde nasıl abone olacağınızı, ham sayaçları kullanışlı oranlara nasıl çevireceğinizi ve kısa bir Python betiğiyle bunları nasıl kaydedip grafiğe dökeceğinizi göreceksiniz.

$SYS topic'leri nedir?

MQTT spesifikasyonu $ ile başlayan topic'leri broker'ın iç kullanımına ayırır. İstemciler buraya publish etmemelidir ve wildcard'lar bu topic'leri yanlışlıkla yakalamaz: # veya +/foo aboneliği hiçbir zaman $SYS mesajı almaz. $SYS/# topic'ine (ya da daha dar bir filtreye) bilerek abone olmanız gerekir.

İçerik standart değildir. Mosquitto, EMQX, VerneMQ ve diğerleri kendi ağaçlarını yayınlar; bazı yönetilen servisler bu ağacı tamamen gizler veya ACL ile kısıtlar. Topic ağacının eksiksiz bir referansı için MQTT $SYS topic'leri yazısına bakın. Bu yazı ise bu topic'leri izleme için kullanmayı anlatıyor.

İzlemeye değer topic'ler

Aşağıdakiler Mosquitto (2.x) adlarıdır. Diğer broker'lar benzer değerleri farklı yollar altında yayınlar.

Burada iki tür sayı vardır. Gauge'lar (bağlı istemciler, abonelikler) olduğu gibi anlamlıdır. Sayaçlar (mesajlar, byte'lar) yalnızca artar; bir oran elde etmek için iki örnek arasındaki farka ihtiyacınız vardır. Mosquitto ayrıca $SYS/broker/load/... altında hazır hareketli ortalamalar yayınlar. Hızlı bir bakış için kullanışlıdırlar ama zaman penceresi üzerindeki kontrolü kaybedersiniz.

EMQX'te ağaç node bazındadır: $SYS/brokers/<node>/version, $SYS/brokers/<node>/uptime, ayrıca stats/... ve metrics/... alt ağaçları ile isteğe bağlı istemci bağlanma/kopma olayları. Yayın aralığı sys_msg_interval ile ayarlanır. EMQX'te dashboard'lar için çoğu zaman yerleşik Prometheus endpoint'i daha iyi bir kaynaktır. Ayrıntılar için Prometheus ve Grafana ile izleme yazısına bakın.

$SYS'e abone olmak

En hızlı kontrol mosquitto_sub ile yapılır. Topic'i tek tırnak içine alın, yoksa shell $SYS'i bir değişken gibi genişletmeye çalışır:

mosquitto_sub -h broker.connectmq.com -p 1883 \
  -u "$MQTT_USER" -P "$MQTT_PASS" \
  -v -t '$SYS/#'

Yalnızca birkaç değeri izlemek için birkaç dar filtreye abone olun:

mosquitto_sub -h broker.connectmq.com -p 1883 \
  -u "$MQTT_USER" -P "$MQTT_PASS" -v \
  -t '$SYS/broker/clients/connected' \
  -t '$SYS/broker/load/messages/+/1min' \
  -t '$SYS/broker/uptime'

Aynısını tarayıcıda da yapabilirsiniz: online MQTT istemcisini açın, bağlanın ve $SYS/# topic'ine abone olun. Hiçbir şey gelmiyorsa şu üç şeyi sırayla kontrol edin:

1. Broker'da $SYS kapalıdır. Mosquitto'da sys_interval 0 bu özelliği kapatır.
2. Bir ACL erişimi engelliyordur. Birçok production yapılandırması $SYS'i yalnızca bir yönetici kullanıcının okumasına izin verir. MQTT 5 istemcileri 0x87 Not authorized reason code'lu bir SUBACK alır. Bunu reason code aracı ile çözümleyebilirsiniz.
3. Yalnızca bir aralık beklediniz. Bazı değerler ancak bir sonraki sys_interval tick'inde görünür.

Güncelleme aralığını ayarlamak

Mosquitto'da aralık tek bir seçenektir:

# Publish $SYS updates every 10 seconds (default). 0 disables $SYS.
sys_interval 10

Daha kısa bir aralık daha ayrıntılı grafikler verir ama $SYS/# topic'ine abone olan her istemci için biraz CPU ve bant genişliği harcar. Dashboard'lar için 10–30 saniye fazlasıyla yeterlidir. Yapılandırmanızı sıfırdan oluşturuyorsanız mosquitto.conf oluşturucu bu seçeneği de içerir.

Sayaçları orana çevirmek

Sayaçlar yalnızca fark olarak işe yarar. messages/received saat 12:00:00'da 1.204.500 ve 12:00:10'da 1.210.500 ise broker o pencerede saniyede (1210500 − 1204500) / 10 = 600 mesaj almıştır. İki kural hesabı dürüst tutar:

• Örneği aldığınız zamanı kullanın, nominal aralığı değil. Ağ jitter'ı ve broker yükü tick'leri kaydırır.
• Sıfırlanmaları ele alın. Broker yeniden başladığında sayaçlar sıfırdan başlar. Yeni değer öncekinden küçükse, devasa bir negatif sıçrama çizmek yerine o farkı atın. uptime değerinin düşmesi yeniden başlatmayı doğrular.

Python ile $SYS kaydetmek ve grafiğe dökmek

Bu betik (paho-mqtt 2.x) birkaç $SYS topic'ine abone olur, sayaçlardan oranları hesaplar, her güncellemede bir satır yazdırır ve her şeyi daha sonra grafiğe dökebileceğiniz bir CSV dosyasına ekler:

import csv, os, time
import paho.mqtt.client as mqtt

HOST, PORT = "broker.connectmq.com", 1883
TOPICS = {
    "$SYS/broker/clients/connected": "gauge",
    "$SYS/broker/subscriptions/count": "gauge",
    "$SYS/broker/messages/received": "counter",
    "$SYS/broker/messages/sent": "counter",
    "$SYS/broker/bytes/received": "counter",
    "$SYS/broker/bytes/sent": "counter",
}
last = {}  # topic -> (zaman damgası, değer)
out = csv.writer(open("sys_metrics.csv", "a", newline=""))

def on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print("bağlantı başarısız:", reason_code)
        return
    client.subscribe([(t, 0) for t in TOPICS])

def on_message(client, userdata, msg):
    now = time.time()
    try:
        value = float(msg.payload.decode().split()[0])
    except ValueError:
        return
    name = msg.topic.removeprefix("$SYS/broker/")
    if TOPICS[msg.topic] == "counter":
        prev = last.get(msg.topic)
        last[msg.topic] = (now, value)
        if prev is None or value < prev[1]:   # ilk örnek veya broker yeniden başladı
            return
        value = (value - prev[1]) / (now - prev[0])
        name += "/s"
    out.writerow([int(now), name, round(value, 2)])
    print(f"{time.strftime('%H:%M:%S')}  {name:<24} {value:>12.1f}")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="testmqtt-sys-monitor")
client.username_pw_set(os.getenv("MQTT_USER"), os.getenv("MQTT_PASS"))
client.on_connect = on_connect
client.on_message = on_message
client.connect(HOST, PORT, keepalive=30)
client.loop_forever()

on_connect içinde abone olmak önemlidir: böylece yeniden bağlanmadan sonra abonelikler otomatik olarak geri yüklenir. CSV'yi grafiğe dökmek için birkaç satır pandas ve matplotlib yeterlidir:

import pandas as pd
import matplotlib.pyplot as plt

df = pd.read_csv("sys_metrics.csv", names=["ts", "metric", "value"])
df["ts"] = pd.to_datetime(df["ts"], unit="s")
df.pivot_table(index="ts", columns="metric", values="value").plot(subplots=True, figsize=(10, 8))
plt.tight_layout()
plt.show()

Geçmişi ve alarmları olan kalıcı bir dashboard için aynı değerleri CSV yerine Prometheus'a veya InfluxDB'ye gönderin.

İzlemenizin tepki verdiğini test edin

Hiç değişiklik görmemiş bir monitör test edilmemiş demektir. Bilinen bir yük üretin ve sayıların değiştiğini doğrulayın:

for i in $(seq 1 1000); do echo "{\"i\":$i}"; done | \
  mosquitto_pub -h broker.connectmq.com -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/sys-monitor/load -l

Test sırasında messages/received/s aşağı yukarı publish hızınız kadar artmalıdır. Online istemcide ya da bir cihaz simülatörü ile on ek bağlantı açtığınızda clients/connected ona kadar artmalıdır. İkisi de değişmiyorsa büyük ihtimalle yazdığınız node'dan farklı bir node'u okuyorsunuzdur ya da bir cluster node bazında değer raporluyordur.

$SYS ile izlemenin sınırları

• Geçmiş yok. $SYS yalnızca anlık durumu gösterir. Saklama işini sizin yapmanız gerekir.
• Alarm yok. Prometheus, Grafana veya size haber veren bir betikle birlikte kullanın. Dışarıdan yapılan kontroller için broker health check yazısına bakın.
• Broker'a özgüdür. Topic adları ve anlamları broker'dan broker'a, hatta bazen sürümden sürüme değişir.
• İzlediği sistem tarafından izlenir. Broker çökerse $SYS susar. Susan bir monitör sakin bir gün gibi görünmemeli, alarm üretmelidir.

$SYS topic'leri, bir MQTT broker'ının sunduğu en ucuz gözlemlenebilirliktir. Broker'ınızın ne yayınladığını görmek için bir kez $SYS/# topic'ine abone olun, sayaçları orana çeviren küçük bir betik tutun ve ağacı yalnızca izleme hesabınızın okuyabilmesi için erişimi kısıtlayın.

Sıkça sorulan sorular