container deployment working

Reverse proxy based container worker working

dispatch_module/dispatch_daemon/config.json

-{"id":"10.130.150.246","master_node":"10.129.6.5"}
\ No newline at end of file
+{"id":"192.168.31.51","master_node":"10.129.6.5"}
\ No newline at end of file

dispatch_module/dispatch_daemon/execute.js

@@ -102,6 +102,8 @@ function runContainer(imageName, port) {
             } else {
                 const process_checkContainer = spawn('docker', ['container', 'inspect', imageName]);
                 process_checkContainer.on('close', (code) => {
+                    console.log("container starting at port", port);
+                    
                     if (code != 0) {
                         const process = spawn('docker', ["run", "-p", `${port}:5000`, "--name", imageName, registry_url + imageName]);
                         let result = "";
@@ -124,14 +126,16 @@ function runContainer(imageName, port) {
                             workerEvent.emit('end', port, "container");
                         })
                     } else {
-                        const process = spawn('docker', ["start", "-a", imageName]);
+                        const clean_container = spawn('docker', ['rm', imageName])
+                        clean_container.on('close', code => {
+                            const process = spawn('docker', ["run", "-p", `${port}:5000`, "--name", imageName, registry_url + imageName]);
                             let result = "";
                             timeStart = Date.now()
                             process.stdout.on('data', (data) => {
                                 console.log(`stdout: ${data}`);
                                 let timeDifference = Math.ceil((Date.now() - timeStart))
                                 console.log("container run time taken: ", timeDifference);
-                            result += data;
+                                // result += data;
                                 workerEvent.emit('start', imageName, port, "container")
                                 resolve(result);
                             });
@@ -144,6 +148,8 @@ function runContainer(imageName, port) {
                             process.on('close', (code) => {
                                 workerEvent.emit('end', port, "container");
                             })
+                        })
+                        
                     }
                 })
                 

dispatch_module/dispatch_daemon/index.js

@@ -55,6 +55,8 @@ libSupport.makeTopic(node_id).then(() => {
 
 function startWorker(local_repository, functionHash,function_id, producer, runtime) {
     let port = libSupport.getPort(usedPort)
+    console.log("Using port", port, "for functionHash", functionHash);
+    
     usedPort.set(port, functionHash)
     if (runtime === "isolate")
         execute.runIsolate(local_repository + functionHash).then(result => {
@@ -123,17 +125,16 @@ execute.workerEvent.on("start", (functionHash, port, runtime) => {
     
 })
 
-execute.workerEvent.on('end', (port) => {
+execute.workerEvent.on('end', (port, runtime) => {
     let functionHash = usedPort.get(port)
     usedPort.delete(port)
     producer.send(
         [{
             topic: "removeWorker",
             messages: JSON.stringify({ functionHash, port, runtime, node_id })
-        }], () => { })
-    console.log("Ending worker for function", functionHash, usedPort, functionToPort);
-
-    
+        }], () => {
+            console.log("Ending worker for function", functionHash, usedPort);
+         })
 })
 
 setInterval(heartbeat, 1000);
\ No newline at end of file

dispatch_module/dispatcher/index.js

@@ -25,7 +25,10 @@ let kafka = require('kafka-node'),
     Consumer = kafka.Consumer,
     consumer = new Consumer(client,
         [
-            { topic: 'response', partition: 0, offset: 0}, { topic: 'heartbeat' }, {topic: "deployed"}
+            { topic: 'response', partition: 0, offset: 0}, 
+            { topic: 'heartbeat' }, 
+            {topic: "deployed"},
+            { topic: "removeWorker"}
         ],
         [
             { autoCommit: true }
@@ -148,8 +151,7 @@ app.post('/serverless/execute/:id', (req, res) => {
         
         let forwardTo = functionToPort.get(req.params.id + runtime)
         console.log("resource found", forwardTo);
-        res.redirect(307, `http://${forwardTo.node_id}:${forwardTo.port}/serverless/function/execute`);
-        console.log("sending request to:", `http://${forwardTo.node_id}:${forwardTo.port}/serverless/function/execute`);
+        libSupport.reverseProxy(req, res, `http://${forwardTo.node_id}:${forwardTo.port}/serverless/function/execute`)
         
     } else {
         requestQueue.push({ req, res })
@@ -180,7 +182,7 @@ function dispatch() {
             partition: 0
         }]
         producer.send(payload, () => {})
-        db.set(function_id, res)
+        db.set(functionHash + runtime, {req, res})
         
     }
 }
@@ -189,48 +191,22 @@ function getAddress() {
     return workerNodes[Math.floor(Math.random() * workerNodes.length)];
 }
 
-
-// client.on('message', function (topic, message) {
-
-//     if (topic === "response") {
-//         message = JSON.parse(message)
-//         console.log(message);
-//         let res = db.get(message.function_id)
-//         res.json({
-//             "status": "success",
-//             "reply": message.result
-//         })
-//         db.delete(message.function_id)
-//     } else if (topic === "heartbeat") {
-//         message = JSON.parse(message)
-//         if (workerNodes.indexOf(message.address) === -1)  {
-//             workerNodes.push(message.address)
-//             console.log(workerNodes);
-// 	}
-//     }
-// })
-
 app.listen(port, () => console.log(`Server listening on port ${port}!`))
 
-// client.on('connect', function () {
-//     client.subscribe("response")
-//     client.subscribe("heartbeat")
-// })
-
 consumer.on('message', function (message) {
     // console.log(message);
     let topic = message.topic
         message = message.value
     if (topic === "response") {
-        message = JSON.parse(message)
-        console.log(message);
-        let res = db.get(message.function_id)
-        if (res != null)
-            res.json({
-                "status": "success",
-                "reply": message.result
-            })
-        db.delete(message.function_id)
+        // message = JSON.parse(message)
+        // console.log(message);
+        // let {req, res} = db.get(message.function_id)
+        // if (res != null)
+        //     res.json({
+        //         "status": "success",
+        //         "reply": message.result
+        //     })
+        // db.delete(message.function_id)
     } else if (topic === "heartbeat") {
         message = JSON.parse(message)
         if (Date.now() - message.timestamp < 300)
@@ -239,14 +215,25 @@ consumer.on('message', function (message) {
                 console.log(workerNodes);
             }
     } else if (topic == "deployed") {
-        let res = db.get(message.function_id)
+        
         message = JSON.parse(message)
-        console.log(message);
+        
+        if (db.has(message.functionHash + message.runtime)) {
+            let { req, res } = db.get(message.functionHash + message.runtime)
             // res.redirect(307, `http://${forwardTo.node_id}:${forwardTo.port}/serverless/function/execute`);
-        functionToPort.set(message.functionHash + message.runtime, {port: parseInt(message.port), 
-            node_id: message.node_id})
-        console.log(functionToPort);
+            if (parseInt(message.port) != -1)
+                functionToPort.set(message.functionHash + message.runtime, {
+                    port: parseInt(message.port),
+                    node_id: message.node_id
+                })
             
+            libSupport.reverseProxy(req, res, `http://${message.node_id}:${message.port}/serverless/function/execute`)
+        }
+    } else if (topic == "removeWorker") {
+        console.log("removing metadata", message);
+        
+        message = JSON.parse(message)
+        functionToPort.delete(message.functionHash + message.runtime)
     }
 });
 

dispatch_module/dispatcher/lib.js

 const crypto = require('crypto');
 const fs = require('fs')
+var rp = require('request-promise');
 function makeid(length) {
     var result           = '';
     var characters       = 'abcdefghijklmnopqrstuvwxyz0123456789';
@@ -26,5 +27,27 @@ function makeid(length) {
     return hash
  }
 
+ function reverseProxy(req, res, url) {
+    console.log("requesting reverseproxy");
+    
+    var options = {
+       method: 'POST',
+       uri: url,
+       body: req.body,
+       json: true // Automatically stringifies the body to JSON
+    };
+
+    rp(options)
+       .then(function (parsedBody) {
+          console.log("parsed body:", parsedBody);
+          res.json(parsedBody)
+       })
+       .catch(function (err) {
+          console.log("error", err.message);
+          res.json(err.message).status(err.statusCode)
+       });
+ }
+ 
  module.exports.makeid = makeid;
  module.exports.generateExecutor = generateExecutor;
+ module.exports.reverseProxy = reverseProxy;
\ No newline at end of file

dispatch_module/dispatcher/package.json

@@ -19,6 +19,8 @@
     "mqtt": "^3.0.0",
     "nano": "^8.1.0",
     "redis": "^2.8.0",
+    "request": "^2.88.0",
+    "request-promise": "^4.2.5",
     "save": "^2.4.0"
   }
 }

dispatch_module/dispatcher/repository/worker_env/env.js

@@ -5,7 +5,7 @@ const app = express()
 app.use(bodyParser.urlencoded({ extended: true }))
 app.use(bodyParser.json())
 let port = 5000, lastRequest = Date.now()
-app.post('/serverless/function/execute/:id', (req, res) => {
+app.post('/serverless/function/execute/', (req, res) => {
     let payload = req.body
     lastRequest = Date.now()
     executor(payload).then((result) => {