MQTT para InfluxDB com Telegraf: guia de configuração do mqtt_consumer

O caminho mais curto de um broker MQTT até um banco de séries temporais é o Telegraf: o input mqtt_consumer assina seus topics, um parser transforma cada payload em uma métrica e o output influxdb_v2 grava no InfluxDB. Sem código, um arquivo de configuração, e ele roda tranquilo ao lado do banco ou em um gateway de borda.

Este guia monta um pipeline funcional com Docker Compose, interpreta payloads JSON com json_v2, mapeia partes do topic para tags e mostra como comprovar cada etapa antes de apontar dispositivos reais para ele.

O pipeline em resumo

O dispositivo de exemplo publica JSON como este em testmqtt/telegraf/sensor-01/telemetry:

{"ts": 1760700000000, "site": "lab", "temperature": 21.4, "humidity": 40.2, "status": "ok"}

Suba o InfluxDB e o Telegraf

Este arquivo Compose inicia o InfluxDB 2.x com organização, bucket e token iniciais, e o Telegraf com a sua configuração montada como somente leitura:

services:
  influxdb:
    image: influxdb:2.7
    ports: ["8086:8086"]
    environment:
      DOCKER_INFLUXDB_INIT_MODE: setup
      DOCKER_INFLUXDB_INIT_USERNAME: admin
      DOCKER_INFLUXDB_INIT_PASSWORD: change-me-please
      DOCKER_INFLUXDB_INIT_ORG: testmqtt
      DOCKER_INFLUXDB_INIT_BUCKET: mqtt
      DOCKER_INFLUXDB_INIT_ADMIN_TOKEN: dev-token-change-me
    volumes:
      - influx-data:/var/lib/influxdb2

  telegraf:
    image: telegraf:1.32
    depends_on: [influxdb]
    environment:
      MQTT_USER: ${MQTT_USER}
      MQTT_PASS: ${MQTT_PASS}
      INFLUX_TOKEN: dev-token-change-me
    volumes:
      - ./telegraf.conf:/etc/telegraf/telegraf.conf:ro

volumes:
  influx-data:

As variáveis de inicialização só valem na primeira inicialização com volume vazio. Em produção, crie um token dedicado, só de escrita, para o Telegraf em vez de reutilizar o token de admin. Para um stack local maior, com broker e dashboards, veja o stack MQTT com Docker Compose.

Configure o mqtt_consumer

[agent]
  interval = "10s"
  flush_interval = "5s"
  metric_batch_size = 1000
  metric_buffer_limit = 50000
  omit_hostname = true

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.connectmq.com:8883"]
  topics = ["testmqtt/telegraf/+/telemetry"]
  qos = 1
  client_id = "telegraf-ingest-01"
  persistent_session = true
  username = "${MQTT_USER}"
  password = "${MQTT_PASS}"
  max_undelivered_messages = 1000
  topic_tag = ""
  data_format = "json_v2"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "testmqtt/telegraf/+/telemetry"
    tags = "_/_/device/_"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "telemetry"
    timestamp_path = "ts"
    timestamp_format = "unix_ms"

    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "site"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "temperature"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "humidity"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "status"
      type = "string"

[[outputs.influxdb_v2]]
  urls = ["http://influxdb:8086"]
  token = "${INFLUX_TOKEN}"
  organization = "testmqtt"
  bucket = "mqtt"

Os ajustes que mais importam:

• servers: tcp://host:1883 sem criptografia, ssl://host:8883 para TLS. O Telegraf valida o certificado do servidor contra o repositório de CAs do sistema; adicione tls_ca = "/path/ca.pem" para uma CA privada.
• topics: filtros MQTT padrão, então + e # funcionam. Teste um filtro contra nomes de topic reais com o topic matcher. Mais sobre filtros: topics e wildcards.
• qos = 1 + persistent_session = true + um client_id fixo: o broker mantém uma sessão para o Telegraf e enfileira mensagens QoS 1 enquanto ele reinicia. Sem um client ID estável, a sessão não pode ser retomada. Veja sessões persistentes.
• max_undelivered_messages: quantas mensagens QoS 1/2 o Telegraf lê antes de elas serem gravadas no output. É o backpressure do Telegraf: se o InfluxDB ficar lento, o Telegraf para de confirmar e o broker segura o restante.
• topic_tag = "": desativa a tag topic padrão. Guardar o topic completo como tag multiplica a cardinalidade de séries; o topic_parsing extrai só a parte de que você precisa.

Como o json_v2 mapeia o payload

Com a configuração acima, o payload de exemplo vira este line protocol:

telemetry,device=sensor-01,site=lab temperature=21.4,humidity=40.2,status="ok" 1760700000000000000

Duas regras mantêm o InfluxDB saudável: tags para o que você filtra e tem poucos valores distintos, fields para medições. E sempre envie números como números JSON: "21.4" entre aspas é texto e será descartado ou criará um conflito de tipo no field. Se os payloads às vezes chegam malformados, leia erros de parsing JSON no MQTT.

Se os dispositivos não enviam timestamp, remova timestamp_path e o Telegraf usa a hora de chegada. Isso serve para dados ao vivo, mas dados que chegam atrasados de dispositivos que estavam offline vão ficar com horário tardio.

Verifique cada etapa

Teste o pipeline na ordem em que os dados fluem, assim você sempre sabe qual etapa está falhando.

1. Os dados estão no broker? Assine testmqtt/telegraf/# no cliente MQTT online e publique uma leitura de teste:

mosquitto_pub -h broker.connectmq.com -p 8883 --capath /etc/ssl/certs \
  -u "$MQTT_USER" -P "$MQTT_PASS" -q 1 \
  -t testmqtt/telegraf/sensor-01/telemetry \
  -m "{\"ts\":$(date +%s000),\"site\":\"lab\",\"temperature\":21.4,\"humidity\":40.2,\"status\":\"ok\"}"

2. O Telegraf interpreta? Rode o Telegraf em modo de teste. Ele imprime as métricas no stdout em vez de gravá-las; --test-wait mantém o input MQTT escutando tempo suficiente para receber mensagens:

docker compose run --rm telegraf telegraf --config /etc/telegraf/telegraf.conf --test --test-wait 15

Publique de novo enquanto ele espera. Você deve ver a linha telemetry,device=sensor-01,.... Nenhuma linha significa que a assinatura ou o parser está errado; uma linha com campos faltando significa que os paths ou tipos do json_v2 não batem.

3. Está no InfluxDB? Suba o stack com docker compose up -d e consulte o bucket:

docker compose exec influxdb influx query --org testmqtt \
  'from(bucket: "mqtt") |> range(start: -10m) |> filter(fn: (r) => r._measurement == "telemetry") |> last()'

Acompanhe o topic de telemetria no navegador

Erros comuns

Consulte os reason codes do broker com a ferramenta de reason codes. Erros do parser aparecem com o prefixo E! [inputs.mqtt_consumer] e geralmente indicam que o payload não é JSON válido ou que um campo declarado tem outro tipo. Rode o Telegraf com --debug para ver detalhes de conexão e de flush.

Escala e retenção

• Mais throughput: o Telegraf processa muitas mensagens por instância, mas se você precisar de várias instâncias, use uma assinatura compartilhada como $share/telegraf/testmqtt/telegraf/+/telemetry para que o broker divida o fluxo em vez de duplicá-lo. Cada instância precisa de seu próprio client_id.
• Retenção: defina um período de retenção no bucket (pela interface ou com influx bucket update --id <bucket-id> --retention 30d) e reduza a resolução dos dados antigos com uma task do InfluxDB. Como planejar por quanto tempo guardar a telemetria bruta está em retenção e armazenamento de dados MQTT.
• Visualização: aponte o Grafana para o InfluxDB para o histórico e adicione o MQTT data source do Grafana se também quiser um painel ao vivo direto do broker.

O Telegraf transforma "MQTT para InfluxDB" em uma revisão de configuração em vez de um projeto de software. Escolha bem suas tags, use QoS 1 com sessão persistente no seu próprio broker, valide com --test antes de ir para produção e mantenha a configuração no controle de versão junto com seus dashboards.

Perguntas frequentes