Add support for passing librdkafka logs to the standard logging module (#148)

Author: edenhill

confluent_kafka/src/Consumer.c

@@ -1363,6 +1363,11 @@ static int Consumer_init (PyObject *selfobj, PyObject *args, PyObject *kwargs) {
                 return -1;
         }
 
+        /* Forward log messages to main queue which is then forwarded
+         * to the consumer queue */
+        if (self->logger)
+                rd_kafka_set_log_queue(self->rk, NULL);
+
         rd_kafka_poll_set_consumer(self->rk);
 
         self->u.Consumer.rkqu = rd_kafka_queue_get_consumer(self->rk);

confluent_kafka/src/Producer.c

@@ -588,6 +588,10 @@ static int Producer_init (PyObject *selfobj, PyObject *args, PyObject *kwargs) {
                 return -1;
         }
 
+        /* Forward log messages to poll queue */
+        if (self->logger)
+                rd_kafka_set_log_queue(self->rk, NULL);
+
         return 0;
 }
 

confluent_kafka/src/confluent_kafka.c

@@ -1076,6 +1076,39 @@ static int stats_cb(rd_kafka_t *rk, char *json, size_t json_len, void *opaque) {
 	return 0;
 }
 
+static void log_cb (const rd_kafka_t *rk, int level,
+                    const char *fac, const char *buf) {
+        Handle *h = rd_kafka_opaque(rk);
+        PyObject *result;
+        CallState *cs;
+        static const int level_map[8] = {
+                /* Map syslog levels to python logging levels */
+                [0] = 50, /* LOG_EMERG   -> logging.CRITICAL */
+                [1] = 50, /* LOG_ALERT   -> logging.CRITICAL */
+                [2] = 50, /* LOG_CRIT    -> logging.CRITICAL */
+                [3] = 40, /* LOG_ERR     -> logging.ERROR */
+                [4] = 30, /* LOG_WARNING -> logging.WARNING */
+                [5] = 20, /* LOG_NOTICE  -> logging.INFO */
+                [6] = 20, /* LOG_INFO    -> logging.INFO */
+                [7] = 10, /* LOG_DEBUG   -> logging.DEBUG */
+        };
+
+        cs = CallState_get(h);
+        result = PyObject_CallMethod(h->logger, "log", "issss",
+                                     level_map[level],
+                                     "%s [%s] %s",
+                                     fac, rd_kafka_name(rk), buf);
+
+        if (result)
+                Py_DECREF(result);
+        else {
+                CallState_crash(cs);
+                rd_kafka_yield(h->rk);
+        }
+
+        CallState_resume(cs);
+}
+
 /****************************************************************************
  *
  *
@@ -1098,6 +1131,8 @@ void Handle_clear (Handle *h) {
 	if (h->stats_cb)
 		Py_DECREF(h->stats_cb);
 
+        Py_XDECREF(h->logger);
+
         if (h->initiated)
                 PyThread_delete_key(h->tlskey);
 }
@@ -1443,7 +1478,20 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
                         Py_XDECREF(ks8);
 			Py_DECREF(ks);
 			continue;
-		}
+                } else if (!strcmp(k, "logger")) {
+                        if (h->logger) {
+                                Py_DECREF(h->logger);
+                                h->logger = NULL;
+                        }
+
+                        if (vo != Py_None) {
+                                h->logger = vo;
+                                Py_INCREF(h->logger);
+                        }
+                        Py_XDECREF(ks8);
+                        Py_DECREF(ks);
+                        continue;
+                }
 
 		/* Special handling for certain config keys. */
 		if (ktype == RD_KAFKA_PRODUCER)
@@ -1509,6 +1557,13 @@ rd_kafka_conf_t *common_conf_setup (rd_kafka_type_t ktype,
 	if (h->stats_cb)
 		rd_kafka_conf_set_stats_cb(conf, stats_cb);
 
+        if (h->logger) {
+                /* Write logs to log queue (which is forwarded
+                 * to the polled queue in the Producer/Consumer constructors) */
+                rd_kafka_conf_set(conf, "log.queue", "true", NULL, 0);
+                rd_kafka_conf_set_log_cb(conf, log_cb);
+        }
+
 	rd_kafka_topic_conf_set_opaque(tconf, h);
 	rd_kafka_conf_set_default_topic_conf(conf, tconf);
 

confluent_kafka/src/confluent_kafka.h

@@ -163,6 +163,8 @@ typedef struct {
         int tlskey;  /* Thread-Local-Storage key */
         rd_kafka_type_t type; /* Producer or consumer */
 
+        PyObject *logger;
+
 	union {
 		/**
 		 * Producer

docs/index.rst

@@ -77,7 +77,7 @@ providing a dict of configuration properties to the instance constructor, e.g.::
           'group.id': 'mygroup', 'session.timeout.ms': 6000,
           'on_commit': my_commit_callback,
           'default.topic.config': {'auto.offset.reset': 'smallest'}}
-  consumer = confluent_kafka.Consumer(**conf)
+  consumer = confluent_kafka.Consumer(conf)
 
 The supported configuration values are dictated by the underlying
 librdkafka C library. For the full range of configuration properties
@@ -108,16 +108,12 @@ The Python bindings also provide some additional configuration properties:
 * ``on_commit(kafka.KafkaError, list(kafka.TopicPartition))`` (**Consumer**): Callback used to indicate success or failure
   of commit requests.
 
-Changelog
-=========
-
-Version 3.0.1
-^^^^^^^^^^^^^
-
-* `PR-3 <https://github.com/confluentinc/confluent-kafka-python/pull/3>`_ - Add /usr/local/lib to library_dirs in setup
-* `PR-4 <https://github.com/confluentinc/confluent-kafka-python/pull/4>`_ - Py3: use bytes for Message payload and key
-* `PR-5 <https://github.com/confluentinc/confluent-kafka-python/pull/5>`_ - Removed hard coded c extentions lib/include paths
-* `PR-9 <https://github.com/confluentinc/confluent-kafka-python/pull/9>`_ - Use consistent syntax highlighting (e.g. prefix commands with `$`)
-* `PR-17 <https://github.com/confluentinc/confluent-kafka-python/pull/17>`_ - Version bump to 0.9.1.2
-
+* ``logger=logging.Handler`` kwarg: forward logs from the Kafka client to the
+  provided ``logging.Handler`` instance.
+  To avoid spontaneous calls from non-Python threads the log messages
+  will only be forwarded when ``client.poll()`` is called.
 
+  mylogger = logging.getLogger()
+  mylogger.addHandler(logging.StreamHandler())
+  producer = confluent_kafka.Producer({'bootstrap.servers': 'mybroker.com'},
+                                      logger=mylogger)

examples/consumer.py

@@ -22,6 +22,7 @@
 import sys
 import getopt
 import json
+import logging
 from pprint import pformat
 
 
@@ -70,8 +71,16 @@ def print_usage_and_exit(program_name):
         conf['stats_cb'] = stats_cb
         conf['statistics.interval.ms'] = int(opt[1])
 
+    # Create logger for consumer (logs will be emitted when poll() is called)
+    logger = logging.getLogger('consumer')
+    logger.setLevel(logging.DEBUG)
+    handler = logging.StreamHandler()
+    handler.setFormatter(logging.Formatter('%(asctime)-15s %(levelname)-8s %(message)s'))
+    logger.addHandler(handler)
+
     # Create Consumer instance
-    c = Consumer(**conf)
+    # Hint: try debug='fetch' to generate some log messages
+    c = Consumer(conf, logger=logger)
 
     def print_assignment(consumer, partitions):
         print('Assignment:', partitions)

tests/test_log.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python
+
+import confluent_kafka
+import logging
+
+
+class CountingFilter(logging.Filter):
+    def __init__(self, name):
+        self.name = name
+        self.cnt = 0
+
+    def filter(self, record):
+        print(self.name, record.getMessage())
+        self.cnt += 1
+        print(record)
+
+
+def test_logging_consumer():
+    """ Tests that logging works """
+
+    logger = logging.getLogger('consumer')
+    logger.setLevel(logging.DEBUG)
+    f = CountingFilter('consumer')
+    logger.addFilter(f)
+
+    kc = confluent_kafka.Consumer({'group.id': 'test',
+                                   'debug': 'all'},
+                                  logger=logger)
+    while f.cnt == 0:
+        kc.poll(timeout=0.5)
+
+    print('%s: %d log messages seen' % (f.name, f.cnt))
+
+    kc.close()
+
+
+def test_logging_producer():
+    """ Tests that logging works """
+
+    logger = logging.getLogger('producer')
+    logger.setLevel(logging.DEBUG)
+    f = CountingFilter('producer')
+    logger.addFilter(f)
+
+    p = confluent_kafka.Producer({'debug': 'all'}, logger=logger)
+
+    while f.cnt == 0:
+        p.poll(timeout=0.5)
+
+    print('%s: %d log messages seen' % (f.name, f.cnt))
+
+
+def test_logging_constructor():
+    """ Verify different forms of constructors """
+
+    for how in ['dict', 'dict+kwarg', 'kwarg']:
+        logger = logging.getLogger('producer: ' + how)
+        logger.setLevel(logging.DEBUG)
+        f = CountingFilter(logger.name)
+        logger.addFilter(f)
+
+        if how == 'dict':
+            p = confluent_kafka.Producer({'debug': 'all', 'logger': logger})
+        elif how == 'dict+kwarg':
+            p = confluent_kafka.Producer({'debug': 'all'}, logger=logger)
+        elif how == 'kwarg':
+            conf = {'debug': 'all', 'logger': logger}
+            p = confluent_kafka.Producer(**conf)
+        else:
+            raise RuntimeError('Not reached')
+
+        print('Test %s with %s' % (p, how))
+
+        while f.cnt == 0:
+            p.poll(timeout=0.5)
+
+        print('%s: %s: %d log messages seen' % (how, f.name, f.cnt))