feat(kafka): add MessageBroker/Kafka/Cluster/{id}/Topic/{topic} metrics

newrelic/core/config.py

@@ -164,6 +164,10 @@ class GCRuntimeMetricsSettings(Settings):
     enabled = False
 
 
+class KafkaSettings(Settings):
+    cluster_metrics_enabled = False
+
+
 class MemoryRuntimeMetricsSettings(Settings):
     pass
 
@@ -660,6 +664,7 @@ class OpentelemetryTracesSettings(Settings):
 _settings.event_harvest_config.harvest_limits = EventHarvestConfigHarvestLimitSettings()
 _settings.event_loop_visibility = EventLoopVisibilitySettings()
 _settings.gc_runtime_metrics = GCRuntimeMetricsSettings()
+_settings.kafka = KafkaSettings()
 _settings.memory_runtime_pid_metrics = MemoryRuntimeMetricsSettings()
 _settings.heroku = HerokuSettings()
 _settings.infinite_tracing = InfiniteTracingSettings()
@@ -991,6 +996,8 @@ def default_otlp_host(host):
 _settings.gc_runtime_metrics.enabled = _environ_as_bool("NEW_RELIC_GC_RUNTIME_METRICS_ENABLED", default=False)
 _settings.gc_runtime_metrics.top_object_count_limit = 5
 
+_settings.kafka.cluster_metrics_enabled = _environ_as_bool("NEW_RELIC_KAFKA_CLUSTER_METRICS_ENABLED", default=False)
+
 _settings.memory_runtime_pid_metrics.enabled = _environ_as_bool(
     "NEW_RELIC_MEMORY_RUNTIME_PID_METRICS_ENABLED", default=True
 )

newrelic/hooks/messagebroker_confluentkafka.py

@@ -13,6 +13,9 @@
 # limitations under the License.
 import logging
 import sys
+import threading
+import time
+import weakref
 
 from newrelic.api.application import application_instance
 from newrelic.api.error_trace import wrap_error_trace
@@ -23,6 +26,7 @@
 from newrelic.api.transaction import current_transaction
 from newrelic.common.object_wrapper import function_wrapper, wrap_function_wrapper
 from newrelic.common.package_version_utils import get_package_version
+from newrelic.core.config import global_settings
 
 _logger = logging.getLogger(__name__)
 
@@ -33,6 +37,69 @@
 HEARTBEAT_SESSION_TIMEOUT = "MessageBroker/Kafka/Heartbeat/SessionTimeout"
 HEARTBEAT_POLL_TIMEOUT = "MessageBroker/Kafka/Heartbeat/PollTimeout"
 
+KAFKA_CLUSTER_METRIC_PRODUCE = "MessageBroker/Kafka/Cluster/{0}/Topic/{1}/Produce"
+KAFKA_CLUSTER_METRIC_CONSUME = "MessageBroker/Kafka/Cluster/{0}/Topic/{1}/Consume"
+
+
+_CLUSTER_ID_TTL_SECONDS = 60 * 60
+
+_nr_cluster_id_cache = {}
+_nr_cluster_id_cache_lock = threading.Lock()
+
+
+def _fetch_cluster_id(instance):
+    settings = global_settings()
+    if not getattr(getattr(settings, "kafka", None), "cluster_metrics_enabled", False):
+        return
+    servers = getattr(instance, "_nr_bootstrap_servers", None)
+    # Sort so that equivalent broker sets with different orderings share the same key.
+    cache_key = ",".join(sorted(servers)) if servers else None
+
+    if cache_key:
+        with _nr_cluster_id_cache_lock:
+            cached = _nr_cluster_id_cache.get(cache_key)
+            if isinstance(cached, tuple):
+                cluster_id, ts = cached
+                if time.monotonic() - ts <= _CLUSTER_ID_TTL_SECONDS:
+                    instance._nr_cluster_id = cluster_id
+                    instance._nr_cluster_id_fetched_at = ts
+                    return
+                # Expired — fall through to start a new fetch.
+            elif cached is not None:
+                # "" sentinel — fetch already in progress.
+                return
+            _nr_cluster_id_cache[cache_key] = ""
+
+    # Hold only a weak reference so the thread closure does not extend the
+    # lifetime of a Producer/Consumer that the caller has already abandoned.
+    instance_ref = weakref.ref(instance)
+
+    def _run():
+        inst = instance_ref()
+        if inst is None:
+            # Instance was GC'd before the thread ran; clean up sentinel and exit.
+            if cache_key:
+                with _nr_cluster_id_cache_lock:
+                    _nr_cluster_id_cache.pop(cache_key, None)
+            return
+        try:
+            meta = inst.list_topics(timeout=5)
+            cluster_id = getattr(meta, "cluster_id", None)
+            if cluster_id:
+                now = time.monotonic()
+                inst._nr_cluster_id = cluster_id
+                inst._nr_cluster_id_fetched_at = now
+                if cache_key:
+                    with _nr_cluster_id_cache_lock:
+                        _nr_cluster_id_cache[cache_key] = (cluster_id, now)
+        except Exception:
+            _logger.debug("NR Kafka cluster ID fetch failed", exc_info=True)
+            if cache_key:
+                with _nr_cluster_id_cache_lock:
+                    _nr_cluster_id_cache.pop(cache_key, None)
+
+    threading.Thread(target=_run, daemon=True, name="NR-Kafka-ClusterId").start()
+
 
 def wrap_Producer_produce(wrapped, instance, args, kwargs):
     transaction = current_transaction()
@@ -63,6 +130,24 @@ def wrap_Producer_produce(wrapped, instance, args, kwargs):
         for server_name in instance._nr_bootstrap_servers:
             transaction.record_custom_metric(f"MessageBroker/Kafka/Nodes/{server_name}/Produce/{topic}", 1)
 
+    cluster_id = getattr(instance, "_nr_cluster_id", None)
+    if cluster_id:
+        if time.monotonic() - getattr(instance, "_nr_cluster_id_fetched_at", 0.0) > _CLUSTER_ID_TTL_SECONDS:
+            _fetch_cluster_id(instance)  # background re-fetch; use stale value below
+    elif hasattr(instance, "_nr_bootstrap_servers"):
+        _cache_key = ",".join(sorted(instance._nr_bootstrap_servers))
+        _cached = _nr_cluster_id_cache.get(_cache_key)
+        if isinstance(_cached, tuple):
+            cluster_id, _ts = _cached
+            instance._nr_cluster_id = cluster_id
+            instance._nr_cluster_id_fetched_at = _ts
+            if time.monotonic() - _ts > _CLUSTER_ID_TTL_SECONDS:
+                _fetch_cluster_id(instance)  # background re-fetch
+    if cluster_id:
+        transaction.record_custom_metric(
+            KAFKA_CLUSTER_METRIC_PRODUCE.format(cluster_id, topic), 1
+        )
+
     with MessageTrace(
         library="Kafka", operation="Produce", destination_type="Topic", destination_name=topic, source=wrapped
     ):
@@ -171,6 +256,23 @@ def wrap_Consumer_poll(wrapped, instance, args, kwargs):
                     transaction.record_custom_metric(
                         f"MessageBroker/Kafka/Nodes/{server_name}/Consume/{destination_name}", 1
                     )
+            cluster_id = getattr(instance, "_nr_cluster_id", None)
+            if cluster_id:
+                if time.monotonic() - getattr(instance, "_nr_cluster_id_fetched_at", 0.0) > _CLUSTER_ID_TTL_SECONDS:
+                    _fetch_cluster_id(instance)  # background re-fetch; use stale value below
+            elif hasattr(instance, "_nr_bootstrap_servers"):
+                _cache_key = ",".join(sorted(instance._nr_bootstrap_servers))
+                _cached = _nr_cluster_id_cache.get(_cache_key)
+                if isinstance(_cached, tuple):
+                    cluster_id, _ts = _cached
+                    instance._nr_cluster_id = cluster_id
+                    instance._nr_cluster_id_fetched_at = _ts
+                    if time.monotonic() - _ts > _CLUSTER_ID_TTL_SECONDS:
+                        _fetch_cluster_id(instance)  # background re-fetch
+            if cluster_id:
+                transaction.record_custom_metric(
+                    KAFKA_CLUSTER_METRIC_CONSUME.format(cluster_id, destination_name), 1
+                )
             transaction.add_messagebroker_info("Confluent-Kafka", get_package_version("confluent-kafka"))
 
     return record
@@ -213,6 +315,16 @@ def wrap_SerializingProducer_init(wrapped, instance, args, kwargs):
     if hasattr(instance, "_value_serializer") and callable(instance._value_serializer):
         instance._value_serializer = wrap_serializer("Serialization/Value", "MessageBroker")(instance._value_serializer)
 
+    try:
+        conf = kwargs.get("conf") or (args[0] if args else {})
+        servers = conf.get("bootstrap.servers") if isinstance(conf, dict) else None
+        if servers:
+            instance._nr_bootstrap_servers = servers.split(",")
+    except Exception:
+        pass
+
+    _fetch_cluster_id(instance)
+
 
 def wrap_DeserializingConsumer_init(wrapped, instance, args, kwargs):
     wrapped(*args, **kwargs)
@@ -223,6 +335,16 @@ def wrap_DeserializingConsumer_init(wrapped, instance, args, kwargs):
     if hasattr(instance, "_value_deserializer") and callable(instance._value_deserializer):
         instance._value_deserializer = wrap_serializer("Deserialization/Value", "Message")(instance._value_deserializer)
 
+    try:
+        conf = kwargs.get("conf") or (args[0] if args else {})
+        servers = conf.get("bootstrap.servers") if isinstance(conf, dict) else None
+        if servers:
+            instance._nr_bootstrap_servers = servers.split(",")
+    except Exception:
+        pass
+
+    _fetch_cluster_id(instance)
+
 
 def wrap_Producer_init(wrapped, instance, args, kwargs):
     wrapped(*args, **kwargs)
@@ -236,6 +358,8 @@ def wrap_Producer_init(wrapped, instance, args, kwargs):
     except Exception:
         pass
 
+    _fetch_cluster_id(instance)
+
 
 def wrap_Consumer_init(wrapped, instance, args, kwargs):
     wrapped(*args, **kwargs)
@@ -249,6 +373,8 @@ def wrap_Consumer_init(wrapped, instance, args, kwargs):
     except Exception:
         pass
 
+    _fetch_cluster_id(instance)
+
 
 def wrap_immutable_class(module, class_name):
     # Wrap immutable binary extension class with a mutable Python subclass