switching to post metrics to influxdb

Author: Soichi Hayashi

api/config/index.js

@@ -109,6 +109,19 @@ exports.metrics = {
     api: "http://10.0.0.10:2080", //TODO
 }
 
+exports.influxdb = {
+    connection: {
+        url: "http://brainlife_influxdb_1:8086",
+        token: "mydevtoken",
+    },
+    org: "brainlife",
+    bucket: "brainlife",
+    location: "localhost",
+
+    countInterval: 10*1000, 
+    healthInterval: 10*1000, 
+}
+
 //for event handler
 exports.event = {
     amqp: {
@@ -221,6 +234,7 @@ exports.groupanalysis = {
     },
 }
 
+
 /*
 function connect_dc(cb) {
     var conn = new ssh2.Client();

bin/daily_metrics.js

@@ -1,27 +1,22 @@
 #!/usr/bin/env node
 
-const winston = require('winston');
-
 const config = require('../api/config');
-config.logger.winston.transports[0].level = 'error';
 const db = require('../api/models');
+const influx = require('@influxdata/influxdb-client');
 
 const mongoose = require("mongoose");
 mongoose.set("debug", false); //suppress log
 
 let graphite_prefix = process.argv[2];
-if(!graphite_prefix) {
-    //console.error("usage: bl_metrics.js <graphite_prefix>");
-    graphite_prefix = "dev"
-}
+if(!graphite_prefix) graphite_prefix = "dev"
 
 function count_apps(d) {
     return new Promise((resolve, reject)=>{
         db.Apps.count({create_date: {$lt: d}, removed: false}, (err, count)=>{
             if(err) return reject(err);
             const time = Math.round(d.getTime()/1000);
             console.log(graphite_prefix+".app.count "+count+" "+time);
-            resolve();
+            resolve(count);
         });
     });
 }
@@ -42,7 +37,7 @@ function count_dataset(d) {
                 if(err) return reject(err);
                 const time = Math.round(d.getTime()/1000);
                 console.log(graphite_prefix+".dataset.count "+count+" "+time);
-                resolve();
+                resolve(count);
             });
         });
     });
@@ -54,7 +49,7 @@ function count_project(d) {
             if(err) return reject(err);
             const time = Math.round(d.getTime()/1000);
             console.log(graphite_prefix+".project.count "+count+" "+time);
-            resolve();
+            resolve(count);
         });
     });
 }
@@ -66,7 +61,7 @@ function count_public_project(d) {
             if(err) return reject(err);
             const time = Math.round(d.getTime()/1000);
             console.log(graphite_prefix+".project.public "+count+" "+time);
-            resolve();
+            resolve(count);
         });
     });
 }
@@ -77,19 +72,28 @@ function count_private_project(d) {
             if(err) return reject(err);
             const time = Math.round(d.getTime()/1000);
             console.log(graphite_prefix+".project.private "+count+" "+time);
-            resolve();
+            resolve(count);
         });
     });
 }
 
 db.init(async function(err) {
     if(err) throw err;
     let today = new Date();
-    await count_apps(today); 
-    await count_dataset(today); 
-    await count_project(today); 
-    await count_public_project(today); 
-    await count_private_project(today); 
+
+    const writeApi = new influx.InfluxDB(config.influxdb.connection)
+        .getWriteApi(config.influxdb.org, config.influxdb.bucket, 'ns')
+    writeApi.useDefaultTags({location: config.influxdb.location})
+    const point = new influx.Point("warehouse");
+    point.timestamp(today);
+    point.intField("app", await count_apps(today));
+    point.intField("dataset", await count_dataset(today));
+    point.intField("all_project", await count_project(today));
+    point.intField("public_project", await count_public_project(today));
+    point.intField("private_project", await count_private_project(today));
+    writeApi.writePoint(point);
+    writeApi.close();
+
     db.disconnect();
 });
 

bin/event_handler.js

@@ -9,6 +9,7 @@ const rp = require('request-promise-native');
 const redis = require('redis');
 const fs = require('fs');
 const child_process = require('child_process');
+const influx = require('@influxdata/influxdb-client');
 
 const pkg = require('../package.json');
 
@@ -19,6 +20,7 @@ const common = require('../api/common');
 // TODO  Look for failed tasks and report to the user/dev?
 let acon;
 
+
 console.log("connected to mongo");
 db.init(async err=>{
     common.connectAMQP((err, conn)=>{
@@ -28,8 +30,9 @@ db.init(async err=>{
     });
 
     await common.connectRedis();
-    setInterval(emit_counts, 1000*config.metrics.counts.interval);  //usually 24 hours?
-    setInterval(emit_health_counts, 1000*config.metrics.health_counts.interval);  //usually 5min
+
+    setInterval(emit_counts, config.influxdb.countInterval);  //usually 24 hours?
+    setInterval(emit_health_counts, config.influxdb.countInterval);  //usually 5min
 });
 
 function subscribe() {
@@ -129,7 +132,9 @@ function subscribe() {
     });
 }
 
-let counts = {};
+let counts = {
+    test: 10,
+};
 function inc_count(path) {
     if(counts[path] === undefined) counts[path] = 0;
     counts[path]++;
@@ -142,18 +147,31 @@ function isValidationTask(task) {
 
 function emit_counts() {
     //emit graphite metrics
+    /*
     let out = "";
     for(let key in counts) {
         out += config.metrics.counts.prefix+"."+key+" "+counts[key]+" "+new Date().getTime()/1000+"\n";
     }
     fs.writeFileSync(config.metrics.counts.path, out);
+    */
+    const writeApi = new influx.InfluxDB(config.influxdb.connection)
+        .getWriteApi(config.influxdb.org, config.influxdb.bucket, 'ns')
+
+    writeApi.useDefaultTags({location: config.influxdb.location})
+    const point = new influx.Point("warehouse.counts");
+    point.timestamp(new Date());
+    for(let key in counts) {
+        point.intField(key, counts[key]);
+    }
+    writeApi.writePoint(point);
+    writeApi.close();
 
     counts = {}; //reset all counters
 }
 
 const health_counts = {
     tasks: 0,
-    instanceS: 0,
+    instances: 0,
 }
 async function emit_health_counts() {
     var report = {
@@ -176,12 +194,25 @@ async function emit_health_counts() {
     await common.redisClient.set("health.warehouse.event."+process.env.HOSTNAME+"-"+process.pid, JSON.stringify(report));
 
     //emit graphite metrics
+    /*
     const time = new Date().getTime()/1000;
     let out = `
 ${config.metrics.health_counts.prefix}.health.tasks ${health_counts.tasks} ${time}
 ${config.metrics.health_counts.prefix}.health.instances ${health_counts.instances} ${time}
 `;
     fs.writeFileSync(config.metrics.health_counts.path, out);
+    */
+
+    const writeApi = new influx.InfluxDB(config.influxdb.connection)
+        .getWriteApi(config.influxdb.org, config.influxdb.bucket, 'ns')
+    writeApi.useDefaultTags({location: config.influxdb.location})
+
+    const point = new influx.Point("warehouse.health");
+    point.intField('tasks', health_counts.tasks);
+    point.intField('instances', health_counts.instances);
+    point.timestamp(new Date());
+    writeApi.writePoint(point);
+    writeApi.close();
 
     health_counts.tasks = 0;
     health_counts.instances = 0;

package-lock.json

@@ -21,6 +21,7 @@
   "homepage": "https://github.com/brain-life/warehouse#readme",
   "dependencies": {
     "@cdxoo/dbscan": "^1.0.2",
+    "@influxdata/influxdb-client": "^1.29.0",
     "amqp": "^0.2.7",
     "archiver": "^5.3.1",
     "async": "^3.2.3",