Add support for the Strimzi Metrics Reporter to Kafka brokers/controllers components

CHANGELOG.md

@@ -11,6 +11,7 @@
 * Support for MirrorMaker 1 has been removed
 * Support for storage class overrides has been removed
 * Added support to configure `dnsPolicy` and `dnsConfig` using the `template` sections.
+* Added support for Strimzi Metrics Reporter to the Kafka brokers/controllers component.
 * Store Kafka node certificates in separate Secrets, one Secret per pod.
 * Allow configuring `ssl.principal.mapping.rules` and custom trusted CAs in Kafka brokers with `type: custom` authentication
 * Moved HTTP bridge configuration to the ConfigMap setup by the operator.

api/src/main/java/io/strimzi/api/kafka/model/common/metrics/JmxPrometheusExporterMetrics.java

@@ -5,7 +5,6 @@
 package io.strimzi.api.kafka.model.common.metrics;
 
 import com.fasterxml.jackson.annotation.JsonInclude;
-import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 import io.strimzi.api.kafka.model.common.Constants;
 import io.strimzi.api.kafka.model.common.ExternalConfigurationReference;
@@ -15,7 +14,7 @@
 import lombok.ToString;
 
 /**
- * JMX Prometheus Exporter metrics config
+ * Prometheus JMX Exporter metrics config
  */
 @Buildable(
         editableEnabled = false,
@@ -30,8 +29,7 @@ public class JmxPrometheusExporterMetrics extends MetricsConfig {
 
     private ExternalConfigurationReference valueFrom;
 
-    @Description("ConfigMap entry where the Prometheus JMX Exporter configuration is stored. ")
-    @JsonProperty(required = true)
+    @Description("ConfigMap entry where the Prometheus JMX Exporter configuration is stored.")
     public ExternalConfigurationReference getValueFrom() {
         return valueFrom;
     }

api/src/main/java/io/strimzi/api/kafka/model/common/metrics/MetricsConfig.java

@@ -8,6 +8,7 @@
 import com.fasterxml.jackson.annotation.JsonSubTypes;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import io.strimzi.api.kafka.model.common.UnknownPropertyPreserving;
+import io.strimzi.crdgenerator.annotations.CelValidation;
 import io.strimzi.crdgenerator.annotations.Description;
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
@@ -23,14 +24,24 @@
         property = "type")
 @JsonSubTypes({
     @JsonSubTypes.Type(name = JmxPrometheusExporterMetrics.TYPE_JMX_EXPORTER, value = JmxPrometheusExporterMetrics.class),
+    @JsonSubTypes.Type(name = StrimziMetricsReporter.TYPE_STRIMZI_METRICS_REPORTER, value = StrimziMetricsReporter.class)
 })
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @EqualsAndHashCode
 @ToString
+@CelValidation(rules = {
+    @CelValidation.CelValidationRule(
+            rule = "self.type != 'jmxPrometheusExporter' || has(self.valueFrom)",
+            message = "valueFrom property is required"
+        )
+})
 public abstract class MetricsConfig implements UnknownPropertyPreserving {
     private Map<String, Object> additionalProperties;
 
-    @Description("Metrics type. Only 'jmxPrometheusExporter' supported currently.")
+    @Description("Metrics type. " +
+            "The supported types are `jmxPrometheusExporter` and `strimziMetricsReporter`. " +
+            "Type `jmxPrometheusExporter` uses the Prometheus JMX Exporter to expose Kafka JMX metrics in Prometheus format through an HTTP endpoint. " +
+            "Type `strimziMetricsReporter` uses the Strimzi Metrics Reporter to directly expose Kafka metrics in Prometheus format through an HTTP endpoint.")
     public abstract String getType();
 
     @Override

api/src/main/java/io/strimzi/api/kafka/model/common/metrics/StrimziMetricsReporter.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.api.kafka.model.common.metrics;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import io.strimzi.api.kafka.model.common.Constants;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.sundr.builder.annotations.Buildable;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+/**
+ * Strimzi Metrics Reporter.
+ */
+@Buildable(
+        editableEnabled = false,
+        builderPackage = Constants.FABRIC8_KUBERNETES_API
+)
+@JsonPropertyOrder({"type", "values"})
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@EqualsAndHashCode(callSuper = true)
+@ToString(callSuper = true)
+public class StrimziMetricsReporter extends MetricsConfig {
+    public static final String TYPE_STRIMZI_METRICS_REPORTER = "strimziMetricsReporter";
+
+    private StrimziMetricsReporterValues values;
+
+    @Description("Must be `" + TYPE_STRIMZI_METRICS_REPORTER + "`")
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @Override
+    public String getType() {
+        return TYPE_STRIMZI_METRICS_REPORTER;
+    }
+
+    @Description("Configuration values for the Strimzi Metrics Reporter.")
+    public StrimziMetricsReporterValues getValues() {
+        return values;
+    }
+
+    public void setValues(StrimziMetricsReporterValues values) {
+        this.values = values;
+    }
+}

api/src/main/java/io/strimzi/api/kafka/model/common/metrics/StrimziMetricsReporterValues.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright Strimzi authors.
+ * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
+ */
+package io.strimzi.api.kafka.model.common.metrics;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+import io.strimzi.api.kafka.model.common.Constants;
+import io.strimzi.api.kafka.model.common.UnknownPropertyPreserving;
+import io.strimzi.crdgenerator.annotations.Description;
+import io.sundr.builder.annotations.Buildable;
+import lombok.EqualsAndHashCode;
+import lombok.ToString;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Strimzi Metrics Reporter configuration.
+ */
+@Buildable(
+        editableEnabled = false,
+        builderPackage = Constants.FABRIC8_KUBERNETES_API
+)
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@JsonPropertyOrder({"allowList"})
+@EqualsAndHashCode
+@ToString
+public class StrimziMetricsReporterValues implements UnknownPropertyPreserving {
+    private List<String> allowList;
+    private Map<String, Object> additionalProperties;
+
+    @Description("A list of regex patterns to filter the metrics to collect. Should contain at least one element.")
+    @JsonInclude(value = JsonInclude.Include.NON_NULL)
+    public List<String> getAllowList() {
+        return allowList;
+    }
+
+    public void setAllowList(List<String> allowList) {
+        this.allowList = allowList;
+    }
+
+    @Override
+    public Map<String, Object> getAdditionalProperties() {
+        return this.additionalProperties != null ? this.additionalProperties : Map.of();
+    }
+
+    @Override
+    public void setAdditionalProperty(String name, Object value) {
+        if (this.additionalProperties == null) {
+            this.additionalProperties = new HashMap<>(2);
+        }
+        this.additionalProperties.put(name, value);
+    }
+}

api/src/main/java/io/strimzi/api/kafka/model/connect/AbstractKafkaConnectSpec.java

@@ -23,6 +23,7 @@
 import io.strimzi.api.kafka.model.common.jmx.KafkaJmxOptions;
 import io.strimzi.api.kafka.model.common.metrics.MetricsConfig;
 import io.strimzi.api.kafka.model.common.tracing.Tracing;
+import io.strimzi.crdgenerator.annotations.CelValidation;
 import io.strimzi.crdgenerator.annotations.Description;
 import io.strimzi.crdgenerator.annotations.KubeLink;
 import io.sundr.builder.annotations.Buildable;
@@ -155,8 +156,18 @@ public void setJmxOptions(KafkaJmxOptions jmxOptions) {
         this.jmxOptions = jmxOptions;
     }
 
-    @Description("Metrics configuration.")
+    @Description("Metrics configuration. Only `jmxPrometheusExporter` can be configured, as this component does not yet support `strimziMetricsReporter`.")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    @CelValidation(rules = {
+        @CelValidation.CelValidationRule(
+                rule = "self.type != 'jmxPrometheusExporter' || has(self.valueFrom)",
+                message = "valueFrom property is required"
+            ),
+        @CelValidation.CelValidationRule(
+                rule = "self.type != 'strimziMetricsReporter'",
+                message = "value type not supported"
+            )
+    })
     @Override
     public MetricsConfig getMetricsConfig() {
         return metricsConfig;

api/src/main/java/io/strimzi/api/kafka/model/kafka/KafkaClusterSpec.java

@@ -64,7 +64,8 @@ public class KafkaClusterSpec implements HasConfigurableMetrics, HasConfigurable
             + "cruise.control.metrics.topic, cruise.control.metrics.reporter.bootstrap.servers, "
             + "node.id, process.roles, controller., metadata.log.dir, zookeeper.metadata.migration.enable, " // KRaft options
             + "client.quota.callback.static.kafka.admin., client.quota.callback.static.produce, client.quota.callback.static.fetch, "
-            + "client.quota.callback.static.storage.per.volume.limit.min.available., client.quota.callback.static.excluded.principal.name.list";
+            + "client.quota.callback.static.storage.per.volume.limit.min.available., client.quota.callback.static.excluded.principal.name.list, "
+            + "prometheus.metrics.reporter.";
 
     public static final String FORBIDDEN_PREFIX_EXCEPTIONS = "zookeeper.connection.timeout.ms, sasl.server.max.receive.size, "
             + "ssl.cipher.suites, ssl.protocol, ssl.enabled.protocols, ssl.secure.random.implementation, "

api/src/main/java/io/strimzi/api/kafka/model/kafka/cruisecontrol/CruiseControlSpec.java

@@ -19,6 +19,7 @@
 import io.strimzi.api.kafka.model.common.UnknownPropertyPreserving;
 import io.strimzi.api.kafka.model.common.metrics.MetricsConfig;
 import io.strimzi.api.kafka.model.kafka.entityoperator.TlsSidecar;
+import io.strimzi.crdgenerator.annotations.CelValidation;
 import io.strimzi.crdgenerator.annotations.Description;
 import io.strimzi.crdgenerator.annotations.DescriptionFile;
 import io.strimzi.crdgenerator.annotations.KubeLink;
@@ -125,8 +126,18 @@ public void setApiUsers(CruiseControlApiUsers apiUsers) {
         this.apiUsers = apiUsers;
     }
 
-    @Description("Metrics configuration.")
+    @Description("Metrics configuration. Only `jmxPrometheusExporter` can be configured, as this component does not support `strimziMetricsReporter`.")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    @CelValidation(rules = {
+        @CelValidation.CelValidationRule(
+                rule = "self.type != 'jmxPrometheusExporter' || has(self.valueFrom)",
+                message = "valueFrom property is required"
+            ),
+        @CelValidation.CelValidationRule(
+                rule = "self.type != 'strimziMetricsReporter'",
+                message = "value type not supported"
+            )
+        })
     @Override
     public MetricsConfig getMetricsConfig() {
         return metricsConfig;

api/src/test/java/io/strimzi/api/kafka/model/kafka/KafkaCrdIT.java

@@ -74,8 +74,8 @@ public void testKafkaWithNullMaintenance() {
                 () -> createDeleteCustomResource("Kafka-with-null-maintenance.yaml"));
 
         assertThat(exception.getMessage(), anyOf(
-                        containsStringIgnoringCase("invalid: spec.maintenanceTimeWindows: Invalid value: \"null\": spec.maintenanceTimeWindows in body must be of type string: \"null\""),
-                        containsStringIgnoringCase("invalid: spec.maintenanceTimeWindows[0]: Invalid value: \"null\": spec.maintenanceTimeWindows[0] in body must be of type string: \"null\"")));
+                        containsStringIgnoringCase("spec.maintenanceTimeWindows: Invalid value: \"null\": spec.maintenanceTimeWindows in body must be of type string: \"null\""),
+                        containsStringIgnoringCase("spec.maintenanceTimeWindows[0]: Invalid value: \"null\": spec.maintenanceTimeWindows[0] in body must be of type string: \"null\"")));
     }
 
     @Test

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/ConfigMapUtils.java

@@ -8,7 +8,7 @@
 import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
 import io.fabric8.kubernetes.api.model.OwnerReference;
 import io.strimzi.operator.cluster.model.logging.SupportsLogging;
-import io.strimzi.operator.cluster.model.metrics.MetricsModel;
+import io.strimzi.operator.cluster.model.metrics.JmxPrometheusExporterModel;
 import io.strimzi.operator.cluster.model.metrics.SupportsMetrics;
 import io.strimzi.operator.common.Reconciliation;
 import io.strimzi.operator.common.model.Labels;
@@ -67,10 +67,10 @@ public static Map<String, String> generateMetricsAndLogConfigMapData(Reconciliat
             data.put(supportsLogging.logging().configMapKey(), supportsLogging.logging().loggingConfiguration(reconciliation, metricsAndLogging.loggingCm()));
         }
 
-        if (model instanceof SupportsMetrics supportMetrics) {
-            String parseResult = supportMetrics.metrics().metricsJson(reconciliation, metricsAndLogging.metricsCm());
+        if (model instanceof SupportsMetrics supportsMetrics && supportsMetrics.metrics() instanceof JmxPrometheusExporterModel jmxMetrics) {
+            String parseResult = jmxMetrics.metricsJson(reconciliation, metricsAndLogging.metricsCm());
             if (parseResult != null) {
-                data.put(MetricsModel.CONFIG_MAP_KEY, parseResult);
+                data.put(JmxPrometheusExporterModel.CONFIG_MAP_KEY, parseResult);
             }
         }