feat: add influxdb

package.json

@@ -23,6 +23,7 @@
   "author": "Trickfilm400",
   "license": "ISC",
   "dependencies": {
+    "@influxdata/influxdb-client": "^1.35.0",
     "convict": "^6.2.4",
     "mqtt": "^5.10.1",
     "mysql": "^2.18.1",

src/config/index.ts

@@ -24,6 +24,38 @@ const config = convict({
       env: 'VANTAGE_PORT',
     },
   },
+  influxdb: {
+    enabled: {
+      doc: 'if influxdb should be used',
+      format: Boolean,
+      default: false,
+      env: 'INFLUXDB_ENABLED',
+    },
+    url: {
+      doc: 'The URL Endpoint for InfluxDB',
+      format: String,
+      default: '',
+      env: 'INFLUXDB_URL',
+    },
+    api_token: {
+      doc: 'The API Token for InfluxDB',
+      format: String,
+      default: '',
+      env: 'INFLUXDB_API_TOKEN',
+    },
+    organisation: {
+      doc: 'The organisation for InfluxDB',
+      format: String,
+      default: '',
+      env: 'INFLUXDB_ORGANISATION',
+    },
+    bucket: {
+      doc: 'The bucket for InfluxDB',
+      format: String,
+      default: '',
+      env: 'INFLUXDB_BUCKET',
+    },
+  },
   mysql: {
     enabled: {
       doc: 'The IP address Mysql should connect to',

src/dataHandler/influxdb.ts

@@ -0,0 +1,38 @@
+import { DataReceiver } from '../lib/dataReceiver';
+import { DataPackage } from '../interfaces/IPackage';
+import { InfluxDB, Point, WriteApi } from '@influxdata/influxdb-client';
+import config from '../config';
+import { logger } from '../core/logger';
+
+export class Influxdb extends DataReceiver<DataPackage> {
+  private writerClient: WriteApi | undefined;
+  constructor() {
+    super();
+    if (config.get('influxdb.enabled')) {
+      const influxDB = new InfluxDB({
+        url: config.get('influxdb.url'),
+        token: config.get('influxdb.api_token'),
+      });
+      this.writerClient = influxDB.getWriteApi(
+        config.get('influxdb.organisation'),
+        config.get('influxdb.bucket')
+      );
+    }
+  }
+  onData(data: DataPackage): void {
+    logger.debug('Influxdb data', data);
+    if (!this.writerClient) return;
+
+    let key: keyof DataPackage;
+    const points: Point[] = [];
+    for (key in data) {
+      const point = new Point(key);
+      point.floatField('value', data[key]);
+      points.push(point);
+    }
+    this.writerClient.writePoints(points);
+  }
+  cleanup() {
+    this.writerClient?.close();
+  }
+}

src/index.ts

@@ -16,6 +16,8 @@ import Mqtt from './dataHandler/mqtt';
 const mqtt = new Mqtt();
 import { Prometheus } from './dataHandler/prometheus';
 const prometheus = new Prometheus();
+import { Influxdb } from './dataHandler/influxdb';
+const influxdb = new Influxdb();
 
 process.on('SIGINT', () => {
   console.log('\n');
@@ -24,6 +26,7 @@ process.on('SIGINT', () => {
     .then(() => socket.cleanup())
     .then(() => mqtt.cleanup())
     .then(() => prometheus.cleanup())
+    .then(() => influxdb.cleanup())
     .then(() => httpServer.cleanup())
     .then(() => {
       process.exit(0);