Add Cluster Operator integration bits

Implementation of "Integrate Bridge with Metrics Reporter" proposal.

Signed-off-by: Federico Valeri <[email protected]>

CHANGELOG.md

@@ -8,7 +8,7 @@
   * The `/openapi/v2` endpoint returns HTTP 410 Gone.
   * Both the `/openapi` and `/openapi/v3` endpoints return the OpenAPI v3 definition of the bridge REST API.
 * 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`.
+  * This is a Kafka plugin that directly 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`.
 
 ### Changes, deprecations and removals

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

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

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.
 +

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);
+            }
         }
     }
 

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
      */

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