Moved longterm metrics to Exponential Average

dispatch_system/constants.json

@@ -3,7 +3,6 @@
     "master_port": 8080,
     "master_address": "localhost",
     "grunt_host": "https://www.namandixit.net/lovecraftian_nightmares/grunt",
-    "log_channel": "LOG_COMMON",
     "couchdb_host": "localhost:5984",
     "couchdb_db_name": "serverless",
     "network": {
@@ -21,7 +20,8 @@
         "deployed": "deployed",
         "remove_worker": "removeWorker",
         "response_rm_2_dm": "RESPONSE_RM_2_DM_DUMMY",
-        "hscale": "hscale"
+        "hscale": "hscale",
+        "log_channel": "LOG_COMMON"
     },
     "autoscalar_metrics": {
         "open_request_threshold": 100

dispatch_system/dispatch_manager/index.js

@@ -23,7 +23,7 @@ const app = express()
 const libSupport = require('./lib')
 const logger = libSupport.logger
 let date = new Date();
-let log_channel = constants.log_channel
+let log_channel = constants.topics.log_channel
 
 let usedPort = new Map(), // TODO: remove after integration with RM
     db = new Map(), // queue holding request to be dispatched

dispatch_system/dispatch_manager/lib.js

@@ -283,7 +283,7 @@ function logBroadcast(message, resource_id, resourceMap) {
             message.function_id = resource.functionHash
          }
          let log = [{
-            topic: constants.log_channel,
+            topic: constants.topics.log_channel,
             messages: JSON.stringify(message),
             partition: 0
          }]

dispatch_system/dispatch_manager/metrics.js

 const constants = require('.././constants.json');
-
-let log_channel = constants.log_channel,
+const alpha = 0.99
+let log_channel = constants.topics.log_channel,
     metrics = {  }
 
 let kafka = require('kafka-node'),
@@ -11,8 +11,15 @@ let kafka = require('kafka-node'),
     }),
     producer = new Producer(client)
 
+/**
+ * Function called to report metric data related to functions
+ * @param {JSON} metric 
+ */
 function collectMetrics(metric) {
-    
+    /**
+     * If metrics for a new function comes in, 
+     * provision required structure for the function
+     */
     if (!(metric.functionHash in metrics)) {
         metrics[metric.functionHash] = {
             shortterm: {
@@ -36,6 +43,16 @@ function collectMetrics(metric) {
     
 }
 
+
+/**
+ * Run periodically to calculate average runtime metrics like coldstart and
+ * warmstart latencies.
+ * The module provides two granularities for metrics - shortterm and longterm
+ * shortterm - realtime data at a granularity of 5s (set in dispatch_manager/lib.js)
+ *             shortterm data is calculated using Simple Moving Average (SMA)
+ * longterm - longterm data is held and averaged out over a period of time.
+ *            longterm data is calculated using Expontential Moving Average (EMA)
+ */
 function broadcastMetrics() {
     
     if (Object.keys(metrics).length !== 0) {
@@ -44,31 +61,26 @@ function broadcastMetrics() {
             if (metric.longterm === undefined) {
                 metric.longterm = {
                         coldstart: 0,
-                        coldstart_total_request: 0,
-                        warm_total_request: 0,
                         warmstart: 0
                 }
             }
-            
-            metric.longterm.coldstart = metric.longterm.coldstart 
-                            * metric.longterm.coldstart_total_request
-                            + metric.shortterm.coldstart
-            
-            metric.longterm.coldstart_total_request += metric.shortterm.coldstart_total_request
-            metric.longterm.coldstart /= (metric.longterm.coldstart_total_request != 0)? 
-                        metric.longterm.coldstart_total_request: 1
-
-            metric.longterm.warmstart = metric.longterm.warmstart
-                            * metric.longterm.warm_total_request
-                            + metric.shortterm.warmstart
-            metric.longterm.warm_total_request += metric.shortterm.warm_total_request
-            metric.longterm.warmstart /= (metric.longterm.warm_total_request != 0)? 
-                        metric.longterm.warm_total_request: 1
+            /**
+             * Shortterm moving average
+             */
+            metric.shortterm.coldstart /= (metric.shortterm.coldstart_total_request != 0) ?
+                metric.shortterm.coldstart_total_request : 1
+            metric.shortterm.warmstart /= (metric.shortterm.warm_total_request != 0) ?
+                metric.shortterm.warm_total_request : 1
+            /**
+             * Longterm exponential moving average
+             */
+            if (metric.shortterm.coldstart != 0)
+                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.shortterm.warmstart * (1 - alpha) : metric.shortterm.warmstart
 
-            metric.shortterm.coldstart /= (metric.shortterm.coldstart_total_request != 0)? 
-                metric.shortterm.coldstart_total_request: 1
-            metric.shortterm.warmstart /= (metric.shortterm.warm_total_request != 0)? 
-                metric.shortterm.warm_total_request: 1
             metric.timestamp = Date.now()
         }
         

dispatch_system/log_listener.js

+const constants = require('./constants.json')
+const util = require('util')
+let kafka = require('kafka-node'),
+    Producer = kafka.Producer,
+    client = new kafka.KafkaClient({
+        kafkaHost: constants.network.external.kafka_host,
+        autoConnect: true
+    }),
+    producer = new Producer(client),
+    Consumer = kafka.Consumer,
+    consumer = new Consumer(client,
+        [
+            { topic: constants.topics.log_channel }
+        ])
+
+consumer.on('message', function (message) {
+    message = JSON.parse(message.value)
+    console.log(util.inspect(message, false, null, true /* enable colors */))
+    
+})
\ No newline at end of file