From 964611caba55bfc45e7318b9973717554fe1a65b Mon Sep 17 00:00:00 2001
From: Yuwei Jiao <yjiao@brex.com>
Date: Mon, 15 Aug 2022 17:36:00 -0700
Subject: [PATCH] prepend tag names in kafka metric if using prometheus
 reporter

---
 kafka/build.gradle                            |  2 ++
 .../metrics/AbstractKafkaMetricsReporter.java | 19 ++++++++++++++++++-
 2 files changed, 20 insertions(+), 1 deletion(-)

diff --git a/kafka/build.gradle b/kafka/build.gradle
index dc67f1e8f..ab8ddecd4 100644
--- a/kafka/build.gradle
+++ b/kafka/build.gradle
@@ -14,6 +14,7 @@ dependencies {
     api libs.managed.kafka.clients
 
     compileOnly mn.micronaut.micrometer.core
+    compileOnly mn.micronaut.micrometer.registry.prometheus
     compileOnly libs.opentracing.kafka.client
     compileOnly libs.zipkin.brave.kafka.clients
     compileOnly mn.micronaut.graal
@@ -27,4 +28,5 @@ dependencies {
 
     testRuntimeOnly mn.micronaut.micrometer.registry.statsd
     testRuntimeOnly mn.micronaut.tracing.core
+    testRuntimeOnly mn.micronaut.micrometer.registry.prometheus
 }
diff --git a/kafka/src/main/java/io/micronaut/configuration/kafka/metrics/AbstractKafkaMetricsReporter.java b/kafka/src/main/java/io/micronaut/configuration/kafka/metrics/AbstractKafkaMetricsReporter.java
index 2c87b94e3..545ba5811 100644
--- a/kafka/src/main/java/io/micronaut/configuration/kafka/metrics/AbstractKafkaMetricsReporter.java
+++ b/kafka/src/main/java/io/micronaut/configuration/kafka/metrics/AbstractKafkaMetricsReporter.java
@@ -19,6 +19,7 @@
 import io.micrometer.core.instrument.Tag;
 import io.micrometer.core.instrument.binder.MeterBinder;
 import io.micrometer.core.lang.NonNull;
+import io.micrometer.prometheus.PrometheusMeterRegistry;
 import io.micronaut.configuration.kafka.metrics.builder.KafkaMetricMeterTypeBuilder;
 import io.micronaut.core.annotation.Internal;
 import org.apache.kafka.common.MetricName;
@@ -99,7 +100,7 @@ public void close() {
 
     private void registerMetric(MeterRegistry meterRegistry, KafkaMetric metric) {
         KafkaMetricMeterTypeBuilder.newBuilder()
-                .prefix(getMetricPrefix())
+                .prefix(getMetricPrefix(meterRegistry, metric))
                 .metric(metric)
                 .tagFunction(getTagFunction())
                 .registry(meterRegistry)
@@ -129,6 +130,22 @@ protected Set<String> getIncludedTags() {
         return tags;
     }
 
+    /**
+     * Customize metric prefix with tag names
+     * @param meterRegistry
+     * @param metric
+     * @return
+     */
+    private String getMetricPrefix(MeterRegistry meterRegistry, KafkaMetric metric) {
+        String prefix = getMetricPrefix();
+        if (meterRegistry instanceof PrometheusMeterRegistry) {
+            StringBuilder sb = new StringBuilder(prefix);
+            metric.metricName().tags().keySet().forEach(v -> sb.append(".").append(v));
+            return sb.toString();
+        }
+        return prefix;
+    }
+
     /**
      * Abstract method to implement with the metric prefix for the reporter.
      *