Parse JSON from MQTT [input MQTT] plugin

Telegraf
,
1

I’m trying to parse a JSON file received by a MQTT Broker (Mosquitto) connection to feed an already create database. I’m using telegraf and influxDB on a VM(Ubuntu 18.04). Both services work well, however, influxdb doesn’t receive the tags from the JSON.

This is the JSON file:
{ "message":{ "basic_container":{ "confidence":{ "altitude":15, "semi_major_confidence":0, "semi_major_orientation":0, "semi_minor_confidence":0 }, "reference_position":{ "altitude":0, "latitude":414066546, "longitude":21750345 }, "station_type":5 }, "generation_delta_time":45315, "high_frequency_container":{ "confidence":{ "curvature":7, "heading":127, "longitudinal_acceleration":102, "speed":127, "vehicle_length":4, "yaw_rate":8 }, "curvature":1023, "curvature_calculation_mode":2, "drive_direction":2, "heading":3601, "longitudinal_acceleration":161, "speed":500, "vehicle_length":1023, "vehicle_width":62, "yaw_rate":32767 }, "protocol_version":2, "station_id":224 }, "origin":"on_board_application", "source_uuid":"cttc_car_224", "type":"cam", "version":"1.0.0", "timestamp":1605886891388 }

The part of telegraf.conf is this:

` [[inputs.mqtt_consumer]]

servers = [“tcp://127.0.0.1:1883”]
topics = [
“5GCroCo/inQueue/v2x/cam/#”,
“5GCroCo/inQueue/v2x/denm/#”,
]
data_format = “json”
tag_keys = [
“station_type”,
]`

I’ve tried to get just a simple tag “station_type” but it is not shown at the influx measurements.
This is the influxdb part:

[[outputs.influxdb]] urls = ["http://127.0.0.1:8086"] database = "v2x_manager"

This is the influx output

`rparada@rparada-VirtualBox:~$ influx
Connected to http://localhost:8086 version 1.8.3
InfluxDB shell version: 1.8.3

use v2x_manager
Using database v2x_manager
show measurements
name: measurements
name

cpu
disk
diskio
kernel
mem
processes
swap
system

What I’m missing?

2

Hello @rparada,
I’m not sure. Can you please include debug=true in your telegraf config?
I get the following line protocol

file,host=Anais.attlocal.net message_generation_delta_time=45315,message_high_frequency_container_curvature=1023,message_high_frequency_container_speed=500,message_high_frequency_container_drive_direction=2,message_high_frequency_container_confidence_speed=127,message_high_frequency_container_longitudinal_acceleration=161,message_station_id=224,message_basic_container_confidence_semi_minor_confidence=0,message_high_frequency_container_confidence_longitudinal_acceleration=102,message_basic_container_reference_position_longitude=21750345,message_high_frequency_container_heading=3601,message_high_frequency_container_curvature_calculation_mode=2,message_basic_container_reference_position_altitude=0,message_high_frequency_container_confidence_heading=127,message_high_frequency_container_confidence_yaw_rate=8,message_high_frequency_container_confidence_curvature=7,message_high_frequency_container_yaw_rate=32767,message_basic_container_confidence_altitude=15,message_basic_container_station_type=5,message_high_frequency_container_vehicle_width=62,message_high_frequency_container_vehicle_length=1023,message_protocol_version=2,message_high_frequency_container_confidence_vehicle_length=4,message_basic_container_confidence_semi_major_orientation=0,message_basic_container_confidence_semi_major_confidence=0,message_basic_container_reference_position_latitude=414066546 1020156975269009408

With

[[inputs.file]]
  files = ["./test.json"]

  ## Data format to consume.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
  data_format = "json"

  ## When strict is true and a JSON array is being parsed, all objects within the
  ## array must be valid
  json_strict = true

  ## Query is a GJSON path that specifies a specific chunk of JSON to be
  ## parsed, if not specified the whole document will be parsed.
  ##
  ## GJSON query paths are described here:
  ##   https://github.com/tidwall/gjson/tree/v1.3.0#path-syntax
  json_query = ""

  ## Tag keys is an array of keys that should be added as tags.  Matching keys
  ## are no longer saved as fields.
  tag_keys = [
    "station_type"
  ]

  ## Array of glob pattern strings keys that should be added as string fields.
  json_string_fields = ["latitude"]

  ## Name key is the key to use as the measurement name.
  json_name_key = ""

  ## Time key is the key containing the time that should be used to create the
  ## metric.
  json_time_key = "timestamp"

  ## Time format is the time layout that should be used to interpret the json_time_key.
  ## The time must be `unix`, `unix_ms`, `unix_us`, `unix_ns`, or a time in the
  ## "reference time".  To define a different format, arrange the values from
  ## the "reference time" in the example to match the format you will be
  ## using.  For more information on the "reference time", visit
  ## https://golang.org/pkg/time/#Time.Format
  ##   ex: json_time_format = "Mon Jan 2 15:04:05 -0700 MST 2006"
  ##       json_time_format = "2006-01-02T15:04:05Z07:00"
  ##       json_time_format = "01/02/2006 15:04:05"
  ##       json_time_format = "unix"
  ##       json_time_format = "unix_ms"
  json_time_format = "unix"

  ## Timezone allows you to provide an override for timestamps that
  ## don't already include an offset
  ## e.g. 04/06/2016 12:41:45
  ##
  ## Default: "" which renders UTC
  ## Options are as follows:
  ##   1. Local               -- interpret based on machine localtime
  ##   2. "America/New_York"  -- Unix TZ values like those found in https://en.wikipedia.org/wiki/List_of_tz_database_time_zones
  ##   3. UTC                 -- or blank/unspecified, will return timestamp in UTC
  json_timezone = "UTC"

  [[outputs.file]]
  ## Files to write to, "stdout" is a specially handled file.
  files = ["stdout"]

  ## Data format to output.
  ## Each data format has its own unique set of configuration options, read
  ## more about them here:
  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
  data_format = "influx"

You can also use fieldpass`` and taginclude``` to limit what gets parsed.