Added Kafka transport

napalm_logs/config/__init__.py

@@ -21,6 +21,7 @@
 LOG_FORMAT = '%(asctime)s,%(msecs)03.0f [%(name)-17s][%(levelname)-8s] %(message)s'
 LOG_FILE = os.path.join(ROOT_DIR, 'var', 'log', 'napalm', 'logs')
 LOG_FILE_CLI_OPTIONS = ('cli', 'screen')
+KAFKA_TOPIC = "napalm-logs"
 
 LOGGING_LEVEL = {
     'debug': logging.DEBUG,

napalm_logs/transport/__init__.py

@@ -8,6 +8,8 @@
 # Import napalm-logs pkgs
 from napalm_logs.transport.base import TransportBase
 from napalm_logs.transport.zeromq import ZMQTransport
+from napalm_logs.transport.kafka import KafkaTransport
+from napalm_logs.transport.kafka import HAS_KAFKA
 from napalm_logs.transport.cli import CLITransport
 from napalm_logs.transport.log import LogTransport
 # from napalm_logs.transport.kafka import KafkaTransport
@@ -20,12 +22,13 @@
     'print': CLITransport,
     'console': CLITransport,
     'log': LogTransport,
-    # 'kafka': KafkaTransport,
     # 'rmq': RabbitMQransport,
     # 'rabbitmq': RabbitMQransport,
     '*': ZMQTransport  # default transport
 }
 
+if HAS_KAFKA:
+    TRANSPORT_LOOKUP['kafka'] = KafkaTransport
 
 def get_transport(name):
     '''

napalm_logs/transport/kafka.py

@@ -0,0 +1,47 @@
+# -*- coding: utf-8 -*-
+'''
+Kafka transport for napalm-logs.
+'''
+from __future__ import absolute_import
+from __future__ import unicode_literals
+
+# Import stdlib
+import json
+import logging
+
+# Import third party libs
+try:
+    import kafka
+    HAS_KAFKA = True
+except ImportError as err:
+    HAS_KAFKA = False
+
+# Import napalm-logs pkgs
+from napalm_logs.exceptions import NapalmLogsException
+from napalm_logs.transport.base import TransportBase
+from napalm_logs.config import KAFKA_TOPIC
+
+log = logging.getLogger(__name__)
+
+
+class KafkaTransport(TransportBase):
+    '''
+    Kafka transport class.
+    '''
+    def __init__(self, addr, port):
+        self.bootstrap_servers = '{addr}:{port}'.format(addr=addr, port=port)
+        self.topic = KAFKA_TOPIC
+
+    def start(self):
+        try:
+            self.producer = kafka.KafkaProducer(bootstrap_servers=self.bootstrap_servers)
+        except kafka.errors.NoBrokersAvailable as err:
+            log.error(err, exc_info=True)
+            raise NapalmLogsException(err)
+
+    def publish(self, obj):
+        self.producer.send(self.topic, obj)
+
+    def stop(self):
+        if hasattr(self, 'producer'):
+            self.producer.close()