Feature: Added MySQL database telemetry log support

src/dispatcher/dispatcher.js

@@ -2,6 +2,7 @@ const winston = require('winston');
     require('winston-daily-rotate-file');
     require('./kafka-dispatcher');
     require('./cassandra-dispatcher');
+    require('./mysql-dispatcher');
 
 const defaultFileOptions = {
     filename: 'dispatcher-%DATE%.log',
@@ -27,6 +28,9 @@ class Dispatcher {
         } else if (this.options.dispatcher === 'cassandra') {
             this.logger.add(winston.transports.Cassandra, this.options);
             console.log('Cassandra transport enabled !!!');
+        } else if (this.options.dispatcher === 'mysql') {
+            this.logger.add(winston.transports.mysql, this.options);
+            console.log('mysql transport enabled !!!');
         } else { // Log to console
             this.options.dispatcher = 'console'
             const config = Object.assign({json: true,stringify: (obj) => JSON.stringify(obj)}, this.options);

src/dispatcher/mysql-dispatcher.js

@@ -0,0 +1,83 @@
+"use strict";
+Object.defineProperty(exports, "__esModule", { value: true });
+var mysql = require('mysql');
+const winston = require('winston');
+
+class MysqlDispatcher extends winston.Transport {
+
+    constructor(options) {
+        super();
+
+        this.options = options;
+
+        this.connection = mysql.createConnection({
+            host: this.options.mysqlHost,
+            port: this.options.mysqlPort,
+            user: this.options.mysqlUser,
+            password: this.options.mysqlPassword,
+            database: this.options.mysqlDatabase
+        });
+
+        // check mysql connection
+        this.connection.connect(function (err) {
+
+            // check if any issue with connection
+            if (err) {
+                console.log("Mysql Connection Error",err)
+                process.exit(1);
+            } else {
+                console.log("MySQL DB Connected!");
+            }
+
+        });
+
+        if (this.connection) {
+
+            let table_name = this.options.table;
+
+            // At the first time run, This command will run to create table if table not exist
+            let sql = "CREATE TABLE IF NOT EXISTS ?? (id int(11) NOT NULL AUTO_INCREMENT,api_id varchar(255) NOT NULL,ver varchar(15) NOT NULL,params longtext NOT NULL CHECK (json_valid(params)),ets bigint(20) NOT NULL,events longtext NOT NULL CHECK(json_valid(events)),channel varchar(50) NOT NULL,pid varchar(50) NOT NULL,mid varchar(50) NOT NULL,syncts bigint(20) NOT NULL,PRIMARY KEY (id))";
+
+            // for run table creation query
+            this.connection.query(sql,[table_name], function (err, result) {
+                if (err) throw console.log("error while creating db");
+                console.log("Telemetry table created");
+            });
+        }
+    }
+
+    log(level, msg, meta, callback) {
+        console.log("msg", msg);
+        let msgData = JSON.parse(msg);
+        console.log("msgData", msgData);
+
+        let promises = [];
+        let table_name = this.options.table;
+
+        try {
+            for (const iterator of msgData.events) {
+
+                //insert query to mysql table one by one
+                promises.push(
+                    this.connection.query("INSERT INTO ?? (api_id, ver, params, ets, events, channel, pid, mid, syncts) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)", [table_name, msgData.id, msgData.ver, JSON.stringify(msgData.params), msgData.ets, JSON.stringify(iterator), iterator.context.channel, iterator.context.pdata.pid, msgData.mid, msgData.syncts])
+                );
+
+            }
+
+        } catch (err) {
+            return callback(err);
+        }
+
+        Promise.all(promises)
+            .then(() => {
+                callback()
+            }).catch((err) => {
+                console.log("Unable to insert data", err);
+                return callback(err);
+            })
+        }
+}
+
+winston.transports.mysql = MysqlDispatcher;
+
+module.exports = { MysqlDispatcher }

src/envVariables.js

@@ -19,6 +19,12 @@ const envVariables = {
     contactPoints: (process.env.telemetry_cassandra_contactpoints || 'localhost').split(','),
     cassandraTtl: process.env.telemetry_cassandra_ttl,
     port: process.env.telemetry_service_port || 9001,
-    threads: process.env.telemetry_service_threads || os.cpus().length
+    threads: process.env.telemetry_service_threads || os.cpus().length,
+    table: process.env.telemetry_table_name || "telemetry",
+    mysqlHost: process.env.MYSQL_HOST,
+    mysqlUser: process.env.MYSQL_USER,
+    mysqlPort: process.env.MYSQL_PORT,
+    mysqlDatabase: process.env.MYSQL_DATABASE,
+    mysqlPassword: process.env.MYSQL_PASSWORD,
 }
 module.exports = envVariables;

src/package.json

@@ -18,6 +18,7 @@
     "kafka-node": "~2.6.1",
     "lodash": "^4.17.10",
     "morgan": "~1.9.0",
+    "mysql": "^2.18.1",
     "request": "2.87.0",
     "uuid": "~3.2.1",
     "winston": "~2.4.3",
@@ -33,4 +34,4 @@
     "chai-http": "^4.0.0",
     "codacy-coverage": "3.0.0"
   }
-}
+}