MQTT İstemci Uygulamalarını Kubernetes'te Çalıştırmak

Kubernetes üzerinde bir MQTT istemcisi çalıştırmak, iki replikaya ölçekleyene kadar sıradan bir Deployment gibi görünür. Sonra pod'lar birbirini broker'dan atmaya başlar, her replika her mesajı işler ve rolling update'ler veri kaybeder. Üç kural neredeyse her şeyi çözer: her pod'a benzersiz bir client ID verin, işi replikalar arasında bölmek için shared subscription kullanın ve MQTT bağlantısını liveness'a değil readiness'a bağlayın.

Bu eğitimde üç kuralı da uygulayan küçük bir Node.js tüketicisi (MQTT.js 5) ve bir Deployment manifesti oluşturuyoruz; buna temiz kapanış ve tarayıcınızdan çalıştırabileceğiniz bir test de dahil.

MQTT istemcileri Kubernetes'te neden farklı davranır

HTTP servisleri durumsuzdur: herhangi bir pod herhangi bir isteğe yanıt verebilir ve Service bağlantıları dağıtır. MQTT istemcileri bunun tersidir. Her pod broker'a doğru tek bir uzun ömürlü bağlantı açar ve hangi istemcinin hangi mesajı alacağına Kubernetes değil broker karar verir.

Kural 1: pod başına bir client ID

MQTT, her client ID için tam olarak bir aktif oturuma izin verir. Aynı ID ile ikinci bir bağlantı geldiğinde broker ilkini kapatır (MQTT 5 reason code 0x8E Session taken over). Aynı şekilde yapılandırılmış iki replika sonsuza kadar sırayla birbirini koparır; belirtiler için session taken over yazısına bakın.

En temiz çözüm downward API'dir: pod adını ortam değişkeni olarak enjekte edin ve ID'yi ondan türetin. Pod adları namespace içinde benzersizdir ve pod'un ömrü boyunca sabittir.

env:
  - name: POD_NAME
    valueFrom:
      fieldRef:
        fieldPath: metadata.name

Uygulamada clientId = "ingest-" + process.env.POD_NAME. ID'leri kısa ve broker'ınızın kabul ettiği karakterlerle sınırlı tutun; client ID oluşturucu ve client ID en iyi uygulamaları sınırları açıklıyor.

Kural 2: shared subscription işi böler

Normal bir abonelikte aynı tüketicinin üç replikası her mesajı ayrı ayrı alır ve her ölçümü üç kez kaydedersiniz. Bir shared subscription replikaları bir grupta toplar ve broker her mesajı yalnızca bir üyeye teslim eder:

$share/<group>/<topic filter>
$share/ingest/testmqtt/k8s/+/telemetry

Tüm pod'lar aynı $share/ingest/... filtresine abone olur. Deployment'ı büyütmek gruba üye ekler, küçültmek üye çıkarır ve broker dengelemeyi kendisi yapar. Shared subscription bir MQTT 5 özelliğidir; birçok broker $share/ önekini MQTT 3.1.1 istemcilerinden de kabul eder, ancak kendi broker'ınızda kontrol edin. Broker'ın üye seçme biçimi dahil tüm mekanizma MQTT shared subscriptions yazısında.

İki şeyi aklınızda tutun: sıralama yalnızca üye başına garanti edilir, grubun tamamında değil; ve bir mesajın her pod'a ulaşması gerekiyorsa (örneğin yapılandırmayı yeniden yükleme), o topic için ayrı bir normal abonelik kullanın.

Tüketici uygulama

Uygulama pod'a özel bir ID ile bağlanır, paylaşılan gruba katılır ve 8080 portunda /healthz (süreç ayakta) ile /readyz (broker'a bağlı) uç noktalarını sunar.

import http from 'node:http';
import mqtt from 'mqtt';

const pod = process.env.POD_NAME ?? `local-${Math.random().toString(16).slice(2, 8)}`;
const client = mqtt.connect(process.env.MQTT_URL ?? 'mqtts://broker.connectmq.com:8883', {
  clientId: `ingest-${pod}`,
  username: process.env.MQTT_USER,
  password: process.env.MQTT_PASS,
  protocolVersion: 5,
  clean: true,
  keepalive: 30,
  reconnectPeriod: 2000 + Math.floor(Math.random() * 3000), // pod'lar arasında jitter
});

let ready = false;

client.on('connect', async () => {
  await client.subscribeAsync('$share/ingest/testmqtt/k8s/+/telemetry', { qos: 1 });
  ready = true;
  console.log(`[${pod}] bağlandı ve abone oldu`);
});
client.on('close', () => { ready = false; });
client.on('error', (err) => console.error(`[${pod}] mqtt hatası:`, err.message));

client.on('message', (topic, payload) => {
  // Asıl iş burada yapılır: veritabanına yazmak, bir API çağırmak, ...
  console.log(`[${pod}] ${topic} ${payload.toString()}`);
});

http.createServer((req, res) => {
  if (req.url === '/healthz') return res.writeHead(200).end('ok');
  if (req.url === '/readyz') return res.writeHead(ready ? 200 : 503).end(ready ? 'ready' : 'not connected');
  res.writeHead(404).end();
}).listen(8080);

process.on('SIGTERM', async () => {
  console.log(`[${pod}] SIGTERM: bağlantı kapatılıyor`);
  ready = false;
  await client.endAsync(); // yoldaki mesajları bekler, sonra DISCONNECT gönderir
  process.exit(0);
});

Rastgele reconnectPeriod önemlidir: broker yeniden başladığında aynı anda yeniden bağlanan elli pod, onu tekrar devirebilecek bir yük sıçraması oluşturur. Backoff stratejileri için yeniden bağlanma fırtınaları yazısına bakın.

Kural 3: kesintiyi büyütmeyen probe'lar

• Readiness — MQTT bağlantısı koptuğu sürece 503 döner. Kubernetes pod'u kullanılabilir saymayı bırakır; böylece bir rollout, sağlıklı pod'ları bağlantısız olanlarla değiştirmek yerine duraklar.
• Liveness — yalnızca sürecin yanıt verip vermediğini kontrol eder. Broker'a ulaşılamıyorsa pod'u yeniden başlatmak sorunu çözmez; tüm pod'ları aynı anda yeniden başlatmak ise bir kesintiyi yeniden bağlanma fırtınasına çevirir.
• Startup (isteğe bağlı) — yavaş açılan uygulamalara liveness devreye girmeden önce zaman tanır.

HTTP trafiği olmayan bir worker'ın Service'e ihtiyacı yoktur; readiness probe yine rollout'ları kontrol eder ve kubectl get pods hangi replikaların bağlı olduğunu bir bakışta gösterir.

Deployment manifesti

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ingest
spec:
  replicas: 3
  selector:
    matchLabels: { app: ingest }
  strategy:
    rollingUpdate: { maxUnavailable: 0, maxSurge: 1 }
  template:
    metadata:
      labels: { app: ingest }
    spec:
      terminationGracePeriodSeconds: 30
      containers:
        - name: ingest
          image: ghcr.io/your-org/mqtt-ingest:1.0.0
          ports: [{ containerPort: 8080 }]
          env:
            - name: POD_NAME
              valueFrom: { fieldRef: { fieldPath: metadata.name } }
            - name: MQTT_URL
              value: mqtts://broker.connectmq.com:8883
            - name: MQTT_USER
              valueFrom: { secretKeyRef: { name: mqtt-credentials, key: username } }
            - name: MQTT_PASS
              valueFrom: { secretKeyRef: { name: mqtt-credentials, key: password } }
          readinessProbe:
            httpGet: { path: /readyz, port: 8080 }
            periodSeconds: 5
            failureThreshold: 2
          livenessProbe:
            httpGet: { path: /healthz, port: 8080 }
            periodSeconds: 10
            failureThreshold: 3
          resources:
            requests: { cpu: 50m, memory: 64Mi }
            limits: { memory: 256Mi }

Secret'ı bir kez kubectl create secret generic mqtt-credentials --from-literal=username="$MQTT_USER" --from-literal=password="$MQTT_PASS" ile oluşturun. Kimlik bilgilerini asla imajın içine gömmeyin.

Temiz kapanış ve rolling update'ler

Bir pod sonlandırılırken Kubernetes SIGTERM gönderir, terminationGracePeriodSeconds kadar bekler ve ardından SIGKILL gönderir. Yukarıdaki handler pod'u hazır değil olarak işaretler, MQTT.js'in yoldaki onayları tamamlamasına izin verir ve temiz bir DISCONNECT gönderir. İki sonucu var:

• Temiz bir DISCONNECT'te broker istemcinin Last Will mesajını yayınlamaz. Will'i planlı yeniden başlatmalar için değil, çökmeler için kullanın; Last Will and Testament yazısına bakın.
• QoS 1'de, pod öldüğü anda teslim edilmiş ama henüz onaylanmamış bir mesaj gruptaki başka bir üyeye yeniden teslim edilir (davranış broker'a göre değişir). Bir kopyanın zarar vermemesi için handler'ınızı idempotent yapın.

maxUnavailable: 0 ile maxSurge: 1, eski bir pod kaldırılmadan önce yeni bir pod'un hazır — yani bağlı — olması gerektiği anlamına gelir; böylece paylaşılan grup rollout sırasında asla küçülmez.

Broker nerede çalışıyor

İstemci pod'ları için broker'ın cluster içinde mi dışında mı olduğu fark etmez; yalnızca doğru porttan (TLS için 8883) çıkış erişimine ihtiyaç duyarlar. Broker'ı cluster içinde çalıştırmak, işletmeniz gereken bir durumlu workload daha demektir; Kubernetes'te Mosquitto bunun neler gerektirdiğini gösteriyor. ConnectMQ gibi yönetilen bir broker cluster'ı durumsuz tutar: aynı manifest her ortamda çalışır, yalnızca URL ve Secret değişir.

Cluster'ınızda kısıtlayıcı bir NetworkPolicy varsa broker host'una ve portuna çıkışa açıkça izin verin; engellenmiş bir port, loglarda bağlantı zaman aşımları ve hiç hazır olmayan bir pod olarak görünür.

Test etmek

1. Manifesti uygulayın ve kubectl get pods -l app=ingest üç hazır pod gösterene kadar bekleyin.
2. Online MQTT istemcisini açın, broker.connectmq.com adresine bağlanın ve testmqtt/k8s/dev-1/telemetry, testmqtt/k8s/dev-2/telemetry gibi topic'lere birkaç mesaj publish edin.
3. kubectl logs -l app=ingest --prefix -f çalıştırın. Her mesaj replikalara dağılmış şekilde yalnızca bir pod'un logunda görünmelidir.
4. Publish ederken kubectl rollout restart deployment/ingest çalıştırın. Readiness, yeni pod'lar bağlanana kadar rollout'u tutmalı ve QoS 1'de hiçbir mesaj kaybolmamalıdır.
5. 1'e ve tekrar 3'e ölçekleyin ve mesajların yeniden dağıldığını doğrulayın.

Tarayıcınızdan test telemetrisi publish edin

Paylaşılan test uç noktası kalıcı oturum tutmaz; kuyruğa alınan mesajlarla StatefulSet tarzı oturumları test etmek için aynı adımları kendi özel broker instance'ınızda uygulayın.

Kubernetes pod'ları birbirinin yerine geçebilir görür; MQTT broker'ları ise client ID'leri kimlik olarak görür. İkisini pod'dan türetilmiş ID'ler, shared subscription'lar ve readiness güdümlü rollout'larla bağdaştırın; tüketicileriniz diğer durumsuz servisler gibi ölçeklensin.

Sıkça sorulan sorular