added queue settings to js-executor

msa/js-executor/config/custom-environment-variables.yml

@@ -17,30 +17,34 @@
 service-type: "TB_SERVICE_TYPE"
 request_topic: "REMOTE_JS_EVAL_REQUEST_TOPIC"
 
+js:
+  response_poll_interval: "REMOTE_JS_RESPONSE_POLL_INTERVAL_MS"
+
 kafka:
   bootstrap:
     # Kafka Bootstrap Servers
     servers: "TB_KAFKA_SERVERS"
+  replication_factor: "TB_QUEUE_KAFKA_REPLICATION_FACTOR"
+  topic-properties: "TB_QUEUE_KAFKA_JE_TOPIC_PROPERTIES"
 
 pubsub:
   project_id: "TB_QUEUE_PUBSUB_PROJECT_ID"
   service_account: "TB_QUEUE_PUBSUB_SERVICE_ACCOUNT"
+  queue-properties: "TB_QUEUE_PUBSUB_JE_QUEUE_PROPERTIES"
 
 aws_sqs:
   access_key_id: "TB_QUEUE_AWS_SQS_ACCESS_KEY_ID"
   secret_access_key: "TB_QUEUE_AWS_SQS_SECRET_ACCESS_KEY"
   region: "TB_QUEUE_AWS_SQS_REGION"
+  queue-properties: "TB_QUEUE_AWS_SQS_JE_QUEUE_PROPERTIES"
 
 rabbitmq:
-  exchange_name: "TB_QUEUE_RABBIT_MQ_EXCHANGE_NAME"
   host: "TB_QUEUE_RABBIT_MQ_HOST"
   port: "TB_QUEUE_RABBIT_MQ_PORT"
   virtual_host: "TB_QUEUE_RABBIT_MQ_VIRTUAL_HOST"
   username: "TB_QUEUE_RABBIT_MQ_USERNAME"
   password: "TB_QUEUE_RABBIT_MQ_PASSWORD"
-  automatic_recovery_enabled: "TB_QUEUE_RABBIT_MQ_AUTOMATIC_RECOVERY_ENABLED"
+  queue-properties: "TB_QUEUE_RABBIT_MQ_JE_QUEUE_PROPERTIES"
-  connection_timeout: "TB_QUEUE_RABBIT_MQ_CONNECTION_TIMEOUT"
-  handshake_timeout: "TB_QUEUE_RABBIT_MQ_HANDSHAKE_TIMEOUT"
 
 logger:
   level: "LOGGER_LEVEL"

msa/js-executor/config/default.yml

@@ -15,23 +15,31 @@
 #
 
 service-type: "kafka"
-request_topic: "js.eval.requests"
+request_topic: "js_eval.requests"
+
+js:
+  response_poll_interval: "25"
 
 kafka:
   bootstrap:
     # Kafka Bootstrap Servers
     servers: "localhost:9092"
+  replication_factor: "1"
+  topic-properties: "retention.ms:604800000;segment.bytes:26214400;retention.bytes:104857600"
+
+pubsub:
+  queue-properties: "ackDeadlineInSec:30;messageRetentionInSec:604800"
+
+aws_sqs:
+  queue-properties: "VisibilityTimeout:30;MaximumMessageSize:262144;MessageRetentionPeriod:604800"
 
 rabbitmq:
-  exchange_name: ""
   host: "localhost"
   port: "5672"
   virtual_host: "/"
-  username: "YOUR_USERNAME"
+  username: "admin"
-  password: "YOUR_PASSWORD"
+  password: "password"
-  automatic_recovery_enabled: "false"
+  queue-properties: "x-max-length-bytes:1048576000;x-message-ttl:604800000"
-  connection_timeout: "60000"
-  handshake_timeout: "10000"
 
 
 logger:

msa/js-executor/package.json

@@ -14,7 +14,7 @@
   "dependencies": {
     "config": "^3.2.2",
     "js-yaml": "^3.12.0",
-    "kafkajs": "^1.11.0",
+    "kafkajs": "^1.12.0",
     "@google-cloud/pubsub": "^1.7.1",
     "aws-sdk": "^2.663.0",
     "amqplib": "^0.5.5",

msa/js-executor/queue/awsSqsTemplate.js

@@ -26,10 +26,13 @@ const accessKeyId = config.get('aws_sqs.access_key_id');
 const secretAccessKey = config.get('aws_sqs.secret_access_key');
 const region = config.get('aws_sqs.region');
 const AWS = require('aws-sdk');
+const queueProperties = config.get('aws_sqs.queue-properties');
+const poolInterval = config.get('js.response_poll_interval');
 
+let queueAttributes = {FifoQueue: 'true', ContentBasedDeduplication: 'true'};
 let sqsClient;
-let queueURL;
+let requestQueueURL;
-let responseTopics = new Map();
+let queueUrls = new Map();
 let stopped = false;
 
 function AwsSqsProducer() {
@@ -41,11 +44,11 @@ function AwsSqsProducer() {
                 headers: headers
             });
 
-        let responseQueueUrl = responseTopics.get(responseTopic);
+        let responseQueueUrl = queueUrls.get(topicToSqsQueueName(responseTopic));
 
         if (!responseQueueUrl) {
             responseQueueUrl = await createQueue(responseTopic);
-            responseTopics.set(responseTopic, responseQueueUrl);
+            queueUrls.set(responseTopic, responseQueueUrl);
         }
 
         let params = {MessageBody: msgBody, QueueUrl: responseQueueUrl, MessageGroupId: scriptId};
@@ -69,13 +72,27 @@ function AwsSqsProducer() {
 
         sqsClient = new AWS.SQS({apiVersion: '2012-11-05'});
 
-        queueURL = await createQueue(requestTopic);
+        const queues = await getQueues();
+
+        queues.forEach(queueUrl => {
+            const delimiterPosition = queueUrl.lastIndexOf('/');
+            const queueName = queueUrl.substring(delimiterPosition + 1);
+            queueUrls.set(queueName, queueUrl);
+        })
+
+        parseQueueProperties();
+
+        requestQueueURL = queueUrls.get(topicToSqsQueueName(requestTopic));
+        if (!requestQueueURL) {
+            requestQueueURL = await createQueue(requestTopic);
+        }
+
         const messageProcessor = new JsInvokeMessageProcessor(new AwsSqsProducer());
 
         const params = {
             MaxNumberOfMessages: 10,
-            QueueUrl: queueURL,
+            QueueUrl: requestQueueURL,
-            WaitTimeSeconds: 0.025
+            WaitTimeSeconds: poolInterval / 1000
         };
         while (!stopped) {
             const messages = await new Promise((resolve, reject) => {
@@ -100,7 +117,7 @@ function AwsSqsProducer() {
                 });
 
                 const deleteBatch = {
-                    QueueUrl: queueURL,
+                    QueueUrl: requestQueueURL,
                     Entries: entries
                 };
                 sqsClient.deleteMessageBatch(deleteBatch, function (err, data) {
@@ -120,14 +137,9 @@ function AwsSqsProducer() {
 })();
 
 function createQueue(topic) {
-    let queueName = topic.replace(/\./g, '_') + '.fifo';
+    let queueName = topicToSqsQueueName(topic);
-    let queueParams = {
+    let queueParams = {QueueName: queueName, Attributes: queueAttributes};
-        QueueName: queueName, Attributes: {
-            FifoQueue: 'true',
-            ContentBasedDeduplication: 'true'
 
-        }
-    };
     return new Promise((resolve, reject) => {
         sqsClient.createQueue(queueParams, function (err, data) {
             if (err) {
@@ -139,6 +151,30 @@ function createQueue(topic) {
     });
 }
 
+function getQueues() {
+    return new Promise((resolve, reject) => {
+        sqsClient.listQueues(function (err, data) {
+            if (err) {
+                reject(err);
+            } else {
+                resolve(data.QueueUrls);
+            }
+        });
+    });
+}
+
+function topicToSqsQueueName(topic) {
+    return topic.replace(/\./g, '_') + '.fifo';
+}
+
+function parseQueueProperties() {
+    const props = queueProperties.split(';');
+    props.forEach(p => {
+        const delimiterPosition = p.indexOf(':');
+        queueAttributes[p.substring(0, delimiterPosition)] = p.substring(delimiterPosition + 1);
+    });
+}
+
 process.on('exit', () => {
     stopped = true;
     logger.info('Aws Sqs client stopped.');

msa/js-executor/queue/kafkaTemplate.js

@@ -19,13 +19,28 @@ const config = require('config'),
     JsInvokeMessageProcessor = require('../api/jsInvokeMessageProcessor'),
     logger = require('../config/logger')._logger('kafkaTemplate'),
     KafkaJsWinstonLogCreator = require('../config/logger').KafkaJsWinstonLogCreator;
+const replicationFactor = config.get('kafka.replication_factor');
+const topicProperties = config.get('kafka.topic-properties');
 
 let kafkaClient;
+let kafkaAdmin;
 let consumer;
 let producer;
 
+const topics = [];
+const configEntries = [];
+
 function KafkaProducer() {
     this.send = async (responseTopic, scriptId, rawResponse, headers) => {
+
+        if (!topics.includes(responseTopic)) {
+            let createResponseTopicResult = await createTopic(responseTopic);
+            topics.push(responseTopic);
+            if (createResponseTopicResult) {
+                logger.info('Created new topic: %s', requestTopic);
+            }
+        }
+
         let headersData = headers.data;
         headersData = Object.fromEntries(Object.entries(headersData).map(([key, value]) => [key, Buffer.from(value)]));
         return producer.send(
@@ -47,10 +62,10 @@ function KafkaProducer() {
         logger.info('Starting ThingsBoard JavaScript Executor Microservice...');
 
         const kafkaBootstrapServers = config.get('kafka.bootstrap.servers');
-        const kafkaRequestTopic = config.get('request_topic');
+        const requestTopic = config.get('request_topic');
 
         logger.info('Kafka Bootstrap Servers: %s', kafkaBootstrapServers);
-        logger.info('Kafka Requests Topic: %s', kafkaRequestTopic);
+        logger.info('Kafka Requests Topic: %s', requestTopic);
 
         kafkaClient = new Kafka({
             brokers: kafkaBootstrapServers.split(','),
@@ -58,12 +73,23 @@ function KafkaProducer() {
             logCreator: KafkaJsWinstonLogCreator
         });
 
+        parseTopicProperties();
+
+        kafkaAdmin = kafkaClient.admin();
+        await kafkaAdmin.connect();
+
+        let createRequestTopicResult = await createTopic(requestTopic);
+
+        if (createRequestTopicResult) {
+            logger.info('Created new topic: %s', requestTopic);
+        }
+
         consumer = kafkaClient.consumer({groupId: 'js-executor-group'});
         producer = kafkaClient.producer();
         const messageProcessor = new JsInvokeMessageProcessor(new KafkaProducer());
         await consumer.connect();
         await producer.connect();
-        await consumer.subscribe({topic: kafkaRequestTopic});
+        await consumer.subscribe({topic: requestTopic});
 
         logger.info('Started ThingsBoard JavaScript Executor Microservice.');
         await consumer.run({
@@ -90,12 +116,37 @@ function KafkaProducer() {
     }
 })();
 
+function createTopic(topic) {
+    return kafkaAdmin.createTopics({
+        topics: [{
+            topic: topic,
+            replicationFactor: replicationFactor,
+            configEntries: configEntries
+        }]
+    });
+}
+
+function parseTopicProperties() {
+    const props = topicProperties.split(';');
+    props.forEach(p => {
+        const delimiterPosition = p.indexOf(':');
+        configEntries.push({name: p.substring(0, delimiterPosition), value: p.substring(delimiterPosition + 1)});
+    });
+}
+
 process.on('exit', () => {
     exit(0);
 });
 
 async function exit(status) {
     logger.info('Exiting with status: %d ...', status);
+
+    if (kafkaAdmin) {
+        logger.info('Stopping Kafka Admin...');
+        await kafkaAdmin.disconnect();
+        logger.info('Kafka Admin stopped.');
+    }
+
     if (consumer) {
         logger.info('Stopping Kafka Consumer...');
         let _consumer = consumer;

msa/js-executor/queue/pubSubTemplate.js

@@ -24,11 +24,21 @@ const {PubSub} = require('@google-cloud/pubsub');
 const projectId = config.get('pubsub.project_id');
 const credentials = JSON.parse(config.get('pubsub.service_account'));
 const requestTopic = config.get('request_topic');
+const queueProperties = config.get('pubsub.queue-properties');
 
 let pubSubClient;
 
+const topics = [];
+const subscriptions = [];
+let queueProps = [];
+
 function PubSubProducer() {
     this.send = async (responseTopic, scriptId, rawResponse, headers) => {
+
+        if (!(subscriptions.includes(responseTopic) && topics.includes(requestTopic))) {
+            await createTopic(requestTopic);
+        }
+
         let data = JSON.stringify(
             {
                 key: scriptId,
@@ -45,6 +55,28 @@ function PubSubProducer() {
         logger.info('Starting ThingsBoard JavaScript Executor Microservice...');
         pubSubClient = new PubSub({projectId: projectId, credentials: credentials});
 
+        parseQueueProperties();
+
+        const topicList = await pubSubClient.getTopics();
+
+        if (topicList) {
+            topicList[0].forEach(topic => {
+                topics.push(getName(topic.name));
+            });
+        }
+
+        const subscriptionList = await pubSubClient.getSubscriptions();
+
+        if (subscriptionList) {
+            topicList[0].forEach(sub => {
+                subscriptions.push(getName(sub.name));
+            });
+        }
+
+        if (!(subscriptions.includes(requestTopic) && topics.includes(requestTopic))) {
+            await createTopic(requestTopic);
+        }
+
         const subscription = pubSubClient.subscription(requestTopic);
 
         const messageProcessor = new JsInvokeMessageProcessor(new PubSubProducer());
@@ -64,6 +96,36 @@ function PubSubProducer() {
     }
 })();
 
+async function createTopic(topic) {
+    if (!topics.includes(topic)) {
+        await pubSubClient.createTopic(topic);
+        topics.push(topic);
+        logger.info('Created new Pub/Sub topic: %s', topic);
+    }
+    await createSubscription(topic)
+}
+
+async function createSubscription(topic) {
+    if (!subscriptions.includes(topic)) {
+        await pubSubClient.topic(topic).createSubscription(topic);
+        subscriptions.push(topic);
+        logger.info('Created new Pub/Sub subscription: %s', topic);
+    }
+}
+
+function parseQueueProperties() {
+    const props = queueProperties.split(';');
+    props.forEach(p => {
+        const delimiterPosition = p.indexOf(':');
+        queueProps[p.substring(0, delimiterPosition)] = p.substring(delimiterPosition + 1);
+    });
+}
+
+function getName(fullName) {
+    const delimiterPosition = fullName.lastIndexOf('/');
+    return fullName.substring(delimiterPosition + 1);
+}
+
 process.on('exit', () => {
     exit(0);
 });

msa/js-executor/queue/rabbitmqTemplate.js

@@ -21,7 +21,17 @@ const config = require('config'),
     logger = require('../config/logger')._logger('rabbitmqTemplate');
 
 const requestTopic = config.get('request_topic');
+const host = config.get('rabbitmq.host');
+const port = config.get('rabbitmq.port');
+const vhost = config.get('rabbitmq.virtual_host');
+const username = config.get('rabbitmq.username');
+const password = config.get('rabbitmq.password');
+const queueProperties = config.get('rabbitmq.queue-properties');
+const poolInterval = config.get('js.response_poll_interval');
+
 const amqp = require('amqplib/callback_api');
+
+let queueParams = {durable: false, exclusive: false, autoDelete: false};
 let connection;
 let channel;
 let stopped = false;
@@ -58,10 +68,11 @@ function RabbitMqProducer() {
 (async () => {
     try {
         logger.info('Starting ThingsBoard JavaScript Executor Microservice...');
+        const url = `amqp://${host}:${port}${vhost}`;
 
-        amqp.credentials.amqplain('admin', 'password');
+        amqp.credentials.amqplain(username, password);
         connection = await new Promise((resolve, reject) => {
-            amqp.connect('amqp://localhost:5672/', function (err, connection) {
+            amqp.connect(url, function (err, connection) {
                 if (err) {
                     reject(err);
                 } else {
@@ -80,6 +91,8 @@ function RabbitMqProducer() {
             });
         });
 
+        parseQueueProperties();
+
         await createQueue(requestTopic);
 
         const messageProcessor = new JsInvokeMessageProcessor(new RabbitMqProducer());
@@ -98,6 +111,8 @@ function RabbitMqProducer() {
             if (message) {
                 messageProcessor.onJsInvokeMessage(message.content.toString('utf8'));
                 channel.ack(message);
+            } else {
+                await sleep(poolInterval);
             }
         }
     } catch (e) {
@@ -107,10 +122,17 @@ function RabbitMqProducer() {
     }
 })();
 
+function parseQueueProperties() {
+    const props = queueProperties.split(';');
+    props.forEach(p => {
+        const delimiterPosition = p.indexOf(':');
+        queueParams[p.substring(0, delimiterPosition)] = p.substring(delimiterPosition + 1);
+    });
+}
+
 function createQueue(topic) {
-    let params = {durable: false};
     return new Promise((resolve, reject) => {
-        channel.assertQueue(topic, params, function (err, data) {
+        channel.assertQueue(topic, queueParams, function (err, data) {
             if (err) {
                 reject(err);
             } else {
@@ -120,6 +142,12 @@ function createQueue(topic) {
     });
 }
 
+function sleep(ms) {
+    return new Promise((resolve) => {
+        setTimeout(resolve, ms);
+    });
+}
+
 process.on('exit', () => {
     exit(0);
 });
@@ -146,4 +174,4 @@ async function exit(status) {
     } else {
         process.exit(status);
     }
-}
+}