Merge pull request #28756 from redpanda-data/manual-backport-28728-v25.2.x-612

[v25.2.x] cluster: fix topic label aggregation on restart

Author: ballard26

src/v/cluster/topic_metrics_watcher.cc

@@ -23,6 +23,8 @@ topic_metrics_watcher::topic_metrics_watcher(
     tt.register_topic_delta_notification(
       [this](const auto&) { maybe_aggregate_topic_label(); });
     _topic_agg_limit.watch([this] { maybe_aggregate_topic_label(); });
+
+    maybe_aggregate_topic_label();
 }
 
 ss::future<> topic_metrics_watcher::stop() { return _gate.close(); }

tests/rptest/tests/topic_label_aggregation_test.py

@@ -7,6 +7,7 @@
 # the Business Source License, use of this software will be governed
 # by the Apache License, Version 2.0
 
+from typing import Any
 from ducktape.utils.util import wait_until
 
 from rptest.clients.types import TopicSpec
@@ -19,34 +20,50 @@ class TopicLabelAggregationTest(RedpandaTest):
     TOPIC_LABEL = "topic"
     PARTITION_LABEL = "partition"
 
-    def __init__(self, *args, **kwargs):
+    def __init__(self, *args: Any, **kwargs: Any):
         super().__init__(num_brokers=3, *args, **kwargs)
 
+    def _aggregated_metrics_have_topic_label(self) -> bool:
+        for node in self.redpanda.nodes:
+            for family in self.redpanda.metrics(
+                node, metrics_endpoint=MetricsEndpoint.METRICS
+            ):
+                for sample in family.samples:
+                    # If the partition label isn't aggregated then neither will the topic label.
+                    if (
+                        self.TOPIC_LABEL in sample.labels.keys()
+                        and self.PARTITION_LABEL not in sample.labels.keys()
+                    ):
+                        return True
+
+        return False
+
+    def _unaggregated_metrics_have_topic_label(self) -> bool:
+        for node in self.redpanda.nodes:
+            for family in self.redpanda.metrics(
+                node, metrics_endpoint=MetricsEndpoint.METRICS
+            ):
+                for sample in family.samples:
+                    # Only consider series that would have their topic label aggregated
+                    # when topic label aggregation is enabled.
+                    if (
+                        self.TOPIC_LABEL in sample.labels.keys()
+                        and self.PARTITION_LABEL in sample.labels.keys()
+                    ):
+                        return True
+
+        return False
+
+    def _create_topics(self, topics: list[TopicSpec]):
+        for t in topics:
+            self.client().create_topic(t)
+
+    def _delete_topics(self, topics: list[TopicSpec]):
+        for t in topics:
+            self.client().delete_topic(t.name)
+
     @cluster(num_nodes=3)
     def test_topic_label_aggregation(self):
-        def metrics_have_topic_label() -> bool:
-            for node in self.redpanda.nodes:
-                for family in self.redpanda.metrics(
-                    node, metrics_endpoint=MetricsEndpoint.METRICS
-                ):
-                    for sample in family.samples:
-                        # If the partition label isn't aggregated then neither will the topic label.
-                        if (
-                            self.TOPIC_LABEL in sample.labels.keys()
-                            and self.PARTITION_LABEL not in sample.labels.keys()
-                        ):
-                            return True
-
-            return False
-
-        def create_topics(topics: list[TopicSpec]):
-            for t in topics:
-                self.client().create_topic(t)
-
-        def delete_topics(topics: list[TopicSpec]):
-            for t in topics:
-                self.client().delete_topic(t.name)
-
         initial_topic_count = len(self.client().describe_topics())
         topics_to_create = [TopicSpec() for _ in range(50)]
         topic_aggregation_limit = len(topics_to_create) + initial_topic_count - 1
@@ -57,11 +74,13 @@ def delete_topics(topics: list[TopicSpec]):
                 "topic_label_aggregation_limit": topic_aggregation_limit,
             }
         )
-        assert metrics_have_topic_label(), "topic label shouldn't be aggregated yet"
+        assert self._aggregated_metrics_have_topic_label(), (
+            "topic label shouldn't be aggregated yet"
+        )
 
-        create_topics(topics_to_create)
+        self._create_topics(topics_to_create)
         wait_until(
-            lambda: not metrics_have_topic_label(),
+            lambda: not self._aggregated_metrics_have_topic_label(),
             timeout_sec=60,
             backoff_sec=10,
             err_msg="topic label wasn't aggregated",
@@ -77,32 +96,99 @@ def delete_topics(topics: list[TopicSpec]):
         topic_count_to_delete = int(current_topic_count - agg_limit_lower_bound) - 1
         assert topic_count_to_delete > 1
 
-        delete_topics(topics_to_create[:topic_count_to_delete])
-        assert not metrics_have_topic_label()
+        self._delete_topics(topics_to_create[:topic_count_to_delete])
+        assert not self._aggregated_metrics_have_topic_label()
 
         # Have topic count fall bellow the 95% limit.
-        delete_topics(
+        self._delete_topics(
             topics_to_create[topic_count_to_delete : (topic_count_to_delete + 2)]
         )
         wait_until(
-            lambda: metrics_have_topic_label(),
+            lambda: self._aggregated_metrics_have_topic_label(),
             timeout_sec=60,
             backoff_sec=10,
             err_msg="topic label wasn't un-aggregated",
         )
 
-        create_topics(topics_to_create[: (topic_count_to_delete + 2)])
+        self._create_topics(topics_to_create[: (topic_count_to_delete + 2)])
         wait_until(
-            lambda: not metrics_have_topic_label(),
+            lambda: not self._aggregated_metrics_have_topic_label(),
             timeout_sec=60,
             backoff_sec=10,
             err_msg="topic label wasn't aggregated",
         )
 
         self.redpanda.set_cluster_config({"topic_label_aggregation_limit": None})
         wait_until(
-            lambda: metrics_have_topic_label(),
+            lambda: self._aggregated_metrics_have_topic_label(),
             timeout_sec=60,
             backoff_sec=10,
             err_msg="topic label wasn't un-aggregated",
         )
+
+    @cluster(num_nodes=3)
+    def test_topic_aggregate_on_toggle(self):
+        """
+        Topic labels are not aggregated when metrics aggregation is turned off in general.
+        This tests that they are aggregated when metrics aggregation is enabled. Assuming
+        that the topic count meets the aggregation threshold.
+        """
+        initial_topic_count = len(self.client().describe_topics())
+        topics_to_create = [TopicSpec() for _ in range(50)]
+        topic_aggregation_limit = len(topics_to_create) + initial_topic_count - 1
+
+        self.redpanda.set_cluster_config(
+            {
+                "aggregate_metrics": False,
+                "topic_label_aggregation_limit": topic_aggregation_limit,
+            }
+        )
+
+        self._create_topics(topics_to_create)
+        assert self._unaggregated_metrics_have_topic_label(), (
+            "topic label shouldn't be aggregated yet"
+        )
+
+        self.redpanda.set_cluster_config(
+            {
+                "aggregate_metrics": True,
+            }
+        )
+        wait_until(
+            lambda: not self._aggregated_metrics_have_topic_label(),
+            timeout_sec=60,
+            backoff_sec=10,
+            err_msg="topic label wasn't aggregated",
+        )
+
+    @cluster(num_nodes=3)
+    def test_topic_aggregate_on_restart(self):
+        """
+        Tests that topic labels remain aggregated when a node restarts.
+        """
+        initial_topic_count = len(self.client().describe_topics())
+        topics_to_create = [TopicSpec() for _ in range(50)]
+        topic_aggregation_limit = len(topics_to_create) + initial_topic_count - 1
+
+        self.redpanda.set_cluster_config(
+            {
+                "aggregate_metrics": True,
+                "topic_label_aggregation_limit": topic_aggregation_limit,
+            }
+        )
+
+        self._create_topics(topics_to_create)
+        wait_until(
+            lambda: not self._aggregated_metrics_have_topic_label(),
+            timeout_sec=60,
+            backoff_sec=10,
+            err_msg="topic label wasn't aggregated",
+        )
+
+        self.redpanda.rolling_restart_nodes(self.redpanda.nodes)
+        wait_until(
+            lambda: not self._aggregated_metrics_have_topic_label(),
+            timeout_sec=60,
+            backoff_sec=10,
+            err_msg="topic label wasn't aggregated",
+        )