[Bug] MM2 should have its own default Strimzi Metrics configuration (#12180) (#12277)

Signed-off-by: Azeez Syed <syedazeez337@gmail.com>

Author: syedazeez337

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

@@ -92,18 +92,16 @@
 @SuppressWarnings({"checkstyle:ClassFanOutComplexity"})
 public class KafkaConnectCluster extends AbstractModel implements SupportsMetrics, SupportsLogging, SupportsJmx {
     /**
-     * Default Strimzi Metrics Reporter allowlist.
+     * Default Strimzi Metrics Reporter allowlist for Kafka Connect.
      * Check example dashboard compatibility in case of changes to existing regexes.
      */
-    private static final List<String> DEFAULT_METRICS_ALLOW_LIST = List.of(
+    protected static final List<String> DEFAULT_METRICS_ALLOW_LIST = List.of(
             "kafka_admin_client_admin_client_metrics_connection_count",
             "kafka_connect_connect_coordinator_metrics.*",
             "kafka_connect_connector_metrics.*",
             "kafka_connect_connector_task_metrics.*",
             "kafka_connect_connect_worker_metrics_.*",
             "kafka_connect_connect_worker_rebalance.*",
-            "kafka_connect_mirror_mirrorcheckpointconnector.*",
-            "kafka_connect_mirror_mirrorsourceconnector.*",
             "kafka_connect_task_error_metrics.*",
             "kafka_consumer_consumer_coordinator_metrics.*",
             "kafka_consumer_consumer_fetch_manager_metrics.*",
@@ -113,6 +111,16 @@ public class KafkaConnectCluster extends AbstractModel implements SupportsMetric
             "kafka_producer_producer_topic.*"
     );
 
+    /**
+     * Returns the default metrics allow list for this component.
+     * Can be overridden by subclasses to provide component-specific defaults.
+     *
+     * @return List of default metrics allow list patterns
+     */
+    protected List<String> getDefaultMetricsAllowList() {
+        return DEFAULT_METRICS_ALLOW_LIST;
+    }
+
     /**
      * Port of the Kafka Connect REST API
      */
@@ -287,7 +295,7 @@ protected static <C extends KafkaConnectCluster> C fromSpec(Reconciliation recon
         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);
+            result.metrics = new StrimziMetricsReporterModel(spec, result.getDefaultMetricsAllowList());
         }
 
         result.logging = new LoggingModel(spec, result.getClass().getSimpleName());

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

@@ -28,6 +28,7 @@
 
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -65,6 +66,19 @@ public class KafkaMirrorMaker2Cluster extends KafkaConnectCluster {
         }
     }
 
+    /**
+     * Default Strimzi Metrics Reporter allowlist for MirrorMaker 2.
+     * Inherits from Kafka Connect metrics and adds MM2-specific metrics.
+     * Check example dashboard compatibility in case of changes to existing regexes.
+     */
+    private static final List<String> DEFAULT_METRICS_ALLOW_LIST;
+    static {
+        List<String> list = new ArrayList<>(KafkaConnectCluster.DEFAULT_METRICS_ALLOW_LIST);
+        list.add("kafka_connect_mirror_mirrorcheckpointconnector.*");
+        list.add("kafka_connect_mirror_mirrorsourceconnector.*");
+        DEFAULT_METRICS_ALLOW_LIST = Collections.unmodifiableList(list);
+    }
+
     private KafkaMirrorMaker2Connectors connectors;
     private Collection<KafkaMirrorMaker2ClusterSpec> clusters;
 
@@ -450,4 +464,15 @@ public KafkaMirrorMaker2Connectors connectors() {
     public Collection<KafkaMirrorMaker2ClusterSpec> clusters() {
         return clusters;
     }
+
+    /**
+     * Override the default metrics allow list to include MM2-specific metrics.
+     * MirrorMaker 2 needs both Kafka Connect metrics and MM2-specific metrics.
+     *
+     * @return List of default metrics allow list patterns for MirrorMaker 2
+     */
+    @Override
+    protected List<String> getDefaultMetricsAllowList() {
+        return DEFAULT_METRICS_ALLOW_LIST;
+    }
 }

cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaConnectClusterTest.java

@@ -2418,4 +2418,29 @@ public void testRoleAbdRoleBindingWithSecrets() {
         assertThat(rb.getRoleRef().getKind(), is("Role"));
         assertThat(rb.getRoleRef().getName(), is(kc.componentName));
     }
+
+    @Test
+    public void testDefaultMetricsConfigurationDoesNotIncludeMM2Metrics() {
+        // Test that Kafka Connect default metrics do not include MM2-specific metrics
+        MetricsConfig metrics = new StrimziMetricsReporterBuilder().build();
+
+        KafkaConnect kafkaConnect = new KafkaConnectBuilder(RESOURCE)
+                .editSpec()
+                    .withMetricsConfig(metrics)
+                .endSpec()
+                .build();
+
+        KafkaConnectCluster kc = KafkaConnectCluster.fromCrd(Reconciliation.DUMMY_RECONCILIATION, kafkaConnect, VERSIONS, SHARED_ENV_PROVIDER);
+
+        assertThat(kc.metrics(), is(notNullValue()));
+        String allowList = ((StrimziMetricsReporterModel) kc.metrics()).getAllowList();
+        
+        // Verify MM2 metrics are NOT included in Kafka Connect default configuration
+        assertThat(allowList, not(containsString("mirrorcheckpointconnector")));
+        assertThat(allowList, not(containsString("mirrorsourceconnector")));
+        
+        // Verify Connect metrics ARE included
+        assertThat(allowList, containsString("kafka_connect_connector_metrics"));
+        assertThat(allowList, containsString("kafka_connect_connect_worker_metrics_"));
+    }
 }

cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaMirrorMaker2ClusterTest.java

@@ -2547,4 +2547,29 @@ public void testRoleAbdRoleBindingWithSecrets() {
         assertThat(rb.getRoleRef().getKind(), is("Role"));
         assertThat(rb.getRoleRef().getName(), is(kmm2.componentName));
     }
+
+    @Test
+    public void testDefaultMetricsConfigurationIncludesBothConnectAndMM2Metrics() {
+        // Test that MirrorMaker 2 default metrics include both Connect and MM2-specific metrics
+        MetricsConfig metrics = new StrimziMetricsReporterBuilder().build();
+
+        KafkaMirrorMaker2 kafkaMirrorMaker2 = new KafkaMirrorMaker2Builder(RESOURCE)
+                .editSpec()
+                    .withMetricsConfig(metrics)
+                .endSpec()
+                .build();
+
+        KafkaMirrorMaker2Cluster kmm2 = KafkaMirrorMaker2Cluster.fromCrd(Reconciliation.DUMMY_RECONCILIATION, kafkaMirrorMaker2, VERSIONS, SHARED_ENV_PROVIDER);
+
+        assertThat(kmm2.metrics(), is(notNullValue()));
+        String allowList = ((StrimziMetricsReporterModel) kmm2.metrics()).getAllowList();
+        
+        // Verify MM2 metrics ARE included in MirrorMaker 2 default configuration
+        assertThat(allowList, containsString("mirrorcheckpointconnector"));
+        assertThat(allowList, containsString("mirrorsourceconnector"));
+        
+        // Verify Connect metrics ARE also included
+        assertThat(allowList, containsString("kafka_connect_connector_metrics"));
+        assertThat(allowList, containsString("kafka_connect_connect_worker_metrics_"));
+    }
 }