MQTT-Client-Apps auf Kubernetes betreiben

Ein MQTT-Client auf Kubernetes sieht aus wie jedes andere Deployment – bis du auf zwei Replicas skalierst. Dann werfen sich die Pods gegenseitig vom Broker, jede Replica verarbeitet jede Nachricht, und Rolling Updates verlieren Daten. Drei Regeln lösen fast alles: gib jedem Pod eine eindeutige Client-ID, teile die Arbeit mit Shared Subscriptions zwischen den Replicas auf und koppel die MQTT-Verbindung an Readiness, nicht an Liveness.

Dieses Tutorial baut einen kleinen Node.js-Consumer (MQTT.js 5) und ein Deployment-Manifest, das alle drei Regeln umsetzt – dazu einen sauberen Shutdown und einen Test, den du im Browser ausführen kannst.

Warum sich MQTT-Clients auf Kubernetes anders verhalten

HTTP-Services sind zustandslos: Jeder Pod kann jede Anfrage beantworten, und der Service verteilt die Verbindungen. MQTT-Clients sind das Gegenteil. Jeder Pod baut eine langlebige Verbindung zum Broker auf, und der Broker – nicht Kubernetes – entscheidet, welcher Client welche Nachricht bekommt.

Regel 1: eine Client-ID pro Pod

MQTT erlaubt genau eine aktive Session pro Client-ID. Kommt eine zweite Verbindung mit derselben ID, schließt der Broker die erste (MQTT-5-Reason-Code 0x8E Session taken over). Zwei identisch konfigurierte Replicas trennen sich also abwechselnd bis in alle Ewigkeit – die Symptome beschreibt Session taken over.

Die sauberste Lösung ist die Downward API: Injiziere den Pod-Namen als Umgebungsvariable und baue die ID daraus. Pod-Namen sind innerhalb eines Namespace eindeutig und über die Lebensdauer des Pods stabil.

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

In der App: clientId = "ingest-" + process.env.POD_NAME. Halte IDs kurz und bei den Zeichen, die dein Broker akzeptiert; der Client-ID-Generator und die Best Practices für Client-IDs erklären die Grenzen.

Regel 2: Shared Subscriptions teilen die Arbeit auf

Mit einer normalen Subscription bekommen drei Replicas desselben Consumers jeweils jede Nachricht – du speicherst jeden Messwert dreimal. Eine Shared Subscription fasst die Replicas zu einer Gruppe zusammen, und der Broker stellt jede Nachricht nur einem Mitglied zu:

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

Alle Pods abonnieren denselben Filter $share/ingest/.... Hochskalieren fügt Gruppenmitglieder hinzu, Herunterskalieren entfernt sie, und der Broker verteilt selbstständig neu. Shared Subscriptions sind ein MQTT-5-Feature; viele Broker akzeptieren das Präfix $share/ auch von MQTT-3.1.1-Clients, prüfe das aber bei deinem. Die komplette Mechanik, auch wie Broker ein Mitglied auswählen, steht in MQTT Shared Subscriptions.

Zwei Dinge solltest du beachten: Die Reihenfolge ist nur pro Mitglied garantiert, nicht über die ganze Gruppe. Und wenn eine Nachricht jeden Pod erreichen muss (etwa ein Konfigurations-Reload), nutze dafür eine separate normale Subscription.

Die Consumer-App

Die App verbindet sich mit einer Pod-spezifischen ID, tritt der Shared-Gruppe bei und stellt auf Port 8080 /healthz (Prozess lebt) und /readyz (mit dem Broker verbunden) bereit.

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), // Jitter zwischen Pods
});

let ready = false;

client.on('connect', async () => {
  await client.subscribeAsync('$share/ingest/testmqtt/k8s/+/telemetry', { qos: 1 });
  ready = true;
  console.log(`[${pod}] verbunden und abonniert`);
});
client.on('close', () => { ready = false; });
client.on('error', (err) => console.error(`[${pod}] MQTT-Fehler:`, err.message));

client.on('message', (topic, payload) => {
  // Hier passiert die eigentliche Arbeit: in eine Datenbank schreiben, eine API aufrufen, ...
  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: trenne Verbindung`);
  ready = false;
  await client.endAsync(); // wartet auf laufende Nachrichten und sendet dann DISCONNECT
  process.exit(0);
});

Der zufällige reconnectPeriod ist wichtig: Startet der Broker neu, erzeugen fünfzig Pods, die im selben Moment reconnecten, eine Spitze, die ihn erneut umwerfen kann. Backoff-Strategien findest du unter Reconnect-Stürme.

Regel 3: Probes, die Ausfälle nicht verschlimmern

• Readiness – liefert 503, solange die MQTT-Verbindung weg ist. Kubernetes zählt den Pod dann nicht mehr als verfügbar, und ein Rollout pausiert, statt gesunde Pods durch getrennte zu ersetzen.
• Liveness – prüft nur, ob der Prozess antwortet. Ist der Broker nicht erreichbar, hilft ein Neustart des Pods nicht, und alle Pods gleichzeitig neu zu starten macht aus einem Ausfall einen Reconnect-Sturm.
• Startup (optional) – gibt langsam startenden Apps Zeit, bevor die Liveness greift.

Ein Worker ohne HTTP-Traffic braucht keinen Service; die Readiness-Probe steuert trotzdem die Rollouts, und kubectl get pods zeigt auf einen Blick, welche Replicas verbunden sind.

Das Deployment-Manifest

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 }

Lege das Secret einmalig mit kubectl create secret generic mqtt-credentials --from-literal=username="$MQTT_USER" --from-literal=password="$MQTT_PASS" an. Credentials gehören nie ins Image.

Sauberer Shutdown und Rolling Updates

Beim Beenden eines Pods sendet Kubernetes SIGTERM, wartet bis zu terminationGracePeriodSeconds und schickt dann SIGKILL. Der Handler oben markiert den Pod als nicht bereit, lässt MQTT.js laufende Bestätigungen abschließen und sendet ein sauberes DISCONNECT. Zwei Folgen:

• Bei einem sauberen DISCONNECT veröffentlicht der Broker den Last Will des Clients nicht. Nutze den Will für Abstürze, nicht für geplante Neustarts – siehe Last Will and Testament.
• Bei QoS 1 wird eine Nachricht, die zugestellt, aber beim Sterben des Pods noch nicht bestätigt war, an ein anderes Gruppenmitglied erneut zugestellt (das Verhalten hängt vom Broker ab). Mach deinen Handler idempotent, damit ein Duplikat keinen Schaden anrichtet.

maxUnavailable: 0 mit maxSurge: 1 bedeutet: Ein neuer Pod muss bereit – also verbunden – sein, bevor ein alter entfernt wird. Die Shared-Gruppe schrumpft während eines Rollouts also nie.

Wo der Broker läuft

Den Client-Pods ist egal, ob der Broker im Cluster oder außerhalb läuft; sie brauchen nur Egress auf dem richtigen Port (8883 für TLS). Den Broker im Cluster zu betreiben heißt, einen weiteren zustandsbehafteten Workload zu pflegen – Mosquitto auf Kubernetes zeigt, was dazugehört. Ein Managed Broker wie ConnectMQ hält den Cluster zustandslos: Dasselbe Manifest funktioniert in jeder Umgebung, nur URL und Secret ändern sich.

Hat dein Cluster eine restriktive NetworkPolicy, erlaube Egress zu Broker-Host und -Port explizit; ein blockierter Port zeigt sich als Verbindungs-Timeouts im Log und als Pod, der nie bereit wird.

Testen

1. Wende das Manifest an und warte, bis kubectl get pods -l app=ingest drei bereite Pods zeigt.
2. Öffne den Online-MQTT-Client, verbinde dich mit broker.connectmq.com und publishe ein paar Nachrichten an testmqtt/k8s/dev-1/telemetry, testmqtt/k8s/dev-2/telemetry usw.
3. Führe kubectl logs -l app=ingest --prefix -f aus. Jede Nachricht sollte im Log von genau einem Pod erscheinen, verteilt über die Replicas.
4. Führe während des Publishens kubectl rollout restart deployment/ingest aus. Die Readiness sollte den Rollout zurückhalten, bis die neuen Pods verbunden sind, und bei QoS 1 sollte keine Nachricht fehlen.
5. Skaliere auf 1 und zurück auf 3 und prüfe, dass sich die Nachrichten neu verteilen.

Test-Telemetrie aus dem Browser publishen

Der geteilte Test-Endpoint speichert keine persistenten Sessions. Um StatefulSet-artige Sessions mit gepufferten Nachrichten zu testen, führe dieselben Schritte gegen deine eigene private Broker-Instanz aus.

Kubernetes behandelt Pods als austauschbar, MQTT-Broker behandeln Client-IDs als Identitäten. Verbinde beides mit vom Pod abgeleiteten IDs, Shared Subscriptions und Readiness-gesteuerten Rollouts, und deine Consumer skalieren wie jeder andere zustandslose Service.

Häufig gestellte Fragen