Zeit für etwas Neues :-)

Die Datenbank "heizung" gibt es nun seit etwa 10 Jahren und sie hat eigentlich immer gute Dienste verrichtet. Anfangs haben wir eine Handvoll Messwerte gespeichert, daraus die neuen Schaltzustände für die Aktoren berechnet und diese dann auch an die Aktoren übertragen. Das funktioniert heute noch genauso, aber

• mittlerweile gibt es > 600 Sensoren und Aktoren an 10 Standorten (locations)
• kommen die Messwerte von unterschiedlichster Hardware oder besser gesagt über verschiedene Schnittstellen (php, python, mqtt, http)

und die Wünsche, diese Werte auch Langzeit zu archivieren werden lauter. Insbesondere die Messwerte der Balkonkraftwerke oder auch der Stromverbrauch soll eigentlich immer auch für die Vergangenheit vorhanden und auswertbar sein. Und da kommt unsere MariaDB an ihre Grenzen.

600 Messwerte x 1440 Minuten am Tag = 864.000 Datensätze pro Tag oder 315.360.000 Messwerte pro Jahr

Das lässt sich natürlich auch in MariaDb alles noch speichern, Auswertungen sind aber sagen wir mal etwas "träge".

In MariaDB hatten wir daher die Speicherdauer der Messwerte vor Jahren schon begrenzt, in der Regel auf einen Monat und damit war Schluss mit Langzeitanalysen.

Jetzt kommt influx ins Spiel, mit dem Ziel, ausgewählte Messwerte dauerhaft zu speichern und auch über Jahre hinweg noch schnell analysieren zu können. Als erste Schnittstelle zur Datenbank (mehr ist denkbar) kommt telegraf zum Einsatz.

Nach Experimenten mit einem Raspberry 4, darauf influx in Version 1.8.10 und telegraf in Version 1.26.3 hat der Server oxolon.de nun noch einen kleinen Bruder bekommen unter monitor-oxolon.de. Darauf influx in der Version 2.7.1 und Grafana in Version 10.1.0. Installationsinfos lassen wir hier mal weg, das Netz ist voll davon und dank Jens läuft nun Grafana auch perfekt unter https:// was eine Grundvoraussetzung ist, um Grafana Dashboards später auch in die eigentliche Website (also diese Seiten hier) einbetten zu können.

Etwas schwierig war die Konfiguration von telegraf. Daher will ich hier zumindest mal das erzielte Ergebnis festhalten. Telegraf wird zum "Lauschen" auf einen mqtt-Broker konfiguriert. Dieser bekommt passgenaue Nachrichten, die zuvor in MariaDB aus den vorhandenen Messwerten zusammengestellt werden.

Damit das ganze flexibel und gut konfigurierbar bleibt, waren zunächst einige Strukturänderungen auf der Datenbankebene in MariaDB notwendig und zwar:

1. neue Schlüsseltabelle (_diagram) um Diagramme zu beschreiben.
2. neue Tabelle (_device_diagram), mit der die Geräte (_device) 1:n Diagramm(en) zugeordnet werden können.
3. neue Schlüsseltabelle (_export_topic) um topics zu benennen.
4. neue Tabelle (device_export_topic), mit der die Geräte (_device) 1:n den Export-Topics zugeordnet werden können.

Mit Nr. 1 und 2 wird ermöglicht, Messwerte für bestimmte Diagramme zusammen zu stellen. Das Diagramm selbst kann dann auf diese Auswahl zugreifen und damit kann die bisherige (umständliche und vor allem nicht ganz "saubere") Zusammenstellung der Messwerte für ein Diagramm über Hilfsgrößen wie zum Beispiel die Maßeinheit entfallen. Das war schon lange auf dem Plan und ist nun zumindest in der Tabellenstruktur definiert. Zugriff hierauf über die Oberfläche und damit für die Hand des Anwenders muss folgen.

Nr. 3 und 4 sind notwendig, um bestimmen zu können, welche Messwerte in die Langzeitspeicherung (influx) einbezogen werden sollen. Ich habe mich dafür entschieden, dies nicht nur durch ja/nein zu realisieren sondern gleich etwas komfortabler zu gestalten. Es ist jetzt möglich, einen Messwert einem oder auch mehreren Export-Topics zuzuordnen. Damit kann man also nicht nur bestimmen ob ein Messwert in den Export geht, sondern auch wohin, also in welchen Topic auf dem mqtt-Broker. Und damit kann ein Messwert auch mehreren dieser Export-Topics zugeordnet werden; sprich in der weiteren Verarbeitung auch in verschiedenen Buckets auf Influx gespeichert werden.

Für den Export Richtung Influx ergeben sich durch Nr. 1 und 2 nun auch gleich noch weitere Metadaten für jeden Messwert nämlich die Info, für welche Diagramme der Messwert vorgesehen ist. Da auch hier eine 1:n Zuordnung zwischen Messwert und Diagramm möglich ist, würde das im JSON-Export zu einer kaskadierten Struktur führen, die ich im Moment nicht will, weil ich auch nicht weiß, wie man diese dann über telegraf für influx vernünftig auflöst. Ich habe daher zunächst mal den Weg gewählt, die Diagramm-ID's zu einem String zusammen zu fassen. Ein Messwert der in den Diagrammen Nr. 7, 12 und 15 erscheinen soll bekäme daher den String 7#12#5 mit auf den Weg.

Zunächst das Select Statement:
insert into mqtt_out (topic, payload, status)
SELECT concat(_export_topic.topic,'/',_device.location,'/', _device.unit, '/', _device.id) as topic_out,
JSON_OBJECT('mysql_device_id', _device.id, 'type', type, 'model', _device.model, 'short_description', _device.short_description, 'position', position, 'unit', _device_model.unit, 'location_unit_int', _device.unit,
'last_value', sf_last_value_device_id(_device.id), 'utc_read', sf_last_utc_read_device_id(_device.id), 'diagrams', sf_group_concat_diagram(_device.id)) AS json_data, 'open' as status
FROM _device
JOIN _device_location ON _device.location = _device_location.location
JOIN _device_model ON _device.model = _device_model.model
JOIN _device_export_topic ON _device.id = _device_export_topic.device_id
JOIN _export_topic ON _device_export_topic.export_topic_id = _export_topic.id
where sf_last_utc_read_device_id(_device.id) is not null

Dieses Statement wird in MariaDB jede Minute als Ereignis ausgelöst und somit im Ergebnis jede Minute ein aktueller Wert (wenn vorhanden) auf den mqtt-Broker geschrieben. Nachfolgend mal die Ausgabe dazu, damit ihr mal die Nachricht seht, wie sie an den Broker gesendet wird:

influx_prod/mysql/DE_RO_AK3/0/10	{"mysql_device_id": 10, "type": "sensor", "model": "DS18B20", "short_description": "Bienenhaus", "position": "bienenhaus", "unit": "Celsius", "location_unit_int": 0, "last_value": "15.30", "utc_read": "2023-10-29 16:37:41", "diagrams": null}

# oder hier etwas besser lesbar:

topic_out = influx_prod/mysql/DE_RO_AK3/0/10
payload = 
{
  "mysql_device_id": 10,
  "type": "sensor",
  "model": "DS18B20",
  "short_description": "Bienenhaus",
  "position": "bienenhaus",
  "unit": "Celsius",
  "location_unit_int": 0,
  "last_value": "15.30",
  "utc_read": "2023-10-29 16:37:41",
  "diagrams": null
}
status = open

Lasst euch von der Spalte "status" nicht verwirren, diese wird nicht an den Broker gesendet sondern dient dazu, gesendete von noch "wartenden" Datensätzen in der Tabelle mqtt_out (Zwischenspeicher innerhalb der Datenbank) zu unterscheiden. Nach erfolgreichem Versenden erhält die Datenbank eine Quittung und "status" wird auf "ready" geändert. Aus dem Zwischenspeicher werden die Datensätze dann nach 24 Stunden gelöscht.

Ihr seht oben, dass neben dem eigentlichen Messwert "last_value" noch jede Menge Metainformationen dazu gegeben werden, um die Daten später mit Grafana auch sinnvoll auswerten zu können. Auch der Zeitstempel wird weitergereicht, denn auch dieser ist in MariaDB bereits vorhanden. Soweit so gut, das war der leichtere Teil der Übung.

Die Konfiguration von telegraf hat mich etwas mehr Nerven gekostet, aber das lag wohl an fehlender Übung in diesem Themenfeld.

Über das Influx Webinterface wird zunächst ein Bucket erstellt. Dabei könnte man dann auch gleich angeben, dass Daten "älter als XXX Stunden/Tage/Jahre" gelöscht werden sollen, was ich aber nicht eingestellt habe. Nach dem Erstellen des Buckets kann man mit "+ADD DATA" einen sogenannten Telegraf Agent anlegen, der dann die Daten in ebendieses Bucket schreibt. Hier wählt man zunächst das mqtt-Plugin aus. Es öffnet sich ein Editor-Fenster, in das die nachfolgende Konfiguration eingetragen wird. Nicht vergessen dann den Server, Usernamen und Passwort zu konkretisieren.

# Erklärung immer unterhalb der Zeile !!!
[[inputs.mqtt_consumer]]
   servers = ["tcp://mqtt_broker:1883"]
   username = "mqtt_username"
   password = "mqtt_password"
# die 3 obigen Zeilen muss man natürlich anpassen
   topics = ["influx_prod/mysql/#"]
# der topic, auf den das plugin lauscht !!! diese Daten gehen in das gleichnamige Bucket:"influx_prod_mysql" !!!
   data_format = "json_v2"
# es soll json im neuen Format v2 aufgelöst werden
   [[inputs.mqtt_consumer.topic_parsing]]
     topic = "influx_prod/+/+/+/+"
# alle topics mit der Pfadtiefe 5 werden betrachtet
     tags = "_/datasource/location/location_unit/topic_device_id"
# der erste Teil des Topics, also "influx_prod" wird durch die Angabe "_/" überlesen
# der 2. Teil "/+" als "datasource", 
# der 3. Teil "/.../+" als tag "location",
# der 4. Teil "/.../.../+" als tag "location_unit und 
# der 5. Teil "/.../.../.../+" als tag "topic_device_id"  an Influx DB gegeben
  [[inputs.mqtt_consumer.json_v2]]
# jetzt wird der payload aufgelöst
    timestamp_path = "utc_read"
# der zeitstempel wird im json-Object mit dem Namen "utc_read" übergeben
    timestamp_format = "2006-01-02 15:04:05"
# ganz cool, man schreibt einfach irgendein Datum in der Form rein, die auch im json enthalten ist
    timestamp_timezone = "UTC"
# und das Zeitformat noch dazu geben
    [[inputs.mqtt_consumer.json_v2.tag]]
       path = "mysql_device_id"
# hier wird auf den Eintrag "mysql_device_id" im obigen json-Objekt referenziert
       type = "int"
# und der passende Datentyp dazu festgelegt
     [[inputs.mqtt_consumer.json_v2.tag]]
       path = "type"
# das gleiche noch einmal für "type"
# diesmal ohne spezielle Typ-Angabe, also passt da dann auch Text hinein
     [[inputs.mqtt_consumer.json_v2.tag]]
       path = "model"
     [[inputs.mqtt_consumer.json_v2.tag]]
       path = "short_description"
     [[inputs.mqtt_consumer.json_v2.tag]]
       path = "position"
     [[inputs.mqtt_consumer.json_v2.tag]]
       path = "unit"
     [[inputs.mqtt_consumer.json_v2.tag]]
       path = "diagrams"
     [[inputs.mqtt_consumer.json_v2.tag]]
       path = "location_unit_int"
       type = "int"
    [[inputs.mqtt_consumer.json_v2.field]]
      path = "last_value"
      type = "float"
# der letzte Eintrag ist nun der eigentliche Messwert, also kein tag- sondern ein field-Eintrag

Mit "SAVE AND TEST" erhält man gleich noch die passenden Infos, wie dieser Telegraf-Agent gestartet werden kann. Ganz wichtig ist das in diesem Dialog angezeigte Token, dass man unbedingt irgendwo sicher abspeichern muss, da dies !!! nie wieder !!! angezeigt wird.

Gestartet habe ich diesen Telegraf-Agent dann auf einem meiner Rasspberry's, direkt in der Konsole mit genau den beiden Statements, die man im "TEST AND SAVE" Dialog sieht (Token und die ID stimmen hier in dieser Beschreibung natürlich nicht):

export INFLUX_TOKEN=7zgVrnY8tGHYSra7UgzbXdTLfqKh
telegraf --config http://monitor-oxolon.de:8086/api/v2/telegrafs/0c0af07998786e000

Telegraf meldet hier übrigens auch gleich Fehler, wenn in der Konfiguration irgendetwas nicht passt. Wenn kein Fehler ausgegeben wird, hat Telegraf fleißig begonnen, seine Arbeit zu tun und ihr findet in dem Bucket die ersten Daten.

Letzter Schritt ist nun die Erstellung einer Datenquelle in Grafana, um dort dann Dashboards bauen zu können. Als ab in die Grafane Website und dort "Connections --> Data sources --> Add new data source" anklicken. Als nächstes muss man einen "data source type" auswählen, logisch, hier muss InfluxDB gewählt werden. Hier meine Einstellungen, alles was ich nicht beschreibe, muss man auch nicht angeben.

Mit "Save & Test" wird alles gespeichert und auch gleich angezeigt, ob alles funktioniert. Wenn ja steht eurem ersten Dashboard in Grafana nun nicht mehr im Wege.

Und hier das Ergebnis: