Telegraf MQTT Consumer: Ingestão de Dados MQTT no InfluxDB

O plugin Telegraf MQTT consumer é a forma mais simples de levar mensagens MQTT para um banco de dados de séries temporais sem escrever código. Neste tutorial você vai configurar o inputs.mqtt_consumer para assinar topics no broker.connectmq.com, interpretar payloads JSON com timestamps e tags corretos, gravar o resultado no InfluxDB 2 e verificar cada etapa.

Como o Telegraf se encaixa em um pipeline MQTT

Um broker MQTT roteia mensagens; ele não armazena histórico. Para montar um gráfico das temperaturas da semana passada ou gerar um alerta a partir de uma tendência, algo precisa assinar os topics e gravar cada leitura em um banco de dados. O Telegraf é um pequeno agente da InfluxData que faz exatamente isso. Ele é configurado com um único arquivo TOML contendo inputs, processors opcionais e outputs.

O input mqtt_consumer é um plugin de serviço: ele mantém uma conexão persistente com o broker, recebe as mensagens conforme chegam, passa-as por um parser para produzir métricas e as entrega aos outputs, que as gravam em lotes. Do ponto de vista do broker, o Telegraf é só mais um cliente MQTT. Esse também é o padrão honesto para qualquer broker MQTT gerenciado, incluindo o ConnectMQ: o broker transporta as mensagens e uma ferramenta externa como o Telegraf cuida do armazenamento.

Pré-requisitos

Se você ainda não tem o InfluxDB, o caminho mais rápido é o Docker:

docker run -d --name influxdb -p 8086:8086 \
  -e DOCKER_INFLUXDB_INIT_MODE=setup \
  -e DOCKER_INFLUXDB_INIT_USERNAME=admin \
  -e DOCKER_INFLUXDB_INIT_PASSWORD=change-me-please \
  -e DOCKER_INFLUXDB_INIT_ORG=iot \
  -e DOCKER_INFLUXDB_INIT_BUCKET=sensors \
  -e DOCKER_INFLUXDB_INIT_ADMIN_TOKEN=my-super-secret-token \
  influxdb:2.7

Prefere um único comando para a stack inteira? O guia da stack TIG com Docker Compose roda Telegraf, InfluxDB e Grafana juntos.

Design de topics e payload

Este guia parte da estrutura de topics e do payload JSON usados nos nossos tutoriais de Python:

connectmq/demo/<run-id>/<line>/<device>/telemetry

{"device": "sensor-0042", "ts": 1760600000123, "temperature": 21.84, "humidity": 44.2}

Escolha o seu próprio run ID aleatório, por exemplo a1b2c3, para que os seus topics não colidam com os de mais ninguém no servidor MQTT público. Se quiser um fluxo contínuo em vez de mensagens de teste manuais, rode o simulador de dispositivos MQTT em Python, que publica exatamente nesse formato.

Uma configuração completa do Telegraf MQTT consumer

Salve o conteúdo a seguir como telegraf.conf:

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

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/telemetry"]
  qos = 0
  client_id = "telegraf-a1b2c3"
  connection_timeout = "30s"

  # Nome do measurement no InfluxDB (o padrão seria "mqtt_consumer")
  name_override = "sensor"
  # Mantém o topic completo como tag; defina "" para desativar
  topic_tag = "topic"

  # Interpreta payloads JSON
  data_format = "json"
  tag_keys = ["device"]
  json_time_key = "ts"
  json_time_format = "unix_ms"

  # Extrai a linha de produção do topic como uma tag extra
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/telemetry"
    tags = "_/_/_/line/_/_"

[[outputs.influxdb_v2]]
  urls = ["http://localhost:8086"]
  token = "${INFLUX_TOKEN}"
  organization = "iot"
  bucket = "sensors"

# Temporário: também imprime as métricas no stdout durante os testes
[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

O que cada configuração faz

• servers usa tcp:// para MQTT sem criptografia na porta 1883. Para TLS, você troca para ssl:// e a porta 8883, como mostrado mais abaixo.
• topics aceita curingas MQTT. Os dois níveis + correspondem a qualquer linha e qualquer dispositivo, enquanto o sufixo fixo telemetry exclui outros tipos de mensagem.
• client_id deve ser definido explicitamente. Sem ele, o Telegraf gera um ID aleatório a cada inicialização, o que impede sessões persistentes e dificulta identificar o cliente nos logs do broker.
• name_override define o nome do measurement no InfluxDB. Sem ele, todas as métricas vão para um measurement chamado mqtt_consumer.
• tag_keys transforma chaves JSON em tags. Tags são indexadas e adequadas para valores pelos quais você filtra ou agrupa, como IDs de dispositivos. Leituras numéricas viram fields automaticamente.
• json_time_key e json_time_format usam o timestamp do dispositivo em vez do momento em que o Telegraf recebeu a mensagem. unix_ms corresponde aos nossos timestamps em milissegundos.
• topic_parsing mapeia segmentos do topic para tags. Cada _ pula um nível e os segmentos nomeados viram tags, então line recebe o quarto nível do topic.

Uma palavra sobre a tag topic: ela é prática durante a depuração, mas cada topic distinto cria uma nova série no InfluxDB. Com algumas centenas de dispositivos isso é inofensivo. Com dezenas de milhares de topics, ou topics que contêm IDs de mensagens ou timestamps, a cardinalidade de séries cresce rápido e deixa as consultas lentas. Quando as tags line e device já carregam a mesma informação, definir topic_tag = "" mantém o banco enxuto.

Defina o token como variável de ambiente e rode o Telegraf primeiro em modo de teste. Como o mqtt_consumer é um input de serviço, o modo de teste precisa de um período de espera para coletar mensagens:

export INFLUX_TOKEN=my-super-secret-token
telegraf --config telegraf.conf --test --test-wait 20

Enquanto ele roda, publique uma mensagem de teste a partir de outro terminal:

mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-a1b2c3 \
  -t connectmq/demo/a1b2c3/line1/sensor-0001/telemetry \
  -m "{\"device\":\"sensor-0001\",\"ts\":$(date +%s%3N),\"temperature\":21.7,\"humidity\":44.1}"

A chamada date +%s%3N gera um timestamp em milissegundos com o GNU date no Linux; no macOS, instale o coreutils e use gdate, ou digite um valor recente fixo. O Telegraf deve imprimir uma linha de line protocol do InfluxDB parecida com esta:

sensor,device=sensor-0001,line=line1,topic=connectmq/demo/a1b2c3/line1/sensor-0001/telemetry humidity=44.1,temperature=21.7 1760600000123000000

Quando isso funcionar, inicie o Telegraf normalmente com telegraf --config telegraf.conf, remova o bloco outputs.file assim que confiar no pipeline e consulte o InfluxDB para confirmar que os dados chegaram:

influx query 'from(bucket: "sensors") |> range(start: -15m) |> filter(fn: (r) => r._measurement == "sensor") |> limit(n: 5)' \
  --org iot --token my-super-secret-token

Protegendo a conexão para produção

O servidor MQTT público serve apenas para experimentos; tudo o que está nele pode ser lido por outras pessoas. Para dados reais, como a telemetria de uma usina ou de uma planta de saneamento, use um broker privado com credenciais e TLS. Com um broker MQTT gerenciado como o ConnectMQ, o bloco de input muda assim:

[[inputs.mqtt_consumer]]
  servers = ["ssl://broker.connectmq.com:8883"]
  topics = ["connectmq/factory1/production/+/+/telemetry"]
  qos = 1
  client_id = "YOUR_CLIENT_ID"
  username = "${MQTT_USERNAME}"
  password = "${MQTT_PASSWORD}"
  persistent_session = true
  max_undelivered_messages = 1000

  name_override = "sensor"
  data_format = "json"
  tag_keys = ["device"]
  json_time_key = "ts"
  json_time_format = "unix_ms"

O broker apresenta um certificado Let's Encrypt, no qual o repositório de CAs do sistema operacional confia, então nenhuma configuração tls_ca é necessária. persistent_session = true, combinado com QoS 1 e um client ID fixo, pede ao broker que mantenha a sessão do Telegraf e enfileire as mensagens durante reinícios curtos do Telegraf. Guarde as credenciais em variáveis de ambiente, e não no arquivo de configuração, principalmente se o arquivo estiver no controle de versão.

Solução de problemas do Telegraf MQTT consumer

Rode o Telegraf com --debug para ver os eventos de conexão e os erros de parse à medida que acontecem. Para payloads com objetos aninhados ou arrays, o parser JSON legado logo chega aos seus limites; o guia do parser json_v2 do Telegraf mostra como mapear estruturas aninhadas e segmentos de topic com precisão. A lista completa de opções está na documentação oficial do plugin mqtt_consumer.

Próximos passos

Com os dados fluindo para o InfluxDB, adicione o Grafana para dashboards e alertas. Se você prefere código a configuração, a ponte Python para InfluxDB faz o mesmo trabalho em um script curto, com controle total sobre o mapeamento. E, se ainda estiver escolhendo as ferramentas de cliente para os próprios dispositivos, veja o guia de bibliotecas de cliente MQTT.

Comece a usar o ConnectMQ

Você pode testar toda esta configuração no servidor MQTT público em poucos minutos. Para dados de produção, o ConnectMQ oferece um broker privado com TLS na porta 8883, MQTT 5 e um portal com explorador de topics ao vivo para conferir o que o Telegraf vai receber. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito, e cole as credenciais geradas na configuração do Telegraf. Se estiver planejando uma implantação maior, leia mais sobre como operar um broker MQTT na nuvem.

Perguntas frequentes

O Telegraf MQTT consumer suporta MQTT 5?

As versões recentes do Telegraf adicionam uma opção protocol para selecionar o MQTT 5. O padrão é o MQTT 3.1.1, suficiente para assinar e gravar métricas na maioria dos cenários.

Uma única instância do Telegraf pode assinar vários topics?

Sim. Liste vários filtros em topics ou adicione vários blocos mqtt_consumer com parsers diferentes quando os topics carregarem formatos de payload diferentes.

Por que as minhas strings JSON não são armazenadas?

O parser json legado ignora valores string, a menos que você os liste em json_string_fields ou tag_keys. O parser json_v2 trata strings de forma explícita.

O Telegraf deve usar QoS 0 ou QoS 1?

QoS 0 serve para telemetria de alta frequência, em que perder uma leitura ocasional não importa. Use QoS 1 com uma sessão persistente quando todas as leituras precisarem ser armazenadas.