Set influxdb time via mqtt

Telegraf
1

Hi,

Maybe somebody can tell me if this is even possible.

I’m storing a series of data in an influxdb database (1.8 for the moment). Each datapoint has an associated timestamp, when it was last acquired. I want to use that timestamp (in unix time format) as the time field in the measurement. I was able to do that if I saved the value directly to influxdb, but now I need to send that data via mqtt (in json format) to a telegraf listener and save it to the influxdb measurement.

However, I have not found a way so far. The field “time” it just not saved into the DB.

I also tried setting this option in telegraf.conf
json_time_key = “unixtime”

Which seems to work for that specific case, but it complains for the rest of the messages (which I cannot control, since they come from zigbee2mqtt) that
2022-02-23T02:57:30Z E! [inputs.mqtt_consumer] Error in plugin: JSON time key could not be found

Any ideas?

2

Please show us the config of your input plugin and an example snippet of the mqtt payload.

1 Like
3

Hi,

This is a sample mqtt payload:

{"measurement": "airquality", "status": "ok", "date": "2022-02-23 06:00:00", "unixtime": 1645596000, "location": "NewYork", "t": 15.5, "p": 1012.7, "h": 87.0, "w": 0.4, "co": 4.8, "no2": 16.6, "o3": 60.9, "pm10": "", "so2": "", "wg": "", "pm25": 41.0}

The measurement field will be the name of the measurement in the influxdb database, and I’d like to use the ‘unixtime’ field as the timestamp.

However, I also use zigbee2mqtt for other sensors in my home, and their payloads look like this (by default, if I don’t specify a measurement, it goes to mqtt_consumer):
{"battery":100,"humidity":41.01,"linkquality":123,"pressure":1022.9,"temperature":25.21,"voltage":3085}
So whatever configuration I use, it must be compatible with both.

This is what my telegraf.conf looks like

# # Read metrics from MQTT topic(s)
[[inputs.mqtt_consumer]]

  servers = ["tcp://192.168.1.105:1883"]
  topics = [
    "telegraf/host01/cpu",
    "telegraf/+/mem",
    "sensors/#",
    "zigbee2mqtt/#",
    "network/#"
  ]

  data_format = "json"
 
  #json_time_key = "unixtime"
  #json_time_format = "unix"
  
  tag_keys = [
    "measurement", 
    "host",
    "hddname",
    "ip",
    "user",
    "method",
    "location"
  ]

(If I uncomment the json_time_* lines, the second payload (zigbee2mqtt) stops working since it can’t find it’s timestamp.)

And also this part in processor plugins, so the measurements tag is used as the name of the measurement:

[[processors.converter]]
  [processors.converter.tags]
    measurement = ["measurement"]
4

I don’t think it’s possible to have a “generic” decoder, especially when time is part of the information you expect (if specified explicitly, then it’s usually mandatory).

Can’t you just split it into two different inputs? (I expect messages to be structurally standardized by topic)
something like

[[inputs.mqtt_consumer]]
  alias = 'Mqtt1'
  servers = ["tcp://192.168.1.105:1883"]
  topics = [
    "telegraf/host01/cpu",
    "telegraf/+/mem",
    "sensors/#",
    "network/#"
  ]
  {...etc...}

[[inputs.mqtt_consumer]]
  alias = 'Mqtt2'
  servers = ["tcp://192.168.1.105:1883"]
  topics = [
    "zigbee2mqtt/#",
  ]
  {...etc...}
1 Like