From be264a7f1bd28fe3298af1a8c13db74f0fe25983 Mon Sep 17 00:00:00 2001
From: Federico Valeri <fedevaleri@gmail.com>
Date: Thu, 19 Dec 2024 09:30:32 +0100
Subject: [PATCH] Add Cluster Operator integration bits

Implementation of "Integrate Bridge with Metrics Reporter" proposal.

Signed-off-by: Federico Valeri <fedevaleri@gmail.com>
---
 CHANGELOG.md                                  |  3 +-
 bin/docker/kafka_bridge_config_generator.sh   | 20 ++++++++++++
 config/application.properties                 |  2 ++
 ...-configuring-kafka-bridge-jmx-metrics.adoc |  3 ++
 .../io/strimzi/kafka/bridge/Application.java  | 32 ++++++++++++-------
 .../kafka/bridge/config/BridgeConfig.java     | 15 +++++++++
 .../kafka/bridge/config/ConfigTest.java       |  5 +++
 7 files changed, 67 insertions(+), 13 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a4f21fec..57b6948f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,8 +5,7 @@
 * Dependency updates (JMX exporter 1.1.0)
 * Dropped support for Java 11 and replaced with Java 17.
 * Added support for the [Strimzi Metrics Reporter](https://github.com/strimzi/metrics-reporter) metrics.
-  This is a Kafka plugin that exports metrics in Prometheus format without passing through JMX, and can be enabled by setting `bridge.metrics=strimziMetricsReporter`.
-  The JMX Exporter metrics can still be enabled by setting `bridge.metrics=jmxPrometheusExporter`.
+  This is a Kafka plugin that directly exports metrics in Prometheus format, and can be enabled by setting `bridge.metrics=strimziMetricsReporter`.
 
 ### Changes, deprecations and removals
 
diff --git a/bin/docker/kafka_bridge_config_generator.sh b/bin/docker/kafka_bridge_config_generator.sh
index a136ca53..9dac70f7 100755
--- a/bin/docker/kafka_bridge_config_generator.sh
+++ b/bin/docker/kafka_bridge_config_generator.sh
@@ -11,10 +11,27 @@ fi
 BRIDGE_PROPERTIES=$(cat <<-EOF
 #Bridge configuration
 bridge.id=${KAFKA_BRIDGE_ID}
+${BRIDGE_METRICS}
 ${BRIDGE_TRACING}
 EOF
 )
 
+if [ -n "$KAFKA_BRIDGE_METRICS_JMX_CONFIG" ]; then
+    METRICS_JMX_PROPERTIES=$(cat <<EOF
+#JMX Exporter configuration
+bridge.metrics.jmx.exporter.config.path=${KAFKA_BRIDGE_METRICS_JMX_CONFIG}
+EOF
+)
+fi
+
+if [ -n "$KAFKA_BRIDGE_METRICS_SMR_CONFIG" ]; then
+    METRICS_SMR_PROPERTIES=$(cat <<EOF
+#Strimzi Reporter configuration
+${KAFKA_BRIDGE_METRICS_SMR_CONFIG}
+EOF
+)
+fi
+
 SECURITY_PROTOCOL=PLAINTEXT
 
 if [ "$KAFKA_BRIDGE_TLS" = "true" ]; then
@@ -167,6 +184,9 @@ EOF
 PROPERTIES=$(cat <<EOF
 $BRIDGE_PROPERTIES
 
+$METRICS_JMX_PROPERTIES
+$METRICS_SMR_PROPERTIES
+
 $KAFKA_PROPERTIES
 
 $ADMIN_CLIENT_PROPERTIES
diff --git a/config/application.properties b/config/application.properties
index 192f54c1..28905ca6 100644
--- a/config/application.properties
+++ b/config/application.properties
@@ -3,6 +3,8 @@ bridge.id=my-bridge
 
 # uncomment the following line to enable JMX Exporter metrics, check the documentation for more details
 #bridge.metrics=jmxPrometheusExporter
+# optionally, you can also set a custom configuration file
+#bridge.metrics.jmx.exporter.config.path=/path/to/my-jmx-exporter-config.yaml
 
 # uncomment the following lines to enable Strimzi Reporter metrics, check the documentation for more details
 #bridge.metrics=strimziMetricsReporter
diff --git a/documentation/modules/proc-configuring-kafka-bridge-jmx-metrics.adoc b/documentation/modules/proc-configuring-kafka-bridge-jmx-metrics.adoc
index 0c774409..3933bce2 100644
--- a/documentation/modules/proc-configuring-kafka-bridge-jmx-metrics.adoc
+++ b/documentation/modules/proc-configuring-kafka-bridge-jmx-metrics.adoc
@@ -18,6 +18,9 @@ Enable metrics for the Kafka Bridge by setting the `bridge.metrics` configuratio
 ----
 bridge.metrics=jmxPrometheusExporter
 ----
++
+Optionally, you can set a custom JMX Exporter configuration file using the `bridge.metrics.jmx.exporter.config.path` property.
+If this is not set, a default configuration will be applied.
 
 . Run the Kafka Bridge script to enable metrics.
 +
diff --git a/src/main/java/io/strimzi/kafka/bridge/Application.java b/src/main/java/io/strimzi/kafka/bridge/Application.java
index 4619f42a..1c31efa5 100644
--- a/src/main/java/io/strimzi/kafka/bridge/Application.java
+++ b/src/main/java/io/strimzi/kafka/bridge/Application.java
@@ -36,6 +36,7 @@
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
 import java.util.EnumSet;
 import java.util.HashSet;
 import java.util.Map;
@@ -135,7 +136,7 @@ private static MetricsReporter getMetricsReporter(BridgeConfig bridgeConfig)
             throws MalformedObjectNameException, IOException {
         if (bridgeConfig.getMetrics() != null) {
             if (bridgeConfig.getMetrics().equals(MetricsType.JMX_EXPORTER.toString())) {
-                return new MetricsReporter(getJmxCollectorRegistry());
+                return new MetricsReporter(getJmxCollectorRegistry(bridgeConfig));
             } else if (bridgeConfig.getMetrics().equals(MetricsType.STRIMZI_REPORTER.toString())) {
                 return new MetricsReporter(new StrimziCollectorRegistry());
             }
@@ -144,23 +145,32 @@ private static MetricsReporter getMetricsReporter(BridgeConfig bridgeConfig)
     }
 
     /**
-     * Return a JmxCollectorRegistry instance with the YAML configuration filters
+     * Return a JmxCollectorRegistry instance with the YAML configuration filters.
+     * This is loaded from a custom config file if present, or from the default configuration file.
      *
      * @return JmxCollectorRegistry instance
      * @throws MalformedObjectNameException
      * @throws IOException
      */
-    private static JmxCollectorRegistry getJmxCollectorRegistry() throws MalformedObjectNameException, IOException {
-        InputStream is = Application.class.getClassLoader().getResourceAsStream("jmx_metrics_config.yaml");
-        if (is == null) {
-            return null;
-        }
-
-        try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
-            String yaml = reader
+    private static JmxCollectorRegistry getJmxCollectorRegistry(BridgeConfig bridgeConfig) throws MalformedObjectNameException, IOException {
+        if (bridgeConfig.getJmxExporterConfigPath() != null && Files.exists(bridgeConfig.getJmxExporterConfigPath())) {
+            // read custom configuration file
+            LOGGER.info("Loading custom JMX Exporter configuration from {}", bridgeConfig.getJmxExporterConfigPath());
+            String yaml = Files.readString(bridgeConfig.getJmxExporterConfigPath(), StandardCharsets.UTF_8);
+            return new JmxCollectorRegistry(yaml);
+        } else {
+            // fallback to default configuration
+            LOGGER.info("Loading default JMX Exporter configuration");
+            InputStream is = Application.class.getClassLoader().getResourceAsStream("jmx_metrics_config.yaml");
+            if (is == null) {
+                return null;
+            }
+            try (BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) {
+                String yaml = reader
                     .lines()
                     .collect(Collectors.joining("\n"));
-            return new JmxCollectorRegistry(yaml);
+                return new JmxCollectorRegistry(yaml);
+            }
         }
     }
 
diff --git a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
index d8047073..2c913eff 100644
--- a/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
+++ b/src/main/java/io/strimzi/kafka/bridge/config/BridgeConfig.java
@@ -10,6 +10,7 @@
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
+import java.nio.file.Path;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
@@ -28,6 +29,9 @@ public class BridgeConfig extends AbstractConfig {
 
     /** Metrics system to be used in the bridge */
     public static final String METRICS_TYPE = BRIDGE_CONFIG_PREFIX + "metrics";
+
+    /** JMX Exporter configuration file path */
+    public static final String JMX_EXPORTER_CONFIG_PATH = METRICS_TYPE + ".jmx.exporter.config.path";
     
     /** Tracing system to be used in the bridge */
     public static final String TRACING_TYPE = BRIDGE_CONFIG_PREFIX + "tracing";
@@ -124,6 +128,17 @@ public String getMetrics() {
                 ? MetricsType.JMX_EXPORTER.toString() : null);
     }
 
+    /**
+     * @return the JMX Exporter configuration file path
+     */
+    public Path getJmxExporterConfigPath() {
+        if (config.get(BridgeConfig.JMX_EXPORTER_CONFIG_PATH) == null) {
+            return null;
+        } else {
+            return Path.of((String) config.get(BridgeConfig.JMX_EXPORTER_CONFIG_PATH));
+        }
+    }
+
     /**
      * @return the tracing system to be used in the bridge
      */
diff --git a/src/test/java/io/strimzi/kafka/bridge/config/ConfigTest.java b/src/test/java/io/strimzi/kafka/bridge/config/ConfigTest.java
index fe341b7d..abe15656 100644
--- a/src/test/java/io/strimzi/kafka/bridge/config/ConfigTest.java
+++ b/src/test/java/io/strimzi/kafka/bridge/config/ConfigTest.java
@@ -10,6 +10,7 @@
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.junit.jupiter.api.Test;
 
+import java.nio.file.Path;
 import java.util.HashMap;
 import java.util.Map;
 
@@ -86,16 +87,20 @@ public void testHttpDefaults() {
 
     @Test
     public void testJmxExporterMetricsType() {
+        String configFilePah = "/tmp/my-jmx-exporter-config.yaml";
+        
         Map<String, Object> map = Map.of(
             "bridge.id", "my-bridge",
             "kafka.bootstrap.servers", "localhost:9092",
             "bridge.metrics", "jmxPrometheusExporter",
+            "bridge.metrics.jmx.exporter.config.path", configFilePah,
             "http.host", "0.0.0.0",
             "http.port", "8080"
         );
 
         BridgeConfig bridgeConfig = BridgeConfig.fromMap(map);
         assertThat(bridgeConfig.getMetrics(), is(MetricsType.JMX_EXPORTER.toString()));
+        assertThat(bridgeConfig.getJmxExporterConfigPath(), is(Path.of(configFilePah)));
     }
 
     @Test