Add support for Strimzi Metrics Reporter to the Strimzi Kafka Bridge (strimzi#11708)

Signed-off-by: OwenCorrigan76 <owencorrigan76@gmail.com>
Signed-off-by: Federico Valeri <fedevaleri@gmail.com>
Co-authored-by: Federico Valeri <fedevaleri@gmail.com>

Author: 2 people

CHANGELOG.md

@@ -6,6 +6,7 @@
   The `strimzi.io/node-pools` and `strimzi.io/kraft` annotations are not required anymore and will be ignored if set.
 * Make properties `broker.session.timeout.ms`, `broker.heartbeat.interval.ms` and `controller.socket.timeout.ms` configurable
 * Add monitoring of custom resources using [kubernetes-state-metrics (KSM)](https://github.com/kubernetes/kube-state-metrics) (see [Strimzi proposal 087](https://github.com/strimzi/proposals/blob/main/087-monitoring-of-custom-resources.md))
+* Added support for Strimzi Metrics Reporter to Kafka Connect, Mirror Maker 2 and Kafka Bridge.
 
 ### Major changes, deprecations and removals
 
@@ -14,6 +15,7 @@
 * Disable Cruise Control network resource goals when resource capacities are not set.
 * The `strimzi_resource_state` metric in the Cluster Operator is deprecated and is planned to be removed in Strimzi 0.51.
   Use kube-state-metrics based metrics from the [examples](https://github.com/strimzi/strimzi-kafka-operator/tree/main/examples/metrics/kube-state-metrics) as a replacement.
+* The field `.spec.enableMetrics` in KafkaBridge is now deprecated and replaced by `.spec.metricsConfig`.
 
 ## 0.47.0
 

api/src/main/java/io/strimzi/api/kafka/model/bridge/KafkaBridgeSpec.java

@@ -8,9 +8,11 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 import io.fabric8.kubernetes.api.model.ResourceRequirements;
+import io.strimzi.api.annotations.DeprecatedProperty;
 import io.strimzi.api.kafka.model.common.ClientTls;
 import io.strimzi.api.kafka.model.common.Constants;
 import io.strimzi.api.kafka.model.common.HasConfigurableLogging;
+import io.strimzi.api.kafka.model.common.HasConfigurableMetrics;
 import io.strimzi.api.kafka.model.common.HasLivenessProbe;
 import io.strimzi.api.kafka.model.common.HasReadinessProbe;
 import io.strimzi.api.kafka.model.common.JvmOptions;
@@ -19,11 +21,13 @@
 import io.strimzi.api.kafka.model.common.Rack;
 import io.strimzi.api.kafka.model.common.Spec;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthentication;
+import io.strimzi.api.kafka.model.common.metrics.MetricsConfig;
 import io.strimzi.api.kafka.model.common.tracing.Tracing;
 import io.strimzi.crdgenerator.annotations.Description;
 import io.strimzi.crdgenerator.annotations.DescriptionFile;
 import io.strimzi.crdgenerator.annotations.KubeLink;
 import io.strimzi.crdgenerator.annotations.Minimum;
+import io.strimzi.crdgenerator.annotations.PresentInVersions;
 import io.sundr.builder.annotations.Buildable;
 import lombok.EqualsAndHashCode;
 import lombok.ToString;
@@ -37,10 +41,10 @@
 @JsonPropertyOrder({
     "replicas", "image", "bootstrapServers", "tls", "authentication", "http", "adminClient", "consumer",
     "producer", "resources", "jvmOptions", "logging", "clientRackInitImage", "rack",
-    "enableMetrics", "livenessProbe", "readinessProbe", "template", "tracing"})
+    "enableMetrics", "metricsConfig", "livenessProbe", "readinessProbe", "template", "tracing"})
 @EqualsAndHashCode(callSuper = true)
 @ToString(callSuper = true)
-public class KafkaBridgeSpec extends Spec implements HasConfigurableLogging, HasLivenessProbe, HasReadinessProbe {
+public class KafkaBridgeSpec extends Spec implements HasConfigurableLogging, HasConfigurableMetrics, HasLivenessProbe, HasReadinessProbe {
     private static final int DEFAULT_REPLICAS = 1;
 
     private int replicas = DEFAULT_REPLICAS;
@@ -57,6 +61,7 @@ public class KafkaBridgeSpec extends Spec implements HasConfigurableLogging, Has
     private JvmOptions jvmOptions;
     private Logging logging;
     private boolean enableMetrics;
+    private MetricsConfig metricsConfig;
     private Probe livenessProbe;
     private Probe readinessProbe;
     private KafkaBridgeTemplate template;
@@ -76,6 +81,10 @@ public void setReplicas(int replicas) {
         this.replicas = replicas;
     }
 
+    @Deprecated
+    @DeprecatedProperty(movedToPath = ".spec.metricsConfig",
+            description = "The `enableMetrics` configuration is deprecated and will be removed in the future.")
+    @PresentInVersions("v1alpha1-v1beta2")
     @JsonInclude(JsonInclude.Include.NON_NULL)
     @Description("Enable the metrics for the Kafka Bridge. Default is false.")
     public boolean getEnableMetrics() {
@@ -86,6 +95,18 @@ public void setEnableMetrics(boolean enableMetrics) {
         this.enableMetrics = enableMetrics;
     }
 
+    @Description("Metrics configuration.")
+    @JsonInclude(JsonInclude.Include.NON_EMPTY)
+    @Override
+    public MetricsConfig getMetricsConfig() {
+        return metricsConfig;
+    }
+
+    @Override
+    public void setMetricsConfig(MetricsConfig metricsConfig) {
+        this.metricsConfig = metricsConfig;
+    }
+
     @Description("Logging configuration for Kafka Bridge.")
     @JsonInclude(value = JsonInclude.Include.NON_NULL)
     @Override

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

@@ -35,6 +35,8 @@
 import io.strimzi.api.kafka.model.common.JvmOptions;
 import io.strimzi.api.kafka.model.common.Rack;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthentication;
+import io.strimzi.api.kafka.model.common.metrics.JmxPrometheusExporterMetrics;
+import io.strimzi.api.kafka.model.common.metrics.StrimziMetricsReporter;
 import io.strimzi.api.kafka.model.common.template.ContainerTemplate;
 import io.strimzi.api.kafka.model.common.template.DeploymentStrategy;
 import io.strimzi.api.kafka.model.common.template.DeploymentTemplate;
@@ -46,6 +48,10 @@
 import io.strimzi.operator.cluster.ClusterOperatorConfig;
 import io.strimzi.operator.cluster.model.logging.LoggingModel;
 import io.strimzi.operator.cluster.model.logging.SupportsLogging;
+import io.strimzi.operator.cluster.model.metrics.JmxPrometheusExporterModel;
+import io.strimzi.operator.cluster.model.metrics.MetricsModel;
+import io.strimzi.operator.cluster.model.metrics.StrimziMetricsReporterModel;
+import io.strimzi.operator.cluster.model.metrics.SupportsMetrics;
 import io.strimzi.operator.cluster.model.securityprofiles.ContainerSecurityProviderContextImpl;
 import io.strimzi.operator.cluster.model.securityprofiles.PodSecurityProviderContextImpl;
 import io.strimzi.operator.common.Reconciliation;
@@ -65,7 +71,17 @@
  * Kafka Bridge model class
  */
 @SuppressWarnings({"checkstyle:ClassFanOutComplexity"})
-public class KafkaBridgeCluster extends AbstractModel implements SupportsLogging {
+public class KafkaBridgeCluster extends AbstractModel implements SupportsLogging, SupportsMetrics {
+    /**
+     * Default Strimzi Metrics Reporter allow list.
+     * Check example dashboards compatibility in case of changes to existing regexes.
+     */
+    private static final List<String> DEFAULT_METRICS_ALLOW_LIST = List.of(
+            "kafka_consumer_consumer_metrics.*",
+            "kafka_producer_kafka_metrics_count_count",
+            "kafka_producer_producer_metrics.*"
+    );
+
     /**
      * HTTP port configuration
      */
@@ -77,19 +93,17 @@ public class KafkaBridgeCluster extends AbstractModel implements SupportsLogging
     protected static final String PASSWORD_VOLUME_MOUNT = "/opt/strimzi/bridge-password/";
     protected static final String ENV_VAR_KAFKA_INIT_INIT_FOLDER_KEY = "INIT_FOLDER";
     private static final String KAFKA_BRIDGE_CONFIG_VOLUME_NAME = "kafka-bridge-configurations";
-    private static final String KAFKA_BRIDGE_CONFIG_VOLUME_MOUNT = "/opt/strimzi/custom-config/";
 
     // Cluster Operator environment variables for custom discovery labels and annotations
     protected static final String CO_ENV_VAR_CUSTOM_SERVICE_LABELS = "STRIMZI_CUSTOM_KAFKA_BRIDGE_SERVICE_LABELS";
     protected static final String CO_ENV_VAR_CUSTOM_SERVICE_ANNOTATIONS = "STRIMZI_CUSTOM_KAFKA_BRIDGE_SERVICE_ANNOTATIONS";
 
     // Kafka Bridge configuration keys (EnvVariables)
     protected static final String ENV_VAR_PREFIX = "KAFKA_BRIDGE_";
-    protected static final String ENV_VAR_KAFKA_BRIDGE_METRICS_ENABLED = "KAFKA_BRIDGE_METRICS_ENABLED";
     protected static final String ENV_VAR_KAFKA_BRIDGE_TRUSTED_CERTS = "KAFKA_BRIDGE_TRUSTED_CERTS";
     protected static final String OAUTH_TLS_CERTS_BASE_VOLUME_MOUNT = "/opt/strimzi/oauth-certs/";
     protected static final String OAUTH_SECRETS_BASE_VOLUME_MOUNT = "/opt/strimzi/oauth/";
-
+    protected static final String KAFKA_BRIDGE_CONFIG_VOLUME_MOUNT = "/opt/strimzi/custom-config/";
     protected static final String CO_ENV_VAR_CUSTOM_BRIDGE_POD_LABELS = "STRIMZI_CUSTOM_KAFKA_BRIDGE_LABELS";
     protected static final String INIT_VOLUME_MOUNT = "/opt/strimzi/init";
 
@@ -106,8 +120,9 @@ public class KafkaBridgeCluster extends AbstractModel implements SupportsLogging
     private KafkaBridgeAdminClientSpec kafkaBridgeAdminClient;
     private KafkaBridgeConsumerSpec kafkaBridgeConsumer;
     private KafkaBridgeProducerSpec kafkaBridgeProducer;
-    private boolean isMetricsEnabled = false;
+    private boolean isLegacyMetricsConfigEnabled = false;
     private LoggingModel logging;
+    private MetricsModel metrics;
 
     // Templates
     private PodDisruptionBudgetTemplate templatePodDisruptionBudget;
@@ -151,7 +166,7 @@ private KafkaBridgeCluster(Reconciliation reconciliation, HasMetadata resource,
      * @param sharedEnvironmentProvider Shared environment provider
      * @return KafkaBridgeCluster instance
      */
-    @SuppressWarnings({"checkstyle:NPathComplexity"})
+    @SuppressWarnings({"checkstyle:NPathComplexity", "deprecation"})
     public static KafkaBridgeCluster fromCrd(Reconciliation reconciliation,
                                              KafkaBridge kafkaBridge,
                                              SharedEnvironmentProvider sharedEnvironmentProvider) {
@@ -182,7 +197,14 @@ public static KafkaBridgeCluster fromCrd(Reconciliation reconciliation,
 
         result.readinessProbeOptions = ProbeUtils.extractReadinessProbeOptionsOrDefault(spec, ProbeUtils.DEFAULT_HEALTHCHECK_OPTIONS);
         result.livenessProbeOptions = ProbeUtils.extractLivenessProbeOptionsOrDefault(spec, ProbeUtils.DEFAULT_HEALTHCHECK_OPTIONS);
-        result.isMetricsEnabled = spec.getEnableMetrics();
+
+        if (spec.getMetricsConfig() instanceof JmxPrometheusExporterMetrics) {
+            result.metrics = new JmxPrometheusExporterModel(spec);
+        } else if (spec.getMetricsConfig() instanceof StrimziMetricsReporter) {
+            result.metrics = new StrimziMetricsReporterModel(spec, DEFAULT_METRICS_ALLOW_LIST);
+        } else {
+            result.isLegacyMetricsConfigEnabled = spec.getEnableMetrics();
+        }
 
         result.setTls(spec.getTls() != null ? spec.getTls() : null);
 
@@ -312,7 +334,7 @@ protected List<VolumeMount> getVolumeMounts() {
 
         return volumeMountList;
     }
-    
+
     private List<VolumeMount> getInitContainerVolumeMounts() {
         List<VolumeMount> volumeMountList = new ArrayList<>();
         volumeMountList.add(VolumeUtils.createVolumeMount(INIT_VOLUME_NAME, INIT_VOLUME_MOUNT));
@@ -396,7 +418,6 @@ private Container createContainer(ImagePullPolicy imagePullPolicy) {
 
     protected List<EnvVar> getEnvVars() {
         List<EnvVar> varList = new ArrayList<>();
-        varList.add(ContainerUtils.createEnvVar(ENV_VAR_KAFKA_BRIDGE_METRICS_ENABLED, String.valueOf(isMetricsEnabled)));
         varList.add(ContainerUtils.createEnvVar(ENV_VAR_STRIMZI_GC_LOG_ENABLED, String.valueOf(gcLoggingEnabled)));
         JvmOptionUtils.javaOptions(varList, jvmOptions);
 
@@ -555,18 +576,23 @@ public ConfigMap generateBridgeConfigMap(MetricsAndLogging metricsAndLogging) {
         // generate the ConfigMap data entries for the metrics and logging configuration
         Map<String, String> data = ConfigMapUtils.generateMetricsAndLogConfigMapData(reconciliation, this, metricsAndLogging);
         // add the ConfigMap data entry for the bridge HTTP and Kafka clients related configuration
-        data.put(
-                BRIDGE_CONFIGURATION_FILENAME,
+        KafkaBridgeConfigurationBuilder builder =
                 new KafkaBridgeConfigurationBuilder(reconciliation, cluster, bootstrapServers)
                         .withTracing(tracing)
                         .withTls(tls)
                         .withAuthentication(authentication)
                         .withKafkaAdminClient(kafkaBridgeAdminClient)
                         .withKafkaProducer(kafkaBridgeProducer)
                         .withKafkaConsumer(kafkaBridgeConsumer)
-                        .withHttp(http, kafkaBridgeProducer, kafkaBridgeConsumer)
-                        .build()
-        );
+                        .withHttp(http, kafkaBridgeProducer, kafkaBridgeConsumer);
+
+        if ((metrics instanceof JmxPrometheusExporterModel) || isLegacyMetricsConfigEnabled) {
+            builder.withJmxPrometheusExporter((JmxPrometheusExporterModel) metrics, isLegacyMetricsConfigEnabled);
+        } else if (metrics instanceof StrimziMetricsReporterModel) {
+            builder.withStrimziMetricsReporter((StrimziMetricsReporterModel) metrics);
+        }
+
+        data.put(BRIDGE_CONFIGURATION_FILENAME, builder.build());
 
         return ConfigMapUtils
                 .createConfigMap(
@@ -591,4 +617,11 @@ public int getReplicas() {
     public LoggingModel logging()   {
         return logging;
     }
+
+    /**
+     * @return Metrics Model instance for configuring Prometheus metrics
+     */
+    public MetricsModel metrics() {
+        return metrics;
+    }
 }

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

@@ -18,13 +18,19 @@
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationScramSha256;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationScramSha512;
 import io.strimzi.api.kafka.model.common.authentication.KafkaClientAuthenticationTls;
+import io.strimzi.api.kafka.model.common.metrics.JmxPrometheusExporterMetrics;
 import io.strimzi.api.kafka.model.common.tracing.Tracing;
+import io.strimzi.operator.cluster.model.metrics.JmxPrometheusExporterModel;
+import io.strimzi.operator.cluster.model.metrics.StrimziMetricsReporterConfig;
+import io.strimzi.operator.cluster.model.metrics.StrimziMetricsReporterModel;
 import io.strimzi.operator.common.Reconciliation;
 
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.stream.Collectors;
 
+import static io.strimzi.api.kafka.model.common.metrics.StrimziMetricsReporter.TYPE_STRIMZI_METRICS_REPORTER;
+import static io.strimzi.operator.cluster.model.KafkaBridgeCluster.KAFKA_BRIDGE_CONFIG_VOLUME_MOUNT;
 import static io.strimzi.operator.cluster.model.KafkaBridgeCluster.OAUTH_SECRETS_BASE_VOLUME_MOUNT;
 
 /**
@@ -346,6 +352,53 @@ public KafkaBridgeConfigurationBuilder withHttp(KafkaBridgeHttpConfig http, Kafk
         return this;
     }
 
+    /**
+     * Configures the Strimzi Metrics Reporter. It is set only if user enables Strimzi Metrics Reporter.
+     *
+     * @param model     Strimzi Metrics Reporter configuration
+     *
+     * @return Returns the builder instance
+     */
+    public KafkaBridgeConfigurationBuilder withStrimziMetricsReporter(StrimziMetricsReporterModel model)   {
+        if (model != null) {
+            printSectionHeader("Strimzi Metrics Reporter configuration");
+            writer.println("bridge.metrics=" + TYPE_STRIMZI_METRICS_REPORTER);
+            // the kafka. prefix is required by the Bridge to pass Kafka client configurations
+            writer.println("kafka.metric.reporters=" + StrimziMetricsReporterConfig.KAFKA_CLASS);
+            writer.println("kafka." + StrimziMetricsReporterConfig.LISTENER_ENABLE + "=false");
+            writer.println("kafka." + StrimziMetricsReporterConfig.ALLOW_LIST + "=" + model.getAllowList());
+            writer.println();
+        }
+        return this;
+    }
+
+    /**
+     * Configures the JMX Prometheus Metrics Exporter.
+     *
+     * @param model JMX Prometheus Metrics Exporter configuration
+     * @param isLegacyMetricsConfigEnabled Flag which indicates whether the metrics are enabled or not in legacy mode.
+     *
+     * @return Returns the builder instance
+     */
+    public KafkaBridgeConfigurationBuilder withJmxPrometheusExporter(
+            JmxPrometheusExporterModel model, boolean isLegacyMetricsConfigEnabled) {
+        if (model != null || isLegacyMetricsConfigEnabled) {
+            printSectionHeader("Prometheus JMX Exporter configuration");
+            writer.println("bridge.metrics=" + JmxPrometheusExporterMetrics.TYPE_JMX_EXPORTER);
+
+            // if isLegacyMetricsConfigEnabled is not used, we pass the path of the config file.
+            // If it is used, the Bridge will use the fallback config.
+            if (!isLegacyMetricsConfigEnabled) {
+                writer.println("bridge.metrics.exporter.config.path="
+                        + KAFKA_BRIDGE_CONFIG_VOLUME_MOUNT + JmxPrometheusExporterModel.CONFIG_MAP_KEY);
+            }
+
+            writer.println();
+        }
+
+        return this;
+    }
+
     /**
      * Prints the file header which is on the beginning of the configuration file.
      */

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaBridgeAssemblyOperator.java

@@ -103,7 +103,7 @@ protected Future<KafkaBridgeStatus> createOrUpdate(Reconciliation reconciliation
             .compose(i -> bridgeInitClusterRoleBinding(reconciliation, initCrbName, initCrb))
             .compose(i -> deploymentOperations.scaleDown(reconciliation, namespace, bridge.getComponentName(), bridge.getReplicas(), operationTimeoutMs))
             .compose(scale -> serviceOperations.reconcile(reconciliation, namespace, KafkaBridgeResources.serviceName(bridge.getCluster()), bridge.generateService()))
-            .compose(i -> MetricsAndLoggingUtils.metricsAndLogging(reconciliation, configMapOperations, bridge.logging(), null))
+            .compose(i -> MetricsAndLoggingUtils.metricsAndLogging(reconciliation, configMapOperations, bridge.logging(), bridge.metrics()))
             .compose(metricsAndLogging -> {
                 ConfigMap configMap = bridge.generateBridgeConfigMap(metricsAndLogging);
                 podAnnotations.put(Annotations.ANNO_STRIMZI_IO_CONFIGURATION_HASH, Util.hashStub(configMap.getData().get(KafkaBridgeCluster.BRIDGE_CONFIGURATION_FILENAME)));