Exécuter des applications clientes MQTT sur Kubernetes

Un client MQTT sur Kubernetes ressemble à n'importe quel Deployment… jusqu'à ce que vous passiez à deux réplicas. Les pods commencent alors à s'éjecter mutuellement du broker, chaque réplica traite chaque message et les rolling updates perdent des données. Trois règles règlent presque tout : donnez à chaque pod un client ID unique, utilisez les shared subscriptions pour répartir le travail entre réplicas et branchez la connexion MQTT sur la readiness, pas sur la liveness.

Ce tutoriel construit un petit consommateur Node.js (MQTT.js 5) et un manifeste de Deployment qui applique ces trois règles, avec un arrêt propre et un test réalisable depuis votre navigateur.

Pourquoi les clients MQTT se comportent différemment sur Kubernetes

Les services HTTP sont sans état : n'importe quel pod peut répondre à n'importe quelle requête et le Service répartit les connexions. Les clients MQTT sont l'inverse. Chaque pod ouvre une connexion longue durée vers le broker, et c'est le broker — pas Kubernetes — qui décide quel client reçoit quel message.

Règle 1 : un client ID par pod

MQTT n'autorise qu'une seule session active par client ID. Quand une deuxième connexion arrive avec le même ID, le broker ferme la première (reason code MQTT 5 0x8E Session taken over). Deux réplicas configurés à l'identique se déconnectent donc à tour de rôle indéfiniment — voir session taken over pour les symptômes.

La solution la plus propre est la downward API : injectez le nom du pod dans une variable d'environnement et construisez l'ID à partir de lui. Les noms de pod sont uniques dans un namespace et stables pendant toute la vie du pod.

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

Dans l'application : clientId = "ingest-" + process.env.POD_NAME. Gardez des IDs courts, avec des caractères acceptés par votre broker ; le générateur de client ID et les bonnes pratiques de client ID détaillent les limites.

Règle 2 : les shared subscriptions répartissent le travail

Avec un abonnement classique, trois réplicas du même consommateur reçoivent chacun tous les messages : vous enregistrez chaque mesure trois fois. Une shared subscription regroupe les réplicas, et le broker remet chaque message à un seul membre :

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

Tous les pods s'abonnent au même filtre $share/ingest/.... Augmenter le nombre de réplicas ajoute des membres au groupe, le réduire en retire, et le broker rééquilibre tout seul. Les shared subscriptions sont une fonctionnalité de MQTT 5 ; beaucoup de brokers acceptent aussi le préfixe $share/ depuis des clients MQTT 3.1.1, mais vérifiez pour le vôtre. Le mécanisme complet, y compris la façon dont le broker choisit un membre, est décrit dans MQTT shared subscriptions.

Gardez deux points en tête : l'ordre n'est garanti que par membre, pas sur l'ensemble du groupe, et si un message doit atteindre tous les pods (un rechargement de configuration, par exemple), utilisez un abonnement classique séparé pour ce topic.

L'application consommatrice

L'application se connecte avec un ID propre au pod, rejoint le groupe partagé et expose /healthz (processus vivant) et /readyz (connecté au broker) sur le port 8080.

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 entre pods
});

let ready = false;

client.on('connect', async () => {
  await client.subscribeAsync('$share/ingest/testmqtt/k8s/+/telemetry', { qos: 1 });
  ready = true;
  console.log(`[${pod}] connecté et abonné`);
});
client.on('close', () => { ready = false; });
client.on('error', (err) => console.error(`[${pod}] erreur mqtt :`, err.message));

client.on('message', (topic, payload) => {
  // Le vrai travail se fait ici : écrire en base, appeler une API, ...
  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 : déconnexion`);
  ready = false;
  await client.endAsync(); // attend les messages en vol puis envoie DISCONNECT
  process.exit(0);
});

Le reconnectPeriod aléatoire compte : quand le broker redémarre, cinquante pods qui se reconnectent au même instant créent un pic capable de le refaire tomber. Voir tempêtes de reconnexion pour les stratégies de backoff.

Règle 3 : des probes qui n'aggravent pas les pannes

• Readiness — renvoie 503 tant que la connexion MQTT est coupée. Kubernetes ne compte plus le pod comme disponible, et un rollout se met en pause au lieu de remplacer des pods sains par des pods déconnectés.
• Liveness — vérifie seulement que le processus répond. Si le broker est injoignable, redémarrer le pod ne règle rien, et redémarrer tous les pods à la fois transforme une panne en tempête de reconnexions.
• Startup (facultatif) — laisse du temps aux applications lentes à démarrer avant que la liveness n'intervienne.

Un worker sans trafic HTTP n'a pas besoin de Service ; la readiness probe pilote quand même les rollouts, et kubectl get pods montre d'un coup d'œil quels réplicas sont connectés.

Le manifeste du Deployment

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 }

Créez le Secret une fois avec kubectl create secret generic mqtt-credentials --from-literal=username="$MQTT_USER" --from-literal=password="$MQTT_PASS". N'intégrez jamais les identifiants dans l'image.

Arrêt propre et rolling updates

Quand un pod est arrêté, Kubernetes envoie SIGTERM, attend jusqu'à terminationGracePeriodSeconds, puis envoie SIGKILL. Le handler ci-dessus marque le pod comme non prêt, laisse MQTT.js terminer les acquittements en cours et envoie un DISCONNECT propre. Deux conséquences :

• Avec un DISCONNECT propre, le broker ne publie pas le Last Will du client. Réservez le will aux plantages, pas aux redémarrages planifiés — voir Last Will and Testament.
• En QoS 1, un message remis mais pas encore acquitté au moment où le pod meurt est redistribué à un autre membre du groupe (le comportement varie selon le broker). Rendez votre handler idempotent pour qu'un doublon soit sans conséquence.

maxUnavailable: 0 avec maxSurge: 1 signifie qu'un nouveau pod doit être prêt — donc connecté — avant qu'un ancien ne soit retiré : le groupe partagé ne rétrécit jamais pendant un rollout.

Où tourne le broker

Peu importe aux pods clients que le broker soit dans le cluster ou à l'extérieur ; ils ont seulement besoin d'un egress sur le bon port (8883 pour TLS). Faire tourner le broker dans le cluster ajoute un workload avec état à exploiter — Mosquitto sur Kubernetes montre ce que cela implique. Un broker managé comme ConnectMQ garde le cluster sans état : le même manifeste fonctionne dans tous les environnements, seuls l'URL et le Secret changent.

Si votre cluster applique une NetworkPolicy restrictive, autorisez explicitement l'egress vers l'hôte et le port du broker ; un port bloqué se traduit par des timeouts de connexion dans les logs et un pod qui ne devient jamais prêt.

Tester

1. Appliquez le manifeste et attendez que kubectl get pods -l app=ingest affiche trois pods prêts.
2. Ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com et publiez quelques messages sur testmqtt/k8s/dev-1/telemetry, testmqtt/k8s/dev-2/telemetry, etc.
3. Lancez kubectl logs -l app=ingest --prefix -f. Chaque message doit apparaître dans le log d'un seul pod, répartis entre les réplicas.
4. Lancez kubectl rollout restart deployment/ingest pendant que vous publiez. La readiness doit retenir le rollout jusqu'à ce que les nouveaux pods soient connectés, et aucun message ne doit manquer en QoS 1.
5. Passez à 1 réplica puis revenez à 3, et vérifiez que les messages se redistribuent.

Publier de la télémétrie de test depuis le navigateur

L'endpoint de test partagé ne conserve pas de sessions persistantes : pour tester des sessions de type StatefulSet avec des messages en file, refaites ces étapes sur votre propre instance de broker privée.

Kubernetes considère les pods comme interchangeables ; les brokers MQTT considèrent les client IDs comme des identités. Réconciliez les deux avec des IDs dérivés du pod, des shared subscriptions et des rollouts pilotés par la readiness, et vos consommateurs scaleront comme n'importe quel autre service sans état.

Questions fréquentes