Datenbank¶

Für die Persistenz der Telemetrie-Daten und der Geräteinformationen wird eine PostgreSQL-Datenbank eingesetzt. Entscheidend ist dabei die Unterstützung von JSONB-Spalten, die das Datenmodell flexibel und erweiterbar halten. Dadurch lassen sich unterschiedliche Geräte mit verschiedenen Capabilities und auch neue Capability-Typen abbilden, ohne dass das Datenbankschema angepasst werden muss.

Schema¶

Das Schema der Datenbank ist in drei Tabellen unterteilt: telemetry, devices und rules.

Tabellen¶

Geräte¶

In der Tabelle devices werden die Geräteinformationen gespeichert. Jedes Gerät hat eine eindeutige UUID, die bei der Registrierung des Geräts generiert und im System der MAC-Adresse zugeordnet wird. Die Tabelle enthält außerdem Informationen über den Gerätetyp, die MAC-Adresse, das Thing Model mit Metadaten, Capabilities sowie deren Optionen, den aktuellen Status des Geräts (online/offline) und die Zeitpunkte der ersten Registrierung und der letzten Aktualisierung.

Telemetrie¶

Die Tabelle telemetry speichert die Telemetrie-Daten der Geräte. Jeder Telemetrie-Datenpunkt ist mit der UUID des Geräts verknüpft, das die Daten gesendet hat. Die Tabelle enthält zudem den Zeitstempel der Daten. Die Telemetrie-Daten werden in der Spalte data als JSONB-Objekt gespeichert, was eine flexible Speicherung von verschiedenen Datentypen und Strukturen ermöglicht. Es ist somit egal wie viele und welche Capabilities ein Gerät hat, die Daten werden immer in der gleichen Spalte gespeichert.

Regeln¶

Die Tabelle rules bildet die Grundlage für eine regelbasierte Automatisierung. Jede Regel besitzt eine eindeutige UUID sowie einen Namen und eine Beschreibung. Die eigentliche Logik ist in zwei JSONB-Feldern hinterlegt: trigger beschreibt die Bedingung, die eine Regel auslöst, und action die daraufhin auszuführende Aktion. Über das Feld enabled lässt sich eine Regel aktivieren oder deaktivieren. Zusätzlich werden der Erstellungszeitpunkt (created_at) und der Zeitpunkt der letzten Auslösung (last_triggered_at) gespeichert.

TimescaleDB¶

Da die Telemetrie-Daten in der Regel sehr viele Datenpunkte pro Gerät und pro Tag erzeugen, wird die TimescaleDB-Erweiterung für PostgreSQL verwendet. Diese Erweiterung ist speziell auf die Speicherung und Abfrage von Zeitreihendaten optimiert und bietet Funktionen wie automatische Aggregation, Komprimierung und effiziente Abfragen von Zeitreihen.

Dabei werden sogenannte Hypertables verwendet, die die Daten in kleinere Abschnitte (Chunks) unterteilen und so die Abfragegeschwindigkeit erhöhen. Eine solche Tabelle wird für die Telemetrie-Daten verwendet.

Retention Policies¶

Die TimescaleDB bietet die Möglichkeit, Retention Policies zu definieren, um alte Daten automatisch zu löschen oder zu archivieren. Dies bietet die Möglichkeit, die Datenbankgröße zu kontrollieren und die Leistung der Abfragen zu verbessern. Zurzeit sind zwei Retention Policies definiert:

1. Telemetrie-Daten werden nach 7 Tagen komprimiert.
2. Telemetrie-Daten werden nach 30 Tagen gelöscht.

Queue¶

Im Backend wird eine Queue verwendet, um die Telemetrie-Daten der Geräte asynchron in die Datenbank zu schreiben. Das verhindert, dass das Backend und speziell die Datenbank durch die vielen Schreiboperationen blockiert werden und z.B. Leseoperationen von Diensten wie Grafana verlangsamt werden.

Die Queue wird mit BullMQ implementiert und verwendet Redis als Speicher. Redis hält die Daten im Arbeitsspeicher und ermöglicht dadurch eine hohe Durchsatzrate für die Queue. Im Betrieb legt der MQTT-Handler im Express.js-Backend einen Job in die Queue ab, der die Daten asynchron in die Datenbank schreibt. Darüber hinaus ist die Queue so eingerichtet, dass maximal fünf Schreiboperationen gleichzeitig ausgeführt werden, um die Datenbank nicht zu überlasten. Schlägt ein Schreibvorgang fehl, wird er bis zu dreimal wiederholt.

Damit das Frontend die Daten nicht verzögert angezeigt bekommt, wird vor dem Schreiben der Daten in die Datenbank ein Event mit diesen Daten über die WebSocket-Schnittstelle an das Frontend gesendet.

export async function handleTelemetry(mac, payload) {
  const uuid = await deviceRepo.getUuidByMac(mac);

  /* ...
    Error-Handling, Timestamp-Handling, Logging, etc. 
    ...
  */

  emitTelemetry(uuid, { uuid, timestamp: isoTimestamp, data });

  telemetryQueue.add(
    'telemetry',
    {
      uuid,
      payload: { timestamp: isoTimestamp, data }
    },
    {
      removeOnComplete: true,
      attempts: 3,
      backoff: {
        type: 'exponential',
        delay: 1000
      }
    }
  ).then(
    () => log.debug(`Telemetry queued for ${uuid} (${Object.keys(data).length} keys)`),
    (err) => log.error(`Failed to enqueue telemetry for ${uuid}:`, err)
  );
}