Started Improved logging

dispatch_module/dispatch_daemon/execute.js

@@ -18,7 +18,6 @@ function runIsolate(local_repository, functionHash, port, resource_id) {
             console.log("worker exited");
         })
         worker.on('online', () => {
-            workerEvent.emit('start', functionHash, port, "isolate")
             resolve()
         })
         
@@ -132,5 +131,4 @@ function runContainer(imageName, port, resource_id) {
 
 module.exports.runContainer = runContainer;
 module.exports.runProcess = runProcess;
-module.exports.runIsolate = runIsolate;
-module.exports.workerEvent = workerEvent;
\ No newline at end of file
+module.exports.runIsolate = runIsolate;
\ No newline at end of file

dispatch_module/dispatch_daemon/index.js

@@ -65,7 +65,6 @@ libSupport.makeTopic(node_id).then(() => {
 function startWorker(local_repository, functionHash, resource_id, producer, runtime, port) {
     console.log("Using port", port, "for functionHash", functionHash);
     
-    usedPort.set(port, functionHash)
     fs.writeFile('./local_repository/config.json', JSON.stringify({port, functionHash, resource_id, runtime}), () => {
         if (runtime === "isolate")
             execute.runIsolate(local_repository, functionHash, port, resource_id)

dispatch_module/dispatch_daemon/log/.gitignore

+*
+!.gitignore
\ No newline at end of file

dispatch_module/dispatch_daemon/package.json

@@ -17,6 +17,7 @@
     "kafka-node": "^5.0.0",
     "morgan": "^1.9.1",
     "mqtt": "^3.0.0",
-    "redis": "^2.8.0"
+    "redis": "^2.8.0",
+    "winston": "^3.2.1"
   }
 }

dispatch_module/dispatcher/index.js

@@ -7,11 +7,10 @@ const constants = require('.././constants.json')
 const fs = require('fs')
 const { spawn } = require('child_process');
 const morgan = require('morgan')
-const mqtt = require('mqtt')
-// const client = mqtt.connect('mqtt://' + constants.mqtt_url)
+
 const app = express()
 const libSupport = require('./lib')
-
+const logger = libSupport.logger
 /**
  * functionToPort maps the function and its respective port mapping
  * TODO: change this to hold a list of mappings of horizontal scaling
@@ -71,7 +70,7 @@ app.post('/serverless/deploy', (req, res) => {
     file.mv(file_path + functionHash, function (err) {
         functionHash = libSupport.generateExecutor(file_path, functionHash)
         if (err) {
-            console.log(err);
+            logger.error(err)
             res.send("error").status(400)
         }
         else {
@@ -117,27 +116,28 @@ function deployContainer(path, imageName) {
             ENTRYPOINT ["node", "${imageName}.js"]`
             , function (err) {
                 if (err) {
-                    console.log("failed", err);
+                    logger.error("failed", err);
+
                     
                     reject(err);
                 }
                 else {
-                    console.log('Dockerfile created');
+                    logger.info('Dockerfile created');
                     const process = spawn('docker', ["build", "-t", registry_url + imageName, path]);
 
                     process.stdout.on('data', (data) => {
-                        console.log(`stdout: ${data}`);
+                        logger.info(`stdout: ${data}`);
 
                     });
 
                     process.stderr.on('data', (data) => {
-                        console.error(`stderr: ${data}`);
+                        logger.error(`stderr: ${data}`);
                     });
 
                     process.on('close', (code) => {
-                        console.log(`child process exited with code ${code}`);
+                        logger.warn(`child process exited with code ${code}`);
                         let timeDifference = Math.ceil((Date.now() - buildStart))
-                        console.log("image build time taken: ", timeDifference);
+                        logger.info("image build time taken: ", timeDifference);
                         const process_push = spawn('docker', ["push", registry_url + imageName]);
 
                         process_push.stdout.on('data', (data) => {
@@ -146,11 +146,11 @@ function deployContainer(path, imageName) {
                         });
 
                         process_push.stderr.on('data', (data) => {
-                            console.error(`stderr: ${data}`);
+                            logger.error(`stderr: ${data}`);
                         });
 
                         process_push.on('close', (code) => {
-                            console.log("image pushed to repository");
+                            logger.info("image pushed to repository");
                             resolve();
                         })
                         
@@ -171,7 +171,7 @@ app.post('/serverless/execute/:id', (req, res) => {
          */
         let forwardTo = functionToResource.get(req.params.id + runtime) 
         let resource = resourceMap.get(forwardTo.resource_id)
-        console.log("resource found", forwardTo, resource);
+        logger.info("resource found", forwardTo, resource);
         libSupport.reverseProxy(req, res, `http://${resource.node_id}:${resource.port}/serverless/function/execute`)
         
     } else {
@@ -196,18 +196,18 @@ function dispatch() {
     let lookbackWindow = Math.min(WINDOW_SIZE, requestQueue.length)
     for (let i = 0; i < lookbackWindow; i++) {
         let {req, res} = requestQueue.shift()
-	    console.log(req.body)
+	    logger.info(req.body)
         let runtime = req.body.runtime
         let functionHash = req.params.id
         
         let resource_id = libSupport.makeid(20) // each function resource request is associated with an unique ID
-        console.log("Dispatching function with Id", resource_id, runtime);
+        logger.info("Dispatching function with Id", resource_id, runtime);
         let node_id = getAddress() // Requests the RM for address and other metadata for function placement
         let port = libSupport.getPort(usedPort) // TODO: will be provided by the RM
         resourceMap.set(resource_id, {
             runtime, functionHash, port, node_id
         })
-        console.log(resourceMap);
+        logger.info(resourceMap);
         
         let payload = [{
             topic: node_id,
@@ -245,17 +245,15 @@ function getAddress() {
     return workerNodes[Math.floor(Math.random() * workerNodes.length)];
 }
 
-app.listen(port, () => console.log(`Server listening on port ${port}!`))
-
 consumer.on('message', function (message) {
     
     let topic = message.topic
         message = message.value
     if (topic === "response") {
-        console.log("response", message);
+        logger.info("response", message);
         
         // message = JSON.parse(message)
-        // console.log(message);
+        // logger.info(message);
         // let {req, res} = db.get(message.function_id)
         // if (res != null)
         //     res.json({
@@ -268,7 +266,7 @@ consumer.on('message', function (message) {
         if (Date.now() - message.timestamp < 300)
             if (workerNodes.indexOf(message.address) === -1)  {
                 workerNodes.push(message.address)
-                console.log(workerNodes);
+                logger.warn(workerNodes);
             }
     } else if (topic == "deployed") {
         try {
@@ -277,7 +275,7 @@ consumer.on('message', function (message) {
             // process.exit(0)
         }
         
-        console.log("deployed", message);
+        logger.info("deployed", message);
         
         if (db.has(message.functionHash + message.runtime)) {
             let { req, res } = db.get(message.functionHash + message.runtime)
@@ -292,7 +290,7 @@ consumer.on('message', function (message) {
                 })
         }
     } else if (topic == "removeWorker") {
-        console.log("removing metadata", message);
+        logger.info("removing metadata", message);
         try {
             message = JSON.parse(message)
         } catch(e) {
@@ -305,15 +303,15 @@ consumer.on('message', function (message) {
     } else if (topic == "RESPONSE_ARBITER_2_DISPATCHER") {
         
         message = JSON.parse(message)
-        console.log(message);
+        logger.info(message);
         let payload = rmQueue.get(message.id)
         if (payload != null) {
             payload[0].topic = getAddress()
-            console.log(payload);
+            logger.info(payload);
 
             producer.send(payload, () => { })
         } else {
-            console.log("something went wrong");
+            logger.error("something went wrong");
             
         }
         
@@ -321,6 +319,7 @@ consumer.on('message', function (message) {
 });
 
 setInterval(dispatch, 2000);
+app.listen(port, () => logger.info(`Server listening on port ${port}!`))
 // {
 //     id: "!!!!!",
     

dispatch_module/dispatcher/lib.js

 const crypto = require('crypto');
 const fs = require('fs')
-var rp = require('request-promise');
+const rp = require('request-promise');
+const winston = require('winston')
+const { createLogger, format, transports } = winston;
+
 
 /**
  * Generates unique IDs of arbitrary length
@@ -83,6 +86,31 @@ function makeid(length) {
    return port
 }
 
+
+const logger = winston.createLogger({
+   level: 'info',
+   format: winston.format.combine(
+      format.timestamp(),
+      format.json()
+   ),
+   defaultMeta: { module: 'Dispatch Manager' },
+   transports: [
+      //
+      // - Write to all logs with level `info` and below to `combined.log` 
+      // - Write all logs error (and below) to `error.log`.
+      //
+      new winston.transports.File({ filename: 'log/error.log', level: 'error' }),
+      new winston.transports.File({ filename: 'log/combined.log' }),
+      new winston.transports.Console({
+         format: winston.format.combine(
+            format.colorize({ all: true }),
+            format.timestamp(),
+            format.simple()
+         )
+      })
+   ]
+});
+
  module.exports = {
-    makeid, generateExecutor, reverseProxy, getPort
+    makeid, generateExecutor, reverseProxy, getPort, logger
  }
\ No newline at end of file

dispatch_module/dispatcher/log/.gitignore

+*
+!.gitignore
\ No newline at end of file

dispatch_module/dispatcher/package.json

@@ -21,6 +21,7 @@
     "redis": "^2.8.0",
     "request": "^2.88.0",
     "request-promise": "^4.2.5",
-    "save": "^2.4.0"
+    "save": "^2.4.0",
+    "winston": "^3.2.1"
   }
 }