Add usage tracking metrics for Kafka clients.

newrelic/api/transaction.py

@@ -186,6 +186,7 @@ def __init__(self, application, enabled=None, source=None):
         self._loop_time = 0.0
 
         self._frameworks = set()
+        self._message_brokers = set()
 
         self._frozen_path = None
 
@@ -541,6 +542,10 @@ def __exit__(self, exc, value, tb):
             for framework, version in self._frameworks:
                 self.record_custom_metric("Python/Framework/%s/%s" % (framework, version), 1)
 
+        if self._message_brokers:
+            for message_broker, version in self._message_brokers:
+                self.record_custom_metric("Python/MessageBroker/%s/%s" % (message_broker, version), 1)
+
         if self._settings.distributed_tracing.enabled:
             # Sampled and priority need to be computed at the end of the
             # transaction when distributed tracing or span events are enabled.
@@ -1666,6 +1671,10 @@ def add_framework_info(self, name, version=None):
         if name:
             self._frameworks.add((name, version))
 
+    def add_messagebroker_info(self, name, version=None):
+        if name:
+            self._message_brokers.add((name, version))
+
     def dump(self, file):
         """Dumps details about the transaction to the file object."""
 

newrelic/hooks/messagebroker_confluentkafka.py

@@ -33,6 +33,12 @@
 HEARTBEAT_POLL_TIMEOUT = "MessageBroker/Kafka/Heartbeat/PollTimeout"
 
 
+def confluent_kafka_version():
+    import confluent_kafka
+
+    return getattr(confluent_kafka, "__version__", None)
+
+
 def wrap_Producer_produce(wrapped, instance, args, kwargs):
     transaction = current_transaction()
     if transaction is None:
@@ -56,6 +62,8 @@ def wrap_Producer_produce(wrapped, instance, args, kwargs):
     else:
         topic = kwargs.get("topic", None)
 
+    transaction.add_messagebroker_info("Confluent-Kafka", confluent_kafka_version())
+
     with MessageTrace(
         library="Kafka",
         operation="Produce",

newrelic/hooks/messagebroker_kafkapython.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 import sys
 
+import kafka
 from kafka.serializer import Serializer
 
 from newrelic.api.application import application_instance
@@ -39,6 +40,10 @@ def _bind_send(topic, value=None, key=None, headers=None, partition=None, timest
     return topic, value, key, headers, partition, timestamp_ms
 
 
+def kafka_python_version():
+    return getattr(kafka, "__version__", None)
+
+
 def wrap_KafkaProducer_send(wrapped, instance, args, kwargs):
     transaction = current_transaction()
 
@@ -48,6 +53,8 @@ def wrap_KafkaProducer_send(wrapped, instance, args, kwargs):
     topic, value, key, headers, partition, timestamp_ms = _bind_send(*args, **kwargs)
     headers = list(headers) if headers else []
 
+    transaction.add_messagebroker_info("Kafka-Python", kafka_python_version())
+
     with MessageTrace(
         library="Kafka",
         operation="Produce",

tests/messagebroker_confluentkafka/test_producer.py

@@ -65,6 +65,8 @@ def producer_callback(err, msg):
 
 
 def test_trace_metrics(topic, send_producer_message):
+    from confluent_kafka import __version__ as version
+
     scoped_metrics = [("MessageBroker/Kafka/Topic/Produce/Named/%s" % topic, 1)]
     unscoped_metrics = scoped_metrics
     txn_name = "test_producer:test_trace_metrics.<locals>.test" if six.PY3 else "test_producer:test"
@@ -73,6 +75,7 @@ def test_trace_metrics(topic, send_producer_message):
         txn_name,
         scoped_metrics=scoped_metrics,
         rollup_metrics=unscoped_metrics,
+        custom_metrics=[("Python/MessageBroker/Confluent-Kafka/%s" % version, 1)],
         background_task=True,
     )
     @background_task()

tests/messagebroker_kafkapython/test_producer.py

@@ -28,6 +28,8 @@
 
 
 def test_trace_metrics(topic, send_producer_message):
+    from kafka.version import __version__ as version
+
     scoped_metrics = [("MessageBroker/Kafka/Topic/Produce/Named/%s" % topic, 1)]
     unscoped_metrics = scoped_metrics
     txn_name = "test_producer:test_trace_metrics.<locals>.test" if six.PY3 else "test_producer:test"
@@ -36,6 +38,7 @@ def test_trace_metrics(topic, send_producer_message):
         txn_name,
         scoped_metrics=scoped_metrics,
         rollup_metrics=unscoped_metrics,
+        custom_metrics=[("Python/MessageBroker/Kafka-Python/%s" % version, 1)],
         background_task=True,
     )
     @background_task()