[add] Added redis-benchmark automation for local/remote runs (#53)

Author: filipecosta90

.gitignore

@@ -3,7 +3,7 @@
 *.json
 *.tar.gz
 *.csv
-
+*.txt
 
 # Byte-compiled / optimized / DLL files
 __pycache__/

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "redisbench-admin"
-version = "0.1.49"
+version = "0.1.50"
 description = "Redis benchmark run helper. A wrapper around Redis and Redis Modules benchmark tools ( ftsb_redisearch, memtier_benchmark, redis-benchmark, aibench, etc... )."
 authors = ["filipecosta90 <[email protected]>"]
 readme = "README.md"

redisbench_admin/run/common.py

@@ -0,0 +1,89 @@
+import logging
+import re
+
+from redisbench_admin.run.redis_benchmark.redis_benchmark import prepareRedisBenchmarkCommand
+from redisbench_admin.run.redisgraph_benchmark_go.redisgraph_benchmark_go import prepareRedisGraphBenchmarkGoCommand
+from redisbench_admin.utils.remote import executeRemoteCommands, getFileFromRemoteSetup
+
+
+def extract_benchmark_tool_settings(benchmark_config):
+    benchmark_tool = None
+    benchmark_min_tool_version = None
+    benchmark_min_tool_version_major = None
+    benchmark_min_tool_version_minor = None
+    benchmark_min_tool_version_patch = None
+
+    for entry in benchmark_config["clientconfig"]:
+        if 'tool' in entry:
+            benchmark_tool = entry['tool']
+        if 'min-tool-version' in entry:
+            benchmark_min_tool_version = entry['min-tool-version']
+            p = re.compile("(\d+)\.(\d+)\.(\d+)")
+            m = p.match(benchmark_min_tool_version)
+            if m is None:
+                logging.error(
+                    "Unable to extract semversion from 'min-tool-version'. Will not enforce version")
+                benchmark_min_tool_version = None
+            else:
+                benchmark_min_tool_version_major = m.group(1)
+                benchmark_min_tool_version_minor = m.group(2)
+                benchmark_min_tool_version_patch = m.group(3)
+    return benchmark_min_tool_version, benchmark_min_tool_version_major, benchmark_min_tool_version_minor, benchmark_min_tool_version_patch, benchmark_tool
+
+
+def prepare_benchmark_parameters(benchmark_config, benchmark_tool, server_plaintext_port, server_private_ip,
+                                 remote_results_file, isremote=False):
+    for entry in benchmark_config["clientconfig"]:
+        if 'parameters' in entry:
+            if benchmark_tool == 'redis-benchmark':
+                command_arr, command_str = prepareRedisBenchmarkCommand(
+                    "redis-benchmark",
+                    server_private_ip,
+                    server_plaintext_port,
+                    entry
+                )
+                redirect_file = ">{}".format(remote_results_file)
+                command_arr.append(redirect_file)
+                command_str = command_str + " " + redirect_file
+
+            if benchmark_tool == 'redisgraph-benchmark-go':
+                command_arr = prepareRedisGraphBenchmarkGoCommand(
+                    "/tmp/redisgraph-benchmark-go",
+                    server_private_ip,
+                    server_plaintext_port,
+                    entry,
+                    remote_results_file,
+                )
+                command_str = " ".join(command_arr)
+    return command_arr, command_str
+
+
+def runRemoteBenchmark(
+        client_public_ip,
+        username,
+        private_key,
+        remote_results_file,
+        local_results_file,
+        command
+):
+    remote_run_result = False
+    res = executeRemoteCommands(client_public_ip, username, private_key, [command])
+    recv_exit_status, stdout, stderr = res[0]
+
+    if recv_exit_status != 0:
+        logging.error("Exit status of remote command execution {}. Printing stdout and stderr".format(recv_exit_status))
+        logging.error("remote process stdout: ".format(stdout))
+        logging.error("remote process stderr: ".format(stderr))
+    else:
+        logging.info("Remote process exited normally. Exit code {}. Printing stdout.".format(recv_exit_status))
+        logging.info("remote process stdout: ".format(stdout))
+        logging.info("Extracting the benchmark results")
+        remote_run_result = True
+        getFileFromRemoteSetup(
+            client_public_ip,
+            username,
+            private_key,
+            local_results_file,
+            remote_results_file,
+        )
+    return remote_run_result

redisbench_admin/run/redis_benchmark/__init__.py

@@ -1,15 +1,21 @@
 import csv
 import logging
+import re
 import shlex
+import subprocess
 
+from redisbench_admin.utils.remote import executeRemoteCommands
 
-def redis_benchmark_from_stdout_csv_to_json(stdout, start_time, start_time_str):
+
+def redis_benchmark_from_stdout_csv_to_json(csv_data, start_time, start_time_str, overloadTestName=None):
     results_dict = {"Tests": {}, "StartTime": int(start_time.strftime("%s")),
                     "StartTimeHuman": start_time_str}
-    csv_data = list(csv.reader(stdout.decode('ascii').splitlines(), delimiter=","))
+    csv_data = list(csv.reader(csv_data.splitlines(), delimiter=","))
     header = csv_data[0]
     for row in csv_data[1:]:
         test_name = row[0]
+        if overloadTestName is not None:
+            test_name = overloadTestName
         results_dict["Tests"][test_name] = {}
         for pos, value in enumerate(row[1:]):
             results_dict["Tests"][test_name][header[pos + 1]] = value
@@ -21,38 +27,89 @@ def prepareRedisBenchmarkCommand(
         server_private_ip: object,
         server_plaintext_port: object,
         benchmark_config: object,
-) -> str:
+):
     """
     Prepares redis-benchmark command parameters
     :param server_private_ip:
     :param server_plaintext_port:
     :param benchmark_config:
-    :return: string containing the required command to run the benchmark given the configurations
+    :return: [string] containing the required command to run the benchmark given the configurations
     """
-    queries_str = [executable_path]
-    queries_str.extend(["-h", "{}".format(server_private_ip)])
-    queries_str.extend(["-p", "{}".format(server_plaintext_port)])
+    command_str = ""
+    command_arr = [executable_path]
+    command_arr.extend(["-h", "{}".format(server_private_ip)])
+    command_arr.extend(["-p", "{}".format(server_plaintext_port)])
 
     # we need the csv output
-    queries_str.extend(["--csv", "-e"])
+    command_arr.extend(["--csv", "-e"])
     last_append = None
     for k in benchmark_config["parameters"]:
         if "clients" in k:
-            queries_str.extend(["-c", "{}".format(k["clients"])])
+            command_arr.extend(["-c", "{}".format(k["clients"])])
         if "requests" in k:
-            queries_str.extend(["-n", "{}".format(k["requests"])])
+            command_arr.extend(["-n", "{}".format(k["requests"])])
         if "threads" in k:
-            queries_str.extend(["--threads", "{}".format(k["threads"])])
+            command_arr.extend(["--threads", "{}".format(k["threads"])])
         if "pipeline" in k:
-            queries_str.extend(["-P", "{}".format(k["pipeline"])])
+            command_arr.extend(["-P", "{}".format(k["pipeline"])])
         # if we have the command keywork then it needs to be at the end of args
         if "command" in k:
+            last_str = k["command"]
             last_append = shlex.split(k["command"])
+    command_str = " ".join(command_arr)
     if last_append is not None:
-        queries_str.extend(last_append)
+        command_arr.extend(last_append)
+        command_str = command_str + " " + last_str
     logging.info(
-        "Running the benchmark with the following parameters: {}".format(
-            " ".join(queries_str)
+        "Running the benchmark with the following parameters:\n\tArgs array: {}\n\tArgs str: {}".format(
+            command_arr, command_str
         )
     )
-    return queries_str
+    return command_arr, command_str
+
+
+def ensure_redis_benchmark_version_from_input(benchmark_min_tool_version, benchmark_min_tool_version_major,
+                                              benchmark_min_tool_version_minor, benchmark_min_tool_version_patch,
+                                              benchmark_tool, stdout):
+    version_output = stdout.decode('ascii').split("\n")[0]
+    logging.info(
+        "Detected benchmark config tool {} with version {}".format(benchmark_tool, version_output))
+    p = re.compile("redis-benchmark (\d+)\.(\d+)\.(\d+) ")
+    m = p.match(version_output)
+    if m is None:
+        raise Exception(
+            "Unable to detect benchmark tool version, and the benchmark requires a min version: {}".format(
+                benchmark_min_tool_version))
+    major = m.group(1)
+    minor = m.group(2)
+    patch = m.group(3)
+    if major < benchmark_min_tool_version_major or (
+            major == benchmark_min_tool_version_major and minor < benchmark_min_tool_version_minor) or (
+            major == benchmark_min_tool_version_major and minor == benchmark_min_tool_version_minor and patch < benchmark_min_tool_version_patch):
+        raise Exception(
+            "Detected benchmark version that is inferior than the minimum required. {} < {}".format(
+                version_output, benchmark_min_tool_version))
+
+
+def redis_benchmark_ensure_min_version_local(benchmark_tool, benchmark_min_tool_version,
+                                             benchmark_min_tool_version_major,
+                                             benchmark_min_tool_version_minor, benchmark_min_tool_version_patch):
+    benchmark_client_process = subprocess.Popen(args=[benchmark_tool, "--version"],
+                                                stdout=subprocess.PIPE,
+                                                stderr=subprocess.STDOUT)
+    (stdout, sterr) = benchmark_client_process.communicate()
+    ensure_redis_benchmark_version_from_input(benchmark_min_tool_version, benchmark_min_tool_version_major,
+                                              benchmark_min_tool_version_minor, benchmark_min_tool_version_patch,
+                                              benchmark_tool, stdout)
+
+
+def redis_benchmark_ensure_min_version_remote(benchmark_tool, benchmark_min_tool_version,
+                                              benchmark_min_tool_version_major,
+                                              benchmark_min_tool_version_minor, benchmark_min_tool_version_patch):
+    benchmark_client_process = subprocess.Popen(args=[benchmark_tool, "--version"],
+                                                stdout=subprocess.PIPE,
+                                                stderr=subprocess.STDOUT)
+    (stdout, sterr) = benchmark_client_process.communicate()
+    ensure_redis_benchmark_version_from_input(benchmark_min_tool_version, benchmark_min_tool_version_major,
+                                              benchmark_min_tool_version_minor, benchmark_min_tool_version_patch,
+                                              benchmark_tool, stdout)

redisbench_admin/run/redis_benchmark/redis_benchmark.py

@@ -0,0 +1,43 @@
+import logging
+
+
+def prepareRedisGraphBenchmarkGoCommand(
+        executable_path: str,
+        server_private_ip: object,
+        server_plaintext_port: object,
+        benchmark_config: object,
+        results_file: object,
+):
+    """
+    Prepares redisgraph-benchmark-go command parameters
+    :param server_private_ip:
+    :param server_plaintext_port:
+    :param benchmark_config:
+    :param results_file:
+    :return: string containing the required command to run the benchmark given the configurations
+    """
+    queries_str = [executable_path]
+    for k in benchmark_config["parameters"]:
+        if "graph" in k:
+            queries_str.extend(["-graph-key", "'{}'".format(k["graph"])])
+        if "clients" in k:
+            queries_str.extend(["-c", "{}".format(k["clients"])])
+        if "requests" in k:
+            queries_str.extend(["-n", "{}".format(k["requests"])])
+        if "rps" in k:
+            queries_str.extend(["-rps", "{}".format(k["rps"])])
+        if "queries" in k:
+            for kk in k["queries"]:
+                query = kk["q"]
+                queries_str.extend(["-query", "'{}'".format(query)])
+                if "ratio" in kk:
+                    queries_str.extend(["-query-ratio", "{}".format(kk["ratio"])])
+    queries_str.extend(["-h", "{}".format(server_private_ip)])
+    queries_str.extend(["-p", "{}".format(server_plaintext_port)])
+    queries_str.extend(["-json-out-file", "{}".format(results_file)])
+    logging.info(
+        "Running the benchmark with the following parameters: {}".format(
+            " ".join(queries_str)
+        )
+    )
+    return queries_str

redisbench_admin/run/redisgraph_benchmark_go/__init__.py

@@ -1,7 +1,3 @@
-import logging
-import re
-import subprocess
-
 from cpuinfo import cpuinfo
 
 
@@ -21,29 +17,3 @@ def run_command_logic(args):
     benchmark_machine_1 = {"machine_info": benchmark_machine_info}
     benchmark_infra["benchmark-machines"]["benchmark-machine-1"] = benchmark_machine_1
     benchmark_infra["total-benchmark-machines"] += 1
-
-
-def redis_benchmark_ensure_min_version(benchmark_tool, benchmark_min_tool_version, benchmark_min_tool_version_major,
-                                       benchmark_min_tool_version_minor, benchmark_min_tool_version_patch):
-    benchmark_client_process = subprocess.Popen(args=[benchmark_tool, "--version"],
-                                                stdout=subprocess.PIPE,
-                                                stderr=subprocess.STDOUT)
-    (stdout, sterr) = benchmark_client_process.communicate()
-    version_output = stdout.decode('ascii').split("\n")[0]
-    logging.info(
-        "Detected benchmark config tool {} with version {}".format(benchmark_tool, version_output))
-    p = re.compile("redis-benchmark (\d+)\.(\d+)\.(\d+) ")
-    m = p.match(version_output)
-    if m is None:
-        raise Exception(
-            "Unable to detect benchmark tool version, and the benchmark requires a min version: {}".format(
-                benchmark_min_tool_version))
-    major = m.group(1)
-    minor = m.group(2)
-    patch = m.group(3)
-    if major < benchmark_min_tool_version_major or (
-            major == benchmark_min_tool_version_major and minor < benchmark_min_tool_version_minor) or (
-            major == benchmark_min_tool_version_major and minor == benchmark_min_tool_version_minor and patch < benchmark_min_tool_version_patch):
-        raise Exception(
-            "Detected benchmark version that is inferior than the minimum required. {} < {}".format(
-                version_output, benchmark_min_tool_version))