initial commit

Author: root

.gitignore

@@ -0,0 +1,5 @@
+.DS_Store
+node_modules
+*.log
+env_producer.sh
+env_consumer.sh

package.json

@@ -0,0 +1,30 @@
+{
+  "name": "postgres_informix_updater",
+  "version": "1.0.0",
+  "description": "Listens to DML triggers from the postgres database and updates the corresponding informix database",
+  "scripts": {
+    "test": "mocha ./test/*.test.js --exit",
+    "lint": "standard --env mocha",
+    "lint:fix": "standard --env mocha --fix",
+    "producer": "node ./src/producer.js",
+    "consumer": "node ./src/consumer.js",
+    "start": "npm run producer & npm run consumer"
+  },
+  "author": "Topcoder",
+  "license": "ISC",
+  "dependencies": {
+    "config": "^3.2.2",
+    "informix-wrapper": "git+https://github.com/appirio-tech/informix-wrapper.git",
+    "no-kafka": "^3.4.3",
+    "pg": "^7.12.1",
+    "standard": "^13.1.0",
+    "underscore": "^1.9.1",
+    "winston": "^3.2.1",
+    "lodash": "^4.17.15",
+    "topcoder-bus-api-wrapper": "^1.0.1"
+  },
+  "devDependencies": {
+    "chai": "^4.2.0",
+    "mocha": "^6.2.0"
+  }
+}

src/common/informixWrapper.js

@@ -0,0 +1,92 @@
+/**
+ * Provides services to execute queries on Informix database
+ */
+const config = require('config')
+const Wrapper = require('informix-wrapper')
+const logger = require('./logger')
+
+const informixOptions = config.get('INFORMIX')
+const writeOperationRegex = new RegExp('^(insert|update|delete|create)', 'i') // Regex to check if sql query is a write operation
+
+/**
+ * Creates a connection to Informix
+ * @param {String} database Name of database to connect to
+ * @param {Boolean} isWrite Whether the query is a write operation or not
+ * @param {Function} reject The callback function called in case of errors
+ * @returns {Object} The connection object
+ */
+function createConnection (database, isWrite, reject) {
+  const jdbc = new Wrapper({ ...informixOptions, ...{ database } })
+  jdbc.on('error', (err) => {
+    if (isWrite) {
+      jdbc.endTransaction(err, (err) => {
+        jdbc.disconnect()
+        return reject(err)
+      })
+    } else {
+      jdbc.disconnect()
+      return reject(err)
+    }
+  })
+  return jdbc.initialize()
+}
+
+/**
+ * Executes a sql query on the given database
+ * @param {String} database Name of database to query on
+ * @param {String} sql The sql query
+ * @param {Object} [params] Optional configuration options
+ * @returns {Promise<Object|Error>} Returns the result of the query on success or the error object on failure
+ */
+function executeQuery (database, sql, params) {
+  return new Promise((resolve, reject) => {
+    const isWrite = writeOperationRegex.test(sql.trim())
+    const connection = createConnection(database, isWrite, reject)
+    connection.connect((err) => {
+      if (err) {
+        connection.disconnect()
+        return reject(err)
+      }
+
+      if (isWrite) { // Write operations are executed inside a transaction
+        connection.beginTransaction(() => {
+          connection.query(sql, (err, res) => {
+            if (err) {
+              return reject(err)
+            }
+            resolve(res)
+          }, {
+            start: (q) => {
+              logger.debug(`Starting to execute ${q}`)
+            },
+            finish: (f) => {
+              connection.endTransaction(null, () => {
+                connection.disconnect()
+                logger.debug(`Finished executing`)
+              })
+            }
+          }).execute(params)
+        })
+      } else {
+        connection.query(sql, (err, res) => {
+          if (err) {
+            return reject(err)
+          }
+          resolve(res)
+        }, {
+          start: (q) => {
+            logger.debug(`Starting to execute ${q}`)
+          },
+          finish: (f) => {
+            connection.disconnect()
+            logger.debug(`Finished executing`)
+          }
+        }).execute(params)
+      }
+    })
+  })
+}
+
+module.exports = {
+  executeQuery
+}

src/common/logger.js

@@ -0,0 +1,36 @@
+/**
+ * This module defines a winston logger instance for the application.
+ */
+const { createLogger, format, transports } = require('winston')
+
+const config = require('config')
+
+const logger = createLogger({
+  format: format.combine(
+    format.json(),
+    format.colorize(),
+    format.printf((data) => `${new Date().toISOString()} - ${data.level}: ${JSON.stringify(data.message, null, 4)}`)
+  ),
+  transports: [
+    new transports.Console({ // Log to console
+      stderrLevels: ['error'],
+      level: config.get('LOG_LEVEL')
+    }),
+    new transports.File({ // Log to file
+      filename: config.get('LOG_FILE'),
+      level: config.get('LOG_LEVEL')
+    })
+  ]
+})
+
+/**
+ * Logs complete error message with stack trace if present
+ */
+logger.logFullError = (err) => {
+  if (err && err.stack) {
+    logger.error(err.stack)
+  } else {
+    logger.error(JSON.stringify(err))
+  }
+}
+module.exports = logger

src/consumer.js

@@ -0,0 +1,101 @@
+/**
+ * Receives postgres DML triggger messages from kafka and updates the informix database
+ */
+const config = require('config')
+const Kafka = require('no-kafka')
+const logger = require('./common/logger')
+const informix = require('./common/informixWrapper.js')
+
+const kafkaOptions = config.get('KAFKA')
+const isSslEnabled = kafkaOptions.SSL && kafkaOptions.SSL.cert && kafkaOptions.SSL.key
+const consumer = new Kafka.SimpleConsumer({
+  connectionString: kafkaOptions.brokers_url,
+  ...(isSslEnabled && { // Include ssl options if present
+    ssl: {
+      cert: kafkaOptions.SSL.cert,
+      key: kafkaOptions.SSL.key
+    }
+  })
+})
+
+const terminate = () => process.exit()
+
+/**
+ * Updates informix database with insert/update/delete operation
+ * @param {Object} payload The DML trigger data
+ */
+async function updateInformix (payload) {
+  logger.debug('Starting to update informix with data:')
+  logger.debug(payload)
+  //const operation = payload.operation.toLowerCase()
+  const operation = payload.payload.operation.toLowerCase()
+  console.log("level producer1 ",operation)
+	let sql = null
+
+        const columns = payload.payload.data
+        const primaryKey = payload.payload.Uniquecolumn
+  // Build SQL query
+  switch (operation) {
+    case 'insert':
+      {
+        const columnNames = Object.keys(columns)
+        sql = `insert into ${payload.payload.schema}:${payload.payload.table} (${columnNames.join(', ')}) values (${columnNames.map((k) => `'${columns[k]}'`).join(', ')});` // "insert into <schema>:<table> (col_1, col_2, ...) values (val_1, val_2, ...)"
+      }
+      break
+    case 'update':
+      {
+	  sql = `update ${payload.payload.schema}:${payload.payload.table} set ${Object.keys(columns).map((key) => `${key}='${columns[key]}'`).join(', ')} where ${primaryKey}=${columns[primaryKey]};` // "update <schema>:<table> set col_1=val_1, col_2=val_2, ... where primary_key_col=primary_key_val"
+      }
+      break
+    case 'delete':
+      {
+       // const columns = payload.data
+        //const primaryKey = payload.Uniquecolumn
+        sql = `delete from ${payload.payload.schema}:${payload.payload.table} where ${primaryKey}=${columns[primaryKey]};` // ""delete from <schema>:<table> where primary_key_col=primary_key_val"
+      }
+      break
+    default:
+      throw new Error(`Operation ${operation} is not supported`)
+  }
+
+  const result = await informix.executeQuery(payload.payload.schema, sql, null)
+  return result
+}
+
+/**
+ *
+ * @param {Array} messageSet List of messages from kafka
+ * @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) {
+  for (const m of messageSet) { // Process messages sequentially
+    try {
+      const payload = JSON.parse(m.message.value)
+      logger.debug('Received payload from kafka:')
+      logger.debug(payload)
+      await updateInformix(payload)
+      await consumer.commitOffset({ topic, partition, offset: m.offset }) // Commit offset only on success
+    } catch (err) {
+      logger.error('Could not process kafka message')
+      logger.logFullError(err)
+    }
+  }
+}
+
+/**
+ * Initialize kafka consumer
+ */
+async function setupKafkaConsumer () {
+  try {
+    await consumer.init()
+    await consumer.subscribe(kafkaOptions.topic, kafkaOptions.partition, { time: Kafka.LATEST_OFFSET }, dataHandler)
+    logger.info('Initialized kafka consumer')
+  } catch (err) {
+    logger.error('Could not setup kafka consumer')
+    logger.logFullError(err)
+    terminate()
+  }
+}
+
+setupKafkaConsumer()

src/producer.js

@@ -0,0 +1,74 @@
+/**
+ * Listens to DML trigger notifications from postgres and pushes the trigger data into kafka
+ */
+const config = require('config')
+const pg = require('pg')
+const logger = require('./common/logger')
+
+const busApi = require('topcoder-bus-api-wrapper')
+const _ = require('lodash')
+
+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 busApiClient = busApi(_.pick(config,
+        ['AUTH0_URL', 'AUTH0_AUDIENCE', 'TOKEN_CACHE_TIME',
+                'AUTH0_CLIENT_ID', 'AUTH0_CLIENT_SECRET', 'BUSAPI_URL',
+                'KAFKA_ERROR_TOPIC', 'AUTH0_PROXY_SERVER_URL']))
+
+busApiClient
+        .getHealth()
+        .then(result => console.log(result.body, result.status))
+        .catch(err => console.log(err))
+
+async function postTopic(payload) {
+        try {
+                await busApiClient.postEvent(payload)
+        } catch (e) {
+                console.log(e)
+        }
+}
+
+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)
+	await postTopic(payload)
+        } else {
+          logger.debug('Ignoring message with incorrect topic or originator')
+        }
+      } catch (err) {
+        logger.error('Could not parse message payload')
+        logger.logFullError(err)
+      }
+    })
+    logger.info('Listening to notifications')
+  } catch (err) {
+    logger.error('Could not setup postgres client')
+    logger.logFullError(err)
+    terminate()
+  }
+}
+
+async function run () {
+  await setupPgClient()
+}
+
+run()