Trying to parse json mqtt message {"${tagName}" : "${value}", "time": "${timestamp}"} using [[inputs.mqtt_consumer.json_v2]] or [[inputs.mqtt_consumer.xpath]] or some other method

Telegraf
1

I’m struggling to parse the message in the title:

{"${tagName}" : "${value}", "time": "${timestamp}"}

Where:

  • tagName = any string
  • value = int, float, bool or string
  • timestamp is in this format: 2025-09-03T14:05:18Z

Ideally, I’d like to not have to change the original message format because it’s already being used in a data pipeline.

I’ve spent three days trying to piece this together using github, questions on this forum, and docs on the plugins I’m trying to use, but it’s beating both me and copilot.

I’ve started the process of writing my own plugin, but I can’t imagine there’s not someone else who’s parsed something like the above already, which theoretically looks really simple.

Please help if you can push me in the right direction!

Here are some configurations I’ve tried so far with errors from logs:

[[inputs.mqtt_consumer.json_v2]]

data_format = "json_v2"
  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "test_measurement"
    [[inputs.mqtt_consumer.json_v2.object]]
      path = "*"

2025-09-03T14:21:10Z E! [agent] Error writing to outputs.influxdb_v2: failed to write metrics to bucket “test_bucket”: 500 Internal Server Error: internal error: An internal error has occurred - check server logs

[[inputs.mqtt_consumer.xpath]]

timestamp = “time”
timestamp_format = “2006-12-25T15:04:01.235Z”

2025-09-03T11:15:09Z E! [inputs.mqtt_consumer] Error in plugin: failed to parse timestamp: parsing time “2025-09-03T11:15:58.308Z” as “2006-12-25T15:04:01.235Z”: cannot parse “-03T11:15:58.308Z” as “2”

timestamp = "time"
timestamp_format = "%Y-%m-%dT%H:%M:%S.%fZ"

[inputs.mqtt_consumer] Error in plugin: failed to parse timestamp: parsing time “2025-09-03T11:25:04.311Z” as “%Y-%m-%dT%H:%M:%S.%fZ”: cannot parse “2025-09-03T11:25:04.311Z” as “%Y-%m-%dT%H:%M:%S.%fZ”

[[inputs.mqtt_consumer.xpath]]
    field_selection = "*"
    [inputs.mqtt_consumer.xpath.tags]

2025-09-03T14:05:18Z E! loading config file /etc/telegraf/telegraf.conf failed: plugin inputs.mqtt_consumer: line 39: configuration specified the fields [“xpath”], but they were not used; this is either a typo or this config option does not exist in this version

2

There are a few possible approaches you could try:

  • Use processors.regex or processors.converter after parsing with json_v2 to restructure the data
  • Apply custom parsing logic within json_v2 using field renaming
  • Use a Starlark processor for more complex transformations

Would you like me to provide specific configuration examples for any of these options? If so, it would really help to know a bit more:

  • A sample of the actual MQTT messages being received
  • Your current Telegraf version
  • The desired output format in InfluxDB

With that information, I can suggest the most accurate configuration for your use case.

3

MQTT message as received by mosquitto: data/machine1 [{test_float:4, time:2025-09-01T12:01:10.657Z}]

MQTT message from telegraf logs: “[{“test_float”:2,“time”:“2025-09-02T10:20:37.257Z”}]”, where topic is data/machine1. Let me know if the actual whole message (ie from wireshark) would be preferable?

On docker, telegraf 1.35.4

Desired output format to influxdb: data,host=machine1, test_float=2 1693737719000000000

Any more detail I can provide, just ask! I appreciate the help understanding my way around telegraf better.
4

Here’s a Telegraf configuration I’d propose. It’s not tested, so you may need to tweak it a bit, but it should give you a solid starting point.

[[inputs.mqtt_consumer]]
  servers = ["tcp://localhost:1883"]
  topics = ["data/+"]
  data_format = "json_v2"
  
  [[inputs.mqtt_consumer.json_v2]]
    measurement_name = "data"
    timestamp_path = "time"
    timestamp_format = "2006-01-02T15:04:05.000Z"
    
    # Parse all fields except 'time' as metric fields
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "@this"
      exclude = ["time"]
      
    # Extract machine name from topic
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "@topic"
      rename = "host"

# Processor to extract machine name from topic path
[[processors.regex]]
  [[processors.regex.tags]]
    key = "host"
    pattern = "^data/(.+)$"
    replacement = "${1}"

# Optional: Processor to handle any remaining cleanup
[[processors.converter]]
  [processors.converter.tags]
    string = []
  [processors.converter.fields]
    integer = []
    float = []

Key points about this configuration:

  1. Uses json_v2 with path = "@this" to capture all fields, excluding the time field
  2. timestamp_path = "time" with the correct Go time format "2006-01-02T15:04:05.000Z"
  3. Uses @topic to capture the full topic, then a regex processor to extract just the machine name
  4. The exclude = ["time"] ensures only the data fields (like test_float) become metrics
1 Like
5

@tomwarburton for completeness my solution with the xpath parser with the data being

[{“test_float”:2,“time”:“2025-09-02T10:20:37.257Z”}]

and the config

[[inputs.file]]
  files = ["test_configs/jsontest_forum_tomwarburton.json"]

  fieldexclude = ["time"]
  data_format = "xpath_json"
  xpath_native_types = true
  [[inputs.file.xpath]]
    metric_name = "'test_measurement'"
    metric_selection = "*"
    field_selection = "*"
    timestamp = "time"
    timestamp_format = "rfc3339"

(please replace inputs.file with inputs.mqtt_consumer) I get

> test_measurement,host=Hugin test_float=2 1756808437000000000

Please note that test_float (and basically every output) is a field as you cannot create measurements without a field in Telegraf!

2 Likes