Merge pull request #30963 from redpanda-data/stephan/swarm-scale

producer_swarm: Add scale factor to wait

Author: StephanDollberg

tests/rptest/redpanda_cloud_tests/omb_validation_test.py

@@ -332,7 +332,7 @@ def assert_no_rejected():
         msg_rate_per_node = messages_per_sec_per_producer * producer_per_swarm_node
 
         # single producer runtime
-        # Each swarm will throttle the client creation rate to about 30 connections/second
+        # Each swarm will throttle the client creation rate to about 30 producers/second
         warm_up_time_s = (
             producer_per_swarm_node * ProducerSwarm.CLIENT_SPAWN_WAIT_MS // 1000
         ) + 60
@@ -430,7 +430,8 @@ def make_swarm():
 
         for s in swarm:
             # wait for the swarm to report that all producers have started (sent at least 1 message)
-            s.wait_for_all_started()
+            broker_scale_factor = max(1, self.num_brokers / 6)
+            s.wait_for_all_started(timeout_scale=broker_scale_factor)
 
         # Now wait for up to five minutes to hit our target connection count: even though all producers
         # have started, it's possible that the connections haven't hit their target yet because some

tests/rptest/services/producer_swarm.py

@@ -127,7 +127,7 @@ def _pre_run_tasks(self):
                 self._node, ProducerSwarm.CUSTOM_PAYLOAD_DIR
             )
 
-    def wait_for_all_started(self):
+    def wait_for_all_started(self, timeout_scale: float = 1.0):
         """Wait until the requested number of producers have started. Note that if the expected
         swarm runtime (messages / rate) is short, this may fail as all producers and start and
         finish before we are able to see this via the metrics endpoint and an exception is thrown."""
@@ -146,6 +146,7 @@ def wait_for_all_started(self):
         # due to leadership transfers which occur in a burst some time between 0 and 5
         # minutes after the topic is created.
         timeout_s = 30 + 3 * ceil(self._producers * self.CLIENT_SPAWN_WAIT_MS / 1000)
+        timeout_s = int(timeout_s * timeout_scale)
 
         def started_count():
             started = self.get_metrics_summary().clients_started