rptest: Updating parameters and test structure

  Test will run OMB with minimal workload to measure idle latency and
  then use target partitions number to measure desired parameters

Author: savex

tests/rptest/redpanda_cloud_tests/high_throughput_test.py

@@ -14,7 +14,7 @@
 import time
 import json
 
-from ducktape.mark import ignore, ok_to_fail
+from ducktape.mark import ignore, ok_to_fail, parametrize
 from ducktape.tests.test import TestContext
 from ducktape.utils.util import wait_until
 from rptest.clients.rpk import RpkTool
@@ -1153,38 +1153,92 @@ def producer_complete():
         consumer.stop()
         consumer.free()
 
-    @cluster(num_nodes=3, log_allow_list=RESTART_LOG_ALLOW_LIST)
-    def test_ht004_minpartomb(self):
-        validator_overrides = {
-            OMBSampleConfigurations.E2E_LATENCY_50PCT:
-            [OMBSampleConfigurations.lte(51)],
-            OMBSampleConfigurations.E2E_LATENCY_AVG:
-            [OMBSampleConfigurations.lte(145)],
-        }
-        partitions_per_topic = self.config.partitions_max_scaled
-        workload = {
+    def _prepare_omb_workload(self, ramp_time, duration, partitions, rate,
+                              msg_size):
+        return {
             "name": "HT004-MINPARTOMB",
             "topics": 1,
-            "partitions_per_topic": partitions_per_topic,
+            "partitions_per_topic": partitions,
             "subscriptions_per_topic": 1,
             "consumer_per_subscription": 3,
             "producers_per_topic": 1,
-            "producer_rate": int(self.config.ingress_rate_scaled / 8),
-            "message_size": 8 * KiB,
+            "producer_rate": rate,
+            "message_size": msg_size,
             "consumer_backlog_size_GB": 0,
-            "test_duration_minutes": 1,
-            "warmup_duration_minutes": 1,
+            "test_duration_minutes": duration,
+            "warmup_duration_minutes": ramp_time,
             "use_randomized_payloads": True,
             "random_bytes_ratio": 0.5,
             "randomized_payload_pool_size": 100,
         }
 
-        benchmark = OpenMessagingBenchmark(
+    def _run_bench(self, workload, validator_overrides):
+        _bench = OpenMessagingBenchmark(
             self._ctx, self.redpanda, "SIMPLE_DRIVER",
             (workload, OMBSampleConfigurations.UNIT_TEST_LATENCY_VALIDATOR
              | validator_overrides))
+        _bench.start()
+        benchmark_time_min = _bench.benchmark_time() + 1
+        _bench.wait(timeout_sec=benchmark_time_min * 60)
+        _metrics = json.loads(_bench.node.account.ssh_output(_bench.chart_cmd))
+        return _bench, list(_metrics.values())[0]
+
+    @cluster(num_nodes=6, log_allow_list=RESTART_LOG_ALLOW_LIST)
+    @parametrize(partitions="min")
+    @parametrize(partitions="max")
+    def test_htt_partitions_omb(self, partitions):
+        def _format_metrics(idle, tier):
+            keys = idle.keys()
+            return "\n".join([f"{k} = {idle[k]} / {tier[k]} " for k in keys])
+
+        # Get values for almost idle cluster load
+        _min_idle_lat = 1000
+        # Assume we have 1 partition per shard,
+        # then number of CPU should be equal to min number of partitions
+        # to get idle-like activity
+        _num_partitions = 8
+
+        if partitions not in ["min", "max"]:
+            raise RuntimeError("Test parameter for partitions invalid")
+
+        idle_validators = {
+            OMBSampleConfigurations.E2E_LATENCY_50PCT:
+            [OMBSampleConfigurations.lte(_min_idle_lat)],
+            OMBSampleConfigurations.E2E_LATENCY_AVG:
+            [OMBSampleConfigurations.lte(_min_idle_lat * 3)],
+        }
+        idle_workload = self._prepare_omb_workload(1, 2, _num_partitions,
+                                                   1 * MiB, 8 * KiB)
+        _, idle_metrics = self._run_bench(idle_workload, idle_validators)
 
-        benchmark.start()
-        benchmark_time_min = benchmark.benchmark_time() + 1
-        benchmark.wait(timeout_sec=benchmark_time_min * 60)
+        # Get values for idle workload
+        k_e2e_50pct = idle_metrics[OMBSampleConfigurations.E2E_LATENCY_50PCT]
+        k_e2e_avg = idle_metrics[OMBSampleConfigurations.E2E_LATENCY_AVG]
+
+        # Calculate target throughput latencies
+        target_e2e_50pct = k_e2e_50pct + 51
+        target_e2e_avg = k_e2e_avg + 145
+
+        # Measure with target load
+        validator_overrides = {
+            OMBSampleConfigurations.E2E_LATENCY_50PCT:
+            [OMBSampleConfigurations.lte(target_e2e_50pct)],
+            OMBSampleConfigurations.E2E_LATENCY_AVG:
+            [OMBSampleConfigurations.lte(target_e2e_avg)],
+        }
+        # Select number of partitions
+        if partitions == "min":
+            _num_partitions = self.tier_config.partitions_min
+        elif partitions == "max":
+            _num_partitions = self.tier_config.partitions_upper_limit
+
+        workload = self._prepare_omb_workload(1, 2, _num_partitions,
+                                              self.tier_config.ingress_rate,
+                                              8 * KiB)
+        benchmark, metrics = self._run_bench(workload, validator_overrides)
         benchmark.check_succeed()
+
+        # Tier metrics should not diviate from idle
+        # metrics more than 145 ms on the average
+        self.logger.info('Workload metrics (idle/tier): '
+                         '"{}"'.format(_format_metrics(idle_metrics, metrics)))