Rodando aplicações cliente MQTT no Kubernetes

Rodar um cliente MQTT no Kubernetes parece um Deployment qualquer até você escalar para duas réplicas. Aí os pods começam a derrubar uns aos outros do broker, cada réplica processa todas as mensagens e os rolling updates perdem dados. Três regras resolvem quase tudo: dê a cada pod um client ID único, use shared subscriptions para dividir o trabalho entre réplicas e ligue a conexão MQTT à readiness, não à liveness.

Este tutorial monta um pequeno consumidor em Node.js (MQTT.js 5) e um manifesto de Deployment que aplica as três regras, além de desligamento limpo e um teste que você pode rodar pelo navegador.

Por que clientes MQTT se comportam diferente no Kubernetes

Serviços HTTP não têm estado: qualquer pod responde a qualquer requisição e o Service balanceia as conexões. Clientes MQTT são o oposto. Cada pod abre uma única conexão de longa duração para o broker, e é o broker — não o Kubernetes — que decide qual cliente recebe cada mensagem.

Regra 1: um client ID por pod

O MQTT permite exatamente uma sessão ativa por client ID. Quando chega uma segunda conexão com o mesmo ID, o broker fecha a primeira (reason code MQTT 5 0x8E Session taken over). Com duas réplicas configuradas iguais, elas se derrubam alternadamente para sempre — veja session taken over para os sintomas.

A solução mais limpa é a downward API: injete o nome do pod como variável de ambiente e monte o ID a partir dele. Nomes de pod são únicos no namespace e estáveis durante a vida do pod.

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

Na aplicação, clientId = "ingest-" + process.env.POD_NAME. Mantenha os IDs curtos e com caracteres que o seu broker aceita; o gerador de client ID e as boas práticas de client ID explicam os limites.

Regra 2: shared subscriptions dividem o trabalho

Com uma assinatura normal, três réplicas do mesmo consumidor recebem cada uma todas as mensagens, e você grava cada leitura três vezes. Uma shared subscription coloca as réplicas em um grupo, e o broker entrega cada mensagem a apenas um membro:

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

Todos os pods assinam o mesmo filtro $share/ingest/.... Escalar o Deployment para cima adiciona membros ao grupo; escalar para baixo remove, e o broker rebalanceia sozinho. Shared subscriptions são um recurso do MQTT 5; muitos brokers também aceitam o prefixo $share/ de clientes MQTT 3.1.1, mas confira no seu. A mecânica completa, incluindo como o broker escolhe um membro, está em MQTT shared subscriptions.

Lembre-se de duas coisas: a ordem só é garantida por membro, não no grupo inteiro, e se uma mensagem precisa chegar a todos os pods (um reload de configuração, por exemplo), use uma assinatura normal separada para esse topic.

A aplicação consumidora

A aplicação conecta com um ID específico do pod, entra no grupo compartilhado e expõe /healthz (processo vivo) e /readyz (conectado ao broker) na porta 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 e inscrito`);
});
client.on('close', () => { ready = false; });
client.on('error', (err) => console.error(`[${pod}] erro mqtt:`, err.message));

client.on('message', (topic, payload) => {
  // O trabalho real vai aqui: gravar no banco, chamar uma 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 as mensagens em trânsito e envia DISCONNECT
  process.exit(0);
});

O reconnectPeriod aleatório importa: quando o broker reinicia, cinquenta pods reconectando no mesmo instante geram um pico que pode derrubá-lo de novo. Veja tempestades de reconexão para estratégias de backoff.

Regra 3: probes que não pioram as quedas

• Readiness — retorna 503 enquanto a conexão MQTT está fora. O Kubernetes deixa de contar o pod como disponível, então um rollout pausa em vez de trocar pods saudáveis por pods desconectados.
• Liveness — só verifica se o processo responde. Se o broker estiver inacessível, reiniciar o pod não resolve, e reiniciar todos os pods de uma vez transforma uma queda em uma tempestade de reconexões.
• Startup (opcional) — dá tempo para aplicações que demoram a subir antes de a liveness entrar em ação.

Um worker sem tráfego HTTP não precisa de Service; a readiness probe continua controlando os rollouts e kubectl get pods mostra de relance quais réplicas estão conectadas.

O manifesto do 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 }

Crie o Secret uma vez com kubectl create secret generic mqtt-credentials --from-literal=username="$MQTT_USER" --from-literal=password="$MQTT_PASS". Nunca coloque credenciais dentro da imagem.

Desligamento limpo e rolling updates

Quando um pod é encerrado, o Kubernetes envia SIGTERM, espera até terminationGracePeriodSeconds e então envia SIGKILL. O handler acima marca o pod como não pronto, deixa o MQTT.js concluir as confirmações em trânsito e envia um DISCONNECT limpo. Duas consequências:

• Com um DISCONNECT limpo, o broker não publica o Last Will do cliente. Use o will para falhas, não para reinícios planejados — veja Last Will and Testament.
• Com QoS 1, uma mensagem entregue mas ainda não confirmada quando o pod morreu é reenviada a outro membro do grupo (o comportamento varia por broker). Deixe seu handler idempotente para que um duplicado não cause problema.

maxUnavailable: 0 com maxSurge: 1 significa que um pod novo precisa estar pronto — ou seja, conectado — antes de um antigo ser removido, então o grupo compartilhado nunca encolhe durante um rollout.

Onde o broker roda

Para os pods cliente tanto faz se o broker está dentro ou fora do cluster; eles só precisam de saída na porta certa (8883 para TLS). Rodar o broker no cluster é mais um workload com estado para operar — Mosquitto no Kubernetes mostra o que isso envolve. Um broker gerenciado como o ConnectMQ mantém o cluster sem estado: o mesmo manifesto funciona em todos os ambientes, só mudam a URL e o Secret.

Se o seu cluster tem uma NetworkPolicy restritiva, libere explicitamente a saída para o host e a porta do broker; uma porta bloqueada aparece como timeouts de conexão nos logs e um pod que nunca fica pronto.

Testando

1. Aplique o manifesto e espere kubectl get pods -l app=ingest mostrar três pods prontos.
2. Abra o cliente MQTT online, conecte em broker.connectmq.com e publique algumas mensagens em testmqtt/k8s/dev-1/telemetry, testmqtt/k8s/dev-2/telemetry e assim por diante.
3. Rode kubectl logs -l app=ingest --prefix -f. Cada mensagem deve aparecer no log de exatamente um pod, distribuídas entre as réplicas.
4. Rode kubectl rollout restart deployment/ingest enquanto publica. A readiness deve segurar o rollout até os novos pods estarem conectados, e nenhuma mensagem deve sumir com QoS 1.
5. Escale para 1 e de volta para 3 e confirme que as mensagens se redistribuem.

Publique telemetria de teste pelo navegador

O endpoint de teste compartilhado não mantém sessões persistentes; para testar sessões no estilo StatefulSet com mensagens enfileiradas, repita os passos na sua própria instância privada de broker.

O Kubernetes trata pods como intercambiáveis; brokers MQTT tratam client IDs como identidades. Una os dois com IDs derivados do pod, shared subscriptions e rollouts guiados pela readiness, e seus consumidores escalam como qualquer outro serviço sem estado.

Perguntas frequentes