Adding initial code for feature "GelfKafkaHandler"

.travis.yml

@@ -9,7 +9,7 @@ python:
   - "pypy"
 install:
   - pip install -e .
-  - pip install requests pytest-cov coveralls
+  - pip install requests pytest-cov coveralls kafka
 before_script:
   - docker-compose -f tests/config/docker-compose.yml up -d
   - sleep 40

README.rst

@@ -25,7 +25,7 @@ Usage
 
 .. code:: python
 
-    from pygelf import GelfTcpHandler, GelfUdpHandler, GelfTlsHandler, GelfHttpHandler
+    from pygelf import GelfTcpHandler, GelfUdpHandler, GelfTlsHandler, GelfHttpHandler, GelfKafkaHandler
     import logging
 
 
@@ -35,6 +35,8 @@ Usage
     logger.addHandler(GelfUdpHandler(host='127.0.0.1', port=9402))
     logger.addHandler(GelfTlsHandler(host='127.0.0.1', port=9403))
     logger.addHandler(GelfHttpHandler(host='127.0.0.1', port=9404))
+    logger.addHandler(GelfKafkaHandler(bootstrap_servers=["127.0.0.1"],
+                                       topic="tpc_gelf"))
 
     logger.info('hello gelf')
 
@@ -102,6 +104,11 @@ HTTP:
 - **path** ('/gelf' by default) - path of the HTTP input (http://docs.graylog.org/en/latest/pages/sending_data.html#gelf-via-http)
 - **timeout** (5 by default) - amount of seconds that HTTP client should wait before it discards the request if the server doesn't respond
 
+Kafka:
+
+- **bootstrap_servers** - addresses of the kafka brokers
+- **topic** - name of the topic in the kafka cluster for GELF messages
+
 Static fields
 =============
 

pygelf/__init__.py

@@ -1 +1 @@
-from .handlers import GelfTcpHandler, GelfUdpHandler, GelfTlsHandler, GelfHttpHandler
+from .handlers import GelfTcpHandler, GelfUdpHandler, GelfTlsHandler, GelfHttpHandler, GelfKafkaHandler

pygelf/handlers.py

@@ -8,6 +8,9 @@
 
 from logging.handlers import SocketHandler, DatagramHandler
 from logging import Handler as LoggingHandler
+
+from kafka import KafkaProducer
+
 from pygelf import gelf
 
 
@@ -159,3 +162,24 @@ def emit(self, record):
         data = self.convert_record_to_gelf(record)
         connection = httplib.HTTPConnection(host=self.host, port=self.port, timeout=self.timeout)
         connection.request('POST', self.path, data, self.headers)
+
+
+class GelfKafkaHandler(BaseHandler, LoggingHandler):
+    """
+    Logging Handler that transforms each record into GELF and sends it over an Kafka-Message Bus
+
+    :param bootstrap_servers: list of kafka brokers
+    :param topic: destination topic for the gelf messages
+
+    """
+
+    def __init__(self, bootstrap_servers, topic, **kwargs):
+        LoggingHandler.__init__(self)
+        BaseHandler.__init__(self, compress=False, **kwargs)
+
+        self.topic = topic
+        self.producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
+
+    def emit(self, record):
+        self.producer.send(self.topic,
+                           self.convert_record_to_gelf(record))