Added Horizontal autoscalar. Ref Issue #2

Added a horizontal autoscalar to scale if number of outstanding requests exceed 100 (currently hardcoded) within a 1s time window.

dispatch_system/dispatch_manager/index.js

@@ -37,11 +37,11 @@ let kafka = require('kafka-node'),
     Consumer = kafka.Consumer,
     consumer = new Consumer(client,
         [
-            { topic: 'response', partition: 0, offset: 0}, 
-            { topic: 'heartbeat' }, 
-            { topic: "deployed" },
-            { topic: "removeWorker" },
-            { topic: "RESPONSE_RM_2_DM"}
+            { topic: 'heartbeat' }, // receives heartbeat messages from workers, also acts as worker join message
+            { topic: "deployed" }, // receives deployment confirmation from workers
+            { topic: "removeWorker" }, // received when a executor environment is blown at the worker
+            { topic: "RESPONSE_RM_2_DM" }, // receives deployment details from RM
+            { topic: "hscale" } // receives signals for horizontal scaling
         ],
         [
             { autoCommit: true }
@@ -236,6 +236,7 @@ function dispatch() {
         let functionHash = req.params.id
         if (!db.has(functionHash + runtime)) {
             db.set(functionHash + runtime, [])
+            db.get(functionHash + runtime).push({ req, res })
             let resource_id = libSupport.makeid(20) // each function resource request is associated with an unique ID
             logger.info(`Generated new resource ID: ${resource_id} for runtime: ${runtime}`);
             
@@ -295,34 +296,18 @@ function getAddress() {
     return workerNodes[Math.floor(Math.random() * workerNodes.length)];
 }
 
-consumer.on('message', function (message) {
-    
-    let topic = message.topic
-        message = message.value
-    if (topic === "response") {
-        logger.info("response " + message);
-        
-        
-    } else if (topic === "heartbeat") {
-        message = JSON.parse(message)
-        if (Date.now() - message.timestamp < 1000)
-            if (workerNodes.indexOf(message.address) === -1)  {
-                workerNodes.push(message.address)
-                logger.warn("New worker discovered. Worker List: " + workerNodes)
-            }
-    } else if (topic == "deployed") {
-        try {
-            message = JSON.parse(message)
-        } catch (e) {
-            // process.exit(0)
-        }
+function postDeploy(message) {
     logger.info("Deployed Resource: " + JSON.stringify(message));
 
-        if (db.has(message.functionHash + message.runtime)) {
-            let sendQueue = db.get(message.functionHash + message.runtime)
     if (message.status == false) {
+        let sendQueue = db.get(message.functionHash + message.runtime)
         // TODO: handle failure
-                res.status(400).json({reason: message.reason})
+        while (sendQueue && sendQueue.length != 0) {
+            let { req, res } = sendQueue.shift()
+            res.status(400).json({ reason: message.reason })
+        }
+        db.delete(message.functionHash + message.runtime)
+        
         return;
     }
     if (functionToResource.has(message.functionHash + message.runtime)) {
@@ -367,9 +352,12 @@ consumer.on('message', function (message) {
     producer.send(confirmRM, () => {
         logger.info(`Confirmed RM for successful deployment resource_id: ${message.resource_id} deployment`)
     })
+
+    if (db.has(message.functionHash + message.runtime)) {
+        let sendQueue = db.get(message.functionHash + message.runtime)
         logger.info("forwarding request via reverse proxy to: " + JSON.stringify(resource));
-            while (sendQueue.length != 0) {
-               let {req, res} = sendQueue.shift()
+        while (sendQueue && sendQueue.length != 0) {
+            let { req, res } = sendQueue.shift()
             libSupport.reverseProxy(req, res, functionToResource, resourceMap)
                 .then(() => {
 
@@ -378,6 +366,31 @@ consumer.on('message', function (message) {
         db.delete(message.functionHash + message.runtime)
     }
 
+}
+
+consumer.on('message', function (message) {
+    
+    let topic = message.topic
+        message = message.value
+    if (topic === "response") {
+        logger.info("response " + message);
+        
+        
+    } else if (topic === "heartbeat") {
+        message = JSON.parse(message)
+        if (Date.now() - message.timestamp < 1000)
+            if (workerNodes.indexOf(message.address) === -1)  {
+                workerNodes.push(message.address)
+                logger.warn("New worker discovered. Worker List: " + workerNodes)
+            }
+    } else if (topic == "deployed") {
+        try {
+            message = JSON.parse(message)
+        } catch (e) {
+            // process.exit(0)
+        }
+        postDeploy(message)
+        
     } else if (topic == "removeWorker") {
         logger.warn("Worker blown: Removing Metadata " + message);
         try {
@@ -403,6 +416,38 @@ consumer.on('message', function (message) {
         }
         
 
+    } else if (topic == "hscale") {
+        message = JSON.parse(message)
+        let resource_id = libSupport.makeid(20), // each function resource request is associated with an unique ID
+            node_id = getAddress(), // Requests the RM for address and other metadata for function placement
+            port = libSupport.getPort(usedPort), // TODO: will be provided by the RM
+            runtime = message.runtime,
+            functionHash = message.functionHash
+
+        let payload = [{
+            topic: node_id,
+            messages: JSON.stringify({
+                "type": "execute", // Request sent to Dispatch Daemon via Kafka for actual deployment at the Worker
+                resource_id,
+                runtime, functionHash,
+                port
+            }),
+            partition: 0
+        }]
+        logger.info("Requesting RM " + JSON.stringify({
+            resource_id,
+            "memory": 332,
+        }))
+
+        /** uncomment when RM is unavailable */
+
+        resourceMap.set(resource_id, {
+            runtime, functionHash, port, node_id
+        })
+        logger.info(resourceMap);
+        producer.send(payload, () => {
+            logger.info(`Resource Deployment request sent to Dispatch Agent`)
+        })
     } else if (topic == "RESPONSE_RM_2_DM") {
         
         logger.info("Response from RM: " + message);
@@ -427,5 +472,24 @@ consumer.on('message', function (message) {
     }
 });
 
-setInterval(dispatch, 2000);
+function autoscalar() {
+    functionToResource.forEach((resourceList, functionKey, map) => {
+        console.log(resourceList);
+        
+        if (resourceList.length > 0 && resourceList[resourceList.length - 1].metric > 100) {
+            let resource = resourceMap.get(resourceList[resourceList.length - 1].resource_id)
+            logger.warn(`resource ${resourceList[resourceList.length - 1]} exceeded autoscalar threshold. Scaling up!`)
+            let payload = [{
+                topic: "hscale",
+                messages: JSON.stringify({ "runtime": resource.runtime, "functionHash": resource.functionHash })
+            }]
+            producer.send(payload, function () { })
+        }
+    });
+
+}
+
+
+setInterval(autoscalar, 1000);
+setInterval(dispatch, 1000);
 app.listen(port, () => logger.info(`Server listening on port ${port}!`))
\ No newline at end of file