Kafka Internal Metrics Instrumentation (#613)

* Kafka Metrics data source

Co-authored-by: Lalleh Rafeei <[email protected]>
Co-authored-by: Hannah Stepanek <[email protected]>

* [Mega-Linter] Apply linters fixes

* Remove breakpoint

* Run linter

* Fix black issues

* Swap @classmethod decorator order.

Co-authored-by: Hannah Stepanek <[email protected]>

* Remove variable named variable

Co-authored-by: Lalleh Rafeei <[email protected]>
Co-authored-by: Hannah Stepanek <[email protected]>
Co-authored-by: TimPansino <[email protected]>
Co-authored-by: Lalleh Rafeei <[email protected]>
Co-authored-by: Hannah Stepanek <[email protected]>

Author: 3 people

newrelic/config.py

@@ -2318,6 +2318,12 @@ def _process_module_builtin_defaults():
         "instrument_kafka_heartbeat",
     )
 
+    _process_module_definition(
+        "kafka.consumer.group",
+        "newrelic.hooks.messagebroker_kafkapython",
+        "instrument_kafka_consumer_group",
+    )
+
     _process_module_definition(
         "logging",
         "newrelic.hooks.logger_logging",

newrelic/hooks/messagebroker_kafkapython.py

@@ -12,11 +12,22 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import logging
+import math
+import threading
+
+from kafka.metrics.metrics_reporter import AbstractMetricsReporter
+
+import newrelic.core.agent
 from newrelic.api.application import application_instance
 from newrelic.api.message_trace import MessageTrace
 from newrelic.api.time_trace import notice_error
 from newrelic.api.transaction import current_transaction
 from newrelic.common.object_wrapper import wrap_function_wrapper
+from newrelic.packages import six
+from newrelic.samplers.decorators import data_source_factory
+
+_logger = logging.getLogger(__name__)
 
 HEARTBEAT_POLL = "MessageBroker/Kafka/Heartbeat/Poll"
 HEARTBEAT_SENT = "MessageBroker/Kafka/Heartbeat/Sent"
@@ -55,11 +66,6 @@ def wrap_KafkaProducer_send(wrapped, instance, args, kwargs):
             raise
 
 
-def instrument_kafka_producer(module):
-    if hasattr(module, "KafkaProducer"):
-        wrap_function_wrapper(module, "KafkaProducer.send", wrap_KafkaProducer_send)
-
-
 def metric_wrapper(metric_name, check_result=False):
     def _metric_wrapper(wrapped, instance, args, kwargs):
         result = wrapped(*args, **kwargs)
@@ -101,3 +107,144 @@ def instrument_kafka_heartbeat(module):
             wrap_function_wrapper(
                 module, "Heartbeat.poll_timeout_expired", metric_wrapper(HEARTBEAT_POLL_TIMEOUT, check_result=True)
             )
+
+
+class KafkaMetricsDataSource(object):
+    _instance = None
+
+    def __init__(self):
+        self.reporters = []
+
+    @classmethod
+    @data_source_factory(name="Kafka Metrics Reporter")
+    def factory(cls, settings=None, environ=None):
+        return cls.singleton()
+
+    @classmethod
+    def singleton(cls, register=True):
+        # If already initialized, exit early
+        if cls._instance:
+            return cls._instance
+
+        # Init and register instance on class
+        instance = cls()
+        cls._instance = instance
+
+        # register_data_source takes a callable so let it rerun singleton to retrieve the instance
+        if register:
+            try:
+                _logger.debug("Registering kafka metrics data source.")
+                newrelic.core.agent.agent_instance().register_data_source(cls.factory)
+            except Exception:
+                _logger.exception(
+                    "Attempt to register kafka metrics data source has failed. Data source will be skipped."
+                )
+
+        return instance
+
+    def register(self, reporter):
+        self.reporters.append(reporter)
+
+    def unregister(self, reporter):
+        if reporter in self.reporters:
+            self.reporters.remove(reporter)
+
+    def start(self):
+        return
+
+    def stop(self):
+        # Clear references to reporters to prevent them from participating in a reference cycle.
+        self.reporters = []
+
+    def __call__(self):
+        for reporter in self.reporters:
+            for name, metric in six.iteritems(reporter.snapshot()):
+                yield name, metric
+
+
+class NewRelicMetricsReporter(AbstractMetricsReporter):
+    def __init__(self, *args, **kwargs):
+        super(NewRelicMetricsReporter, self).__init__(*args, **kwargs)
+
+        # Register with data source for harvesting
+        self.data_source = KafkaMetricsDataSource.singleton()
+        self.data_source.register(self)
+
+        self._metrics = {}
+        self._lock = threading.Lock()
+
+    def close(self, *args, **kwargs):
+        self.data_source.unregister(self)
+        with self._lock:
+            self._metrics = {}
+
+    def init(self, metrics):
+        for metric in metrics:
+            self.metric_change(metric)
+
+    @staticmethod
+    def invalid_metric_value(metric):
+        name, value = metric
+        return not any((math.isinf(value), math.isnan(value), value == 0))
+
+    def snapshot(self):
+        with self._lock:
+            # metric.value can only be called once, so care must be taken when filtering
+            metrics = ((name, metric.value()) for name, metric in six.iteritems(self._metrics))
+            return {
+                "MessageBroker/Kafka/Internal/%s" % name: {"count": value}
+                for name, value in filter(self.invalid_metric_value, metrics)
+            }
+
+    def get_metric_name(self, metric):
+        metric_name = metric.metric_name  # Get MetricName object to work with
+
+        name = metric_name.name
+        group = metric_name.group
+
+        if "topic" in metric_name.tags:
+            topic = metric_name.tags["topic"]
+            return "/".join((group, topic, name))
+        else:
+            return "/".join((group, name))
+
+    def metric_change(self, metric):
+        name = self.get_metric_name(metric)
+        with self._lock:
+            self._metrics[name] = metric
+
+    def metric_removal(self, metric):
+        name = self.get_metric_name(metric)
+        with self._lock:
+            if name in self._metrics:
+                self._metrics.pop(name)
+
+    def configure(self, configs):
+        return
+
+
+def wrap_KafkaProducerConsumer_init(wrapped, instance, args, kwargs):
+    try:
+        if "metric_reporters" in kwargs:
+            metric_reporters = list(kwargs.get("metric_reporters", []))
+            metric_reporters.append(NewRelicMetricsReporter)
+            kwargs["metric_reporters"] = [metric_reporters]
+        else:
+            kwargs["metric_reporters"] = [NewRelicMetricsReporter]
+    except Exception:
+        pass
+
+    return wrapped(*args, **kwargs)
+
+
+def instrument_kafka_producer(module):
+    if hasattr(module, "KafkaProducer"):
+        wrap_function_wrapper(module, "KafkaProducer.__init__", wrap_KafkaProducerConsumer_init)
+
+    if hasattr(module, "KafkaProducer"):
+        wrap_function_wrapper(module, "KafkaProducer.send", wrap_KafkaProducer_send)
+
+
+def instrument_kafka_consumer_group(module):
+    if hasattr(module, "KafkaConsumer"):
+        wrap_function_wrapper(module, "KafkaConsumer.__init__", wrap_KafkaProducerConsumer_init)

tests/messagebroker_kafkapython/conftest.py

@@ -18,14 +18,15 @@
 import kafka
 import pytest
 from testing_support.db_settings import kafka_settings
-from testing_support.fixtures import (  # noqa: F401
+from testing_support.fixtures import (  # noqa: F401, W0611
     code_coverage_fixture,
     collector_agent_registration_fixture,
     collector_available_fixture,
 )
 
 from newrelic.api.transaction import current_transaction
 from newrelic.common.object_wrapper import transient_function_wrapper
+from newrelic.hooks.messagebroker_kafkapython import KafkaMetricsDataSource
 
 DB_SETTINGS = kafka_settings()[0]
 
@@ -54,7 +55,7 @@
 
 
 @pytest.fixture(scope="function")
-def producer():
+def producer(data_source):
     producer = kafka.KafkaProducer(
         bootstrap_servers=BROKER, api_version=(2, 0, 2), value_serializer=lambda v: json.dumps(v).encode("utf-8")
     )
@@ -63,7 +64,7 @@ def producer():
 
 
 @pytest.fixture(scope="function")
-def consumer(topic):
+def consumer(topic, data_source):
     consumer = kafka.KafkaConsumer(
         topic,
         bootstrap_servers=BROKER,
@@ -82,6 +83,17 @@ def topic():
     yield "test-topic-%s" % str(uuid.uuid4())
 
 
+@pytest.fixture(scope="session")
+def data_source():
+    """
+    Must be required by consumer and producer fixtures, or the first one of them to be
+    instantiated will create and register the singleton. We rely on the singleton to
+    not be registered to properly test the output of it without interference from the
+    harvest thread.
+    """
+    return KafkaMetricsDataSource.singleton(register=False)
+
+
 @transient_function_wrapper(kafka.producer.kafka, "KafkaProducer.send.__wrapped__")
 # Place transient wrapper underneath instrumentation
 def cache_kafka_headers(wrapped, instance, args, kwargs):

tests/messagebroker_kafkapython/test_metrics.py

@@ -0,0 +1,123 @@
+# Copyright 2010 New Relic, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+def test_data_source_metrics(data_source, topic, producer, consumer):
+    producer.send(topic, value=1)
+    producer.flush()
+    next(iter(consumer))
+
+    metrics = dict(data_source())
+    metric_names = list(metrics.keys())
+    assert metrics
+
+
+# Example metrics
+
+# MessageBroker/Kafka/Internal/kafka-metrics-count/count
+# MessageBroker/Kafka/Internal/producer-metrics/connection-close-rate
+# MessageBroker/Kafka/Internal/producer-metrics/connection-creation-rate
+# MessageBroker/Kafka/Internal/producer-metrics/select-rate
+# MessageBroker/Kafka/Internal/producer-metrics/io-wait-time-ns-avg
+# MessageBroker/Kafka/Internal/producer-metrics/io-wait-ratio
+# MessageBroker/Kafka/Internal/producer-metrics/io-time-ns-avg
+# MessageBroker/Kafka/Internal/producer-metrics/io-ratio
+# MessageBroker/Kafka/Internal/producer-metrics/connection-count
+# MessageBroker/Kafka/Internal/producer-metrics/batch-size-avg
+# MessageBroker/Kafka/Internal/producer-metrics/batch-size-max
+# MessageBroker/Kafka/Internal/producer-metrics/compression-rate-avg
+# MessageBroker/Kafka/Internal/producer-metrics/record-queue-time-avg
+# MessageBroker/Kafka/Internal/producer-metrics/record-queue-time-max
+# MessageBroker/Kafka/Internal/producer-metrics/record-send-rate
+# MessageBroker/Kafka/Internal/producer-metrics/records-per-request-avg
+# MessageBroker/Kafka/Internal/producer-metrics/byte-rate
+# MessageBroker/Kafka/Internal/producer-metrics/record-size-max
+# MessageBroker/Kafka/Internal/producer-metrics/record-size-avg
+# MessageBroker/Kafka/Internal/producer-metrics/metadata-age
+# MessageBroker/Kafka/Internal/producer-metrics/network-io-rate
+# MessageBroker/Kafka/Internal/producer-metrics/outgoing-byte-rate
+# MessageBroker/Kafka/Internal/producer-metrics/request-rate
+# MessageBroker/Kafka/Internal/producer-metrics/request-size-avg
+# MessageBroker/Kafka/Internal/producer-metrics/request-size-max
+# MessageBroker/Kafka/Internal/producer-metrics/incoming-byte-rate
+# MessageBroker/Kafka/Internal/producer-metrics/response-rate
+# MessageBroker/Kafka/Internal/producer-metrics/request-latency-avg
+# MessageBroker/Kafka/Internal/producer-metrics/request-latency-max
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-bootstrap-0/outgoing-byte-rate
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-bootstrap-0/request-rate
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-bootstrap-0/request-size-avg
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-bootstrap-0/request-size-max
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-bootstrap-0/incoming-byte-rate
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-bootstrap-0/response-rate
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-bootstrap-0/request-latency-avg
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-bootstrap-0/request-latency-max
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-1001/outgoing-byte-rate
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-1001/request-rate
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-1001/request-size-avg
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-1001/request-size-max
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-1001/incoming-byte-rate
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-1001/response-rate
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-1001/request-latency-avg
+# MessageBroker/Kafka/Internal/producer-node-metrics.node-1001/request-latency-max
+# MessageBroker/Kafka/Internal/producer-topic-metrics.test-topic-c962647a-f6cf-4a24-a90b-40ab2364dc55/record-send-rate
+# MessageBroker/Kafka/Internal/producer-topic-metrics.test-topic-c962647a-f6cf-4a24-a90b-40ab2364dc55/byte-rate
+# MessageBroker/Kafka/Internal/producer-topic-metrics.test-topic-c962647a-f6cf-4a24-a90b-40ab2364dc55/compression-rate
+# MessageBroker/Kafka/Internal/consumer-metrics/connection-close-rate
+# MessageBroker/Kafka/Internal/consumer-metrics/connection-creation-rate
+# MessageBroker/Kafka/Internal/consumer-metrics/select-rate
+# MessageBroker/Kafka/Internal/consumer-metrics/io-wait-time-ns-avg
+# MessageBroker/Kafka/Internal/consumer-metrics/io-wait-ratio
+# MessageBroker/Kafka/Internal/consumer-metrics/io-time-ns-avg
+# MessageBroker/Kafka/Internal/consumer-metrics/io-ratio
+# MessageBroker/Kafka/Internal/consumer-metrics/connection-count
+# MessageBroker/Kafka/Internal/consumer-metrics/network-io-rate
+# MessageBroker/Kafka/Internal/consumer-metrics/outgoing-byte-rate
+# MessageBroker/Kafka/Internal/consumer-metrics/request-rate
+# MessageBroker/Kafka/Internal/consumer-metrics/request-size-avg
+# MessageBroker/Kafka/Internal/consumer-metrics/request-size-max
+# MessageBroker/Kafka/Internal/consumer-metrics/incoming-byte-rate
+# MessageBroker/Kafka/Internal/consumer-metrics/response-rate
+# MessageBroker/Kafka/Internal/consumer-metrics/request-latency-avg
+# MessageBroker/Kafka/Internal/consumer-metrics/request-latency-max
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-bootstrap-0/outgoing-byte-rate
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-bootstrap-0/request-rate
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-bootstrap-0/request-size-avg
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-bootstrap-0/request-size-max
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-bootstrap-0/incoming-byte-rate
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-bootstrap-0/response-rate
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-bootstrap-0/request-latency-avg
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-bootstrap-0/request-latency-max
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/fetch-size-avg
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/fetch-size-max
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/bytes-consumed-rate
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/records-per-request-avg
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/records-consumed-rate
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/fetch-latency-avg
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/fetch-latency-max
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/fetch-rate
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/records-lag-max
+# MessageBroker/Kafka/Internal/consumer-coordinator-metrics/assigned-partitions
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-1001/outgoing-byte-rate
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-1001/request-rate
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-1001/request-size-avg
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-1001/request-size-max
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-1001/incoming-byte-rate
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-1001/response-rate
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-1001/request-latency-avg
+# MessageBroker/Kafka/Internal/consumer-node-metrics.node-1001/request-latency-max
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/test-topic-c962647a-f6cf-4a24-a90b-40ab2364dc55/fetch-size-avg
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/test-topic-c962647a-f6cf-4a24-a90b-40ab2364dc55/fetch-size-max
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/test-topic-c962647a-f6cf-4a24-a90b-40ab2364dc55/bytes-consumed-rate
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/test-topic-c962647a-f6cf-4a24-a90b-40ab2364dc55/records-per-request-avg
+# MessageBroker/Kafka/Internal/consumer-fetch-manager-metrics/test-topic-c962647a-f6cf-4a24-a90b-40ab2364dc55/records-consumed-rate