kafka audit update

Author: root

.gitignore

@@ -5,3 +5,4 @@ env_producer.sh
 env_consumer.sh
 *.env
 *.sh
+*.list

src/consumer.js

@@ -7,8 +7,8 @@ const logger = require('./common/logger')
 const updateInformix = require('./services/updateInformix')
 const pushToKafka = require('./services/pushToKafka')
 const healthcheck = require('topcoder-healthcheck-dropin');
+const auditTrail = require('./services/auditTrail');
 const kafkaOptions = config.get('KAFKA')
-//const sleep = require('sleep');
 const isSslEnabled = kafkaOptions.SSL && kafkaOptions.SSL.cert && kafkaOptions.SSL.key
 const consumer = new Kafka.SimpleConsumer({
   connectionString: kafkaOptions.brokers_url,
@@ -21,17 +21,17 @@ const consumer = new Kafka.SimpleConsumer({
 })
 
 
-  const check = function () {
-    if (!consumer.client.initialBrokers && !consumer.client.initialBrokers.length) {
-      return false;
-    }
-    let connected = true;
-    consumer.client.initialBrokers.forEach(conn => {
-      logger.debug(`url ${conn.server()} - connected=${conn.connected}`);
-      connected = conn.connected & connected;
-    });
-    return connected;
-  };
+const check = function () {
+  if (!consumer.client.initialBrokers && !consumer.client.initialBrokers.length) {
+    return false;
+  }
+  let connected = true;
+  consumer.client.initialBrokers.forEach(conn => {
+    logger.debug(`url ${conn.server()} - connected=${conn.connected}`);
+    connected = conn.connected & connected;
+  });
+  return connected;
+};
 
 
 const terminate = () => process.exit()
@@ -41,39 +41,54 @@ const terminate = () => process.exit()
  * @param {String} topic The name of the message topic
  * @param {Number} partition The kafka partition to which messages are written
  */
-async function dataHandler (messageSet, topic, partition) {
+async function dataHandler(messageSet, topic, partition) {
   for (const m of messageSet) { // Process messages sequentially
+    let message
     try {
-      const payload = JSON.parse(m.message.value)
-      logger.debug('Received payload from kafka:')
-      logger.debug(payload)
-      await updateInformix(payload)
- //     await insertConsumerAudit(payload, true, undefined, false)
+      message = JSON.parse(m.message.value)
+      logger.debug('Received message from kafka:')
+      logger.debug(JSON.stringify(message))
+      await updateInformix(message)
       await consumer.commitOffset({ topic, partition, offset: m.offset }) // Commit offset only on success
+      await auditTrail([message.payload.payloadseqid,'scorecard_consumer',message.payload.table,message.payload.Uniquecolumn,
+             message.payload.operation,1,0,"",message.timestamp,new Date(),""],'consumer')
     } catch (err) {
       logger.error('Could not process kafka message')
-      logger.logFullError(err)
-      if (!payload.retryCount) {
-        payload.retryCount = 0
-      }
-      if (payload.retryCount >= config.KAFKA.maxRetry) {
-        await pushToKafka(
-          Object.assign({}, payload, { topic: config.KAFKA.errorTopic, recipients: config.KAFKA.recipients })
-        )
-        return
-      }
-      await pushToKafka(
-        Object.assign({}, payload, { retryCount: payload.retryCount + 1 })
-      )
+      //logger.logFullError(err)
+      try {
+        await consumer.commitOffset({ topic, partition, offset: m.offset }) // Commit success as will re-publish
+        logger.debug('Trying to push same message after adding retryCounter')
+        if (!message.payload.retryCount) {
+          message.payload.retryCount = 0
+          logger.debug('setting retry counter to 0 and max try count is : ', config.KAFKA.maxRetry);
+        }
+        if (message.payload.retryCount >= config.KAFKA.maxRetry) {
+          logger.debug('Recached at max retry counter, sending it to error queue: ', config.KAFKA.errorTopic);
+
+          let notifiyMessage = Object.assign({}, message, { topic: config.KAFKA.errorTopic })
+          notifiyMessage.payload['recipients'] = config.KAFKA.recipients
+          logger.debug('pushing following message on kafka error alert queue:')
+          logger.debug(notifiyMessage)
+          await pushToKafka(notifiyMessage)
+          return
+        }
+        message.payload['retryCount'] = message.payload.retryCount + 1;
+        await pushToKafka(message)
+        logger.debug('pushed same message after adding retryCount')
+      } catch (err) {
+	 //await auditTrail([payload.payload.payloadseqid,'scorecard_consumer',payload.payload.table,payload.payload.Uniquecolumn,
+           //  payload.payload.operation,0,message.payload.retryCount,"re-publish kafka err",payload.timestamp,new Date(),""],'consumer')
+        logger.error("Error occured in re-publishing kafka message", err)
       }
     }
   }
+}
 
 
 /**
  * Initialize kafka consumer
  */
-async function setupKafkaConsumer () {
+async function setupKafkaConsumer() {
   try {
     await consumer.init()
     await consumer.subscribe(kafkaOptions.topic, kafkaOptions.partition, { time: Kafka.LATEST_OFFSET }, dataHandler)
@@ -87,4 +102,3 @@ async function setupKafkaConsumer () {
 }
 
 setupKafkaConsumer()
-

src/producer.js

@@ -8,7 +8,7 @@ const pushToKafka = require('./services/pushToKafka')
 const pgOptions = config.get('POSTGRES')
 const pgConnectionString = `postgresql://${pgOptions.user}:${pgOptions.password}@${pgOptions.host}:${pgOptions.port}/${pgOptions.database}`
 const pgClient = new pg.Client(pgConnectionString)
-
+const auditTrail = require('./services/auditTrail');
 const express = require('express')
 const app = express()
 const port = 3000
@@ -17,28 +17,26 @@ const port = 3000
 async function setupPgClient () {
   try {
     await pgClient.connect()
-    // Listen to each of the trigger functions
     for (const triggerFunction of pgOptions.triggerFunctions) {
       await pgClient.query(`LISTEN ${triggerFunction}`)
     }
     pgClient.on('notification', async (message) => {
-      console.log('Received trigger payload:')
-      logger.debug(`Received trigger payload:`)
-      logger.debug(message)
-      //console.log(message)
       try {
         const payload = JSON.parse(message.payload)
-        console.log("level 0",payload);
 	const validTopicAndOriginator = (pgOptions.triggerTopics.includes(payload.topic)) && (pgOptions.triggerOriginators.includes(payload.originator)) // Check if valid topic and originator
         if (validTopicAndOriginator) {
-         // await pushToKafka(payload)
+     console.log(`${payload.topic} ${payload.payload.table} ${payload.payload.operation} ${payload.timestamp}`);
 	await pushToKafka(payload)
-        } else {
+	} else {
           logger.debug('Ignoring message with incorrect topic or originator')
         }
-      } catch (err) {
+     await auditTrail([payload.payload.payloadseqid,'scorecard_producer',1,payload.topic,payload.payload.table,payload.payload.Uniquecolumn,
+	     payload.payload.operation,"",payload.timestamp,new Date(),JSON.stringify(payload.payload)],'producer')
+      } catch (error) {
         logger.error('Could not parse message payload')
-        logger.logFullError(err)
+     await auditTrail([payload.payload.payloadseqid,'scorecard_producer',0,payload.topic,payload.payload.table,payload.payload.Uniquecolumn,
+	     payload.payload.operation,"error",payload.timestamp,new Date(),JSON.stringify(payload.payload)],'producer')
+	logger.logFullError(error)
       }
     })
     logger.info('Listening to notifications')

src/services/auditTrail.js

@@ -0,0 +1,45 @@
+const config = require('config')
+const pg = require('pg')
+const logger = require('../common/logger')
+
+const pgOptions = config.get('POSTGRES')
+const pgConnectionString = `postgresql://${pgOptions.user}:${pgOptions.password}@${pgOptions.host}:${pgOptions.port}/${pgOptions.database}`
+let pgClient2
+
+async function setupPgClient2 () {
+  pgClient2 = new pg.Client(pgConnectionString)
+  try {
+    await pgClient2.connect()
+	logger.debug('Connected to Pg Client2 Audit:')
+    }
+   catch (err) {
+    logger.error('Could not setup postgres client2')
+    logger.logFullError(err)
+    process.exit()
+  }
+}
+
+async function auditTrail (data,sourcetype) {
+if (!pgClient2) {
+	await setupPgClient2()
+}
+if (sourcetype === 'producer'){
+ sql = 'INSERT INTO tcs_catalog.producer_scorecard_audit(payloadseqid,origin_source,kafka_post_status,topic_name,table_name,Uniquecolumn,operationtype,errormessage,payloadtime,auditdatetime,payload) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)'
+logger.debug(`--Audit Trail update producer--`)
+} else {
+sql = 'INSERT INTO tcs_catalog.consumer_scorecard_audit(payloadseqid,origin_source,table_name,Uniquecolumn,operationtype,dest_db_status, dest_retry_count,errormessage,payloadtime,auditdatetime,dest_operationquery) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)'
+logger.debug(`--Audit Trail  update consumer--`)
+}
+  return pgClient2.query(sql, data, (err, res) => {
+  if (err) {
+  logger.debug(`--Audit Trail  update error-- ${err.stack}`)
+  //pgClient2.end()
+  } else {
+    logger.debug(`--Audit Trail update success-- `)
+  }
+})
+}
+
+
+module.exports = auditTrail
+

src/services/updateInformix.js

@@ -7,15 +7,10 @@ const logger = require('../common/logger')
  * @param {Object} payload The DML trigger data
  */
 async function updateInformix (payload) {
-  logger.debug('Starting to update informix with data:')
-  logger.debug(payload)
-  if (payload.payload.table === 'scorecard_question'){
-  logger.debug('inside scorecard_question')
-  sleep.sleep(2);
-  }
+  logger.debug('=====Starting to update informix with data:====')
   //const operation = payload.operation.toLowerCase()
   const operation = payload.payload.operation.toLowerCase()
-  console.log("level producer1 ",operation)
+  console.log("level 1 informix ",operation)
 	let sql = null
 
         const columns = payload.payload.data