process deployment working

Added long running processes and deployment pipeline bypass support

dispatch_module/dispatch_daemon/execute.js

@@ -28,17 +28,19 @@ function runIsolate(filename) {
 
 }
 
-function runProcess(local_repository, functionHash) {
+function runProcess(local_repository, functionHash, port) {
     let filename = local_repository + functionHash
     return new Promise((resolve, reject) => {
         let timeStart = Date.now()
-        const process = spawn('node', [filename]);
+        const process = spawn('node', [filename, port]);
         let result = "";
         process.stdout.on('data', (data) => {
             console.log(`stdout: ${data}`);
             result += data;
             let timeDifference = Math.ceil((Date.now() - timeStart))
             console.log("process time taken: ", timeDifference);
+            workerEvent.emit('start', functionHash, port, "process")
+            resolve(result);
         });
 
         process.stderr.on('data', (data) => {
@@ -48,7 +50,7 @@ function runProcess(local_repository, functionHash) {
 
         process.on('close', (code) => {
             console.log(`child process exited with code ${code}`);
-            resolve(result);
+            workerEvent.emit('end', port, "process");
         });
     })
 

dispatch_module/dispatch_daemon/index.js

@@ -70,17 +70,18 @@ function startWorker(local_repository, functionHash,function_id, producer, runti
             }], () => { })
         })
     else if (runtime === "process")
-        execute.runProcess(local_repository, functionHash, port).then(result => {
-            producer.send(
-                [{
-                    topic: "response",
-                    messages: JSON.stringify({
-                        status: "success",
-                        result,
-                        function_id
-                    })
-                }], () => { })
-        })
+        execute.runProcess(local_repository, functionHash, port)
+        // .then(result => {
+        //     producer.send(
+        //         [{
+        //             topic: "response",
+        //             messages: JSON.stringify({
+        //                 status: "success",
+        //                 result,
+        //                 function_id
+        //             })
+        //         }], () => { })
+        // })
     else if (runtime === "container")
         execute.runContainer(functionHash, port)
         // .then(result => {

dispatch_module/dispatcher/index.js

@@ -217,6 +217,7 @@ consumer.on('message', function (message) {
     } else if (topic == "deployed") {
         
         message = JSON.parse(message)
+        console.log("deployed", message);
         
         if (db.has(message.functionHash + message.runtime)) {
             let { req, res } = db.get(message.functionHash + message.runtime)
@@ -227,7 +228,11 @@ consumer.on('message', function (message) {
                     node_id: message.node_id
                 })
             
-            libSupport.reverseProxy(req, res, `http://${message.node_id}:${message.port}/serverless/function/execute`)
+            libSupport.reverseProxy(req, res, 
+                `http://${message.node_id}:${message.port}/serverless/function/execute`)
+                .then(() => {
+                    db.delete(message.functionHash + message.runtime)
+                })
         }
     } else if (topic == "removeWorker") {
         console.log("removing metadata", message);

dispatch_module/dispatcher/lib.js

@@ -28,24 +28,28 @@ function makeid(length) {
  }
 
  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
-    };
+    return new Promise((resolve, reject) => {
+       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)
-       });
+       rp(options)
+          .then(function (parsedBody) {
+             console.log("parsed body:", parsedBody);
+             res.json(parsedBody)
+             resolve()
+          })
+          .catch(function (err) {
+             console.log("error", err.message);
+             res.json(err.message).status(err.statusCode)
+             resolve()
+          });
+    })
  }
  
  module.exports.makeid = makeid;

dispatch_module/dispatcher/repository/worker_env/env.js

@@ -4,7 +4,7 @@ const bodyParser = require('body-parser')
 const app = express()
 app.use(bodyParser.urlencoded({ extended: true }))
 app.use(bodyParser.json())
-let port = 5000, lastRequest = Date.now()
+let port = (process.argv[2] != null)? parseInt(process.argv[2]): 5000, lastRequest = Date.now()
 app.post('/serverless/function/execute/', (req, res) => {
     let payload = req.body
     lastRequest = Date.now()
@@ -22,7 +22,7 @@ app.listen(port, () => console.log(`Server listening on port ${port}!`))
 
 function shouldDie() {
     
-    if (Date.now() - lastRequest > 300 * 1000) {
+    if (Date.now() - lastRequest > 5 * 1000) {
         console.log("Idle for too long. Exiting");
         
         process.exit(0)