New fields in rmq messages (#101)

* - add `workspaceId` and `containerId` in rabbit messages
- minor refactoring

* logger fixes

* fixes

Author: andkom

.npmignore

@@ -1 +1,5 @@
+.env
+.idea
+.circleci
+mocha_spec
 spec

README.md

@@ -77,7 +77,6 @@ Sailor logger adds the following extra context to log messages:
 - `ELASTICIO_FLOW_ID`
 - `ELASTICIO_FLOW_VERSION`
 - `ELASTICIO_FUNCTION`
-- `ELASTICIO_ORGANIZATION_ID`
 - `ELASTICIO_STEP_ID`
 - `ELASTICIO_TASK_USER_EMAIL`
 - `ELASTICIO_TENANT_ID`

lib/amqp.js

@@ -164,13 +164,10 @@ class Amqp {
     }
 
     async sendHttpReply(data, properties) {
-
         const routingKey = properties.headers.reply_to;
 
         if (!routingKey) {
-            throw new Error(
-                //eslint-disable-next-line no-useless-escape
-                `Component emitted \'httpReply\' event but \'reply_to\' was not found in AMQP headers`);
+            throw new Error(`Component emitted 'httpReply' event but 'reply_to' was not found in AMQP headers`);
         }
         return this.prepareMessageAndSendToExchange(data, properties, routingKey);
     }

lib/logging.js

@@ -24,7 +24,6 @@ const data = Object.assign(
         'ELASTICIO_FLOW_ID',
         'ELASTICIO_FLOW_VERSION',
         'ELASTICIO_FUNCTION',
-        'ELASTICIO_ORGANIZATION_ID',
         'ELASTICIO_STEP_ID',
         'ELASTICIO_TASK_USER_EMAIL',
         'ELASTICIO_TENANT_ID',

lib/sailor.js

@@ -80,9 +80,14 @@ class Sailor {
 
     reportError(err) {
         const headers = {
-            execId: process.env.ELASTICIO_EXEC_ID,
-            taskId: process.env.ELASTICIO_FLOW_ID,
-            userId: process.env.ELASTICIO_USER_ID
+            execId: this.settings.EXEC_ID,
+            taskId: this.settings.FLOW_ID,
+            workspaceId: this.settings.WORKSPACE_ID,
+            containerId: this.settings.CONTAINER_ID,
+            userId: this.settings.USER_ID,
+            stepId: this.settings.STEP_ID,
+            compId: this.settings.COMP_ID,
+            function: this.settings.FUNCTION
         };
         const props = createDefaultAmqpProperties(headers);
         this.amqpConnection.sendError(err, props);
@@ -174,7 +179,7 @@ class Sailor {
         const metaHeaders = _.pick(headers, metaHeaderNames);
         const metaHeadersLowerCased = _.mapKeys(metaHeaders, (value, key) => key.toLowerCase());
 
-        let result = _.pick(headers, ['taskId', 'execId', 'userId']);
+        let result = _.pick(headers, ['taskId', 'execId', 'workspaceId', 'containerId', 'userId', 'stepId', 'compId']);
         result = _.extend(result, metaHeadersLowerCased);
 
         result.threadId = headers.threadId || headers['x-eio-meta-trace-id'];
@@ -240,6 +245,11 @@ class Sailor {
         let outgoingMessageHeaders = _.clone(incomingMessageHeaders);
 
         outgoingMessageHeaders = _.extend(outgoingMessageHeaders, {
+            execId: settings.EXEC_ID,
+            taskId: settings.FLOW_ID,
+            workspaceId: settings.WORKSPACE_ID,
+            containerId: settings.CONTAINER_ID,
+            userId: settings.USER_ID,
             stepId: settings.STEP_ID,
             compId: settings.COMP_ID,
             function: settings.FUNCTION,
@@ -386,11 +396,9 @@ class Sailor {
                     logger.error('Failed to updated keys #%s', deliveryTag);
                     await onError(error);
                 }
-
             }
 
             function onEnd() {
-
                 if (endWasEmitted) {
                     logger.warn({
                         messagesCount: self.messagesCount,
@@ -471,4 +479,3 @@ function createDefaultAmqpProperties(headers) {
 }
 
 exports.Sailor = Sailor;
-

lib/settings.js

@@ -1,17 +1,17 @@
-var _ = require('lodash');
+const _ = require('lodash');
 
 exports.readFrom = readFrom;
 
 function readFrom(envVars) {
-
-    var result = {};
+    const result = {};
 
     // required settings
-
-    var requiredAlways = [
+    const requiredAlways = [
         'FLOW_ID',
         'EXEC_ID',
         'STEP_ID',
+        'CONTAINER_ID',
+        'WORKSPACE_ID',
 
         'USER_ID',
         'COMP_ID',
@@ -22,7 +22,7 @@ function readFrom(envVars) {
         'API_KEY'
     ];
 
-    var requiredForMessageProcessing = [
+    const requiredForMessageProcessing = [
         'AMQP_URI',
         'LISTEN_MESSAGES_ON',
         'PUBLISH_MESSAGES_TO',
@@ -33,7 +33,7 @@ function readFrom(envVars) {
         'SNAPSHOT_ROUTING_KEY'
     ];
 
-    var optional = {
+    const optional = {
         REBOUND_INITIAL_EXPIRATION: 15000,
         REBOUND_LIMIT: 20,
         COMPONENT_PATH: '',
@@ -52,7 +52,7 @@ function readFrom(envVars) {
     };
 
     _.forEach(requiredAlways, function readRequired(key) {
-        var envVarName = 'ELASTICIO_' + key;
+        const envVarName = 'ELASTICIO_' + key;
         result[key] = envVars[envVarName] || throwError(envVarName + ' is missing');
     });
 
@@ -64,7 +64,7 @@ function readFrom(envVars) {
     }
 
     _.forEach(optional, function readOptional(defaultValue, key) {
-        var envVarName = 'ELASTICIO_' + key;
+        const envVarName = 'ELASTICIO_' + key;
         if (typeof defaultValue === 'number' && envVars[envVarName]) {
             result[key] = parseInt(envVars[envVarName]) || defaultValue;
         } else {

mocha_spec/integration_helpers.js

@@ -8,11 +8,10 @@ const nock = require('nock');
 
 const env = process.env;
 
-
 class AmqpHelper extends EventEmitter {
-
     constructor() {
         super();
+
         this.httpReplyQueueName = PREFIX + 'request_reply_queue';
         this.httpReplyQueueRoutingKey = PREFIX + 'request_reply_routing_key';
         this.nextStepQueue = PREFIX + '_next_step_queue';
@@ -37,11 +36,11 @@ class AmqpHelper extends EventEmitter {
         return this.subscriptionChannel.publish(
             env.ELASTICIO_LISTEN_MESSAGES_ON,
             env.ELASTICIO_DATA_ROUTING_KEY,
-            new Buffer(JSON.stringify(message)),
-            {
+            new Buffer(JSON.stringify(message)), {
                 headers: Object.assign({
                     execId: env.ELASTICIO_EXEC_ID,
                     taskId: env.ELASTICIO_FLOW_ID,
+                    workspaceId: env.ELASTICIO_WORKSPACE_ID,
                     userId: env.ELASTICIO_USER_ID,
                     threadId,
                     messageId: parentMessageId
@@ -62,6 +61,7 @@ class AmqpHelper extends EventEmitter {
             durable: true,
             autoDelete: false
         };
+
         yield subscriptionChannel.assertExchange(env.ELASTICIO_LISTEN_MESSAGES_ON, 'direct', exchangeOptions);
         yield publishChannel.assertExchange(env.ELASTICIO_PUBLISH_MESSAGES_TO, 'direct', exchangeOptions);
 
@@ -114,11 +114,13 @@ class AmqpHelper extends EventEmitter {
                 that.consumer.bind(that, that.nextStepQueue),
                 { consumerTag: 'sailor_nodejs_1' }
             );
+
             yield that.publishChannel.consume(
                 that.nextStepErrorQueue,
                 that.consumer.bind(that, that.nextStepErrorQueue),
                 { consumerTag: 'sailor_nodejs_2' }
             );
+
             yield that.publishChannel.consume(
                 that.httpReplyQueueName,
                 that.consumer.bind(that, that.httpReplyQueueName),
@@ -128,7 +130,6 @@ class AmqpHelper extends EventEmitter {
     }
 
     consumer(queue, message) {
-
         this.publishChannel.ack(message);
 
         const emittedMessage = JSON.parse(message.content.toString());
@@ -138,22 +139,19 @@ class AmqpHelper extends EventEmitter {
             body: emittedMessage.body,
             emittedMessage
         };
+
         this.dataMessages.push(data);
         this.emit('data', data, queue);
 
         // publishChannel.cancel('sailor_nodejs');
-
         // done();
-
     }
 }
 
 function amqp() {
-
     const handle = {
         //eslint-disable-next-line no-empty-function
         getMessages() {
-
         }
     };
     return handle;
@@ -166,6 +164,9 @@ function prepareEnv() {
     env.ELASTICIO_STEP_ID = 'step_1';
     env.ELASTICIO_EXEC_ID = 'some-exec-id';
 
+    env.ELASTICIO_WORKSPACE_ID = '5559edd38968ec073600683';
+    env.ELASTICIO_CONTAINER_ID = 'dc1c8c3f-f9cb-49e1-a6b8-716af9e15948';
+
     env.ELASTICIO_USER_ID = '5559edd38968ec0736000002';
     env.ELASTICIO_COMP_ID = '5559edd38968ec0736000456';
 
@@ -177,8 +178,6 @@ function prepareEnv() {
     env.ELASTICIO_FLOW_WEBHOOK_URI = 'https://in.elastic.io/hooks/' + env.ELASTICIO_FLOW_ID;
 
     env.DEBUG = 'sailor:debug';
-
-
 }
 
 function mockApiTaskStepResponse(response) {
@@ -205,4 +204,3 @@ exports.amqp = function amqp() {
 
 exports.prepareEnv = prepareEnv;
 exports.mockApiTaskStepResponse = mockApiTaskStepResponse;
-