index.js

"use strict";

var util = require('util'),
    winston = require('winston'),
    _ = require('lodash'),
    kafka = require('kafka-node');

var Producer = kafka.Producer,
    client,
    producer;

var _isConnected = false;

var KafkaLogger = function (options) {
    this.name = options.name || 'KafkaLogger';
    this.level = options.level || 'info';
    this.meta = options.meta || {};

    /*
        KAFKA Options
    */
    // Zookeeper connection string, default localhost:2181/kafka0.8
    this.connectionString = options.connectionString || 'localhost:2181';

    // This is a user supplied identifier for the client application, default kafka-node-client
    this.clientId = options.clientId;

    // Object, Zookeeper options, see node-zookeeper-client
    this.zkOptions = options.zkOptions;

    this.topic = options.topic;


    // Connect
    client = new kafka.Client(this.connectionString, this.clientId, this.zkOptions);
    producer = new Producer(client);

    producer.on('ready', function () {
        _isConnected = true;
    });

    producer.on('error', function (err) {
        _isConnected = false;
        var msg = 'winston-kafka-logger - Cannot connect to kafka server';
        // throw new Error(msg);
        console.error(msg, err);
    });
};

util.inherits(KafkaLogger, winston.Transport);

KafkaLogger.prototype.log = function (level, msg, meta, callback) {
    if (_isConnected) {
        var payload = {
            msg: msg,
            level: level,
            meta: _.defaults(meta, this.meta),
            timestamp: new Date()
        };

        var payloads = [
            { topic: this.topic, messages: [JSON.stringify(payload)] }
        ];

        try {
            producer.send(payloads, function() {
                // Ignore
            });
        }
        catch(err) {
            console.error('Failed to send log to kafka!!');
        }
    }

    callback(null, true);
};

module.exports = KafkaLogger;