feature(perf): support latte in gradual grow throughput test

Gradual throughput test uses several substitution variables
and 2 of them, called "$threads" and "$throttle", are not compatible "as-is"
with the latte benchmarking tool.

First, Latte has "--threads=N" and "--concurrency=M" parameters
whereas cassandra-stress has only "--threads".
So, latte's values must be multiplied to get CS's value.

Second, latte uses "--rate=100" not "fixed=100/s" or "throttle=100s"
like CS does.

Knowing above, add appropriate parsing of the latte command to support
existing substitution values.

Example:

Following CS command:
  cassandra-stress mixed cl=QUORUM duration=$duration -mode cql3 native \
    -rate 'threads=$threads $throttle' \
    -col 'size=FIXED(1024) n=FIXED(1)' -pop seq=15000001..20000000

Can be simulated by the following Latte command:
  latte run --function=write,read --tag=mixed --sampling=10s --duration=$duration \
  $throttle --threads=7 --concurrency=$threads --consistency=QUORUM
  --start-cycle 15000001 --end-cycle 20000000 \
  -P key_size=10 -P column_size=1024 -P column_count=1 -P row_count=20000000
  data_dir/latte/latte_cs_alike.rn

In case of Latte we specify "--threads" explicitly as (N-1) value
where N is number of CPU cores on loader nodes.

Author: vponomaryov

performance_regression_gradual_grow_throughput.py

@@ -1,3 +1,4 @@
+import math
 import pathlib
 import time
 from enum import Enum
@@ -10,6 +11,7 @@
 from sdcm.utils.common import skip_optional_stage
 from sdcm.sct_events import Severity
 from sdcm.sct_events.system import TestFrameworkEvent
+from sdcm.stress.latte_thread import find_latte_threads_num
 from sdcm.results_analyze import PredefinedStepsTestPerformanceAnalyzer
 from sdcm.utils.decorators import latency_calculator_decorator
 from sdcm.utils.latency import calculate_latency, analyze_hdr_percentiles
@@ -32,6 +34,21 @@ class Workload(NamedTuple):
     step_duration: str
 
 
+def is_latte_command(stress_cmd: str) -> bool:
+    return "latte " in stress_cmd and " run " in stress_cmd
+
+
+def process_stress_threads_number(stress_cmd: str, num_threads: int) -> int:
+    # NOTE: latte has 2 separate options - '--threads' and '--concurrency'.
+    #       CS's threads == latte threads multiple of 'concurrency'
+    #       Example: 620 CS threads = 7 latte threads with concurrency as '89' per each thread (623)
+    if is_latte_command(stress_cmd):
+        latte_threads_num = find_latte_threads_num(stress_cmd)
+        return math.ceil(num_threads / latte_threads_num)
+
+    return num_threads
+
+
 class PerformanceRegressionPredefinedStepsTest(PerformanceRegressionTest):  # pylint: disable=too-many-instance-attributes
     """
     This class presents new performance test that run gradual increased throughput steps.
@@ -191,8 +208,9 @@ def run_step(self, stress_cmds, current_throttle, num_threads, step_duration):
         stress_queue = []
         for stress_cmd in stress_cmds:
             params = {"round_robin": True, "stats_aggregate_cmds": False}
+            current_num_threads = process_stress_threads_number(stress_cmd, num_threads)
             stress_cmd_to_run = stress_cmd.replace(
-                "$threads", f"{num_threads}").replace("$throttle", f"{current_throttle}")
+                "$threads", f"{current_num_threads}").replace("$throttle", f"{current_throttle}")
             if step_duration is not None:
                 stress_cmd_to_run = stress_cmd_to_run.replace("$duration", step_duration)
             params.update({'stress_cmd': stress_cmd_to_run})
@@ -225,7 +243,15 @@ def run_gradual_increase_load(self, workload: Workload, stress_num, num_loaders,
 
         for throttle_step in workload.throttle_steps:
             self.log.info("Run cs command with rate: %s Kops", throttle_step)
-            current_throttle = f"fixed={int(int(throttle_step) // (num_loaders * stress_num))}/s" if throttle_step != "unthrottled" else ""
+            if throttle_step == "unthrottled":
+                current_throttle = ""
+            else:
+                throttle_value = int(int(throttle_step) // (num_loaders * stress_num))
+                if is_latte_command(workload.cs_cmd_tmpl[0]):
+                    current_throttle = f"--rate={throttle_value}"
+                else:
+                    current_throttle = f"fixed={throttle_value}/s"
+
             run_step = ((latency_calculator_decorator(legend=f"Gradual test step {throttle_step} op/s",
                                                       cycle_name=throttle_step))(self.run_step))
             results, _ = run_step(stress_cmds=workload.cs_cmd_tmpl, current_throttle=current_throttle,
@@ -304,7 +330,8 @@ def warmup_cache(self, stress_cmd_templ, num_threads):
         stress_queue = []
         for stress_cmd in stress_cmd_templ:
             params = {"round_robin": True, "stats_aggregate_cmds": False}
-            stress_cmd_to_run = stress_cmd.replace("$threads", str(num_threads))
+            current_num_threads = process_stress_threads_number(stress_cmd, num_threads)
+            stress_cmd_to_run = stress_cmd.replace("$threads", str(current_num_threads))
             params.update({'stress_cmd': stress_cmd_to_run})
             # Run all stress commands
             self.log.debug('RUNNING warm up stress cmd: %s', stress_cmd_to_run)

sdcm/stress/latte_thread.py

@@ -36,6 +36,7 @@
 
 LATTE_FN_NAME_RE = '(?:-f|--function)[ =]([\w\s\d:,]+)|--functions[ =]([\w\s\d:,]+)'
 LATTE_TAG_RE = r'--tag(?:\s+|=)([\w-]+(?:,[\w-]+)*)\b'
+LATTE_THREADS_RE = r"(?:\s|^)(?:-t[= ]|--threads[= ]?)(\d+)(?:\s|$)"
 LOGGER = logging.getLogger(__name__)
 
 
@@ -66,6 +67,14 @@ def find_latte_tags(stress_cmd):
     return tags
 
 
+def find_latte_threads_num(stress_cmd):
+    matches = re.findall(LATTE_THREADS_RE, stress_cmd)
+    if matches:
+        return int(matches[0])
+    # NOTE: threads were not specified, assume latte's default as `1`
+    return 1
+
+
 def get_latte_operation_type(stress_cmd):
     write_found, counter_write, read_found, counter_read = False, False, False, False
     for fn in find_latte_fn_names(stress_cmd):