serverless-openwhisk invoke local using docker for https://github.com/serverless/serverless-openwhisk/issues/119

invokelocal.js

'use strict';

const fs = require('fs');
const rw = require('rw');
const { spawn, execSync } = require('child_process');
const request = require('requestretry');

const OPENWHISK_DEFAULTS = {

    // https://github.com/apache/incubator-openwhisk/blob/master/docs/reference.md#system-limits
    timeoutSec: 60,
    memoryLimitMB: 256,

    // https://github.com/apache/incubator-openwhisk/blob/master/ansible/files/runtimes.json
    // note: openwhisk deployments might have their own versions
    kinds: {
        // "nodejs"     : "openwhisk/nodejsaction:latest", // deprecated, image no longer available
        "nodejs:6"   : "openwhisk/nodejs6action:latest",
        "nodejs:8"   : "openwhisk/action-nodejs-v8:latest",
        "python"     : "openwhisk/python2action:latest",
        "python:2"   : "openwhisk/python2action:latest",
        "python:3"   : "openwhisk/python3action:latest",
        // "swift"      : "openwhisk/swiftaction:latest", // deprecated, image no longer available
        "swift:3"    : "openwhisk/swift3action:latest",   // deprecated, but still available
        "swift:3.1.1": "openwhisk/action-swift-v3.1.1:latest",
        "swift:4.1"  : "openwhisk/action-swift-v4.1:latest",
        "java"       : "openwhisk/java8action:latest",
        "php:7.1"    : "openwhisk/action-php-v7.1:latest",
        "php:7.2"    : "openwhisk/action-php-v7.2:latest",
        "native"     : "openwhisk/dockerskeleton:latest",
    }
}

// the OW runtime sever API isn't officially documented, some links:
// blog post: http://jamesthom.as/blog/2017/01/16/openwhisk-docker-actions/
// OW invoker code, look for /init and /run requests:
// https://github.com/apache/incubator-openwhisk/blob/master/common/scala/src/main/scala/whisk/core/containerpool/Container.scala
const RUNTIME_PORT = 8080;

// retry delay for action /init
const RETRY_DELAY_MS = 100;

function prettyJson(json) {
    return JSON.stringify(json, null, 4);
}

class OpenWhiskInvokeLocalDockerPlugin {

    constructor(serverless, options) {
        super();
        this.serverless = serverless;
        this.options = options;

        this.hooks = {
            'invoke:local:invoke': this.invokeLocal.bind(this),
        };

        this.disableServerlessLog();
    }

    debugLog(msg) {
        if (this.options.verbose) {
            this.log(msg);
        }
    }

    log(msg) {
        this.serverless.cli.log(msg);
    }
    
    disableServerlessLog() {
        // disable annoying webpack logging
        const cmds = this.serverless.processedInput.commands;
        if (cmds.length === 2 && cmds[0] === 'invoke' && cmds[1] === 'local') {
            this.serverlessLog = this.serverless.cli.log;
            this.serverless.cli.log = function() {};
        }
    }

    enableServerlessLog() {
        if (this.serverlessLog) {
            this.serverless.cli.log = this.serverlessLog;
        }
    }

    disableOpenwhiskPlugin() {
        // HACK to prevent the serverless-openwhisk plugin from handling the invoke local as well
        this.serverless.pluginManager.plugins.forEach(plugin => {
            if (plugin.constructor.name == "OpenWhiskInvokeLocal") {
                function noop() {
                    return Promise.resolve();
                }
                plugin.validate = noop;
                plugin.loadEnvVars = noop;
                plugin.invokeLocal = noop;
            }
        });
    }

    docker(args, opts) {
        this.debugLog("> docker " + args);
        const result = execSync("docker " + args, opts);
        if (result) {
            return result.toString().trim();
        } else {
            return '';
        }
    }
    
    dockerSpawn(args) {
        this.debugLog("> docker " + args);
    
        const proc = spawn('docker', args.split(' '));
    
        proc.stdout.on('data', function(data) {
            process.stdout.write(data.toString());
        });
        proc.stderr.on('data', function(data) {
            process.stderr.write(data.toString());
        });

        return proc;
    }
    
    getMainFunction(func) {
        if (!func) {
            return "main";
        }
        const parts = func.handler.split('.');
        if (parts.length < 2) {
          return func;
        }
        return parts[parts.length - 1];
    }

    getActionParams() {
        // if not given as json argument (--data) already
        if (!this.options.data) {
            if (this.options.path) {
                // read from given json file
                this.options.data = require(this.options.path);
            } else {
                // read from std input
                this.options.data = rw.readFileSync("/dev/stdin", "utf8");
            }
        }

        // parse json if necessary
        if (typeof this.options.data !== 'object') {
            try {
                this.options.data = JSON.parse(this.options.data);
            } catch (e) {
                // do nothing if it's a simple string or object already
                this.log(e);
            }
        }

        // merge in default action params
        const defaultParams = this.func.parameters || {};
        this.options.data = Object.assign(defaultParams, this.options.data);
        
        return this.options.data;
    }

    getRuntime() {
        const kind =
            this.func.runtime ||
            this.serverless.service.provider.runtime ||
            "nodejs";
        
        // remove :default suffix if present
        const kind2 = kind.replace(/:default$/, '');

        const image = OPENWHISK_DEFAULTS.kinds[kind2];

        if (!image) {
            throw `Unsupported kind: ${kind}`;
        }

        return image;
    }

    containerName(name) {
        // TODO: docker container names are restricted to [a-zA-Z0-9][a-zA-Z0-9_.-]
        // for now, just replace the slashes if there is an openwhisk package in the name
        name = name.replace('/', '--');
        // add sls (serverless) prefix to containers we start locally
        return `serverless-${name}`;
    }

    containerRuns(name) {
        try {
            this.docker(`inspect -f '{{.State.Running}}' ${name}`, {stdio: 'ignore'});
            return true;
        } catch (e) {
            return false;
        }
    }

    getHost(name) {
        if (this.host) {
            return this.host;
        }
        // prefer the specific container id if available
        if (this.containerId) {
            name = this.containerId;
        }
        this.host = this.docker(`port ${name} ${RUNTIME_PORT}`);
        return this.host;
    }

    invokeLocal() {
        this.disableOpenwhiskPlugin();
        this.enableServerlessLog();

        // common preparations
        this.func = this.serverless.service.getFunction(this.options.function);

        this.actionName = this.func.name || `${this.serverless.service.name}_${this.options.function}`;
        const name = this.containerName(this.actionName);

        if (this.options.status) {
            if (this.containerRuns(name)) {
                this.log(`Container is running: ${name}`);
                process.exit();
            } else {
                this.log(`No running container.`);
                process.exit(1);
            }

        } else if (this.options.start) {
            // build and start container, ready for invocations
            return this.build()
                .then(() => this.startContainer(name))
                .then(() => this.initAction())
                .then(() => this.log(`Started container ${name} ${this.containerId}.`));

        } else if (this.options.stop) {
            // shutdown and remove container
            return this.stopContainer(name)
                .then(() => this.log(`Stopped container ${name}.`));

        } else if (this.containerRuns(name)) {
            // invocation on an existing container
            return this.runAction(name);

        } else {
            // steps for complete single invocation
            return this.build()
                .then(() => this.startContainer(name))
                .then(() => this.initAction())
                .then(() => this.runAction())
                .finally(() => this.stopContainer());
        }
    }

    build() {
        if (this.serverless.service.plugins.includes('serverless-webpack')) {
            // webpack:package
            // serverless-webpack will not package the zip (webpack:package) on invoke local,
            // but we need the zip file for local deployment as well, so we trigger it explicitly here
            return this.serverless.pluginManager.spawn('webpack:package');

        } else {
            return Promise.resolve();
        }
    }

    startContainer(name) {
        return new Promise((resolve, reject) => {
            if (name) {
                try {
                    // make sure a left over container is removed
                    this.docker(`kill ${name}`, {stdio: 'ignore'});
                } catch (ignore) {}
            }

            try {
                const setName     = name ? `--name "${name}"` : '';
                const memoryBytes = (this.func.memory || OPENWHISK_DEFAULTS.memoryLimitMB) * 1024 * 1024;
                const customArgs  = this.options.dockerArgs || "";
                const image       = this.func.image || this.getRuntime();

                this.containerId = this.docker(`run -d --rm ${setName} -p ${RUNTIME_PORT} -m ${memoryBytes} ${customArgs} ${image}`);

                resolve();
            } catch (e) {
                reject(e);
            }
        });
    }

    stopContainer(name) {
        return new Promise((resolve, reject) => {
            try {
                if (this.containerId) {
                    this.docker(`kill ${this.containerId}`);
                } else if (name) {
                    this.docker(`kill ${name}`);
                }
                return resolve();
            } catch (e) {
                return reject(e);
            }
        });
    }

    initAction() {
        return new Promise((resolve, reject) => {
            this.debugLog(`initializing action: POST http://${this.getHost()}/init`);

            const zipFile = '.serverless/' + this.options.function + '.zip';
            if (!fs.existsSync(zipFile)) {
                throw new this.serverless.classes.Error('The packaging produced no action zip at ' + zipFile);
            }

            this.func.timeout = this.func.timeout || OPENWHISK_DEFAULTS.timeoutSec;

            request.post(
                // POST request
                {
                    url: `http://${this.getHost()}/init`,
                    json: {
                        value: {
                            binary: true,
                            main: this.getMainFunction(this.func),
                            code: fs.readFileSync(zipFile).toString('base64'),
                        }
                    },

                    maxAttempts: this.func.timeout * 1000 / RETRY_DELAY_MS,
                    retryDelay: RETRY_DELAY_MS,

                    retryStrategy: (err, response, body) => {
                        if (this.options.verbose) {
                            this.serverless.cli.printDot();
                        }
                        return request.RetryStrategies.NetworkError(err, response, body);
                    }
                },
                // request callback
                (error, response, body) => {
                    // print space after the ... above upon each retry
                    const attempts = response ? response.attempts : error.attempts;
                    if (attempts > 1 && this.options.verbose) {
                        process.stdout.write(' ');
                    }

                    if (error) {
                        console.log(error);
                        return reject();
                    }
                    const ok = (body && body.OK === true);
                    if (!ok) {
                        console.log();
                        this.log("/init failed with:");
                        if (body.error) {
                            console.log(body.error);
                        } else {
                            // unkown error response, print everything
                            console.log(prettyJson(body));
                        }
                        return reject();
                    }

                    this.debugLog('action ready');

                    return resolve();
                }
            )
        });
    }

    runAction(name) {
        return new Promise((resolve, reject) => {
            // show docker logs in real time
            const procDockerLogs = this.dockerSpawn(`logs -f ${this.containerId || name}`);

            this.func.timeout = this.func.timeout || OPENWHISK_DEFAULTS.timeoutSec;
            this.debugLog(`invoking action: POST http://${this.getHost(name)}/run (timeout ${this.func.timeout} seconds)`);

            const params = this.getActionParams();
            if (this.options.verbose) {
                console.log(prettyJson(params));
            }

            request.post(
                // POST request
                {
                    url: `http://${this.getHost(name)}/run`,
                    maxAttempts: 1,
                    timeout: this.func.timeout * 1000,
                    json: {
                        value: params
                    }
                },
                // request callback
                (error, response, body) => {
                    procDockerLogs.kill();

                    if (error) {
                        if (error.code === 'ESOCKETTIMEDOUT') {
                            return reject(`action timed out after ${this.func.timeout} seconds`);
                        } else {
                            console.log(error);
                            return reject(`action invocation failed: ${error.message}`)
                        }
                    } else if (body) {
                        if (body.error) {
                            return reject(`action invocation returned error: ${prettyJson(body.error)}`)
                        }

                        this.log(`result of action ${this.actionName}:`);
                        console.log(prettyJson(body));

                    } else {
                        this.log("action returned empty result");
                    }
                    resolve();
                }
            )
        });
    }
}

module.exports = OpenWhiskInvokeLocalDockerPlugin;