Longterm metrics DB

Added support to save long term metrics to couchDB

dispatch_system/constants.json

@@ -4,7 +4,8 @@
     "master_address": "localhost",
     "grunt_host": "https://www.namandixit.net/lovecraftian_nightmares/grunt",
     "couchdb_host": "localhost:5984",
-    "couchdb_db_name": "serverless",
+    "function_db_name": "serverless",
+    "metrics_db_name": "metrics",
     "network": {
         "network_bridge": "hybrid_kafka-serverless",
         "internal": {

dispatch_system/dispatch_daemon/index.js

@@ -11,7 +11,7 @@ const fs = require('fs')
 const fetch = require('node-fetch');
 
 let metadataDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
-metadataDB = metadataDB + "/" + constants.couchdb_db_name + "/"
+metadataDB = metadataDB + "/" + constants.function_db_name + "/"
 
 const kafka = require('kafka-node')
 const logger = libSupport.logger

dispatch_system/dispatch_manager/explicit_chain_handler.js

@@ -8,7 +8,7 @@ const fetch = require('node-fetch')
 const constants = require('../constants.json')
 const secrets = require('./secrets.json')
 let metadataDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
-metadataDB = metadataDB + "/" + constants.couchdb_db_name + "/"
+metadataDB = metadataDB + "/" + constants.function_db_name + "/"
 const logger = libSupport.logger
 
 const registry_url = constants.registry_url

dispatch_system/dispatch_manager/index.js

@@ -17,7 +17,7 @@ const apiSpec = require('./swagger.json');
  * URL to the couchdb database server used to store function metadata
  */
 let metadataDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
-metadataDB = metadataDB + "/" + constants.couchdb_db_name + "/"
+metadataDB = metadataDB + "/" + constants.function_db_name + "/"
 
 const app = express()
 const libSupport = require('./lib')
@@ -553,18 +553,7 @@ function autoscalar() {
 
 }
 
-function periodicMetricBroadcast() {
-    let message = {}, flag = false
-    functionToResource.forEach((functionHeap, functionHash) => {
-        if (functionHeap.length > 0) {
-            message[functionHash] = functionHeap.length
-            libSupport.metrics.collectMetrics({type: "scale", value: functionHeap.length, functionHash: functionHash})
-        }
-    })
-}
-
 setInterval(libSupport.viterbi, 1000, functionBranchTree)
 setInterval(autoscalar, 1000);
 setInterval(dispatch, 1000);
-// setInterval(periodicMetricBroadcast, 5000)
 app.listen(port, () => logger.info(`Server listening on port ${port}!`))
\ No newline at end of file

dispatch_system/dispatch_manager/metrics.js

 const constants = require('.././constants.json');
+const secrets = require('./secrets.json')
+const fetch = require('node-fetch');
 const alpha = 0.99
 let log_channel = constants.topics.log_channel,
     metrics = {  }
 
+
+let metricsDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
+metricsDB = metricsDB + "/" + constants.metrics_db_name + "/"
 let kafka = require('kafka-node'),
     Producer = kafka.Producer,
     client = new kafka.KafkaClient({
@@ -53,17 +58,12 @@ function collectMetrics(metric) {
  * longterm - longterm data is held and averaged out over a period of time.
  *            longterm data is calculated using Expontential Moving Average (EMA)
  */
-function broadcastMetrics() {
-    
+async function broadcastMetrics() {
     if (Object.keys(metrics).length !== 0) {
-        for (let [functionHash, metric] of Object.entries(metrics)) {
+        for (let [functionHash, metricData] of Object.entries(metrics)) {
+            
+            let {metric, revision} = await fetchData(functionHash, metricData)
             
-            if (metric.longterm === undefined) {
-                metric.longterm = {
-                        coldstart: 0,
-                        warmstart: 0
-                }
-            }
             /**
             * Shortterm moving average
             */
@@ -75,12 +75,26 @@ function broadcastMetrics() {
              * Longterm exponential moving average
              */
             if (metric.shortterm.coldstart != 0)
-                metric.longterm.coldstart = (metric.longterm.coldstart != 0)? metric.longterm.coldstart * alpha
+                metric.longterm.coldstart = (metric.longterm.coldstart != 0) ? metric.longterm.coldstart * alpha
                     + metric.shortterm.coldstart * (1 - alpha) : metric.shortterm.coldstart
             if (metric.shortterm.warmstart != 0)
-                metric.longterm.warmstart = (metric.longterm.warmstart != 0)? metric.longterm.warmstart * alpha
+                metric.longterm.warmstart = (metric.longterm.warmstart != 0) ? metric.longterm.warmstart * alpha
                     + metric.shortterm.warmstart * (1 - alpha) : metric.shortterm.warmstart
             
+            let payload = {
+                method: 'put',
+                body: JSON.stringify({
+                    coldstart: metric.longterm.coldstart,
+                    warmstart: metric.longterm.warmstart,
+                    _rev: revision
+                }),
+                headers: { 'Content-Type': 'application/json' }
+            }
+
+            console.log(payload);
+            
+
+            await fetch(metricsDB + functionHash, payload)
             metric.timestamp = Date.now()
         }
         
@@ -105,6 +119,38 @@ function broadcastMetrics() {
     }
 }
 
+/**
+ * Function to fetch the latest data from metric DB
+ * @param {String} functionHash 
+ * @param {JSON} metric 
+ */
+async function fetchData(functionHash, metric) {
+    let revision
+    let res = await fetch(metricsDB + functionHash)
+    let json = await res.json()
+    console.log(json);
+
+    if (json.error === "not_found") {
+        metric.longterm = {
+            coldstart: 0,
+            warmstart: 0
+        }
+    } else {
+
+        metric.longterm = {
+            coldstart: json.coldstart,
+            warmstart: json.warmstart,
+        }
+        revision = json._rev
+    }
+    console.log(metric);
+    
+    return {
+        metric,
+        revision
+    }
+}
+
 module.exports = {
     collectMetrics, broadcastMetrics
 }
\ No newline at end of file