Parser json_v2 do Telegraf e Interpretação de Topics MQTT Explicados

O parser json_v2 do Telegraf permite decidir exatamente quais partes de um payload MQTT se tornam measurements, tags e fields, enquanto a interpretação de topics extrai identificadores do próprio topic. Neste tutorial você vai configurar os dois no input mqtt_consumer do Telegraf, partindo de um payload aninhado realista, e verificar a saída linha por linha antes que qualquer coisa chegue ao InfluxDB.

Por que o parser JSON legado não basta

O parser json original do Telegraf achata tudo o que encontra em fields, une chaves aninhadas com sublinhados, ignora strings a menos que você as liste e oferece controle limitado sobre os tipos. Isso é aceitável para payloads planos como {"temperature": 21.5}. Payloads industriais e de IoT raramente são tão simples. Um gateway de máquina normalmente publica algo assim:

{
  "ts": 1760600000123,
  "device": { "id": "cnc1", "fw": "2.4.1", "serial": "SN-88231" },
  "status": "running",
  "metrics": { "spindle_rpm": 11800, "temp_c": 41.7, "vibration_mm_s": 2.3 },
  "alarms": []
}

Você quer o firmware como tag, o status como field de string, as três métricas como fields com o tipo correto, o número de série ignorado e o timestamp do dispositivo usado como horário do ponto. O parser json_v2 expressa cada uma dessas decisões de forma explícita com a sintaxe de caminhos GJSON, então nada vai parar no banco de dados por acidente.

O cenário de exemplo

As mensagens chegam em um topic no estilo ISA-95:

connectmq/demo/a1b2c3/factory1/line1/cnc1/telemetry

São sete níveis, dentro do limite de oito do broker público. O topic já carrega site, linha e máquina, então a configuração vai obtê-los do topic em vez de depender do payload.

Publique uma mensagem de teste no broker.connectmq.com com o seu próprio ID aleatório no lugar de a1b2c3:

mosquitto_pub -h broker.connectmq.com -p 1883 -i pub-a1b2c3 \
  -t connectmq/demo/a1b2c3/factory1/line1/cnc1/telemetry \
  -m "{\"ts\":$(date +%s%3N),\"device\":{\"id\":\"cnc1\",\"fw\":\"2.4.1\",\"serial\":\"SN-88231\"},\"status\":\"running\",\"metrics\":{\"spindle_rpm\":11800,\"temp_c\":41.7,\"vibration_mm_s\":2.3},\"alarms\":[]}"

Configuração do parser json_v2 do Telegraf

Salve isto como json_v2.conf:

[agent]
  omit_hostname = true

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/+/telemetry"]
  client_id = "telegraf-json-a1b2c3"
  qos = 0
  topic_tag = ""
  data_format = "json_v2"

  # Site, linha e máquina vêm do topic
  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/telemetry"
    tags = "_/_/_/site/line/asset/_"

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

    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "device.fw"
      rename = "firmware"

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

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.spindle_rpm"
      rename = "spindle_rpm"
      type = "int"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.temp_c"
      rename = "temp_c"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "metrics.vibration_mm_s"
      rename = "vibration_mm_s"
      type = "float"

    [[inputs.mqtt_consumer.json_v2.field]]
      path = "alarms.#"
      rename = "alarm_count"
      type = "int"

[[outputs.file]]
  files = ["stdout"]
  data_format = "influx"

Execute em modo de teste. O mqtt_consumer é um service input, então dê tempo para ele receber uma mensagem e publique a mensagem de teste enquanto ele aguarda:

telegraf --config json_v2.conf --test --test-wait 20

A saída é uma única linha em line protocol:

machine,asset=cnc1,firmware=2.4.1,line=line1,site=factory1 status="running",spindle_rpm=11800i,temp_c=41.7,vibration_mm_s=2.3,alarm_count=0i 1760600000123000000

Como o mapeamento funciona

Os caminhos usam sintaxe GJSON

Cada path é uma expressão GJSON avaliada contra o payload. Um ponto entra em objetos aninhados, então metrics.temp_c lê a temperatura dentro de metrics. O modificador # retorna o tamanho de um array, e é por isso que alarms.# produz uma contagem de alarmes sem armazenar o array em si. O GJSON também suporta índices de array como readings.0.value e consultas como readings.#(sensor=="t1").value quando você precisa escolher um elemento.

Tags, fields e tipos

Declarar os tipos explicitamente evita um problema clássico do InfluxDB: um field gravado uma vez como inteiro e depois como float é rejeitado como conflito de tipo. Escolha um tipo por field e mantenha-o para sempre.

Sem rename, um caminho aninhado vira um nome de field com o caminho achatado, o que tende a produzir nomes longos e desajeitados. Renomear mantém o esquema legível em consultas e dashboards.

A interpretação do topic preenche a hierarquia

O bloco topic_parsing divide o topic por / e o alinha com o padrão em tags. Cada _ pula um nível e cada nome cria uma tag a partir daquela posição. O padrão precisa ter exatamente tantos segmentos quanto o topic, e o filtro topic dentro do bloco decide a quais mensagens a regra se aplica, com os curingas + e # permitidos.

A interpretação do topic também pode definir o measurement e criar fields. Aqui está a mesma regra obtendo o measurement do último nível do topic em vez do measurement_name fixo:

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/+"
    measurement = "_/_/_/_/_/_/measurement"
    tags = "_/_/_/site/line/asset/_"

Com essa regra, mensagens em .../cnc1/telemetry vão para um measurement telemetry e mensagens em .../cnc1/energy para um measurement energy. A interpretação do topic é executada depois do parser do payload, então o measurement vindo do topic substitui o de measurement_name. Para valores numéricos carregados no próprio topic, use fields junto com uma tabela [inputs.mqtt_consumer.topic_parsing.types] para definir os tipos, mas raramente esse é um bom design de topic.

Convertendo arrays em várias métricas

Gateways costumam agrupar várias leituras em uma única mensagem:

{
  "gateway": "gw-07",
  "readings": [
    { "sensor": "t1", "ts": 1760600000100, "value": 21.3 },
    { "sensor": "t2", "ts": 1760600000105, "value": 22.0 },
    { "sensor": "h1", "ts": 1760600000110, "value": 48.5 }
  ]
}

Um bloco object transforma cada elemento do array em uma métrica própria. Coloque-o em um input mqtt_consumer separado que escute o topic de lote, para que ele nunca seja executado contra payloads sem um array readings:

[[inputs.mqtt_consumer]]
  servers = ["tcp://broker.connectmq.com:1883"]
  topics = ["connectmq/demo/a1b2c3/+/+/+/batch"]
  client_id = "telegraf-batch-a1b2c3"
  topic_tag = ""
  data_format = "json_v2"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic = "connectmq/demo/a1b2c3/+/+/+/batch"
    tags = "_/_/_/site/line/gateway/_"

  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "reading"

    [[inputs.mqtt_consumer.json_v2.object]]
      path = "readings"
      tags = ["sensor"]
      timestamp_key = "ts"
      timestamp_format = "unix_ms"
      [inputs.mqtt_consumer.json_v2.object.fields]
        value = "float"

Cada elemento produz uma linha como reading,gateway=gw-07,line=line1,sensor=t1,site=factory1 value=21.3 1760600000100000000. A lista tags promove chaves a tags, timestamp_key lê o horário de cada elemento e a tabela fields fixa os tipos. Observe que a tag do gateway vem do topic, o que mantém o bloco object focado apenas no array.

Checklist de solução de problemas

1. Nenhuma saída no modo de teste. Confirme o topic com mosquitto_sub -h broker.connectmq.com -p 1883 -i dbg-a1b2c3 -t "connectmq/demo/a1b2c3/#" -v e certifique-se de publicar dentro da janela do --test-wait.
2. Field ausente na saída. Confira o caminho GJSON contra o payload real; um erro de digitação em um caminho simplesmente não retorna nada. Versões mais novas do Telegraf aceitam optional = true em fields que nem sempre estão presentes.
3. Erro sobre segmentos na interpretação do topic. Conte os níveis no filtro topic e no padrão tags; eles precisam coincidir.
4. Timestamps em 1970. O payload usa segundos, mas a configuração diz unix_ms, ou o contrário.
5. Erros de conflito de tipo no InfluxDB. Uma configuração anterior gravou o mesmo field com outro tipo. Use um novo nome de field ou um novo bucket ao mudar tipos.
6. Todos os valores viram string. Um type de field foi definido como string ou o dispositivo envia números entre aspas, como "41.7". Definir type = "float" converte números entre aspas.
7. Séries demais. Um valor de alta cardinalidade, como número de série ou ID de mensagem, foi mapeado para uma tag. Transforme-o em field ou descarte-o.

Rodando em produção

Quando o mapeamento estiver certo, remova o bloco outputs.file, adicione um output [[outputs.influxdb_v2]] e mude o input para um broker privado. Com um broker MQTT gerenciado como a ConnectMQ, isso significa servers = ["ssl://broker.connectmq.com:8883"], username, password e o client ID emitido no portal. Antes de escrever regras de interpretação para um novo tipo de dispositivo, o explorador de topics ao vivo do portal é uma forma prática de copiar um payload real. A ConnectMQ também decodifica Sparkplug B e pode republicá-lo em topics ISA-95, então o Telegraf pode assinar topics ISA-95 limpos; inspecione um payload republicado no portal antes de escrever caminhos para ele.

Para o básico do plugin de input, comece pelo tutorial do Telegraf MQTT consumer. Para rodar o pipeline completo com o Grafana, use a configuração da stack TIG com Docker Compose e, para escolher bibliotecas do lado dos dispositivos, veja o guia de bibliotecas de cliente MQTT. A referência oficial do parser json_v2 lista todas as opções.

Comece com a ConnectMQ

Teste suas regras de interpretação contra o servidor MQTT público gratuito com payloads de exemplo e depois passe para credenciais privadas nas máquinas reais. A ConnectMQ oferece um broker MQTT gerenciado compatível com MQTT 5, com TLS e um portal que mostra topics e payloads ao vivo. Crie um broker Serverless gratuito em app.connectmq.com, sem cartão de crédito. Para uma visão mais ampla das opções de hospedagem, veja o guia de broker MQTT na nuvem.

Perguntas frequentes

Qual é a diferença entre os parsers json e json_v2 no Telegraf?

O parser json legado achata os payloads automaticamente, com controle limitado. O parser json_v2 usa caminhos GJSON explícitos para tags, fields, tipos e timestamps, o que se adapta muito melhor a payloads IoT aninhados.

O Telegraf consegue obter tags do topic MQTT?

Sim. Um bloco topic_parsing mapeia níveis do topic para tags, fields ou o nome do measurement. Use _ para os níveis que quiser pular.

Como converto um array JSON em métricas separadas?

Use um bloco object com o caminho do array. Cada elemento vira uma métrica própria, e você pode promover chaves a tags e definir um timestamp por elemento.

Preciso do json_v2 para payloads planos?

Não. Para payloads numéricos planos, o parser legado funciona bem. Passe para o json_v2 quando precisar de strings, tipos explícitos, fields renomeados ou estruturas aninhadas.