index.js

"use strict";

const express = require('express');
const fileUpload = require('express-fileupload');
const constants = require('.././constants.json');
const chainHandler = require('./explicit_chain_handler');
const secrets = require('./secrets.json');
const fs = require('fs');
const { spawn } = require('child_process');
const morgan = require('morgan');
const heap = require('heap');
const fetch = require('node-fetch');
const swStats = require('swagger-stats');
const apiSpec = require('./swagger.json');
const util = require('util')
const sharedStructures = require('./shared_structures')

/**
 * URL to the couchdb database server used to store function metadata
 */
let metadataDB = `http://${secrets.couchdb_username}:${secrets.couchdb_password}@${constants.couchdb_host}`
metadataDB = metadataDB + "/" + constants.function_db_name + "/"

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

const app = express()
const libSupport = require('./lib')
const logger = libSupport.logger
let date = new Date();
let log_channel = constants.topics.log_channel

let usedPort = new Map(), // TODO: remove after integration with RM
    db = sharedStructures.db, // queue holding request to be dispatched
    resourceMap = sharedStructures.resourceMap, // map between resource_id and resource details like node_id, port, associated function etc
    functionToResource = sharedStructures.functionToResource, // a function to resource map. Each map contains a minheap of
                                   // resources associated with the function
    workerNodes = sharedStructures.workerNodes, // list of worker nodes currently known to the DM
    functionBranchTree = sharedStructures.functionBranchTree // a tree to store function branch predictions
    
chainHandler.initialise(functionToResource)
let kafka = require('kafka-node'),
    Producer = kafka.Producer,
    client = new kafka.KafkaClient({ 
        kafkaHost: constants.network.external.kafka_host,
        autoConnect: true
    }),
    producer = new Producer(client),
    Consumer = kafka.Consumer,
    consumer = new Consumer(client,
        [
            { topic: constants.topics.heartbeat }, // receives heartbeat messages from workers, also acts as worker join message
            { topic: constants.topics.deployed }, // receives deployment confirmation from workers
            { topic: constants.topics.remove_worker }, // received when a executor environment is blown at the worker
            { topic: constants.topics.response_rm_2_dm }, // receives deployment details from RM
            { topic: constants.topics.hscale } // receives signals for horizontal scaling
        ],
        [
            { autoCommit: true }
        ])

app.use(morgan('combined', {
    skip: function (req, res) { return res.statusCode < 400 }
}))
app.use(express.json());
app.use(express.urlencoded({ extended: true }));
const file_path = __dirname + "/repository/"
app.use('/repository', express.static(file_path)); // file server hosting deployed functions
app.use(fileUpload())
app.use(swStats.getMiddleware({ swaggerSpec: apiSpec })); // statistics middleware
app.use('/serverless/chain', chainHandler.router); // chain router (explicit_chain_handler.js) for handling explicit chains
let requestQueue = []

const WINDOW_SIZE = 10
const port = constants.master_port
const registry_url = constants.registry_url

app.get('/metrics', (req, res) => {
    res.set('Content-Type', libSupport.metrics.register.contentType);
    res.end(libSupport.metrics.register.metrics());
});

/**
 * REST API to receive deployment requests
 */
app.post('/serverless/deploy', (req, res) => {
    
    let runtime = req.body.runtime
    let file = req.files.serverless

    let functionHash = file.md5

    file.mv(file_path + functionHash, function (err) { // move function file to repository
        functionHash = libSupport.generateExecutor(file_path, functionHash)
        /**
         * Adding meta caching via couchdb
         * This will create / update function related metadata like resource limits etc
         * on a database named "serverless".
         */
        fetch(metadataDB + functionHash).then(res => res.json())
            .then(json => {
                if (json.error === "not_found") {
                    logger.warn("New function, creating metadata")
                    fetch(metadataDB + functionHash, {
                        method: 'put',
                        body: JSON.stringify({
                            memory: req.body.memory
                        }),
                        headers: { 'Content-Type': 'application/json' },
                    }).then(res => res.json())
                    .then(json => console.log(json));
                } else {
                    logger.warn('Repeat deployment, updating metadata')
                    fetch(metadataDB + functionHash, {
                        method: 'put',
                        body: JSON.stringify({
                            memory: req.body.memory,
                            _rev: json._rev
                        }),
                        headers: { 'Content-Type': 'application/json' },
                    }).then(res => res.json())
                    .then(json => console.log(json));
                }
        });
        if (err) {
            logger.error(err)
            res.send("error").status(400)
        }
        else {
            if (runtime === "container") {
                deployContainer(file_path, functionHash)
                    .then(() => {
                        res.json({
                            status: "success",
                            function_id: functionHash
                        })
                    })
                    .catch(err => {
                        res.json({
                            status: "error",
                            reason: err}).status(400)
                    })
            } else {
                res.json({
                    status: "success",
                    function_id: functionHash
                })
            }
        }
    })
    
})

/**
 * Create the docker file, build and push image to remote repository
 * @param {String: Path from where to extract function executor} path 
 * @param {String: Name of the image} imageName 
 */
function deployContainer(path, imageName) {
    return new Promise((resolve, reject) => {
        let buildStart = Date.now()
        /**
         * Generating dockerfile for the received function
         */
        fs.writeFile('./repository/Dockerfile',
            `FROM node:latest
            WORKDIR /app
            COPY ./worker_env/package.json /app
            ADD ./worker_env/node_modules /app/node_modules
            COPY ${imageName}.js /app
            ENTRYPOINT ["node", "${imageName}.js"]`
            , function (err) {
                if (err) {
                    logger.error("failed", err);

                    
                    reject(err);
                }
                else {
                    logger.info('Dockerfile created');
                    const process = spawn('docker', ["build", "-t", registry_url + imageName, path]); // docker build

                    process.stdout.on('data', (data) => {
                        logger.info(`stdout: ${data}`);

                    });

                    process.stderr.on('data', (data) => {
                        logger.error(`stderr: ${data}`);
                    });

                    process.on('close', (code) => {
                        logger.warn(`child process exited with code ${code}`);
                        let timeDifference = Math.ceil((Date.now() - buildStart))
                        logger.info("image build time taken: ", timeDifference);
                        const process_push = spawn('docker', ["push", registry_url + imageName]); // docker push image to local registry

                        process_push.stdout.on('data', (data) => {
                            console.log(`stdout: ${data}`);

                        });

                        process_push.stderr.on('data', (data) => {
                            logger.error(`stderr: ${data}`);
                        });

                        process_push.on('close', (code) => {
                            logger.info("image pushed to repository");
                            resolve();
                        })
                        
                    });
                }
            });
    })
}

/**
 * REST API to receive execute requests
 */
app.post('/serverless/execute/:id', (req, res) => {

    let runtime = req.body.runtime
    let id = req.params.id + runtime
    res.timestamp = Date.now() 
    if (functionToResource.has(id)) {
        res.start = 'warmstart'
        libSupport.reverseProxy(req, res, functionToResource, resourceMap, functionBranchTree)
    } else {
        res.start = 'coldstart'
        /**
         * Requests are queued up before being dispatched. To prevent requests coming in for the 
         * same function from starting too many workers, they are grouped together
         * and one worker is started per group.
         */
        if (db.has(req.params.id + runtime)) {
            db.get(req.params.id + runtime).push({ req, res })
            return;
        }
        requestQueue.push({ req, res })
        /**
         * We store functions for function placement heuristics purposes. This lets us look into the function
         * patterns being received and make intelligent deployment decisions based on it.
         */
        if (requestQueue.length >= WINDOW_SIZE)
            dispatch()
    }
})

/**
 * Send dispatch signal to Worker nodes and deploy resources after consultation with the RM
 */
function dispatch() {
    /**
     * The lookahead window will be used for optimisation purposes
     * Ex. It might be used to co-group similar runtimes on same machines
     */
    let lookbackWindow = Math.min(WINDOW_SIZE, requestQueue.length)
    for (let i = 0; i < lookbackWindow; i++) {
        let {req, res} = requestQueue.shift()
        
	    // logger.info(req.body)
        let runtime = req.body.runtime
        let functionHash = req.params.id
        if (!db.has(functionHash + runtime)) {
            db.set(functionHash + runtime, [])
            db.get(functionHash + runtime).push({ req, res })
            
            let payload = [{
                topic: constants.topics.hscale,
                messages: JSON.stringify({ runtime, functionHash })
            }]
            producer.send(payload, function () { })

            speculative_deployment(req, runtime)
        } else {
            logger.info("deployment process already started waiting")
            db.get(functionHash + runtime).push({ req, res })
        }

        
    }
}

/**
 * Handles post deployment metadata updates and starts reverse-proxying
 * @param {string} message Message received from DD after deployment
 */
function postDeploy(message) {
    logger.info("Deployed Resource: " + JSON.stringify(message));
    let id = message.functionHash + message.runtime
    if (message.status == false) {
        let sendQueue = db.get(id)
        // TODO: handle failure
        while (sendQueue && sendQueue.length != 0) {
            let { req, res } = sendQueue.shift()
            res.status(400).json({ reason: message.reason })
        }
        db.delete(id)
        
        return;
    }
    if (functionToResource.has(id)) {
        let resourceHeap = functionToResource.get(id)
        heap.push(resourceHeap, {
            resource_id: message.resource_id,
            open_request_count: 0
        }, libSupport.compare)
        logger.warn("Horizontally scaling up: " +
            JSON.stringify(functionToResource.get(id)));

    } else {
        /**
        * function to resource map - holds a min heap of resources associated with a function
        * the min heap is sorted based on a metric [TBD] like CPU usage, request count, mem usage etc
        * TODO: decide on metric to use for sorting.
        */
        let resourceHeap = []
        heap.push(resourceHeap, {
            resource_id: message.resource_id,
            open_request_count: 0
        }, libSupport.compare)
        functionToResource.set(id, resourceHeap)
        logger.warn("Creating new resource pool"
            + JSON.stringify(functionToResource.get(id)));

    }
    
    try {
        let resource = resourceMap.get(message.resource_id)
        resource.deployed = true
        libSupport.logBroadcast({
            entity_id: message.entity_id,
            "reason": "deployment",
            "status": true,
            starttime: (Date.now() - resource.deploy_request_time)
        }, message.resource_id, resourceMap)
        
        if (db.has(id)) {
            let sendQueue = db.get(id)
            logger.info("forwarding request via reverse proxy to: " + JSON.stringify(resource));
            while (sendQueue && sendQueue.length != 0) {
                let { req, res } = sendQueue.shift()
                libSupport.reverseProxy(req, res, functionToResource, resourceMap, functionBranchTree)
                    .then(() => {

                    })
            }
            db.delete(id)
        }
        
        libSupport.metrics.collectMetrics({type: "scale", value: 
            functionToResource.get(id).length, 
            functionHash: message.functionHash, runtime: message.runtime, 
            starttime: (Date.now() - resource.deploy_request_time)})
    } catch (e) {
        logger.error(e.message)
    }

}

consumer.on('message', function (message) {
    
    let topic = message.topic
        message = message.value
    // console.log(topic, message)
    if (topic === "response") {
        logger.info("response " + message);
        
        
    } else if (topic === constants.topics.heartbeat) {
        message = JSON.parse(message)
        if (Date.now() - message.timestamp < 1000)
            if (!workerNodes.has(message.address))  {
                workerNodes.set(message.address, message.timestamp)
                logger.warn("New worker discovered. Worker List: ")
                logger.warn(workerNodes)
            }
    } else if (topic == constants.topics.deployed) {
        try {
            message = JSON.parse(message)
        } catch (e) {
            // process.exit(0)
        }
        postDeploy(message)
        
    } else if (topic == constants.topics.remove_worker) {
        logger.warn("Worker blown: Removing Metadata " + message);
        try {
            message = JSON.parse(message)
        } catch(e) {
            // process.exit(0)
        }
        usedPort.delete(message.port)
        let id = message.functionHash + message.runtime
        if (functionToResource.has(id)) {
            let resourceArray = functionToResource.get(id)
            for (let i = 0; i < resourceArray.length; i++)
                if (resourceArray[i].resource_id === message.resource_id) {
                    resourceArray.splice(i, 1);
                    break;
                }

            heap.heapify(resourceArray, libSupport.compare)
            libSupport.metrics.collectMetrics({type: "scale", value: 
                resourceArray.length, 
                functionHash: message.functionHash, runtime: message.runtime})
            libSupport.logBroadcast({
                entity_id: message.entity_id,
                "reason": "terminate",
                "total_request": message.total_request,
                "status": true
            }, message.resource_id, resourceMap)
            .then(() => {
                resourceMap.delete(message.resource_id)
                if (resourceArray.length == 0)
                    functionToResource.delete(id)
            })
            
        }

        

    } else if (topic == constants.topics.hscale) {
        message = JSON.parse(message)
        let resource_id = libSupport.makeid(constants.id_size), // each function resource request is associated with an unique ID
            runtime = message.runtime,
            functionHash = message.functionHash
        logger.info(`Generated new resource ID: ${resource_id} for runtime: ${runtime}`);
        console.log("Resource Status: ", functionToResource);
        if (!functionToResource.has(functionHash + runtime) && !db.has(functionHash + runtime)) {
            console.log("adding db");
            
            db.set(functionHash + runtime, [])
        }
        /**
         * Request RM for resource
         */
        logger.info("Requesting RM " + JSON.stringify({
            resource_id,
            "memory": 332,
        }))

        resourceMap.set(resource_id, {
            runtime, functionHash, port: null, node_id: null,
            deployed: false, deploy_request_time: Date.now()
        })


        let payloadToRM = [{
            topic: constants.topics.request_dm_2_rm, // changing from REQUEST_DM_2_RM
            messages: JSON.stringify({
                resource_id,
                "memory": 332,
            }),
            partition: 0
        }]
        
        producer.send(payloadToRM, () => {
            // db.set(functionHash + runtime, { req, res })
            console.log("sent rm");

        })
    } else if (topic == constants.topics.response_rm_2_dm) {
        
        logger.info("Response from RM: " + message);
        message = JSON.parse(message)
        let resourceChoice = message.nodes[0]
        if (resourceMap.has(message.resource_id)) {
            let resource = resourceMap.get(message.resource_id)
            if (typeof resourceChoice === 'string') {
                resource.port = libSupport.getPort(usedPort)
                resource.node_id = resourceChoice
            } else {
                resource.port = (resourceChoice.port) ? resourceChoice.port : libSupport.getPort(usedPort)
                resource.node_id = resourceChoice.node_id
            }
            
            let payload = [{
                topic: resource.node_id,
                messages: JSON.stringify({
                    "type": "execute", // Request sent to Dispatch Daemon via Kafka for actual deployment at the Worker
                    resource_id: message.resource_id,
                    runtime: resource.runtime, functionHash: resource.functionHash,
                    port: resource.port, resources: {
                        memory: resource.memory
                    }
                }),
                partition: 0
            }]
            // logger.info(resourceMap);
            producer.send(payload, () => {
                logger.info(`Resource Deployment request sent to Dispatch Agent`)
            })
        } else {
            logger.error("something went wrong, resource not found in resourceMap")
        }
        
    }
});

function autoscalar() {
    functionToResource.forEach((resourceList, functionKey, map) => {
        
        if (resourceList.length > 0 && 
            resourceList[resourceList.length - 1].open_request_count > constants.autoscalar_metrics.open_request_threshold) {
            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: constants.topics.hscale,
                messages: JSON.stringify({ "runtime": resource.runtime, "functionHash": resource.functionHash })
            }]
            producer.send(payload, function () { })
        }
    });

}

/**
 * Speculative deployment:
 * If function MLE path is present then deploy those parts of the path which are
 * not already running
 * 
 * FIXME: Currently supports homogenous runtime chain i.e takes runtime as a param. 
 * Change it to also profile runtime
 */
async function speculative_deployment(req, runtime) {
    if (constants.speculative_deployment && req.headers['x-resource-id'] === undefined) {
        // console.log(functionBranchTree, req.params.id);

        if (functionBranchTree.has(req.params.id)) {
            let branchInfo = functionBranchTree.get(req.params.id)
            console.log("mle_path", branchInfo.mle_path);

            if (branchInfo.mle_path && branchInfo.mle_path.length > 1) {
                for (let node of branchInfo.mle_path)
                    node.id = node.node
                let metrics = await libSupport.fetchData(metricsDB + "_bulk_get", {
                    method: 'post',
                    body: JSON.stringify({
                        docs: branchInfo.mle_path
                    }),
                    headers: { 'Content-Type': 'application/json' },
                })
                console.log(util.inspect(metrics, false, null, true /* enable colors */))
                
                for (let node of branchInfo.mle_path) {
                    // console.log(functionToResource);

                    if (!functionToResource.has(node.node + runtime) && !db.has(node.node + runtime)) {
                        console.log("Deploying according to MLE path: ", node.node);

                        let payload = [{
                            topic: constants.topics.hscale,
                            messages: JSON.stringify({ "runtime": "container", "functionHash": node.node })
                        }]
                        producer.send(payload, function () { })
                        db.set(node.node + runtime, [])
                    }
                }
            }
        }
    }
}
setInterval(libSupport.metrics.broadcastMetrics, 5000)
setInterval(libSupport.viterbi, 1000, functionBranchTree)
setInterval(autoscalar, 1000);
setInterval(dispatch, 1000);
app.listen(port, () => logger.info(`Server listening on port ${port}!`))