execute.js

'use strict';
// const isolateBackend = require('./isolate')
const fs = require('fs')
const { spawn } = require('child_process');
const constants = require("../constants.json")
const libSupport = require('./lib')
const { Worker, isMainThread, workerData } = require('worker_threads');
const registry_url = constants.registry_url
const logger = libSupport.logger

function runIsolate(local_repository, metadata) {
    let port = metadata.port,
        functionHash = metadata.functionHash,
        resource_id = metadata.resource_id,
        memory = metadata.resources.memory
    let filename = local_repository + functionHash + ".js"
    return new Promise((resolve, reject) => {

        const worker = new Worker(filename, {
            argv: [resource_id, functionHash, port, "isolate", constants.network.external.kafka_host],
            resourceLimits: {
                maxOldGenerationSizeMb: memory
            }
        });
        worker.on('message', resolve);
        worker.on('error', reject);
        worker.on('exit', (code) => {
            if (code !== 0)
                reject(new Error(`Worker stopped with exit code ${code}`));
            logger.info(`Isolate Worker with resource_id ${resource_id} blown`);
            resolve()
        })
    });

}

function runProcess(local_repository, metadata) {
    let port = metadata.port,
        functionHash = metadata.functionHash,
        resource_id = metadata.resource_id,
        memory = metadata.resources.memory
    let filename = local_repository + functionHash + ".js"
    return new Promise((resolve, reject) => {
        let timeStart = Date.now()
        
        const process = spawn('node', [filename, resource_id, functionHash, port, "process", 
                constants.network.external.kafka_host, `--max-old-space-size=${memory}` ]);
        process.stdout.on('data', (data) => {
            console.log(`stdout: ${data}`);
            let timeDifference = Math.ceil((Date.now() - timeStart))
            console.log("process time taken: ", timeDifference);
            
        });

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

        process.on('close', (code) => {
            resolve(process.pid);
            logger.info(`Process Environment with resource_id ${resource_id} blown`);
        });
    })

}


function runContainer(metadata) {
    let imageName = metadata.functionHash,
        port = metadata.port,
        resource_id = metadata.resource_id,
        memory = metadata.resources.memory

    logger.info(imageName);

    return new Promise((resolve, reject) => {
        let timeStart = Date.now()

        const process_checkImage = spawn('docker', ["inspect", registry_url + imageName])

        process_checkImage.on('close', (code) => {
            if (code != 0) {
                const process_pullImage = spawn('docker', ["pull", registry_url + imageName]);

                process_pullImage.stderr.on('data', (data) => {
                    console.error(`stderr: ${data}`);
                    reject(data);
                });

                process_pullImage.on('close', (code) => {
                    if (code != 0)
                        reject("error")
                    else {
                        const process = spawn('docker', ["run", "--rm", `--network=${constants.network.network_bridge}`, "-p", `${port}:${port}`, 
                            "--name", resource_id, registry_url + imageName,
                            resource_id, imageName, port, "container", constants.network.internal.kafka_host]);
                        let result = "";
                        // timeStart = Date.now()
                        process.stdout.on('data', (data) => {
                            logger.info(`stdout: ${data}`);
                            let timeDifference = Math.ceil((Date.now() - timeStart))
                            logger.info("container run time taken: ", timeDifference);
                            result += data;
                            resolve(resource_id);
                        });

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

                        process.on('close', (code) => {
                            logger.info("Exiting container");
                            
                        })
                    }
                    
                })
            } else {
                    logger.info("container starting at port", port);
                    
                    const process = spawn('docker', ["run", "--rm", `--network=${constants.network.network_bridge}`, 
                            "-p", `${port}:${port}`, "--name", resource_id, 
                        registry_url + imageName, resource_id, imageName, port, "container", constants.network.internal.kafka_host]);
                    let result = "";
                    // timeStart = Date.now()
                    process.stdout.on('data', (data) => {
                        logger.info(`stdout: ${data}`);
                        let timeDifference = Math.ceil((Date.now() - timeStart))
                        logger.info("container run time taken: ", timeDifference);
                        resolve(resource_id);
                    });

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

                    process.on('close', (code) => {
                        logger.info("Exiting container");
                    })
                
            }
            
        })

        
    })

}

module.exports.runContainer = runContainer;
module.exports.runProcess = runProcess;
module.exports.runIsolate = runIsolate;