override throughput + client parameters if load testing

Signed-off-by: Michael Oviedo <[email protected]>

Author: OVI3D0

osbenchmark/benchmark.py

@@ -648,10 +648,9 @@ def add_workload_source(subparser):
         help="Stop executing tests if an error occurs in one of the test iterations (default: false).",
     )
     test_execution_parser.add_argument(
-        "--load-test",
-        action="store_true",
-        help="Run a load test (default: false).",
-        default=False
+        "--load-test-qps",
+        help="Run a load test on your cluster, up to a certain QPS value (default: 0)",
+        default=0
     )
 
     ###############################################################################
@@ -926,7 +925,7 @@ def configure_test(arg_parser, args, cfg):
         "load_worker_coordinator_hosts",
         opts.csv_to_list(args.load_worker_coordinator_hosts))
     cfg.add(config.Scope.applicationOverride, "workload", "test.mode.enabled", args.test_mode)
-    cfg.add(config.Scope.applicationOverride, "workload", "load.test.enabled", args.load_test)
+    cfg.add(config.Scope.applicationOverride, "workload", "load.test.clients", int(args.load_test_qps))
     cfg.add(config.Scope.applicationOverride, "workload", "latency.percentiles", args.latency_percentiles)
     cfg.add(config.Scope.applicationOverride, "workload", "throughput.percentiles", args.throughput_percentiles)
     cfg.add(config.Scope.applicationOverride, "workload", "randomization.enabled", args.randomization_enabled)

osbenchmark/worker_coordinator/worker_coordinator.py

@@ -657,7 +657,15 @@ def start_benchmark(self):
         self.logger.info("Attaching cluster-level telemetry devices.")
         self.telemetry.on_benchmark_start()
         self.logger.info("Cluster-level telemetry devices are now attached.")
-
+        # if load testing is enabled, modify the client + throughput number for the task(s)
+        # target throughput + clients will then be equal to the qps passed in through --load-test
+        load_test_clients = self.config.opts("workload", "load.test.clients", mandatory=False)
+        if load_test_clients:
+            for task in self.test_procedure.schedule:
+                for subtask in task:
+                    subtask.clients = load_test_clients
+                    subtask.params["target-throughput"] = load_test_clients
+            self.logger.info("Load test mode enabled - set client count to %d", load_test_clients)
         allocator = Allocator(self.test_procedure.schedule)
         self.allocations = allocator.allocations
         self.number_of_steps = len(allocator.join_points) - 1
@@ -1634,7 +1642,7 @@ async def __call__(self, *args, **kwargs):
             await asyncio.sleep(rampup_wait_time)
 
         if rampup_wait_time:
-            self.logger.info(f" Client id {self.client_id} is running now.")
+            self.logger.info("Client id [%s] is running now.", self.client_id)
 
         self.logger.debug("Entering main loop for client id [%s].", self.client_id)
         # noinspection PyBroadException