Ejecutar aplicaciones cliente MQTT en Kubernetes

Ejecutar un cliente MQTT en Kubernetes parece un Deployment cualquiera hasta que lo escalas a dos réplicas. Entonces los pods empiezan a echarse mutuamente del broker, cada réplica procesa cada mensaje y los rolling updates pierden datos. Tres reglas lo arreglan casi todo: da a cada pod un client ID único, usa shared subscriptions para repartir el trabajo entre réplicas y conecta el estado MQTT a la readiness, no a la liveness.

Este tutorial construye un pequeño consumidor en Node.js (MQTT.js 5) y un manifiesto de Deployment que aplica las tres reglas, además de un apagado limpio y una prueba que puedes hacer desde el navegador.

Por qué los clientes MQTT se comportan distinto en Kubernetes

Los servicios HTTP no tienen estado: cualquier pod puede responder a cualquier petición y el Service reparte las conexiones. Los clientes MQTT son lo contrario. Cada pod abre una única conexión de larga duración hacia el broker, y es el broker —no Kubernetes— quien decide qué cliente recibe cada mensaje.

Regla 1: un client ID por pod

MQTT permite exactamente una sesión activa por client ID. Cuando llega una segunda conexión con el mismo ID, el broker cierra la primera (reason code MQTT 5 0x8E Session taken over). Con dos réplicas configuradas igual, se desconectan por turnos para siempre; consulta session taken over para ver los síntomas.

La solución más limpia es la downward API: inyecta el nombre del pod como variable de entorno y construye el ID a partir de él. Los nombres de pod son únicos dentro del namespace y estables durante la vida del pod.

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

En la aplicación, clientId = "ingest-" + process.env.POD_NAME. Mantén los IDs cortos y con los caracteres que acepte tu broker; el generador de client ID y las buenas prácticas de client ID explican los límites.

Regla 2: las shared subscriptions reparten el trabajo

Con una suscripción normal, tres réplicas del mismo consumidor reciben cada una todos los mensajes, así que guardas cada lectura tres veces. Una shared subscription agrupa las réplicas y el broker entrega cada mensaje a un solo miembro:

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

Todos los pods se suscriben al mismo filtro $share/ingest/.... Escalar el Deployment hacia arriba añade miembros al grupo; escalarlo hacia abajo los quita, y el broker rebalancea solo. Las shared subscriptions son una funcionalidad de MQTT 5; muchos brokers también aceptan el prefijo $share/ desde clientes MQTT 3.1.1, pero compruébalo en el tuyo. La mecánica completa, incluido cómo elige el broker un miembro, está en MQTT shared subscriptions.

Ten en cuenta dos cosas: el orden solo está garantizado por miembro, no en todo el grupo, y si un mensaje debe llegar a todos los pods (por ejemplo, recargar la configuración) usa una suscripción normal aparte para ese topic.

La aplicación consumidora

La aplicación se conecta con un ID específico del pod, se une al grupo compartido y expone /healthz (proceso vivo) y /readyz (conectado al broker) en el puerto 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}] conectado y suscrito`);
});
client.on('close', () => { ready = false; });
client.on('error', (err) => console.error(`[${pod}] error mqtt:`, err.message));

client.on('message', (topic, payload) => {
  // Aquí va el trabajo real: escribir en una base de datos, llamar a una 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: desconectando`);
  ready = false;
  await client.endAsync(); // espera a los mensajes en vuelo y envía DISCONNECT
  process.exit(0);
});

El reconnectPeriod aleatorio importa: cuando el broker se reinicia, cincuenta pods reconectando en el mismo instante generan un pico que puede volver a tumbarlo. Consulta tormentas de reconexión para estrategias de backoff.

Regla 3: probes que no empeoran las caídas

• Readiness: devuelve 503 mientras la conexión MQTT está caída. Kubernetes deja de contar el pod como disponible, así que un rollout se pausa en lugar de sustituir pods sanos por pods desconectados.
• Liveness: solo comprueba que el proceso responde. Si el broker no es alcanzable, reiniciar el pod no lo arregla, y reiniciar todos los pods a la vez convierte una caída en una tormenta de reconexiones.
• Startup (opcional): da tiempo a las aplicaciones lentas en arrancar antes de que actúe la liveness.

Un worker sin tráfico HTTP no necesita un Service; la readiness probe sigue controlando los rollouts y kubectl get pods muestra de un vistazo qué réplicas están conectadas.

El manifiesto del 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 }

Crea el Secret una vez con kubectl create secret generic mqtt-credentials --from-literal=username="$MQTT_USER" --from-literal=password="$MQTT_PASS". Nunca metas las credenciales en la imagen.

Apagado limpio y rolling updates

Cuando se termina un pod, Kubernetes envía SIGTERM, espera hasta terminationGracePeriodSeconds y después envía SIGKILL. El handler de arriba marca el pod como no listo, deja que MQTT.js termine las confirmaciones en vuelo y envía un DISCONNECT limpio. Dos consecuencias:

• Con un DISCONNECT limpio, el broker no publica el Last Will del cliente. Usa el will para caídas, no para reinicios planificados; consulta Last Will and Testament.
• Con QoS 1, un mensaje entregado pero aún no confirmado cuando murió el pod se reenvía a otro miembro del grupo (el comportamiento varía según el broker). Haz que tu handler sea idempotente para que un duplicado no cause daño.

maxUnavailable: 0 con maxSurge: 1 significa que un pod nuevo debe estar listo —es decir, conectado— antes de retirar uno antiguo, así que el grupo compartido nunca se reduce durante un rollout.

Dónde se ejecuta el broker

A los pods cliente les da igual que el broker esté dentro o fuera del clúster; solo necesitan salida por el puerto correcto (8883 para TLS). Ejecutar el broker en el clúster te deja otro workload con estado que operar; Mosquitto en Kubernetes muestra lo que implica. Un broker gestionado como ConnectMQ mantiene el clúster sin estado: el mismo manifiesto funciona en todos los entornos y solo cambian la URL y el Secret.

Si tu clúster tiene una NetworkPolicy restrictiva, permite explícitamente la salida al host y puerto del broker; un puerto bloqueado aparece como timeouts de conexión en los logs y un pod que nunca llega a estar listo.

Probarlo

1. Aplica el manifiesto y espera a que kubectl get pods -l app=ingest muestre tres pods listos.
2. Abre el cliente MQTT online, conéctate a broker.connectmq.com y publica algunos mensajes en testmqtt/k8s/dev-1/telemetry, testmqtt/k8s/dev-2/telemetry, etc.
3. Ejecuta kubectl logs -l app=ingest --prefix -f. Cada mensaje debe aparecer en el log de exactamente un pod, repartidos entre las réplicas.
4. Ejecuta kubectl rollout restart deployment/ingest mientras publicas. La readiness debe retener el rollout hasta que los pods nuevos estén conectados, y con QoS 1 no debe perderse ningún mensaje.
5. Escala a 1 y de nuevo a 3 y confirma que los mensajes se redistribuyen.

Publica telemetría de prueba desde tu navegador

El endpoint de pruebas compartido no guarda sesiones persistentes, así que para probar sesiones al estilo StatefulSet con mensajes encolados repite los pasos contra tu propia instancia privada de broker.

Kubernetes trata los pods como intercambiables; los brokers MQTT tratan los client IDs como identidades. Une ambos mundos con IDs derivados del pod, shared subscriptions y rollouts guiados por la readiness, y tus consumidores escalarán como cualquier otro servicio sin estado.

Preguntas frecuentes