Merge pull request #14 from oslabs-beta/james/kafka-npm-package

James/kafka npm package

Author: hann7

chronos_npm_package/chronos.js

@@ -6,6 +6,7 @@ const MongoServerWrapper = require('./wrappers/MongoServerWrapper.js');
 const PostgresClientWrapper = require('./wrappers/PostgresClientWrapper.js');
 const PostgresServerWrapper = require('./wrappers/PostgresServerWrapper.js');
 const { validateInput, addNotifications } = require('./controllers/helpers');
+const { initialFetch } = require('./controllers/kafkaHelpers');
 
 let userConfig = {};
 const chronos = {};
@@ -101,6 +102,35 @@ chronos.track = () => {
   }
   return null;
 };
+
+  /**
+ * **********************************************
+ *              COLLECT KAFKA METRICS
+ * Only supports MongoDB and PostgreSQL for now!
+ * **********************************************
+ */
+
+chronos.kafka = function () {
+  const { database, jmxuri } = userConfig;
+  if (jmxuri === undefined){
+    console.log('No specified URI for a JMX Exporter');
+    return;
+  }
+
+  // Ensures that the provided URI returns correctly formatted data.
+  initialFetch(jmxuri);
+
+  if (database.type === 'MongoDB') {
+    mongo.connect(userConfig);
+    mongo.kafka(userConfig);
+  }
+
+  if (database.type === 'PostgreSQL') {
+    postgres.connect(userConfig);
+    postgres.kafka(userConfig);
+  }
+};
+
 /**
  * Wraps the gRPC server object to automatically write logs to user configed DB
  *

chronos_npm_package/controllers/helpers.js

@@ -4,8 +4,25 @@ const helpers = {
    * Throws an error on input valid data types or on missing fields
    * Sets the default interval to 5 seconds and dockerized to false
    */
+
+  /*
+  User Config object:
+  {
+    microservice: string - Name of the microservice. Will be used as a table name in the chronos's db
+    interval: number - The number of milliseconds between every collection of data
+    dockerized: boolean - Should be set to true if the service is running inside of a container
+    NEW! -> jmxuri: string - (optional) The address exposed by the JMX Exporter for collecting Kafka metrics  <- NEW!
+    database: {
+      connection: should be a string and only supports 'REST' and 'gRPC'
+      type: should be a string and only supports 'MongoDB' and 'PostgreSQL'.
+      URI: should be a connection string to the database where you intend Chronos to record metrics
+    }
+    notifications: an array - optional for configuring slack or email notifications
+  }
+
+  */
   validateInput(config) {
-    const { microservice, database, interval, dockerized, connection } = config;
+    const { microservice, database, interval, dockerized, jmxuri } = config;
 
     // Validate all required fields exist and are valid input types
     if (!microservice || typeof microservice !== 'string') {
@@ -25,6 +42,13 @@ const helpers = {
       );
     }
 
+    // validate that the jmxuri is a string
+    if (jmxuri && typeof jmxuri !== 'string') {
+      throw new Error(
+        'Invalid input for "jmxuri" in chronos-config.js: Please provide the address of the JMX Exporter'
+      );
+    }
+
     // Validate database type
     if (database.type !== 'PostgreSQL' && database.type !== 'MongoDB') {
       throw new Error(

chronos_npm_package/controllers/kafkaHelpers.js

@@ -0,0 +1,52 @@
+const fetch = require('node-fetch');
+
+const kafkaHelpers = {};
+
+// Creates an array full of objects that store values for various Kafka metrics
+kafkaHelpers.extractWord = function (str) {
+  const res = [];
+  const arr = str.split('\n'); // `/\n/`
+  for (const element of arr) {
+    if (
+      element &&
+      element.length !== 0 &&
+      element[0] !== '#' &&
+      element.substring(0, 3) !== 'jmx' &&
+      element.substring(0, 4) !== "'jmx"
+    ) {
+      const metric = element.split(' ')[0];
+      const value = Number(element.split(' ')[1]);
+      const time = Date.now();
+      const category = 'Event';
+      res.push({ metric, value, time, category });
+    }
+  }
+  return res;
+};
+
+// executes first to ensure that the provided jmxuri provides legitimate data.
+// if it doesnt then an error is thrown
+kafkaHelpers.initialFetch = function (jmxuri) {
+  fetch(jmxuri)
+    .then(response => {
+      if (response.status !== 200) {
+        throw new Error(`Error: The provided URI for the JMX exporter is invalid`);
+      } else {
+        console.log('Initial fetch to JMX Exporter was successful.');
+      }
+      return response.text();
+    })
+    .then(text => {
+      console.log('\nInitial Fetch Response Text:\n', text);
+    });
+};
+
+// fetches kafka metrics from the user-specified location of JMX prometheus and returns the processed result
+kafkaHelpers.kafkaFetch = function (config) {
+  return fetch(config.jmxuri)
+    .then(data => data.text())
+    .then(data => kafkaHelpers.extractWord(data))
+    .catch(err => console.log(err));
+};
+
+module.exports = kafkaHelpers;

chronos_npm_package/controllers/mongo.js

@@ -5,6 +5,8 @@ const CommunicationModel = require('../models/CommunicationModel');
 const ServicesModel = require('../models/ServicesModel');
 const HealthModelFunc = require('../models/HealthModel');
 const ContainerInfoFunc = require('../models/ContainerInfo');
+const KafkaModel = require('../models/KafkaModel');
+const { kafkaFetch } = require('./kafkaHelpers.js');
 require('../models/ContainerInfo');
 
 // Handle deprecation warnings
@@ -209,6 +211,7 @@ chronos.health = ({ microservice, interval }) => {
       })
       .catch(err => console.log('Error saving health data: ', err.message));
   }, interval);
+  console.log('Interval set, recording health data');
 };
 
 /**
@@ -302,6 +305,29 @@ chronos.docker = ({ microservice, interval }) => {
     });
 };
 
+/*
+ This function takes as a parameter the promise returned from the kafkaFetch().
+It then takes the returned array of metrics, turns them into documents based on
+KafkaModel.js, and inserts them into the db at the provided uri with insertMany()
+*/
+chronos.kafka = function (userConfig) {
+  // fetch the data from Kafka with kafkaFetch()
+  // then take turn each result in the returned array into a kafkaModel doc
+  // insertMany into the the KafkaModel
+  console.log('Starting Kafka Collection');
+  setInterval(() => {
+    kafkaFetch(userConfig)
+      .then(parsedArray => {
+        const documents = [];
+        for (const metric of parsedArray) {
+          documents.push(KafkaModel(metric));
+        }
+        return KafkaModel.insertMany(documents);
+      })
+      .catch(err => console.log('Error inserting kafka documents: ', err));
+  }, userConfig.interval);
+};
+
 // // grabs container data for multiple containers info - TBD
 // chronos.dockerInfo = ({ microservice, interval }) => {
 //   si.dockerInfo()
@@ -311,17 +337,4 @@ chronos.docker = ({ microservice, interval }) => {
 //     .catch(err => console.log('Error saving health data: ', err.message));
 // };
 
-// import data from JMX Exporter port
-// chronos.kafka = () => {
-//   // write fetch request to localhost:12345/metrics
-//   // then store metrics in database
-//   fetch('localhost:12345/metrics')
-//     .then(data => {
-//       console.log(data);
-//       const parsedData = parser.toTree(data);
-//       console.log(parsedData);
-//     })
-//     .catch(err => console.log(err));
-// };
-
 module.exports = chronos;

chronos_npm_package/controllers/postgres.js

@@ -2,6 +2,7 @@
 const si = require('systeminformation');
 const { Client } = require('pg');
 const alert = require('./alert');
+const { kafkaFetch } = require('./kafkaHelpers');
 
 let client;
 
@@ -377,4 +378,63 @@ chronos.docker = function ({ microservice, interval }) {
     });
 };
 
+// Constructs a parameterized query string for inserting multiple data points into
+// the kafkametrics db based on the number of data points;
+function createQueryString(numRows) {
+  let query = `
+    INSERT INTO
+      kafkametrics (metric, value, category, time)
+    VALUES
+  `;
+  for (let i = 0; i < numRows; i++) {
+    const newRow = `($${4 * i + 1}, $${4 * i + 2}, $${4 * i + 3}, TO_TIMESTAMP($${4 * i + 4}))`;
+    query = query.concat(newRow);
+    if (i !== numRows - 1) query = query.concat(',');
+  }
+  query = query.concat(';');
+  return query;
+}
+
+// Places the values being inserted into postgres into an array that will eventually
+// hydrate the parameterized query
+function createQueryArray(dataPointsArray) {
+  const queryArray = [];
+  for (const element of dataPointsArray) {
+    queryArray.push(element.metric);
+    queryArray.push(element.value);
+    queryArray.push(element.category);
+    queryArray.push(element.time / 1000); // Converts milliseconds to seconds to work with postgres
+  }
+  return queryArray;
+}
+
+chronos.kafka = function (userConfig) {
+  // create kafkametrics table if it does not exist
+  const createTableQuery = `
+    CREATE TABLE IF NOT EXISTS kafkametrics (
+      _id SERIAL PRIMARY KEY,
+      metric VARCHAR(200),
+      value FLOAT DEFAULT 0.0,
+      category VARCHAR(200) DEFAULT 'event',
+      time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+    );`;
+
+  client
+    .query(createTableQuery)
+    .catch(err => console.log('Error creating kafkametrics table in PostgreSQL:\n', err));
+
+  setInterval(() => {
+    kafkaFetch(userConfig)
+      .then(parsedArray => {
+        const numDataPoints = parsedArray.length;
+        const queryString = createQueryString(numDataPoints);
+        const queryArray = createQueryArray(parsedArray);
+        //console.log('POSTGRES QUERY STRING: ', queryString);
+        //console.log('POSTGRES QUERY ARRAY', queryArray);
+        return client.query(queryString, queryArray);
+      })
+      .catch(err => console.log('Error inserting kafka metrics into PostgreSQL:\n', err));
+  }, userConfig.interval);
+};
+
 module.exports = chronos;

chronos_npm_package/models/KafkaModel.js

@@ -7,14 +7,15 @@ const KafkaSchema = new Schema({
     type: Date,
     default: Date.now(),
   },
-  activecontrollercount: {
-    type: Number,
+  metric: {
+    type: String,
   },
-  offlinepartitionscount: {
+  value: {
     type: Number,
   },
-  uncleanleaderelectionspersec: {
-    type: Number,
+  category: {
+    type: String,
+    default: '',
   },
 });
 

chronos_npm_package/package.json

@@ -14,6 +14,7 @@
     "axios": "^0.21.1",
     "hpropagate": "0.0.7",
     "mongoose": "^5.11.13",
+    "node-fetch": "^2.6.7",
     "nodemailer": "^6.4.17",
     "pg": "^8.5.1",
     "systeminformation": "^4.34.7"