Parsing and using custom timestamps with json_v2

Telegraf
1

Hello,

I’m trying to parse a json string from an MQTT feed - following the guides from github i can plot the three pressures but i can’t seem to loop through the array and plot against the timestamp - is there something obvious I’m missing?

For these types of data, what would be best -use the default tag names or split these up into new tags and fields?

telegraf.conf

[[inputs.mqtt_consumer]]
  servers = ["tcp://mosquitto:1883"]
  topics = ["device/manufacturer/model/serial_number/pressures"]
  data_format = "json_v2"
  [[inputs.mqtt_consumer.json_v2]]
    [[inputs.mqtt_consumer.json_v2.object]]
      path = "pressures"      
      timestamp_path = "timestamp"
      timestamp_format = "UTC"
    [[inputs.mqtt_consumer.json_v2.object]]
      path = "pressures"

JSON

{    "pressures": [
        {
            "timestamp": "2022-07-28 02:43:33:667",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:33:767",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:33:867",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:33:967",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:067",
            "pump 1": -6.0,
            "pump 2": 26.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:167",
            "pump 1": -5.0,
            "pump 2": 28.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:267",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:367",
            "pump 1": -4.0,
            "pump 2": 27.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:467",
            "pump 1": -6.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:567",
            "pump 1": -6.0,
            "pump 2": 28.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:667",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:767",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:867",
            "pump 1": -5.0,
            "pump 2": 28.0,
            "system": -5.0
        }
    ],}

The JSON file is produced roughly every second and contains between 8-12 pressure points each iteration. I can change the format of the JSON string if necessary as its been produced by a python client. As I’m new to JSON and LP this is my first attempt so this may not be the best JSON structure. The timestamp is currently set to UTC but can bechanged to unix_ms if this is better?

A quick follow on question - is it better to break out the JSON data into single values and instead use multiple MQTT topics to pass data to influx? Are multiple MQTT feeds with a single value more efficient than one MQTT feed with a complex JSON/array object coating all the values?

Many thanks!

2

Hi @om327 Welcome back!

Something like this should parse the 3 pressures from each element of the array with the associated timestamp.

 [[inputs.mqtt_consumer.json_v2]]
    [[inputs.mqtt_consumer.json_v2.object]]
      path = "pressures.@this"
      included_keys  = ["pump 1", "pump 2", "system"]   
      timestamp_path = "timestamp"
      timestamp_format = "UTC"

A full explanation can be found here.

Phill

P.S Extra comma?

3

Hi @phill - thanks again, it seems you’re always bailing me out!!

The extra comma was an error from when I copied the complete JSON I’ve been experimenting with. I have some lab equipment that I’ve created some python drivers for and then made into a basic MQTT client. This can then connect to telegraf or to an OPC UA server I’ve also made in python.

{
    "info": {
        "firmware": 1.115,
        "date": "14-03-2018",
        "serial number": "1047",
        "system type": "R2-Plus",
        "error dection": null,
        "pressure mode": "50bar 50mv"
    },
    "status": {
        "start time": "2022-07-28 02:43:23:267",
        "run state": 0,
        "flow rates": {
            "pump 1": 123,
            "pump 2": 250
        },
        "air locks": {
            "lock 1": 76,
            "lock 2": 60
        },
        "system pressure limit": 30.0,
        "led bitmap": 0,
        "setpoints": {
            "reactor 1": 24,
            "reactor 2": 60,
            "reactor 3": 24,
            "reactor 4": 24
        }
    },
    "flows": [
        {
            "time": "2022-07-28 02:57:40:167",
            "pump 1 flowrate": "250",
            "pump 2 flowrate": "500"
        },
        {
            "time": "2022-07-28 02:57:41:167",
            "pump 1 flowrate": "1500",
            "pump 2 flowrate": "1000"
        }
    ],
    "temepratures": [
        {
            "timestamp": "2022-07-28 04:14:16:767",
            "reactor 1": {
                "state": "heating",
                "temperature": 19.9
            },
            "reactor 2": {
                "state": "unpowered",
                "temperature": null
            },
            "reactor 3": {
                "state": "cooling",
                "temperature": 20.7
            },
            "reactor 4": {
                "state": "unpowered",
                "temperature": null
            }
        },
        {
            "timestamp": "2022-07-28 04:14:17:767",
            "reactor 1": {
                "state": "unpowered",
                "temperature": 19.9
            },
            "reactor 2": {
                "state": "unpowered",
                "temperature": null
            },
            "reactor 3": {
                "state": "unpowered",
                "temperature": 20.7
            },
            "reactor 4": {
                "state": "unpowered",
                "temperature": null
            }
        }
    ],
    "pressures": [
        {
            "timestamp": "2022-07-28 02:43:33:667",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:33:767",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:33:867",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:33:967",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:067",
            "pump 1": -6.0,
            "pump 2": 26.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:167",
            "pump 1": -5.0,
            "pump 2": 28.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:267",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:367",
            "pump 1": -4.0,
            "pump 2": 27.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:467",
            "pump 1": -6.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:567",
            "pump 1": -6.0,
            "pump 2": 28.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:667",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -5.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:767",
            "pump 1": -5.0,
            "pump 2": 27.0,
            "system": -6.0
        },
        {
            "timestamp": "2022-07-28 02:43:34:867",
            "pump 1": -5.0,
            "pump 2": 28.0,
            "system": -5.0
        }
    ],
    "valves": [
        {
            "timestamp": "2022-07-28 03:35:52:167",
            "valve state": 0
        },
        {
            "timestamp": "2022-07-28 03:35:53:167",
            "valve state": 0
        }
    ]
}

For JSON as complex as this, would it be best to separate these into multiple MQTT topics rather than attempt to parse in one go? Not sure what the computational cost of the two methods is and which is more efficient or scalable?

The alternative is to send the data directly to the OPC UA server and then receive data from the OPC UA server in telegraf. Ideally I’d like to get as close to real-time as possible but from a couple of other post it looks like influx is limited to about 1 second updates due to polling limitations. I’m also experimenting with grafana which appears to allow faster rendering ~ 10ms but requires JSON to be simple:

{
   tag1 : value 1
   tag2 : value 2
   tag3 : value 3 
}

Which way would you go?

Thanks again!

4

@om327 So you want the displayed data to be near real time? Like a digital storage scope?

5

Hi @phill Yes if possible, but it is proving a little tricky.

I also just tried your code snippet - unfortunately, it only seems to show output for system and not for pump 1 or pump 2 ? It also only plots one data point and uses influx’s timestamp rather than the JSON one?

[[inputs.mqtt_consumer]]
  servers = ["tcp://mosquitto:1883"]
  topics = ["device/Vapourtec/R2-Plus/1048/pressures"]
  data_format = "json_v2"
 [[inputs.mqtt_consumer.json_v2]]
    [[inputs.mqtt_consumer.json_v2.object]]
      path = "pressures.@this"
      included_keys  = ["pump 1", "pump 2", "system"]   
      timestamp_path = "timestamp"
      timestamp_format = "UTC"

At first I thought I might be the timestamp was not compliant so I’ve changed the time format in my python MQTT client to be compliant with RFC3339

2022-07-29T02:06:26.743351Z

Unfortunately, it still doesn’t seem to like it. Looking at the graph, the intervals only appear to go down to seconds rather than miliseconds - can Influx record and plot at this rate?

6

@om327 My code example is a hack. It should be close (might need to escape the space,
i.e., ["pump\ 1", "pump\ 2"]). There are other ways to parse out the data besides included_keys - refer to the full explanation link above.

As you know timestamps down to nanosec resolution are permitted. Maybe the x-axis scaling?