<filename>index.js
// Node package which wraps KafkaJS and facilitates communication between applications that use Kafka and KafkaMirror
// monitoring tool

const { Kafka, logLevel } = require('kafkajs');
const winston = require('winston');
const { transformLogData } = require('./utilities');

// Create websocket that will send data about the Kafka cluster that Kafka simulator is running to KafkaMirror application
// This is the socket.io "server"
const app = require('express')();
const http = require('http').createServer(app);
const io = require('socket.io')(http, {
  cors: {
    origin: 'http://localhost:8080',
    methods: ['GET', 'POST'],
  },
});

// Function that is invoked to create a connection between kafka-simulator or other Kafka application and KafkaMirror
// at the specified port
const KafkaMirror = (props, port = 3030) => {
  let socket = null;

  io.on('connection', (socketConnection) => {
    socket = socketConnection;
    console.log(`Client is connected [id=${socket.id}]`);
    socket.on('disconnect', () => {
      console.log(`Client disconnected [id=${socket.id}]`);
    });
  });

  http.listen(port, () => {
    console.log(`listening on port ${port}`);
  });

  // specify log levels for winston logger
  const toWinstonLogLevel = (level) => {
    switch (level) {
      case logLevel.ERROR:
      case logLevel.NOTHING:
        return 'error';
      case logLevel.WARN:
        return 'warn';
      case logLevel.INFO:
        return 'info';
      case logLevel.DEBUG:
        return 'debug';
    }
  };

  // Use chunking to send new data through socket io on a periodic basis every second as opposed to real-time to
  // accomodate large amounts of streaming data generated by node.js when parsing the yelp and other large datasets
  let chunk = [];
  setInterval(() => {
    io.sockets.emit('log', JSON.stringify(chunk));
    console.log('chunk length is', chunk.length);
    chunk = [];
  }, 1000);

  // Initialize and configure Winston logger to be incorporated into the establishment of connection with Kafka application
  const WinstonLogCreator = (logLevel) => {
    let size;
    let newID;
    let lastSentID;

    // Winston needs at least one transport in order to register that a log has been recieved
    const logger = winston.createLogger({
      level: toWinstonLogLevel(logLevel),
      transports: [new winston.transports.File({ filename: 'myapp.log' })],
    });

    logger.stream({ start: -1 }).on('log', function (log) {
      // When Kafka produces a log of type Request Produce we pull the size and correlation ID
      // Size of message (e.g., size of yelp review text) is not included in the corresponding Response Produce
      // KafkaMirror pulls the size included in each log message and displays in the "throughput" chart of KafkaMirror
      // correlation ID stored to prevent duplicate produce logs
      if (log.message.indexOf('Request Produce') > -1) {
        size = log.extra.size;
        newID = log.extra.correlationId;
      }

      // When Kafka produces a log of type response produce, we will pull information from the previous
      // request produce and add data to our chuck array
      if (log.message.indexOf('Response Produce') > -1) {
        const data = transformLogData(log);
        data.requestSize = size;
        if (lastSentID !== newID) {
          chunk.push(data);
          lastSentID = newID;
        }
      }
    });

    return ({ namespace, level, label, log }) => {
      const { message, ...extra } = log;
      logger.log({
        level: toWinstonLogLevel(level),
        message,
        extra,
      });
    };
  };

  // Invoked to create a new connection to a Kafka instance using KafkaJS node package
  const kafka = new Kafka({
    ...props,
    logLevel: logLevel.DEBUG,
    logCreator: WinstonLogCreator,
  });

  return kafka;
};

module.exports = KafkaMirror;
