Release/9.1.0 performance

.circleci/config.yml

@@ -278,10 +278,10 @@ workflows:
           context: org-global
           requires:
             - setup
-            - test-unit
-            - test-coverage
-            - vulnerability-check
-            - audit-licenses
+            # - test-unit
+            # - test-coverage
+            # - vulnerability-check
+            # - audit-licenses
           filters:
             tags:
               only: /v[0-9]+(\.[0-9]+)*\-snapshot+((\.[0-9]+)?)/

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@mojaloop/central-services-stream",
-  "version": "8.8.0",
+  "version": "9.1.0-snapshot",
   "description": "Streaming library code for central services.",
   "license": "Apache-2.0",
   "bugs": {

src/kafka/consumer.js

@@ -49,6 +49,8 @@ const RdkafkaStats = require('node-rdkafka-prometheus')
 
 const Protocol = require('./protocol')
 
+const LOG_ENABLED = false
+
 /**
  * Consumer ENUMs
  *
@@ -272,7 +274,7 @@ class Consumer extends EventEmitter {
       // this._setDefaultConsumeTimeout(this._config.options.consumeTimeout)
 
       this._consumer.on('event.log', log => {
-        logger.silly(log.message)
+        if (LOG_ENABLED) logger.silly(log.message)
       })
 
       this._consumer.on('event.error', error => {
@@ -312,7 +314,7 @@ class Consumer extends EventEmitter {
 
       logger.silly('Registering data event..')
       this._consumer.on('data', message => {
-        logger.silly(`Consumer::onData() - message: ${JSON.stringify(message)}`)
+        if (LOG_ENABLED) logger.silly(`Consumer::onData() - message: ${JSON.stringify(message)}`)
         const returnMessage = { ...message }
         // let returnMessage = {}
         // Object.assign(returnMessage, message)
@@ -397,7 +399,7 @@ class Consumer extends EventEmitter {
     // setup queues to ensure sync processing of messages if options.sync is true
     if (this._config.options.sync) {
       this._syncQueue = async.queue((message, callbackDone) => {
-        logger.debug(`Consumer::consume() - Sync Process - ${JSON.stringify(message)}`)
+        if (LOG_ENABLED) logger.debug(`Consumer::consume() - Sync Process - ${JSON.stringify(message)}`)
         let payload
         if (this._config.options.mode === ENUMS.CONSUMER_MODES.flow) {
           payload = message.message
@@ -499,10 +501,12 @@ class Consumer extends EventEmitter {
             const parsedValue = Protocol.parseValue(msg.value, this._config.options.messageCharset, this._config.options.messageAsJSON)
             msg.value = parsedValue
           })
-          if (this._config.options.messageAsJSON) {
-            logger.debug(`Consumer::_consumePoller() - messages[${messages.length}]: ${JSON.stringify(messages)}}`)
-          } else {
-            logger.debug(`Consumer::_consumePoller() - messages[${messages.length}]: ${messages}}`)
+          if (LOG_ENABLED) {
+            if (this._config.options.messageAsJSON) {
+              logger.debug(`Consumer::_consumePoller() - messages[${messages.length}]: ${JSON.stringify(messages)}}`)
+            } else {
+              logger.debug(`Consumer::_consumePoller() - messages[${messages.length}]: ${messages}}`)
+            }
           }
           if (this._config.options.sync) {
             this._syncQueue.push({ error, messages }, function (err) {
@@ -512,7 +516,7 @@ class Consumer extends EventEmitter {
             })
           } else {
             Promise.resolve(workDoneCb(error, messages)).then((response) => {
-              Logger.debug(`Consumer::_consumePoller() - non-sync wokDoneCb response - ${response}`)
+              if (LOG_ENABLED) Logger.debug(`Consumer::_consumePoller() - non-sync wokDoneCb response - ${response}`)
             }).catch((err) => {
               Logger.error(`Consumer::_consumePoller() - non-sync wokDoneCb response - ${err}`)
               super.emit('error', err)
@@ -562,6 +566,7 @@ class Consumer extends EventEmitter {
         }
       } else {
         // lets transform the messages into the desired format
+        const tick = (new Date()).getTime()
         messages.map(msg => {
           const { topic } = msg
           const parsedValue = Protocol.parseValue(msg.value, this._config.options.messageCharset, this._config.options.messageAsJSON)
@@ -574,11 +579,15 @@ class Consumer extends EventEmitter {
             topics: topic
           }, msg.size)
         })
-        if (this._config.options.messageAsJSON) {
-          logger.debug(`Consumer::_consumerRecursive() - messages[${messages.length}]: ${JSON.stringify(messages)}}`)
-        } else {
-          logger.debug(`Consumer::_consumerRecursive() - messages[${messages.length}]: ${messages}}`)
+        if (LOG_ENABLED) {
+          if (this._config.options.messageAsJSON) {
+            logger.debug(`Consumer::_consumerRecursive() - messages[${messages.length}]: ${JSON.stringify(messages)}}`)
+          } else {
+            logger.debug(`Consumer::_consumerRecursive() - messages[${messages.length}]: ${messages}}`)
+          }
         }
+        const toe = (new Date()).getTime()
+        if (LOG_ENABLED) console.log('parse:', toe - tick)
 
         if (this._config.options.sync) {
           this._syncQueue.push({ error, messages }, (error) => {
@@ -595,7 +604,7 @@ class Consumer extends EventEmitter {
             ['success', 'topics']
           ).startTimer()
           Promise.resolve(workDoneCb(error, messages)).then((response) => { // histogram
-            Logger.debug(`Consumer::_consumerRecursive() - non-sync wokDoneCb response - ${response}`)
+            if (LOG_ENABLED) Logger.debug(`Consumer::_consumerRecursive() - non-sync wokDoneCb response - ${response}`)
             !!Metrics.isInitiated() && histTimerEnd({ success: true, topics: topicName })
           }).catch((err) => {
             Logger.error(`Consumer::_consumerRecursive() - non-sync wokDoneCb response - ${err}`)
@@ -632,18 +641,20 @@ class Consumer extends EventEmitter {
       } else {
         const parsedValue = Protocol.parseValue(message.value, this._config.options.messageCharset, this._config.options.messageAsJSON)
         message.value = parsedValue
-        if (this._config.options.messageAsJSON) {
-          logger.debug(`Consumer::_consumerFlow() - message: ${JSON.stringify(message)}`)
-        } else {
-          logger.debug(`Consumer::_consumerFlow() - message: ${message}`)
+        if (LOG_ENABLED) {
+          if (this._config.options.messageAsJSON) {
+            logger.debug(`Consumer::_consumerFlow() - message: ${JSON.stringify(message)}`)
+          } else {
+            logger.debug(`Consumer::_consumerFlow() - message: ${message}`)
+          }
         }
         if (this._config.options.sync) {
           this._syncQueue.push({ error, message }, function (err) {
             if (err) { logger.error(err) }
           })
         } else {
           Promise.resolve(workDoneCb(error, message)).then((response) => {
-            Logger.debug(`Consumer::_consumerFlow() - non-sync wokDoneCb response - ${response}`)
+            if (LOG_ENABLED) Logger.debug(`Consumer::_consumerFlow() - non-sync wokDoneCb response - ${response}`)
           }).catch((err) => {
             Logger.error(`Consumer::_consumerFlow() - non-sync wokDoneCb response - ${err}`)
             super.emit('error', err)
@@ -711,9 +722,9 @@ class Consumer extends EventEmitter {
       ['success', 'topics']
     ).startTimer()
     const { logger } = this._config
-    logger.silly('Consumer::commitMessage() - start')
+    if (LOG_ENABLED) logger.silly('Consumer::commitMessage() - start')
     this._consumer.commitMessage(msg)
-    logger.silly('Consumer::commitMessage() - end')
+    if (LOG_ENABLED) logger.silly('Consumer::commitMessage() - end')
     !!Metrics.isInitiated() && histTimerEnd({ success: true, topics: this._topicsStr })
   }
 
@@ -729,9 +740,9 @@ class Consumer extends EventEmitter {
       ['success', 'topics']
     ).startTimer()
     const { logger } = this._config
-    logger.silly('Consumer::commitSync() - start')
+    if (LOG_ENABLED) logger.silly('Consumer::commitSync() - start')
     this._consumer.commitSync(topicPartitions)
-    logger.silly('Consumer::commitSync() - end')
+    if (LOG_ENABLED) logger.silly('Consumer::commitSync() - end')
     !!Metrics.isInitiated() && histTimerEnd({ success: true })
   }
 
@@ -747,9 +758,9 @@ class Consumer extends EventEmitter {
       ['success', 'topics']
     ).startTimer()
     const { logger } = this._config
-    logger.silly('Consumer::commitMessageSync() - start')
+    if (LOG_ENABLED) logger.silly('Consumer::commitMessageSync() - start')
     this._consumer.commitMessageSync(msg)
-    logger.silly('Consumer::commitMessageSync() - end')
+    if (LOG_ENABLED) logger.silly('Consumer::commitMessageSync() - end')
     !!Metrics.isInitiated() && histTimerEnd({ success: true, topics: this._topicsStr })
   }
 

test/perf/config/default.json

@@ -8,18 +8,18 @@
             ],
             "configs": {
                 "options": {
-                    "pollIntervalMs": 100,
+                    "pollIntervalMs": 10,
                     "messageCharset": "utf8"
                 },
                 "rdkafkaConf": {
                     "metadata.broker.list": "localhost:9092",
                     "client.id": "default-client",
-                    "event_cb": true,
+                    "event_cb": false,
                     "compression.codec": "none",
                     "retry.backoff.ms": 100,
                     "message.send.max.retries": 2,
                     "socket.keepalive.enable": true,
-                    "queue.buffering.max.messages": 1000,
+                    "queue.buffering.max.messages": 20000,
                     "queue.buffering.max.ms": 50,
                     "batch.num.messages": 100,
                     "api.version.request": true,
@@ -41,9 +41,8 @@
             "configs": {
                 "options": {
                     "mode": 2,
-                    "batchSize": 1,
+                    "batchSize": 100,
                     "recursiveTimeout": 100,
-                    "pollFrequency": 0,
                     "messageCharset": "utf8",
                     "messageAsJSON": true,
                     "sync": true,

test/perf/src/index.js

@@ -54,7 +54,7 @@ Program.command('produce') // sub-command name, coffeeType = type, required
   .description('Start Producer') // command description
   .option('--topic <name>', 'Kafka topic')
   .option('--maxMessages <num>', 'Max number of messages to be sent', 1)
-  .option('--messageSize <size>', 'Size of each message to be sent in bytes')
+  .option('--payloadSize <size>', 'Size of payload to be sent in bytes')
   // .option('--batchSize', 'Start the Prepare Handler')
   .option('--api', 'Enable API')
 
@@ -63,7 +63,7 @@ Program.command('produce') // sub-command name, coffeeType = type, required
     // Logger.info(`Program.command('produce').args=${Flatted.stringify(args)}`)
     let topic
     let maxMessages
-    let messageSize
+    let payloadSize
 
     if (args.topic) {
       Logger.info(`Program.command('produce').args.topic=${args.topic}`)
@@ -79,18 +79,18 @@ Program.command('produce') // sub-command name, coffeeType = type, required
       }
     }
 
-    if (args.messageSize) {
-      Logger.info(`Program.command('produce').args.messageSize=${args.messageSize}`)
+    if (args.payloadSize) {
+      Logger.info(`Program.command('produce').args.payloadSize=${args.payloadSize}`)
       try {
-        messageSize = parseInt(args.messageSize)
+        payloadSize = parseInt(args.payloadSize)
       } catch (err) {
         Logger.error(err)
       }
     }
 
     try {
       await Setup(Config.PRODUCER.HOSTNAME, Config.PRODUCER.PORT, !args.api)
-      await KafkaProducer.run(maxMessages, messageSize, topic)
+      await KafkaProducer.run(maxMessages, payloadSize, topic)
     } catch (err) {
       Logger.error(err)
     }

test/perf/src/kafka/producer.js

@@ -37,10 +37,10 @@ const Config = require('@local/config')
 const Logger = require('@mojaloop/central-services-logger')
 const Faker = require('faker')
 
-const runProducer = async (messageNum = 1, messageSize = 10, topicName) => {
+const runProducer = async (messageNum = 1, payloadSize = 10, topicName) => {
   const batchId = uuidv4()
   const batchStart = (new Date()).getTime()
-  Logger.info(`[cid=${batchId}, messageNum=${messageNum}, messageSize=${messageSize}, topicName=${topicName}] ~ Producer::perf::runProducer - START`)
+  Logger.info(`[cid=${batchId}, messageNum=${messageNum}, payloadSize=${payloadSize}, topicName=${topicName}] ~ Producer::perf::runProducer - START`)
   const config = Config.PRODUCER.KAFKA.configs
 
   var topicConf = {
@@ -50,12 +50,12 @@ const runProducer = async (messageNum = 1, messageSize = 10, topicName) => {
   var producerClient = await KafkaHelper.createProducer(config)
 
   if (messageNum > 1) {
-    Logger.info(`[cid=${batchId}, messageNum=${messageNum}, messageSize=${messageSize}, topicName=${topicName}] ~ Producer::perf::runProducer -  Sending ${messageNum} messages for batch '${batchId}'`)
+    Logger.info(`[cid=${batchId}, messageNum=${messageNum}, payloadSize=${payloadSize}, topicName=${topicName}] ~ Producer::perf::runProducer -  Sending ${messageNum} messages for batch '${batchId}'`)
   } else {
-    Logger.info(`[cid=${batchId}, messageNum=${messageNum}, messageSize=${messageSize}, topicName=${topicName}] ~ Producer::perf::runProducer - Sending a single messages`)
+    Logger.info(`[cid=${batchId}, messageNum=${messageNum}, payloadSize=${payloadSize}, topicName=${topicName}] ~ Producer::perf::runProducer - Sending a single messages`)
   }
 
-  const randomPayload = Faker.random.alphaNumeric(messageSize)
+  const randomPayload = Faker.random.alphaNumeric(payloadSize)
 
   for (var i = 0; i < messageNum; i++) {
     var messageValues = {
@@ -67,20 +67,34 @@ const runProducer = async (messageNum = 1, messageSize = 10, topicName) => {
     }
     var message
     var result
+    // const renderedMessage = {
+    //   "content": {
+    //     "id": uuidv4(),
+    //     "batchId": uuidv4(),
+    //     "payload": randomPayload,
+    //     "metrics": {
+    //       "start": `${(new Date()).getTime()}`,
+    //       "batchStart": `${batchStart}`
+    //     }
+    //   },
+    //   "type": "application/json"
+    // }
     try {
-      message = JSON.parse(Mustache.render(Config.TEMPLATES.messages[0], messageValues))
-      Logger.debug(`[cid=${batchId}, messageNum=${messageNum}, messageSize=${messageSize}, topicName=${topicName}, tid=${messageValues.id}] ~ Producer::perf::runProducer - Sending payload [${i + 1}] - ${JSON.stringify(message)}`)
+      const renderedMessage = Mustache.render(Config.TEMPLATES.messages[0], messageValues)
+      message = JSON.parse(renderedMessage)
+      // message = renderedMessage
+      Logger.debug(`[cid=${batchId}, messageNum=${messageNum}, payloadSize=${payloadSize}, messageSize=${Buffer.byteLength(renderedMessage, 'utf8')}, topicName=${topicName}, tid=${messageValues.id}] ~ Producer::perf::runProducer - Sending payload [${i + 1}] - ${JSON.stringify(message)}`)
       result = await producerClient.sendMessage(message, topicConf)
-      Logger.info(`[cid=${batchId}, messageNum=${messageNum}, messageSize=${messageSize}, topicName=${topicName}, tid=${messageValues.id}] ~ Producer::perf::runProducer - Message[${i + 1}] sent with result: ${result}`)
+      Logger.info(`[cid=${batchId}, messageNum=${messageNum}, payloadSize=${payloadSize}, messageSize=${Buffer.byteLength(renderedMessage, 'utf8')}, topicName=${topicName}, tid=${messageValues.id}] ~ Producer::perf::runProducer - Message[${i + 1}] sent with result: ${result}`)
     } catch (err) {
-      Logger.info(`[cid=${batchId}, messageNum=${messageNum}, messageSize=${messageSize}, topicName=${topicName}, tid=${messageValues.id}] ~ Producer::perf::runProducer - Message[${i + 1}] sent with error: ${result}`)
+      Logger.info(`[cid=${batchId}, messageNum=${messageNum}, payloadSize=${payloadSize}, topicName=${topicName}, tid=${messageValues.id}] ~ Producer::perf::runProducer - Message[${i + 1}] sent with error: ${result}`)
       Logger.error(err)
     }
   }
 
-  Logger.info(`[cid=${batchId}, messageNum=${messageNum}, messageSize=${messageSize}, topicName=${topicName}] ~ Producer::perf::runProducer - Producer for '${topicName}' Disconnecting`)
+  Logger.info(`[cid=${batchId}, messageNum=${messageNum}, payloadSize=${payloadSize}, topicName=${topicName}] ~ Producer::perf::runProducer - Producer for '${topicName}' Disconnecting`)
   producerClient.disconnect()
-  Logger.info(`[cid=${batchId}, messageNum=${messageNum}, messageSize=${messageSize}, topicName=${topicName}] ~ Producer::perf::runProducer - END`)
+  Logger.info(`[cid=${batchId}, messageNum=${messageNum}, payloadSize=${payloadSize}, topicName=${topicName}] ~ Producer::perf::runProducer - END`)
 }
 // Logger.debug(`process.argv=${process.argv}`)
 // if(process.argv.length == 3 && !isNaN(process.argv[2])){