prometheus data processing and storage

Author: eishakaushal

chronos_npm_package/chronos.js

@@ -138,6 +138,20 @@ class Chronos {
     }
   }
 
+  async docker () {
+    await utilities.testMetricsQuery(this.config);
+    if (this.config.database.type === 'MongoDB') {
+      mongo.connect(this.config);
+      mongo.serverQuery(this.config);
+      // return mongo.modifyMetrics(this.config);
+    } else if (this.config.database.type === 'PostgreSQL') {
+      postgres.connect(this.config);
+      postgres.serverQuery(this.config);
+    } else {
+      throw new Error('The only allowed database types are MongoDB and PostgreSQL');
+    }
+  }
+
   ServerWrapper(server, proto, methods) {
     /**
      * Wraps the gRPC server object to automatically write logs to provided DB

chronos_npm_package/controllers/mongo.js

@@ -169,12 +169,14 @@ mongo.serverQuery = async config => {
   await mongo.setQueryOnInterval(config);
 };
 
-mongo.saveService = async config => {
+mongo.saveService = config => {
   let microservice;
   if (config.mode === 'kafka') {
     microservice = 'kafkametrics';
   } else if (config.mode === 'kubernetes') {
     microservice = 'kubernetesmetrics';
+  } else if (config.mode === 'docker') {
+    microservice = `${config.containerName}`;
   } else {
     throw new Error('Unrecongnized mode');
   }
@@ -205,6 +207,11 @@ mongo.setQueryOnInterval = async config => {
   } else if (config.mode === 'kubernetes') {
     model = KubernetesModel;
     metricsQuery = await utilities.promMetricsQuery;
+  } else if (config.mode === 'docker') {
+    model = ContainerInfoFunc(`${config.containerName}`);
+    //console.log('setQueryOnInterval line 212 dockerModel:', ContainerInfoFunc(`${config.containerName}`));
+    metricsQuery = utilities.promMetricsQuery;
+    //console.log('setQueryOnInterval line 214 metricsQuery:', metricsQuery);
   } else {
     throw new Error('Unrecognized mode');
   }
@@ -230,24 +237,25 @@ mongo.setQueryOnInterval = async config => {
           ///////
           length = await mongo.addMetrics(parsedArray, config.mode, currentMetricNames, model);
         }
-        // const documents = [];
-        // for (const metric of parsedArray) {
-        //   /**
-        //    * This will check if the current metric in the parsed array
-        //    * evaluates to true within the currentMetricNames object
-        //    * which is updated by the user when they select/deselect metrics on the electron app
-        //    * helping to avoid overloading the db with unnecessary data.
-        //    */
-
-        //   if (currentMetricNames[metric.metric]) documents.push(model(metric));
-        // }
-        // await model.insertMany(parsedArray, err => {
-        //   if (err) {
-        //     console.error(err)
-        //   } else {
-        //     console.log(`${config.mode} metrics recorded in MongoDB`)
-        //   }
-        // });
+        const documents = [];
+        for (const metric of parsedArray) {
+          /**
+           * This will check if the current metric in the parsed array
+           * evaluates to true within the currentMetricNames object
+           * which is updated by the user when they select/deselect metrics on the electron app
+           * helping to avoid overloading the db with unnecessary data.
+           */
+
+          if (currentMetricNames[metric.metric]) documents.push(model(metric));
+        }
+        await model.insertMany(parsedArray, err => {
+          if (err) {
+            console.error(err)
+          } else {
+            console.log(`${config.mode} metrics recorded in MongoDB`)
+          }
+        });
+        
         let allMetrics = await model.find({});
         console.log('allMetrics.length: ', allMetrics.length);
         console.log("🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 start creating dashboards 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡 🟡")

chronos_npm_package/controllers/utilities.js

@@ -55,11 +55,11 @@ const helpers = {
       );
     }
 
-    const modeTypes = ['kafka', 'kubernetes', 'microservices'];
+    const modeTypes = ['kafka', 'kubernetes', 'microservices', 'docker'];
 
     if (!mode || !modeTypes.includes(mode)) {
       throw new Error(
-        'You must input a mode into your chronos.config file. The mode may either be "kubernetes", "kafka", or "microservice"'
+        'You must input a mode into your chronos.config file. The mode may either be "kubernetes", "kafka", "microservice", or "docker"'
       );
     }
 
@@ -69,7 +69,7 @@ const helpers = {
       );
     }
 
-    if (mode === 'kubernetes') {
+    if (mode === 'kubernetes' || mode === 'docker') {
       if (
         !promService ||
         typeof promService !== 'string' ||
@@ -139,7 +139,7 @@ const helpers = {
   getMetricsURI: config => {
     if (config.mode === 'kafka') {
       return config.jmxuri;
-    } else if (config.mode === 'kubernetes') {
+    } else if (config.mode === 'kubernetes' || config.mode === 'docker') {
       return `http://${config.promService}:${config.promPort}/api/v1/query?query=`;
     } else {
       throw new Error('Unrecognized mode');
@@ -153,14 +153,14 @@ const helpers = {
    */
   testMetricsQuery: async config => {
     let URI = helpers.getMetricsURI(config);
-    if (config.mode === 'kubernetes') URI += 'up';
+    URI += 'up';
     try {
       const response = await axios.get(URI);
-      if (response.status !== 200) console.error('Invalid response from metrics server:', URI);
+      if (response.status !== 200) console.error('Invalid response from metrics server:', URI, response.status, response.data);
       else console.log('Successful initial response from metrics server:', URI);
       return response;
     } catch (error) {
-      console.log(error);
+      console.error(error);
       throw new Error('Unable to query metrics server: ' + URI);
     }
   },
@@ -226,38 +226,30 @@ const helpers = {
    */
   promMetricsQuery: async config => {
     const URI = helpers.getMetricsURI(config);
-    const query = URI + encodeURIComponent('{__name__=~".+",container=""}');
-    try {
-      const response = await axios.get(query);
-      //console.log("response is: ", response);
-      return helpers.parseProm(response.data.data.result);
-    } catch (error) {
-      return console.error(config.mode, '|', 'Error fetching from URI:', URI, '\n', error);
+    let query;
+    if (config.mode === 'docker') {
+      query = URI + encodeURIComponent(`{__name__=~".+",name="${config.containerName}"}`);
+    } else {
+      query = URI + encodeURIComponent('{__name__=~".+",container=""}');
     }
-  },
-
-  promMetrics: async config => {
-    const URI = `http://${config.promService}:${config.promPort}/api/v1/query?query=`;
-    const query = URI + encodeURIComponent('{__name__=~".+",container=""}');
     try {
       const response = await axios.get(query);
-      //console.log("response is: ", response);
-      return response.data.data.result;
+      //console.log('promMetricsQuery line 236:', response.data.data.result);
+      return helpers.parseProm(config, response.data.data.result);
     } catch (error) {
-      return console.error('Error fetching from URI:', URI, '\n', error);
+      return console.error(config.mode, '|', 'Error fetching from URI:', URI, '\n', error);
     }
   },
 
-
   /**
    * Parses response from Prometheus request and returns object with
    * @param {*} data
    * @returns bject with the gathered metric, value, time gathered, and category of event
    */
-  parseProm: data => {
+  parseProm: (config, data) => {
     const res = [];
     const time = Date.now();
-    const category = 'Event';
+    const category = config.mode === 'docker' ? `${config.containerName}`: 'Event';
 
     /**
      * Opportunity for improvement: Prometheus may query metrics that have the same job + instance + metric
@@ -272,10 +264,18 @@ const helpers = {
     const names = new Set();
 
     for (const info of data) {
-      if (!info.metric.job) continue;
-      // Set the base name using the job, IP, and metric __name__
-      let wholeName = info.metric.job + '/' + info.metric.instance + '/' + info.metric['__name__'];
-      let name = wholeName.replace(/.*\/.*\//g, '');
+      let wholeName;
+      let name;
+      if (config.mode === 'docker'){
+        if (!info.metric.name) continue;
+        wholeName = info.metric['__name__'];
+        name = wholeName.replace(/.*\/.*\//g, '');
+      } else {
+        if (!info.metric.job) continue;
+        // Set the base name using the job, IP, and metric __name__
+        wholeName = info.metric.job + '/' + info.metric.instance + '/' + info.metric['__name__'];
+        name = wholeName.replace(/.*\/.*\//g, '');
+      }
       if (names.has(name)) continue;
       else {
         names.add(name);
@@ -303,6 +303,8 @@ const helpers = {
     return res;
   },
 
+  
+
   createGrafanaDashboard: async (
     metric,
     datasource,

chronos_npm_package/models/ContainerInfo.js

@@ -2,53 +2,24 @@ const mongoose = require('mongoose');
 
 const { Schema } = mongoose;
 
-const ContainerSchema = new Schema({
-  // Added additional schema for Docker container stats (9).
-  containerid: {
-    type: String,
-  },
-  containername: {
-    type: String,
-  },
-  platform: {
-    type: String,
-  },
-  starttime: {
+const DockerSchema = new Schema({
+  time: {
     type: Date,
+    default: Date.now(),
   },
-  memoryusage: {
-    type: Number, // bytes
-  },
-  memorylimit: {
-    type: Number,
-  },
-  memorypercent: {
-    type: Number,
+  metric: {
+    type: String,
   },
-  cpupercent: {
+  value: {
     type: Number,
   },
-  networkreceived: {
-    type: Number, // bytes
-    default: 0,
-  },
-  networksent: {
-    type: Number, // bytes
-    default: 0,
-  },
-  processcount: {
-    type: Number, // count
-  },
-  restartcount: {
-    type: Number, // count
-  },
-  time: {
-    type: Date,
-    default: Date.now(),
+  category: {
+    type: String,
+    default: '',
   },
 });
 
 module.exports = ContainerName => {
   console.log('Inside Docker Schema ContainerInfo.js LN52', ContainerName)
-  return mongoose.model(ContainerName, ContainerSchema);
+  return mongoose.model(ContainerName, DockerSchema);
 };

examples/docker/books/BookServer.js

@@ -14,7 +14,7 @@ chronos.propagate();
 const app = express();
 
 app.use(express.json());
-app.use('/', chronos.track());
+chronos.docker();
 
 app.use(cors());
 app.use('/', express.static(path.resolve(__dirname, '../frontend')));

examples/docker/books/chronos-config.js

@@ -6,9 +6,11 @@ module.exports = {
   microservice: 'books',
   interval: 5000,
 
-  // Mode Specific
-  mode: 'microservices',
-  dockerized: true,
+ // Mode Specific
+ mode: 'docker',
+ promService: 'docker.for.mac.localhost',
+ promPort: 9090,
+ containerName: 'books',
 
   database: {
     connection: 'REST',

examples/docker/books/chronos_npm_package/chronos.js

@@ -108,7 +108,9 @@ class Chronos {
     if (this.config.database.type === 'MongoDB') {
       mongo.connect(this.config);
       mongo.serverQuery(this.config);
-    } else if (this.config.database.type === 'PostgreSQL') {
+    }
+
+    else if (this.config.database.type === 'PostgreSQL') {
       postgres.connect(this.config);
       postgres.serverQuery(this.config);
     } else {
@@ -120,6 +122,24 @@ class Chronos {
     // Test metrics server connection
     await utilities.testMetricsQuery(this.config);
 
+    if (this.config.database.type === 'MongoDB') {
+      await mongo.connect(this.config);
+      await mongo.storeGrafanaAPIKey(this.config);
+      //await mongo.createGrafanaDashboards(this.config);
+      mongo.serverQuery(this.config);
+      // return mongo.modifyMetrics(this.config);
+    }
+
+    else if (this.config.database.type === 'PostgreSQL') {
+      postgres.connect(this.config);
+      postgres.serverQuery(this.config);
+    } else {
+      throw new Error('The only allowed database types are MongoDB and PostgreSQL');
+    }
+  }
+
+  async docker () {
+    await utilities.testMetricsQuery(this.config);
     if (this.config.database.type === 'MongoDB') {
       mongo.connect(this.config);
       mongo.serverQuery(this.config);