add changelog, ST, unit test and documentation changes

Signed-off-by: Lukas Kral <lukywill16@gmail.com>

Author: im-konge

CHANGELOG.md

@@ -8,6 +8,7 @@
 * Allow upgrading from unknown Apache Kafka versions (that might be used in Strimzi patch releases)
 * Add support for Kafka 4.1.2
 * Remove PreferredLeaderElectionGoal from Cruise Control's default.goals list
+* Enable configuring `allowList` of Strimzi Metrics Reporter dynamically
 
 ### Major changes, deprecations, and removals
 

cluster-operator/src/test/java/io/strimzi/operator/cluster/operator/resource/KafkaConfigurationDiffTest.java

@@ -396,6 +396,15 @@ public void testPropertiesAllScopesChanged() {
         assertThat(kcd.getConfigDiff(Scope.READ_ONLY).size(), is(1));
     }
 
+    @Test
+    public void testPrometheusMetricsReporterAllowList() {
+        List<ConfigEntry> ces = singletonList(new ConfigEntry("prometheus.metrics.reporter.allowlist", "kafka_controller.*"));
+        KafkaConfigurationDiff kcd = new KafkaConfigurationDiff(Reconciliation.DUMMY_RECONCILIATION, getCurrentConfiguration(ces),
+            getDesiredConfiguration(ces), kafkaVersion, brokerNodeRef, false, true);
+        assertThat(kcd.getDiffSize(), is(0));
+        assertThat(kcd.canBeUpdatedDynamically(), is(true));
+    }
+
     @Test
     public void testAreDoublesEqual() {
         assertThat(KafkaConfigurationDiff.areDoublesEqual("test.option", Map.of(), Map.of("test.option", "0.8")), is(false));

config-model-generator/src/main/java/io/strimzi/build/kafka/metadata/KafkaConfigModelGenerator.java

@@ -59,7 +59,7 @@ private KafkaConfigModelGenerator() {
     public static void main(String[] args) throws Exception {
         String version = kafkaVersion();
         Map<String, ConfigModel> configs = configs(version);
-        addCustomPluginConfig(configs);
+        addPrometheusMetricsReporterAllowListConfig(configs);
 
         ObjectMapper mapper = JsonMapper.builder().enable(SerializationFeature.INDENT_OUTPUT).enable(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY).build();
         ConfigModels root = new ConfigModels();
@@ -135,7 +135,13 @@ private static Map<String, ConfigModel> configs(String version) throws NoSuchMet
         return result;
     }
 
-    private static void addCustomPluginConfig(Map<String, ConfigModel> configs) {
+    /**
+     * Adds `prometheus.metrics.reporter.allowlist` with its configuration to the ConfigModel,
+     * so we are able to update it dynamically in Kafka brokers/controllers.
+     *
+     * @param configs   config models of the Kafka version that should be updated with another config model
+     */
+    private static void addPrometheusMetricsReporterAllowListConfig(Map<String, ConfigModel> configs) {
         String prometheusMetricAllowListName = "prometheus.metrics.reporter.allowlist";
         ConfigModel prometheusMetricConfigModel = new ConfigModel();
         prometheusMetricConfigModel.setScope(Scope.CLUSTER_WIDE);

documentation/modules/con-common-configuration-properties.adoc

@@ -557,6 +557,7 @@ The Strimzi Metrics Reporter offers a lightweight solution for exposing Kafka me
 
 To enable Strimzi Metrics Reporter, set the type to `strimziMetricsReporter`.
 The `allowList` configuration is a comma-separated list of regex patterns to filter the metrics that are collected. This defaults to `.*`,  which allows all metrics.
+Changes to this field will not trigger rolling update, the configuration is dynamically updated for Kafka brokers and controllers.
 
 NOTE: Using `strimziMetricsReporter` is only supported in the Kafka brokers and controllers at the moment.
 

systemtest/src/main/java/io/strimzi/systemtest/utils/specific/MetricsUtils.java

@@ -183,12 +183,12 @@ public static void assertContainsMetric(List<Metric> metrics, String metricName)
         assertThat(String.format("metric '%s' is not present in the list of metrics", metricName), containsMetric, is(true));
     }
 
-    private static List<Double> createPatternAndCollect(BaseMetricsCollector collector, String metric) {
+    public static List<Double> createPatternAndCollect(BaseMetricsCollector collector, String metric) {
         Pattern pattern = Pattern.compile(metric + " ([\\d.^\\n]+)", Pattern.CASE_INSENSITIVE);
         return collector.waitForSpecificMetricAndCollect(pattern);
     }
 
-    private static List<Double> createPatternAndCollectWithoutWait(BaseMetricsCollector collector, String metric) {
+    public static List<Double> createPatternAndCollectWithoutWait(BaseMetricsCollector collector, String metric) {
         Pattern pattern = Pattern.compile(metric + " ([\\d.^\\n]+)", Pattern.CASE_INSENSITIVE);
         return collector.collectSpecificMetric(pattern);
     }

systemtest/src/test/java/io/strimzi/systemtest/metrics/StrimziMetricsReporterST.java

@@ -11,15 +11,18 @@
 import io.skodjob.annotations.TestDoc;
 import io.skodjob.kubetest4j.resources.KubeResourceManager;
 import io.strimzi.api.kafka.model.bridge.KafkaBridgeResources;
+import io.strimzi.api.kafka.model.common.metrics.StrimziMetricsReporter;
 import io.strimzi.api.kafka.model.kafka.KafkaResources;
 import io.strimzi.operator.common.Annotations;
 import io.strimzi.systemtest.AbstractST;
 import io.strimzi.systemtest.Environment;
 import io.strimzi.systemtest.TestConstants;
+import io.strimzi.systemtest.annotations.IsolatedTest;
 import io.strimzi.systemtest.annotations.ParallelTest;
 import io.strimzi.systemtest.docs.TestDocsLabels;
 import io.strimzi.systemtest.kafkaclients.internalClients.BridgeClients;
 import io.strimzi.systemtest.kafkaclients.internalClients.BridgeClientsBuilder;
+import io.strimzi.systemtest.labels.LabelSelectors;
 import io.strimzi.systemtest.performance.gather.collectors.BaseMetricsCollector;
 import io.strimzi.systemtest.resources.operator.SetupClusterOperator;
 import io.strimzi.systemtest.storage.TestStorage;
@@ -31,12 +34,18 @@
 import io.strimzi.systemtest.templates.crd.KafkaTemplates;
 import io.strimzi.systemtest.templates.crd.KafkaTopicTemplates;
 import io.strimzi.systemtest.templates.specific.ScraperTemplates;
+import io.strimzi.systemtest.utils.RollingUpdateUtils;
+import io.strimzi.systemtest.utils.kafkaUtils.KafkaUtils;
 import io.strimzi.systemtest.utils.kubeUtils.objects.NetworkPolicyUtils;
+import io.strimzi.systemtest.utils.kubeUtils.objects.PodUtils;
+import io.strimzi.systemtest.utils.specific.MetricsUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Tag;
 
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.LockSupport;
 
@@ -51,6 +60,8 @@
 import static io.strimzi.systemtest.utils.specific.MetricsUtils.assertMetricValue;
 import static io.strimzi.systemtest.utils.specific.MetricsUtils.assertMetricValueHigherThanOrEqualTo;
 import static io.strimzi.systemtest.utils.specific.MetricsUtils.assertMetricValueNotNull;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assumptions.assumeFalse;
 
 @Tag(SANITY)
@@ -271,6 +282,43 @@ void testKafkaBridgeMetrics() {
         assertMetricValueNotNull(bridgeCollector, "kafka_consumer_consumer_metrics_connection_count\\{.*}");
     }
 
+    @IsolatedTest
+    void testDynamicReconfigurationAllowList() {
+        Map<String, String> kafkaPods = PodUtils.podSnapshot(testStorage.getNamespaceName(), LabelSelectors.allKafkaPodsLabelSelector(testStorage.getClusterName()));
+
+        // change configuration to disable kafka_server.* metrics and add kafka_controller.* metrics
+        KafkaUtils.replace(testStorage.getNamespaceName(), testStorage.getClusterName(), kafka -> {
+                StrimziMetricsReporter config = (StrimziMetricsReporter) kafka.getSpec().getKafka().getMetricsConfig();
+                config.getValues().setAllowList(List.of("kafka_controller.*"));
+
+                kafka.getSpec().getKafka().setMetricsConfig(config);
+            }
+        );
+
+        RollingUpdateUtils.waitForNoRollingUpdate(testStorage.getNamespaceName(), LabelSelectors.allKafkaPodsLabelSelector(testStorage.getClusterName()), kafkaPods);
+
+        LOGGER.info("Check if Kafka Pods are missing the 'kafka_server_' metrics");
+        kafkaCollector.collectMetricsFromPods(TestConstants.METRICS_COLLECT_TIMEOUT);
+
+        assertThat(MetricsUtils.createPatternAndCollectWithoutWait(kafkaCollector, "kafka_server_replicamanager_leadercount").isEmpty(), is(true));
+
+        LOGGER.info("Changing back to previous state");
+        KafkaUtils.replace(testStorage.getNamespaceName(), testStorage.getClusterName(), kafka -> {
+                StrimziMetricsReporter config = (StrimziMetricsReporter) kafka.getSpec().getKafka().getMetricsConfig();
+                config.getValues().setAllowList(List.of("kafka_server.*"));
+
+                kafka.getSpec().getKafka().setMetricsConfig(config);
+            }
+        );
+
+        RollingUpdateUtils.waitForNoRollingUpdate(testStorage.getNamespaceName(), LabelSelectors.allKafkaPodsLabelSelector(testStorage.getClusterName()), kafkaPods);
+
+        LOGGER.info("Check if Kafka Pods are not missing the 'kafka_server_' metrics");
+        kafkaCollector.collectMetricsFromPods(TestConstants.METRICS_COLLECT_TIMEOUT);
+
+        assertThat(MetricsUtils.createPatternAndCollect(kafkaCollector, "kafka_server_replicamanager_leadercount").isEmpty(), is(false));
+    }
+
     @BeforeAll
     void setupEnvironment() {
         testStorage = new TestStorage(KubeResourceManager.get().getTestContext());