Component as service -- pass tests except one time execs

Author: Anton Kotenko

.eslintrc.json

@@ -0,0 +1,46 @@
+{
+    "root": true,
+    "parserOptions": {
+        "ecmaVersion": 2018,
+        "sourceType": "script",
+        "ecmaFeatures": {
+            "impliedStrict": true
+        }
+    },
+    "plugins": [
+        "eslint-plugin-mocha"
+    ],
+    "extends": [
+        "standard"
+    ],
+    "rules": {
+        "mocha/no-exclusive-tests": "error",
+        "mocha/no-skipped-tests": "error",
+        "node/no-deprecated-api": "warn",
+        "semi": [
+            "error",
+            "always"
+        ],
+        "indent": [
+            "error",
+            4
+        ],
+        "max-len": [
+            "warn",
+            120,
+            4,
+            {
+                "ignoreUrls": true,
+                "ignoreTemplateLiterals": true
+            }
+        ],
+        "space-before-function-paren": [
+            "warn",
+            {
+                "anonymous": "always",
+                "named": "never",
+                "asyncArrow": "always"
+            }
+        ]
+    }
+}

.gitignore

@@ -2,3 +2,4 @@ coverage
 node_modules
 .DS_Store
 .idea
+.*.sw[op]

lib/AmqpConnWrapper.js

@@ -0,0 +1,54 @@
+const amqplib = require('amqplib');
+// FIXME TESTS
+class AmqpConnWrapper {
+    constructor(amqpUri, logger) {
+        this._logger = logger;
+        this._amqpUri = amqpUri;
+    }
+    async start() {
+        // FIXME safety for double calls
+        // TODO handle channel errors. Recoonect????
+        this._amqp = await amqplib.connect(this._amqpUri);
+        // FIXME
+        //this.amqp.on('error', this._logger.criticalerrorandExit);
+        //this.amqp.on('close', this._logger.criticalerrorandExit);
+        this._logger.debug('Connected to amqp');
+
+        this._subscribeChannel = await this._amqp.createChannel();
+        // FIXME
+        //this.subscribeChannel.on('error', this._logger.criticalErrorAndExit);
+        this._logger.debug('Opened subscribe channel');
+
+        this._publishChannel = await this._amqp.createConfirmChannel();
+        // FIXME
+        //this.publishChannel.on('error', this._logger.criticalErrorAndExit);
+        this._logger.debug('Opened publish channel');
+    }
+    async stop() {
+        // FIXME safety for double calls
+        this._logger.trace('Close AMQP connections');
+        try {
+            await this._subscribeChannel.close();
+        } catch (alreadyClosed) {
+            this._logger.debug('Subscribe channel is closed already');
+        }
+        try {
+            await this._publishChannel.close();
+        } catch (alreadyClosed) {
+            this._logger.debug('Publish channel is closed already');
+        }
+        try {
+            await this._amqp.close();
+        } catch (alreadyClosed) {
+            this._logger.debug('AMQP connection is closed already');
+        }
+        this._logger.debug('Successfully closed AMQP connections');
+    }
+    getPublishChannel() {
+        return this._publishChannel;
+    }
+    getSubscribeChannel() {
+        return this._subscribeChannel;
+    }
+}
+module.exports = AmqpConnWrapper;

lib/App.js

@@ -0,0 +1,109 @@
+class App {
+    /**
+     * Start application
+     * Implemented as SAFE in terms of exception
+     */
+    async start() {
+        process.on('uncaughtException', (e) => {
+            try {
+                this._safeLogError(e, 'Uncaught exception');
+                this._safeFatalErrorHook(e);
+            } catch (error) {
+                console.error('Error handling uncaught exception', error);
+            } finally {
+                process.exit(this.constructor.EXIT_CODES.UNCAUGHT_EXCEPTION);
+            }
+        });
+        process.on('unhandledRejection', (e) => {
+            try {
+                this._safeLogError(e, 'Uncaught rejection');
+                this._safeFatalErrorHook(e);
+            } catch (error) {
+                console.error('Error handling uncaught rejection', error);
+            } finally {
+                process.exit(this.constructor.EXIT_CODES.UNCAUGHT_REJECTION);
+            }
+        });
+        process.on('SIGTERM', this.stop.bind(this));
+        process.on('SIGINT', this.stop.bind(this));
+
+        try {
+            await this._start();
+        } catch (e) {
+            this._safeLogError(e, 'Failed to start');
+            this._safeFatalErrorHook(e);
+            process.exit(this.constructor.EXIT_CODES.FAILED_TO_START);
+        }
+    }
+
+
+    /**
+     * Gracefully stop application.
+     * part of standard shutdown procedure
+     * Implemented as SAFE in terms of exception
+     */
+    async stop() {
+        try {
+            await this._stop();
+        } catch (e) {
+            this._safeLogError(e, 'Failed to stop gracefully');
+            process.exit(this.constructor.EXIT_CODES.FAILED_TO_STOP);
+        }
+    }
+
+    /**
+     * Start procedure implementation.
+     * Not safe in terms of exceptions. May throw.
+     * In case of error, exception will be caught, logged
+     * and process will be destroyed in "cruel" way
+     */
+    _start() {
+        throw new Error('implement me');
+    }
+
+    /**
+     * Graceful stop procedure implementation.
+     * Not safe in terms of exceptions. May throw.
+     * In case of error, exception will be caught, logged
+     * and process will be destroyed in "cruel" way
+     */
+    _stop() {
+        throw new Error('implement me');
+    }
+
+    _logError() {
+        throw new Error('implement me');
+    }
+
+    async _handleFatalError() {
+        throw new Error('implement me');
+    }
+
+    _safeLogError() {
+        console.error(...arguments);
+        try {
+            this._logError(...arguments);
+        } catch (e) {
+            console.error('Uncaught rejection', e);
+        }
+    }
+
+    _safeFatalErrorHook() {
+        try {
+            this._handleFatalError(...arguments);
+        } catch (e) {
+            this._safeLogError(e, 'Uncaught rejection');
+        }
+    }
+
+    static get EXIT_CODES() {
+        return {
+            SUCCESS: 0,
+            FAILED_TO_START: 1,
+            FAILED_TO_STOP: 2,
+            UNCAUGHT_EXCEPTION: 3,
+            UNCAUGHT_REJECTION: 4
+        };
+    }
+}
+module.exports = App;

lib/Multisailor.js

@@ -0,0 +1,99 @@
+const assert = require('assert');
+const { ComponentLogger } = require('./logging.js');
+const { AmqpCommunicationLayer } = require('./amqp.js');
+const { Sailor } = require('./sailor.js');
+
+
+class MultiSailor {
+    constructor(amqpConn, config, logger) {
+        this._amqpConn = amqpConn;
+        this._logger = logger; // not context aware logger
+        this._sharedCommunicationLayer = new AmqpCommunicationLayer(this._amqpConn, this._logger, config);
+        this._config = config;
+        this._sailors = {};
+    }
+    async start() {
+        await this._sharedCommunicationLayer.listenQueue(this._dispatch.bind(this));
+    }
+    async stop() {
+        await this._sharedCommunicationLayer.unlistenQueue();
+    }
+
+    async _dispatch(msg) {
+        try {
+            const sailor = await this._getSailor(msg);
+            sailor.processMessage(msg);
+        } catch (e) {
+            this._logger.error(e, 'Can not handle message');
+        }
+    }
+
+    async _getSailor(msg) {
+        let sailor;
+        const { taskId: flowId, stepId } = msg.properties.headers;
+        assert(flowId && stepId, 'Incorrect message format, no taskId/stepId');
+        try {
+            this._sailors[flowId] = this._sailors[flowId] || {};
+            if (this._sailors[flowId][stepId]) {
+                return this._sailors[flowId][stepId];
+            }
+            const config = this._restoreExecContext(msg);
+            const logger = new ComponentLogger(config); // FIXME defaults. Prooperly create it
+            const communicationLayer = new AmqpCommunicationLayer(this._amqpConn, logger, config);
+            const sailor = new Sailor(
+                communicationLayer,
+                logger,
+                config
+            );
+            await sailor.prepare();
+            /**
+             * TODO skip atm. Generally it's possible to fetch data from api.
+             * to handle if startup is requried.
+             * Startup is requried if no hooks data and sailor supports startup hook
+             * if (STARTUP_REQUERD)
+             *     await sailor.startup();
+             * }
+            */
+            await sailor.init();
+            // NOTICE do not call sailor.run here.
+            // message dispatching is done in different ways in MultiSailor and SingleSailor modes
+            return this._sailors[flowId][stepId] = sailor;
+        } catch (e) {
+            sailor && sailor.reportError(e);
+            // FIXME try to log error in context of current message
+            throw e;
+        }
+    }
+
+    _restoreExecContext(msg) {
+        const manadatoryValues = Object.assign({}, this._config);
+        manadatoryValues.FLOW_ID = msg.properties.headers.taskId;
+        manadatoryValues.STEP_ID = msg.properties.headers.stepId;
+        manadatoryValues.EXEC_ID = msg.properties.headers.execId;
+        manadatoryValues.USER_ID = msg.properties.headers.userId;
+
+        //// used by communication layer only
+        //manadatoryValues.PUBLISH_MESSAGES_TO
+        //manadatoryValues.DATA_ROUTING_KEY
+        //manadatoryValues.ERROR_ROUTING_KEY
+        //manadatoryValues.REBOUND_ROUTING_KEY
+        //manadatoryValues.SNAPSHOT_ROUTING_KEY
+
+        //manadatoryValues.CONTAINER_ID = //STUB randomize
+
+
+        //manadatoryValues.WORKSPACE_ID = stepData.
+        //manadatoryValues.COMP_ID = stepData.comp_id;
+        //manadatoryValues.FUNCTION = stepData.function;
+
+
+        //manadatoryValues.COMP_NAME = stepData.comp_name; // FIXME
+        //manadatoryValues.EXEC_TYPE = 'flow-step'; // FIXME should match admiral ENV_VARS_CREATOR
+        //manadatoryValues.FLOW_VERSION = stepData.flow_version; // FIXME
+        //manadatoryValues.TENANT_ID = stepData.tenant_id
+        //manadatoryValues.TASK_USER_EMAIL = stepData.XXXX
+        //manadatoryValues.EXECUTION_RESULT_ID  = 'ZZZZ'// FIXME skip ATM one time execs
+        return manadatoryValues;
+    }
+}
+module.exports = MultiSailor;

lib/SingleApp.js

@@ -0,0 +1,55 @@
+const App = require('./App.js');
+
+class SingleApp extends App {
+    async _start() {
+        const { ComponentLogger } = require('./logging.js');
+        const Sailor = require('./sailor.js');
+        const AmqpConnWrapper = require('./AmqpConnWrapper.js');
+        const { AmqpCommunicationLayer } = require('./amqp.js');
+        const { SingleSailorConfig } = require('./settings.js');
+
+        this._config = SingleSailorConfig.fromEnv();
+        this._logger = new ComponentLogger(this._config);
+        this._amqpConn = new AmqpConnWrapper(this._config.AMQP_URI, this._logger);
+        this._communicationLayer = new AmqpCommunicationLayer(this._amqpConn, this._config, this._logger);
+        this._sailor = new Sailor(this._communicationLayer, this._config, this._logger);
+        if (this._config.HOOK_SHUTDOWN) {
+            await this._sailor.prepare();
+            await this._sailor.shutdown();
+            return;
+        } else {
+            await this._amqpConn.start();
+            await this._sailor.prepare();
+
+            if (this._config.STARTUP_REQUIRED) {
+                await this._sailor.startup();
+            }
+
+            await this._sailor.init();
+            await this._sailor.run();
+        }
+    }
+
+    async _stop() {
+        if (this._communicationLayer) {
+            await this._communicationLayer.unlistenQueue();
+        }
+        if (this._amqpConn) {
+            await this._amqpConn.stop();
+            this._amqpConn = null;
+        }
+    }
+
+    async _handleFatalError(error) {
+        if (this._sailor && !this._config.HOOK_SHUTDOWN) {
+            await this._sailor.reportError(error);
+        }
+    }
+
+    _logError() {
+        if (this._logger) {
+            this._logger.error(...arguments);
+        }
+    }
+}
+module.exports = SingleApp;