MQTT, JSON, Telegraf und InfluxDB

[Kai (Gast)]

Hallo,

ich habe bei mir mehrere Sensoren (ESPs mit ESPEasy) die über MQTT Ihre 
Daten an den Broker senden. Mit Hilfe von Telegraf sammel ich diese 
Daten ein und schreibe sie in die InfluxDB.

Als Datenformat habe ich "value" mit Typ "float" konfiguriert.
(Config siehe unten)

Das funktioniert auch alles soweit.

Jetzt zu meinem Problem:
Ich möchte einen weiteren Sensor hinzufügen. Der sendet seine Daten 
allerdings im JSON Format.
Das soll Telegraf auch können, aber irgendwie bekomme ich die 
Konfiguration da nicht hin.


Ich lasse mir alle Daten die ankommen auf der Konsole ausgeben, dort 
kommt aber nur die Fehlermeldung das die JSON Daten nicht als 
"value"/"float" ausgewertet werden können.

[inputs.mqtt_consumer] Error in plugin: strconv.ParseFloat: parsing "07:24\",\"timestamp\":\"2022-10-15T06:19:01Z\",\"device\":\"rtlwmbus[00000001]\",\"rssi_dbm\":92}": invalid syntax]

Hat jemand eine Idee was ich falsch mache oder einen Ansatz zum weiteren 
Vorgehen?


MQTT JSON Nachricht die verarbeitet werden soll:
(die Daten werden aktuell an das Topic "test" gesendet und kommen dort 
auch an)

{"media":"heat cost allocation","meter":"qcaloric","name":"Bad","id":"88134838","status":"OK","current_consumption_hca":7,"set_date":"2022-05-31","consumption_at_set_date_hca":643,"set_date_1":"2022-05-31","consumption_at_set_date_1_hca":643,"error_date":"2127-15-31","device_date_time":"2022-10-15 07:24","timestamp":"2022-10-15T06:19:01Z","device":"rtlwmbus[00000001]","rssi_dbm":92}

Telegraf Config:

[[inputs.mqtt_consumer]]
  servers = ["tcp://localhost:1883"]
  qos = 0
  connection_timeout = "30s"
  topics = [
    "sensors/#",
    "test/#"
  ]
  persistent_session = false
  data_format = "value"
  data_type = "float"

  [[inputs.mqtt_consumer.json_v2]]
    data_format = "json_v2"
    topics = [ "test/#" ]
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "name"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "current_consumption_hca"
      type = "float"

[Ein T. (ein_typ)]

Kai schrieb:
>

> [inputs.mqtt_consumer] Error in plugin: strconv.ParseFloat: parsing 
> "07:24\",\"timestamp\":\"2022-10-15T06:19:01Z\",\"device\":\"rtlwmbus[00000001]\",\"rssi_dbm\":92}": 
> invalid syntax]
>

Ist da ein Zeilenumbruch vor "07:24"? Vielleicht ist der ja das Problem.

Edit: oh, ach ja... 07:24 ist natürlich auch kein Float, ne.

[Sebastian (Gast)]

Kai schrieb:
> Hat jemand eine Idee was ich falsch mache oder einen Ansatz zum weiteren
> Vorgehen?

Die JSON-Nachricht scheint irgendwo "unterwegs" in zwei Fragmente 
zerlegt worden zu sein.

LG, Sebastian

[Georg A. (georga)]

Was sucht eigentlich der data_type="float" in der 
inputs.mqtt_consumer-Section?

BTW: Ich habe selten so eine bescheidene Doku wie bei telegraf gesehen. 
Es ist schon toll, dass es viele Beispiele gibt. Aber offensichtlich 
gibt es NUR Beispiele und keine vernünftige Beschreibung der generellen 
Syntax. Das man den ganz Kram nur immer per Trial&Error testen kann, 
machts nicht besser.

[Kai (Gast)]

Sebastian schrieb:
> Die JSON-Nachricht scheint irgendwo "unterwegs" in zwei Fragmente
> zerlegt worden zu sein.

Ich habe die Nachricht parallel mit dem mosquitto_client mitgeschrieben, 
da kommt die Nachricht an wie oben geposted (ohne Zeilenumbruch).
Wenn die Nachricht fragmentiert wird dann sollte das an Telegraf liegen.
Leider habe ich keine Ansatz wie ich das prüfen kann.

Georg A. schrieb:
> Was sucht eigentlich der data_type="float" in der
> inputs.mqtt_consumer-Section?

Das ist der "Standardteil" für meine anderen Sensoren und nach meinem 
Verständnis ein Konkretisierung von "data_format = "value"".
Sollte der Teil anders aufgebaut sein?

Das hätte ich auch gleich schon in meinem Ursprungspost deutlicher 
schreiben können: Die Konfiguration vor dem 
"[[inputs.mqtt_consumer.json_v2]]" läuft schon seit längerem ohne 
Probleme.
Um die JSON Daten einlesen zu können habe ich den 
"[[inputs.mqtt_consumer.json_v2]]" und alles folgende neu eingefügt und 
komme seit dem nicht weiter.

Georg A. schrieb:
> BTW: Ich habe selten so eine bescheidene Doku wie bei telegraf gesehen.
> Es ist schon toll, dass es viele Beispiele gibt. Aber offensichtlich
> gibt es NUR Beispiele und keine vernünftige Beschreibung der generellen
> Syntax. Das man den ganz Kram nur immer per Trial&Error testen kann,
> machts nicht besser.

Ja leider, aber "schön" zu hören das es nicht nur mir so geht.
Dummerweise kenne ich auch keine Alternative und bisher lief es auch 
gut.

[Georg A. (georga)]

Kai schrieb:
> Georg A. schrieb:
>> Was sucht eigentlich der data_type="float" in der
>> inputs.mqtt_consumer-Section?
>
> Das ist der "Standardteil" für meine anderen Sensoren und nach meinem
> Verständnis ein Konkretisierung von "data_format = "value"".
> Sollte der Teil anders aufgebaut sein?

Ich weiss es nicht, fehlt ja eine gescheite Doku ;) Es könnte aber dazu 
führen, dass per Default alles nach float geparst wird und dann bei 
Strings anschlägt.

Ich habe in mqtt_consumer nur das Allernötigste (server, json_v2 und 
topics) und erst in den den json_v2.fields den Namen und den Typ, den 
ich erwarte.

[Nils B. (hbquax)]

Mir konnte auch noch niemand verständlich erklären, was genau Telegraf 
eigentlich macht, was nicht ein Fünfzeiler in Python auch kann.

[Kai (Gast)]

So, bin heute dazu gekommen etwas weiter rum zu probieren.

Georg A. schrieb:
> Was sucht eigentlich der data_type="float" in der
> inputs.mqtt_consumer-Section?
Habe die einfach mal raus genommen. Wie vermutet kommen dann die 
Tag/Value Paare nicht mehr an.


Georg A. schrieb:
> Ich habe in mqtt_consumer nur das Allernötigste (server, json_v2 und
> topics) und erst in den den json_v2.fields den Namen und den Typ, den
> ich erwarte.
Ich schließe daraus das bei Dir Daten nur im JSON Format ankommen?
Wie sieht da Deine Konfig aus?
Entspricht die diesem hier:

[[inputs.mqtt_consumer]]
  servers = ["tcp://localhost:1883"]
  qos = 0
  connection_timeout = "30s"
  topics = [
    "sensors/#",
    "test/#"
  ]
  persistent_session = false
  data_format = "json_v2"

  [[inputs.mqtt_consumer.json_v2]]
    topics = [ "test/#" ]
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "name"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "current_consumption_hca"
      type = "float"

[Kai (Gast)]

Nils B. schrieb:
> Mir konnte auch noch niemand verständlich erklären, was genau
> Telegraf
> eigentlich macht, was nicht ein Fünfzeiler in Python auch kann.

Für mich sind es folgende Funktionen:
- Läuft als Deamon
- am MQTT Broker "lauschen" auf neue Nachrichten
- Parsen der Nachrichten (Tag/Value, JSON)
- Einspielen der Daten in die Influx DB

Ich bin leider nicht so fit in Python und beim Googlen sah das Thema 
auch nicht nach "mal eben" aus.
Solltest Du die 5 Zeilen zufällig irgendwo rumliegen haben würde ich 
mich freuen.

[Kai (Gast)]

Nachtrag:
Ich habe mal versucht einen 2. Input zu erzeugen.
(https://community.influxdata.com/t/multiple-mqtt-consumer-inputs-how-to-configure/8450/7)
Da hagelt es aber Fehlermeldungen :(

[Georg A. (georga)]

Kai schrieb:
> Ich schließe daraus das bei Dir Daten nur im JSON Format ankommen?

Ja, gibt nur JSON. Das Zeug von tfrec wird in sowas verpackt:

Topic sensors/1a97/msg 
{"Id":"1a97","Time":"1668545893","Temperature":"+11.3","Humidity":"86","Battery":"0","RSSI":"79"}

Config:

[[inputs.mqtt_consumer]]
        servers = ["tcp://127.0.0.1:1883"]
        topics = ["sensors/#"]
        data_format = "json_v2"

        [[inputs.mqtt_consumer.topic_parsing]]
                topic = "sensors/+/msg"
                tags = "_/Id/_" 

        [[inputs.mqtt_consumer.json_v2]]
                measurement_name = "sensors"
                timestamp_path = "Time"
                timestamp_format = "unix"
                [[inputs.mqtt_consumer.json_v2.field]]
                        path = "Temperature"
                        type = "float"
                [[inputs.mqtt_consumer.json_v2.field]]
                        path = "Humidity"
                        type = "float"
                [[inputs.mqtt_consumer.json_v2.field]]
                        path = "Battery"
                        type = "int"
                [[inputs.mqtt_consumer.json_v2.field]]
                        path = "RSSI"
                        type = "int"

Hab da noch eine zweite Config für die DS18B20-Meldungen von Tasmota. Da 
habe ich es nicht geschafft, die DS18B20-* Strukturen aus dem JSON, die 
die ID und die Temperatur beinhalten, mit einem Wildcard einzulesen. Es 
wird mit path = "DS18B20-*" zwar ein Wert gelesen, aber immer nur der 
erste. Hmpf.

Rein kommt:

tele/tasmota_A02014/SENSOR 
{"Time":"2022-11-15T21:59:38",
"COUNTER":{"C1":28146,"C2":36963,"C3":38783},
"DS18B20-1":{"Id":"0000001F0D50","Temperature":24.7},
"DS18B20-2":{"Id":"0000001BBDC8","Temperature":22.7},
"DS18B20-3":{"Id":"0000001D1EAA","Temperature":22.9},
"DS18B20-4":{"Id":"0000001BAD55","Temperature":22.8},
"DS18B20-5":{"Id":"0000001E71AF","Temperature":24.1},
"ESP32":{"Temperature":45.6},"TempUnit":"C"}

Config dazu:

[[inputs.mqtt_consumer]]
        servers = ["tcp://127.0.0.1:1883"]
        data_format = "json_v2"
        topics = [ "tele/+/SENSOR" ]

        [[inputs.mqtt_consumer.topic_parsing]]
                topic =  "tele/+/SENSOR"

        [[inputs.mqtt_consumer.json_v2]]
                measurement_name = "sensors"
                [[inputs.mqtt_consumer.json_v2.object]]
                        path =  "DS18B20-1" 
                        disable_prepend_keys = true
                        included_keys = ["Temperature"]
                        tags = ["Id"]
                [[inputs.mqtt_consumer.json_v2.object]]
                        path =  "DS18B20-2" 
                        disable_prepend_keys = true
                        included_keys = ["Temperature"]
                        tags = ["Id"]
                # usw für DS18B20-3,....

[Kai (Gast)]

Verstehe ich es richtig das zu zwei Input Blöcke 
([[inputs.mqtt_consumer]]) in deiner Konfig hast?
Einen für die Daten von tfrec und den anderen für DS18?

Das hatte ich bei mir gestern nicht ohne Fehler hinbekommen.
Aber wenn es bei Dir läuft muss ich da wohl noch mal schauen ;)

[Georg A. (georga)]

Kai schrieb:
> Verstehe ich es richtig das zu zwei Input Blöcke
> ([[inputs.mqtt_consumer]]) in deiner Konfig hast?
> Einen für die Daten von tfrec und den anderen für DS18?

Ja, sind zwei getrennte Files in /etc/telegraf/telegraf.d. Liegt an 
dieser TOML-Syntax, dass mit dem zweiten [[inputs....]] einfach eine 
neue Instanz erzeugt wird:

https://toml.io/en/v1.0.0#array-of-tables

[Kai (Gast)]

So, heute hatte ich wieder etwas Zeit und habe mir das angeschaut und 
die Konfiguration auf mehrere Dateien verteilt.

Es scheint jetzt zu laufen :)

Konfig für bisherige Geräte, telegraf-mqtt-input-value.conf:

[[inputs.mqtt_consumer]]
  servers = ["tcp://localhost:1883"]
  qos = 0
  connection_timeout = "30s"
  topics = [
    "sensors/#",
  ]
  persistent_session = false
  data_format = "value"
  data_type = "float"

Konfig für neuen Sensor (JSON Datenformat), 
telegraf-mqtt-input-json.conf:

[[inputs.mqtt_consumer]]
  servers = ["tcp://localhost:1883"]
  qos = 0
  connection_timeout = "30s"
  topics = [
    "sensors/+/heating/#",
  ]
  persistent_session = false
  data_format = "json_v2"

  [[inputs.mqtt_consumer.topic_parsing]]
    topic =  "sensors/+/heating/#"
   
  [[inputs.mqtt_consumer.json_v2]]
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "id"
    [[inputs.mqtt_consumer.json_v2.tag]]
      path = "name"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "current_consumption_hca"
      type = "float"

Das einzige Problem das ich jetzt noch habe, ist das Topic 
"sensors/+/heating/#" aus der ersten Konfig raus zu filtern.
Dort ist "sensors/#" abonniert, und da es viele unterschiedliche 
untertopics gibt würde ich lieber das eine raus filtern als alle 
aufzuführen.

Der "Metric" Filter mit namedrop scheint bei mqtt input nicht zu 
funktionieren 
(https://docs.influxdata.com/telegraf/v1.21/administration/configuration/)

Hat da jemand eine Idee?