tests/admin_fuzzer: query all nodes about partition count

Since metadata are eventually consistent we query all nodes about
current partition count of a topic. Only if all values are the same we
allow to execute add partitions operation. This way operation result
will always be deterministic.

Fixes: redpanda-data#8637

Signed-off-by: Michal Maslanka <michal@redpanda.com>
(cherry picked from commit aab3b0c)

tests/rptest/services/admin_ops_fuzzer.py

@@ -14,12 +14,14 @@
 from threading import Event, Condition
 import threading
 import time
+from requests.exceptions import HTTPError
 from ducktape.utils.util import wait_until
 from rptest.clients.types import TopicSpec
 from time import sleep
 
 from rptest.clients.rpk import RpkTool
 from rptest.services.admin import Admin
+from rptest.services.redpanda_installer import VERSION_RE, int_tuple
 
 
 # Operation context (used to save state between invocation of operations)
@@ -204,15 +206,44 @@ def __init__(self, prefix):
         self.topic = None
         self.total = None
 
+    def _current_partition_count(self, ctx):
+
+        per_node_count = set()
+        for n in ctx.redpanda.started_nodes():
+            node_version = int_tuple(
+                VERSION_RE.findall(ctx.redpanda.get_version(n))[0])
+            # do not query nodes with redpanda version prior to v23.1.x
+            if node_version[0] < 23:
+                return None
+            try:
+                partitions = ctx.admin().get_partitions(node=n,
+                                                        topic=self.topic)
+                per_node_count.add(len(partitions))
+            except HTTPError as err:
+                if err.response.status_code == 404:
+                    return None
+                else:
+                    raise
+
+        if len(per_node_count) != 1:
+            ctx.redpanda.logger.info(
+                f"inconsistent partition count for {self.topic}: {per_node_count}"
+            )
+            return None
+
+        return next(iter(per_node_count))
+
     def execute(self, ctx):
         self.topic = _choice_random_topic(ctx, prefix=self.prefix)
         if self.topic is None:
             return False
 
-        rpk = ctx.rpk()
-        current = len(list(rpk.describe_topic(self.topic)))
-        to_add = random.randint(1, 5)
-        self.total = current + to_add
+        if self.current is None:
+            self.current = self._current_partition_count(ctx)
+        if self.current is None:
+            return False
+        if self.total is None:
+            self.total = random.randint(self.current + 1, self.current + 5)
         ctx.redpanda.logger.info(
             f"Updating topic: {self.topic} partitions count, current: {current} adding: {to_add} partitions"
         )