index.js

'use strict';
const constants = require(".././constants.json")
const secrets = require('./secrets.json')
const config = require('./config.json')
const libSupport = require('./lib')
libSupport.updateConfig()
const node_id = config.id
const {spawn } = require('child_process')
const execute = require('./execute')
const fs = require('fs')
const fetch = require('node-fetch');

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

const kafka = require('kafka-node')
const logger = libSupport.logger

const local_repository = __dirname + "/local_repository/"
const host_url = "http://" + constants.master_address + ":" + constants.master_port

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

libSupport.makeTopic(node_id).then(() => {
    logger.info("node topic created")
    let consumer = new Consumer(client,
        [
            { topic: node_id, partition: 0, offset: 0 }
        ],
        [
            { autoCommit: true }
        ])
    consumer.on('message', function (message) {
        logger.info(message);
        let topic = message.topic
        message = message.value
        message = JSON.parse(message)
        let runtime = message.runtime
        let functionHash = message.functionHash
        let resource_id = message.resource_id
        let port = message.port
        /**
         * Download necessary files (function file) and Start resource deployment
         */
        if (message.type === "execute") {
            logger.info("Received Deployment request for resource_id: " + resource_id);
            fetch(metadataDB + functionHash).then(res => res.json())
            .then(json => {
                console.log("metadata", json);
                
                libSupport.download(host_url + "/repository/" + functionHash + ".js", local_repository + functionHash + ".js").then(() => {
                    let metadata = {
                        resource_id, functionHash,
                        runtime, port,
                        resources: {
                            memory: json.memory
                        }
                    }
                    startWorker(local_repository, producer, metadata)
            })
            }).catch(err => {
                logger.error("something went wrong" + err.toString())
            });
            

        }

    })
})

/**
 * download and start grunt
 */
libSupport.download(constants.grunt_host, "grunt", false).then(() => {
    logger.info("Downloaded grunt binary from repository")
    fs.chmod('grunt', 0o755, (err) => {
        logger.info("grunt made executable. Starting grunt")
        let grunt = spawn('./grunt', [node_id])
        grunt.stdout.on('data', data => {
            logger.info(data.toString());

        })

        grunt.stderr.on('data', data => {
            logger.info(data.toString());

        })
        grunt.on('close', (code) => {
            logger.info("Grunt exited with exit code", code);

        })
    })

})

    
/**
 * Start a worker executor of the runtime type
 * @param {String} local_repository 
 * @param {String} functionHash 
 * @param {String} resource_id 
 * @param {String} producer 
 * @param {String} runtime 
 * @param {Number} port 
 */
function startWorker(local_repository, producer, metadata) {
    let runtime = metadata.runtime
    console.log(metadata);
    
    logger.info(`Using port ${metadata.port} for functionHash ${metadata.functionHash}`)
    
    if (runtime === "isolate")
        execute.runIsolate(local_repository, metadata)
        .catch(err => {
            logger.error("=====================deployment failed=========================");
            producer.send([{
                topic: "deployed",
                messages: JSON.stringify({
                    "status": false,
                    resource_id: metadata.resource_id,
                    "reason": "isolate exit"
                })
            }], () => { })
        })
    else if (runtime === "process")
        execute.runProcess(local_repository, metadata)
        .catch(err => {
            logger.error("=====================deployment failed=========================");
            producer.send([{ topic: "deployed",
                messages: JSON.stringify({ 
                "status": false, 
                resource_id: metadata.resource_id,
                "reason": "process exit"
            }) }], () => { })
        })
    else if (runtime === "container")
        execute.runContainer(metadata)
    else {
        producer.send(
            [{
                topic: "response",
                messages: JSON.stringify({ status: "unknown runtime" })
            }], () => { })

        return
    }
    
}

function heartbeat() {
    let payload = [{
        topic: "heartbeat",
        messages: JSON.stringify({"address": node_id, "timestamp": Date.now()})
    }]
    producer.send(payload, function(cb) {})
}


setInterval(heartbeat, 1000);