How to parse json array to influx Db

Telegraf
1

Hi,
I am trying to parse json array from telegraf to influxdb, the source of the json data is NATS server. below is the sample data.
[{“runId”:“R001”,“testName”:“Test”,“nodeName”:“Test-Node”,“errorMessage”:“noData”,“errorResponseBody”:“noData”,“requestName”:“WebSocket Open Connection-1”,“responseCode”:“Websocket I/O error”,“result”:“fail”,“samplerType”:“transaction”,“connectTime”:0,“count”:1,“errorCount”:1,“latency”:0,“processingTime”:0,“receivedBytes”:0,“sentBytes”:0,“responseTime”:81,“timeStamp”:1727323776272,“time”:1727323776352000000,“ObjectType”:“requestsRaw”}, {“runId”:“R001”,“testName”:“Test”,“nodeName”:“Test-Node”,“errorMessage”:“noData”,“errorResponseBody”:“noData”,“requestName”:“WebSocket Open Connection-1”,“responseCode”:“Websocket I/O error”,“result”:“fail”,“samplerType”:“transaction”,“connectTime”:0,“count”:1,“errorCount”:1,“latency”:0,“processingTime”:0,“receivedBytes”:0,“sentBytes”:0,“responseTime”:80,“timeStamp”:1727323776353,“time”:1727323776433000000,“ObjectType”:“requestsRaw”}]

I tried multiple configurations, lastly i used is
[[inputs.nats_consumer]]
servers = [“nats://localhost:4222”]
subjects = [“nats.subject.requestsraw”]
data_format = “json_v2”

[[inputs.nats_consumer.json_v2]]

[[inputs.nats_consumer.json_v2.object]]
path = “@this”
disable_prepend_keys = true
tags = [“runId”, “testName”, “nodeName”, “ObjectType”]

2

@Thrinatha_Reddy,
Welcome!
You’ll want to use the json_v2 format, that’s correct.
I think you’ll want to do something like:

[[inputs.nats_consumer]]
  servers = ["nats://localhost:4222"]
  subjects = ["nats.subject.requestsraw"]
  data_format = "json_v2"

  [[inputs.nats_consumer.json_v2]]
    [[inputs.nats_consumer.json_v2.object]]
      path = "@this"  # Parses each item in the array
      disable_prepend_keys = true  # Prevents prefixing field names with JSON keys
      tags = ["runId", "testName", "nodeName", "ObjectType"]  # Specifies which fields to use as tags

    [[inputs.nats_consumer.json_v2.field]]
      path = "responseTime"  # Response time field
      type = "int"

    [[inputs.nats_consumer.json_v2.field]]
      path = "latency"  # Latency field
      type = "int"

    [[inputs.nats_consumer.json_v2.field]]
      path = "receivedBytes"  # Received bytes field
      type = "int"

    [[inputs.nats_consumer.json_v2.field]]
      path = "sentBytes"  # Sent bytes field
      type = "int"

    [[inputs.nats_consumer.json_v2.field]]
      path = "errorCount"  # Error count field
      type = "int"

    [[inputs.nats_consumer.json_v2.field]]
      path = "connectTime"  # Connect time field
      type = "int"

    [[inputs.nats_consumer.json_v2.field]]
      path = "time"  # Time field for InfluxDB
      type = "int"

    [[inputs.nats_consumer.json_v2.field]]
      path = "result"  # Status result (fail/success)
      type = "string"

Here is a great place to see multiple examples:

3

@Anaisdg Thank you for your valuable time. I have tried the configuration you suggested, but unfortunately, it is not parsing as expected. My issue is that my JSON array contains 2,000 messages, yet Telegraf is treating it as a single message instead of processing all 2,000 individually. I have reviewed the documentation, but I have not been able to find a solution. If you have any insights on how to read all the messages from the array, I would greatly appreciate your assistance.

Thank you!

4

Hi Team,
This issue is solved with xpath_json plugin.

Below is my configuration.

[[inputs.nats_consumer]]
servers = [“nats://localhost:4222”]
subjects = [“nats.subject.requestsraw”]
data_format = “xpath_json”
xpath_native_types = true
fieldexclude = [“subjects”, “runId”, “testName”, “nodeName”, “ObjectType”, “errorMessage”, “errorResponseBody”, “requestName”, “responseCode”, “result”, “samplerType”]

[[inputs.nats_consumer.xpath]]
metric_name = “‘requestsRaw’”
metric_selection = “/"
timestamp = “time”
timestamp_format = “unix_ns”
field_selection = "”
[inputs.nats_consumer.xpath.tags]
runId = “string(/runId)”
testName = “string(/testName)”
nodeName = “string(/nodeName)”
ObjectType = “string(/ObjectType)”
errorMessage = “string(/errorMessage)”
errorResponseBody = “string(/errorResponseBody)”
requestName = “string(/requestName)”
responseCode = “string(/responseCode)”
result = “string(/result)”
samplerType = “string(/samplerType)”
unique_id = “string”

1 Like