lib.js

const crypto = require('crypto');
const fs = require('fs')
const rp = require('request-promise');
const fetch = require('node-fetch');
const winston = require('winston')
const constants = require('.././constants.json')
const secrets = require('./secrets.json')
const metrics = require('./metrics')
const { createLogger, format, transports } = winston;
const heap = require('heap')


let kafka = require('kafka-node'),
   Producer = kafka.Producer,
   client = new kafka.KafkaClient({
      kafkaHost: constants.network.external.kafka_host,
      autoConnect: true
   }),
   producer = new Producer(client)

let implicitChainDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
implicitChainDB = implicitChainDB + "/" + constants.implicit_chain_db_names + "/"

/**
 * Generates unique IDs of arbitrary length
 * @param {Length of the ID} length 
 */
function makeid(length) {
    var result           = '';
    var characters       = 'abcdefghijklmnopqrstuvwxyz0123456789';
    var charactersLength = characters.length;
    for ( var i = 0; i < length; i++ ) {
       result += characters.charAt(Math.floor(Math.random() * charactersLength));
    }
    return result;
 }


/**
 * generates the runtime executor after inserting the received function
 * TODO: make this asynchronous
 * @param {string Path from where to extract the function} functionPath 
 * @param {string Function Hash value} functionHash 
 */
function generateExecutor(functionPath, functionHash) {
   input = fs.readFileSync('./repository/worker_env/env.js')
   functionFile = fs.readFileSync(functionPath + functionHash)
   searchSize = "(resolve, reject) => {".length

   insertIndex = input.indexOf("(resolve, reject) => {") + searchSize

   output = input.slice(0, insertIndex) + functionFile + input.slice(insertIndex)
    
   let hash = crypto.createHash('md5').update(output).digest("hex");
   console.log(hash);
   
    fs.writeFileSync(functionPath + hash + ".js", output)
    return hash
 }

 /**
  * Reverse proxy to take user requests and forward them to appropriate workers using a loadbalacer
  * @param {JSON} req the user request to be forwarded to the worker
  * @param {JSON} res Object to use to return the response to the user
  * @param {Map} functionToResource Function to resource Map
  * @param {Map} resourceMap Map from resource ID to resource metadata
  * @param {Map} functionBranchTree Holds the function path's and related probability distribution
  */
function reverseProxy(req, res, functionToResource, resourceMap, functionBranchTree) {
   branchChainPredictor(req, resourceMap, functionToResource, functionBranchTree)
   return new Promise((resolve, reject) => {
      let runtime = req.body.runtime
      let id = req.params.id + runtime
      /**
       * Bypass deployment pipeline if resource available
       */
      let functionHeap = functionToResource.get(id)
      // loadbalancing by choosing worker with lowest load
      let forwardTo = functionHeap[0]
      let resource = resourceMap.get(forwardTo.resource_id)
      // logger.info(`Choosing resource ${JSON.stringify(forwardTo.resource_id)}` +
      //    "\n forwarding via reverse proxy to: " + JSON.stringify(resource));
      let url = `http://${resource.node_id}:${resource.port}/serverless/function/execute`
   
      // logger.info("Request received at reverseproxy. Forwarding to: " + url);
      forwardTo.open_request_count += 1
      heap.heapify(functionHeap, compare) // maintain loadbalancer by heapifying the Map
      // logger.info(functionHeap);
      
      var options = {
         method: 'POST',
         uri: url,
         body: req.body,
         json: true // Automatically stringifies the body to JSON
      };
      
      
      rp(options)
         .then(function (parsedBody) {
            let serviceTime = Date.now() - res.timestamp
            
            res.json(parsedBody)
            forwardTo.open_request_count -= 1
            heap.heapify(functionHeap, compare)
            
            metrics.collectMetrics({type: res.start, value: serviceTime, functionHash: req.params.id, runtime})
            resolve()
         })
         .catch(function (err) {
            forwardTo.open_request_count -= 1
            heap.heapify(functionHeap, compare)
            logger.error("error" + err);
            res.json(err.message).status(err.statusCode)
            resolve()
         });
      
   })
}

function getPort(usedPort) {
   let port = -1, ctr = 0
   do {
       min = Math.ceil(30000);
       max = Math.floor(60000);
       port = Math.floor(Math.random() * (max - min + 1)) + min;
       ctr += 1;
       if (ctr > 30000) {
           port = -1
           break
       }
   } while (usedPort.has(port))
   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()
            
         )
      })
   ]
});

function compare(a, b) {
   return a.open_request_count - b.open_request_count
}

function branchChainPredictor(req, resourceMap, functionToResource, functionBranchTree) {
   // console.log(req.headers['x-resource-id']);
   
   if (req.headers['x-resource-id'] === undefined) {
      let functionHash = req.params.id
      if (functionBranchTree.has(functionHash)) {
         let branchInfo = functionBranchTree.get(functionHash)
         branchInfo.req_count++

      } else {
         
         let data = {
            req_count: 1,
            parent: true,
            branches: new Map()
         }
         functionBranchTree.set(functionHash, data)
      }
      
   } else {
      let resource_id = req.headers['x-resource-id']
      let resource = resourceMap.get(resource_id)
      let forwardBranch = req.params.id
      if (!functionBranchTree.has(resource.functionHash)) {
         let data = {
            req_count: 1,
            parent: false,
            branches: new Map()
         }
         data.branches.set(forwardBranch, 1)
         functionBranchTree.set(resource.functionHash, data)
      } else {
         let branchInfo = functionBranchTree.get(resource.functionHash)
         if (!branchInfo.parent)
            branchInfo.req_count++
         if (branchInfo.branches.has(forwardBranch)) {
            let branchProb = branchInfo.branches.get(forwardBranch)
            branchProb = (branchProb * (branchInfo.req_count - 1) + 1.0)
            branchInfo.branches.set(forwardBranch, branchProb)
         } else {
            branchInfo.branches.set(forwardBranch, 1.0)
         }
         for (let [branch, prob] of branchInfo.branches.entries()) {
            if (branch !== forwardBranch)
               prob *= (branchInfo.req_count - 1)
            prob /= branchInfo.req_count
            branchInfo.branches.set(branch, prob)
         }
      }
   }
   
   // console.log("branch tree", functionBranchTree);
}

function viterbi(functionBranchTree) {
   
   functionBranchTree.forEach((metadata, node) => {
      if (metadata.parent && metadata.req_count % 5 == 0) {
         let path = []
         let parents = [[node, {
            prob: 1,
            metadata
         }]]
         path.push({node, probability: 1})
         let siblings = new Map()
         while(parents.length > 0) {
            // console.log("parent_group", parents);
            
            for (const parent of parents) {
               // console.log("=========begin==========\n",parent, "\n=============end============");
               // console.log(parent[1].metadata);
               
               if (parent[1].metadata === undefined)
                  continue
               let forwardBranches = parent[1].metadata.branches
               // console.log(forwardBranches);
               
               let parentProbability = parent[1].prob
               
               forwardBranches.forEach((branchProb, subNode) => {
                  let probability = 0
                  if (siblings.has(subNode))
                     probability = siblings.get(subNode)
                  probability += branchProb * parentProbability
                  // console.log("prob", probability);
                  
                  siblings.set(subNode, probability)
               })
               // console.log("siblings", siblings);
               
            }
            parents = []
            let maxSibling, maxProb = 0
            siblings.forEach((prob, sibling) => {
               if (prob > maxProb) {
                  maxSibling = sibling
                  maxProb = prob
               }
            })
            parentIDs = Array.from( siblings.keys() );
            for (const id of parentIDs) {
               let metadata = functionBranchTree.get(id)
               parents.push([
                  id, {
                     prob: siblings.get(id),
                     metadata
                  }
               ])
            }
            if (maxSibling !== undefined)
               path.push({node: maxSibling, probability: maxProb})
            siblings = new Map()
         }
         // if (path.length > 0)
         //    console.log("path", path);
         metadata.mle_path = path
         if (path.length > 1) {
            let payload = {
               method: 'put',
               body: JSON.stringify(path),
               headers: { 'Content-Type': 'application/json' }
            }
            fetch(implicitChainDB + functionHash, payload)
         }
      }
   });
}

function logBroadcast(message, resource_id, resourceMap) {
   return new Promise((resolve, reject) => {
      try {

         message.timestamp = Date.now()
         if (resource_id && resourceMap.has(resource_id)) {
            let resource = resourceMap.get(resource_id)
            message.resource_id = resource_id
            message.node_id = resource.node_id
            message.runtime = resource.runtime
            message.function_id = resource.functionHash
         }
         let log = [{
            topic: constants.topics.log_channel,
            messages: JSON.stringify(message),
            partition: 0
         }]
         producer.send(log, () => {
            resolve()
         })
      } catch (err) {
         console.log(err);
         reject()
      }
   })
   
}

async function fetchData(url, data = null) {
   let res
   if (data === undefined || data === null)
      res = await fetch(url)
   else
      res = await fetch(url, data)
   return await res.json()
}

 module.exports = {
    makeid, generateExecutor, reverseProxy, 
    getPort, logger, compare,
    viterbi, logBroadcast, fetchData, metrics,
    producer
 }