YCSB local run working as expected. Improved remote tool specification on YAML (#75)

* [wip] YCSB local run ( requires post processing of results )

* [add] Made tool source remote config more verbose/easy to use

* Bumping version from 0.1.62 to 0.1.63

Author: filipecosta90

poetry.lock

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "redisbench-admin"
-version = "0.1.62"
+version = "0.1.63"
 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"
@@ -33,6 +33,7 @@ pytest-cov = "^2.9.0"
 codecov = "^2.1.4"
 black = "^20.8b1"
 flake8 = "^3.9.1"
+docker = "^5.0.0"
 
 [build-system]
 requires = ["poetry>=0.12"]

pyproject.toml

@@ -29,6 +29,7 @@
 def extract_benchmark_tool_settings(benchmark_config):
     benchmark_tool = None
     benchmark_tool_source = None
+    benchmark_tool_source_inner_path = None
     benchmark_min_tool_version = None
     benchmark_min_tool_version_major = None
     benchmark_min_tool_version_minor = None
@@ -38,7 +39,12 @@ def extract_benchmark_tool_settings(benchmark_config):
         if "tool" in entry:
             benchmark_tool = entry["tool"]
         if "tool_source" in entry:
-            benchmark_tool_source = entry["tool_source"]
+            for inner_entry in entry["tool_source"]:
+                if "remote" in inner_entry:
+                    benchmark_tool_source = inner_entry["remote"]
+                if "bin_path" in inner_entry:
+                    benchmark_tool_source_inner_path = inner_entry["bin_path"]
+
         if "min-tool-version" in entry:
             benchmark_min_tool_version = entry["min-tool-version"]
             p = re.compile(r"(\d+)\.(\d+)\.(\d+)")
@@ -60,6 +66,7 @@ def extract_benchmark_tool_settings(benchmark_config):
         benchmark_min_tool_version_patch,
         benchmark_tool,
         benchmark_tool_source,
+        benchmark_tool_source_inner_path,
     )
 
 

redisbench_admin/run/common.py

@@ -1,3 +1,7 @@
+import csv
+import re
+
+
 def prepare_ycsb_benchmark_command(
     executable_path: str,
     server_private_ip: object,
@@ -29,7 +33,7 @@ def prepare_ycsb_benchmark_command(
             step = k["step"]
         if "workload" in k:
             workload = k["workload"]
-            if workload.startswith("./"):
+            if current_workdir is not None and workload.startswith("./"):
                 workload = "{}{}".format(current_workdir, workload[1:])
         if "threads" in k:
             threads = k["threads"]
@@ -49,9 +53,36 @@ def prepare_ycsb_benchmark_command(
 
     for prop in override_workload_properties:
         for k, v in prop.items():
-            if type(v) == str and v.startswith("./"):
+            if current_workdir is not None and type(v) == str and v.startswith("./"):
                 v = "{}{}".format(current_workdir, v[1:])
-            command_arr.extend(["-p", '"{}={}"'.format(k, v)])
+            command_arr.extend(["-p", "{}={}".format(k, v)])
 
     command_str = " ".join(command_arr)
     return command_arr, command_str
+
+
+def post_process_ycsb_results(stdout, start_time_ms, start_time_str):
+    results_dict = {
+        "Tests": {},
+        "StartTime": start_time_ms,
+        "StartTimeHuman": start_time_str,
+    }
+    if type(stdout) == bytes:
+        stdout = stdout.decode("ascii")
+    csv_data = list(csv.reader(stdout.splitlines(), delimiter=","))
+    start_row = 0
+    for row in csv_data:
+        if len(row) >= 1:
+            if "[OVERALL]" in row[0]:
+                break
+        start_row = start_row + 1
+    for row in csv_data[start_row:]:
+        if len(row) >= 3:
+            op_group = row[0].strip()[1:-1]
+            metric_name = row[1].strip()
+            metric_name = re.sub("[^0-9a-zA-Z]+", "_", metric_name)
+            value = row[2].strip()
+            if op_group not in results_dict["Tests"]:
+                results_dict["Tests"][op_group] = {}
+            results_dict["Tests"][op_group][metric_name] = value
+    return results_dict

redisbench_admin/run/ycsb/ycsb.py

@@ -3,7 +3,6 @@
 import os
 import pathlib
 import shutil
-import stat
 import subprocess
 import sys
 import tempfile
@@ -22,6 +21,7 @@
     redis_benchmark_from_stdout_csv_to_json,
     redis_benchmark_ensure_min_version_local,
 )
+from redisbench_admin.run.ycsb.ycsb import post_process_ycsb_results
 from redisbench_admin.utils.local import (
     spin_up_local_redis,
     get_local_run_full_filename,
@@ -230,6 +230,19 @@ def post_process_benchmark_results(
         )
         with open(local_benchmark_output_filename, "w") as json_file:
             json.dump(results_dict, json_file, indent=True)
+    if benchmark_tool == "ycsb":
+        logging.info(
+            "Converting ycsb output to json. Storing it in: {}".format(
+                local_benchmark_output_filename
+            )
+        )
+        results_dict = post_process_ycsb_results(
+            stdout.decode("ascii"),
+            start_time_ms,
+            start_time_str,
+        )
+        with open(local_benchmark_output_filename, "w") as json_file:
+            json.dump(results_dict, json_file, indent=True)
 
 
 def check_benchmark_binaries_local_requirements(
@@ -242,6 +255,7 @@ def check_benchmark_binaries_local_requirements(
         benchmark_min_tool_version_patch,
         benchmark_tool,
         tool_source,
+        tool_source_bin_path,
     ) = extract_benchmark_tool_settings(benchmark_config)
     which_benchmark_tool = None
     if benchmark_tool is not None:
@@ -255,57 +269,17 @@ def check_benchmark_binaries_local_requirements(
         logging.info("Checking benchmark tool {} is accessible".format(benchmark_tool))
         which_benchmark_tool = shutil.which(benchmark_tool)
         if which_benchmark_tool is None:
-            if tool_source is not None:
-                logging.info(
-                    "Tool {} was not detected on path. Using remote source to retrieve it: {}".format(
-                        benchmark_tool, tool_source
-                    )
-                )
-                if tool_source.startswith("http"):
-                    if not os.path.isdir(binaries_localtemp_dir):
-                        os.mkdir(binaries_localtemp_dir)
-                    filename = tool_source.split("/")[-1]
-                    full_path = "{}/{}".format(binaries_localtemp_dir, filename)
-                    if not os.path.exists(full_path):
-                        logging.info(
-                            "Retrieving remote file from {} to {}. Using the dir {} as a cache for next time.".format(
-                                tool_source, full_path, binaries_localtemp_dir
-                            )
-                        )
-                        wget.download(tool_source, full_path)
-                    logging.info(
-                        "Decompressing {} into {}.".format(
-                            full_path, binaries_localtemp_dir
-                        )
-                    )
-                    if not os.path.exists(get_decompressed_filename(full_path)):
-                        full_path = decompress_file(full_path, binaries_localtemp_dir)
-                    else:
-                        full_path = get_decompressed_filename(full_path)
-                    benchmark_tool_workdir = os.path.abspath(full_path)
-                    executable = stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH
-                    which_benchmark_tool = which_local(
-                        benchmark_tool, executable, full_path, which_benchmark_tool
-                    )
-                    if which_benchmark_tool is None:
-                        raise Exception(
-                            "Benchmark tool {} was not acessible at {}. Aborting...".format(
-                                benchmark_tool, full_path
-                            )
-                        )
-                    else:
-                        logging.info(
-                            "Reusing cached remote file (located at {} ).".format(
-                                full_path
-                            )
-                        )
-
-            else:
-                raise Exception(
-                    "Benchmark tool {} was not acessible. Aborting...".format(
-                        benchmark_tool
-                    )
-                )
+            (
+                benchmark_tool_workdir,
+                which_benchmark_tool,
+            ) = fetch_benchmark_tool_from_source_to_local(
+                benchmark_tool,
+                benchmark_tool_workdir,
+                binaries_localtemp_dir,
+                tool_source,
+                tool_source_bin_path,
+                which_benchmark_tool,
+            )
         else:
             logging.info(
                 "Tool {} was detected at {}".format(
@@ -332,6 +306,61 @@ def check_benchmark_binaries_local_requirements(
     return benchmark_tool, which_benchmark_tool, benchmark_tool_workdir
 
 
+def fetch_benchmark_tool_from_source_to_local(
+    benchmark_tool,
+    benchmark_tool_workdir,
+    binaries_localtemp_dir,
+    tool_source,
+    bin_path,
+    which_benchmark_tool,
+):
+    if tool_source is not None and bin_path is not None:
+        logging.info(
+            "Tool {} was not detected on path. Using remote source to retrieve it: {}".format(
+                benchmark_tool, tool_source
+            )
+        )
+        if tool_source.startswith("http"):
+            if not os.path.isdir(binaries_localtemp_dir):
+                os.mkdir(binaries_localtemp_dir)
+            filename = tool_source.split("/")[-1]
+            full_path = "{}/{}".format(binaries_localtemp_dir, filename)
+            if not os.path.exists(full_path):
+                logging.info(
+                    "Retrieving remote file from {} to {}. Using the dir {} as a cache for next time.".format(
+                        tool_source, full_path, binaries_localtemp_dir
+                    )
+                )
+                wget.download(tool_source, full_path)
+            logging.info(
+                "Decompressing {} into {}.".format(full_path, binaries_localtemp_dir)
+            )
+            if not os.path.exists(get_decompressed_filename(full_path)):
+                full_path = decompress_file(full_path, binaries_localtemp_dir)
+            else:
+                full_path = get_decompressed_filename(full_path)
+            benchmark_tool_workdir = os.path.abspath(full_path)
+            which_benchmark_tool = os.path.abspath(
+                "{}/{}".format(benchmark_tool_workdir, bin_path)
+            )
+            if which_benchmark_tool is None:
+                raise Exception(
+                    "Benchmark tool {} was not acessible at {}. Aborting...".format(
+                        benchmark_tool, full_path
+                    )
+                )
+            else:
+                logging.info(
+                    "Reusing cached remote file (located at {} ).".format(full_path)
+                )
+
+    else:
+        raise Exception(
+            "Benchmark tool {} was not acessible. Aborting...".format(benchmark_tool)
+        )
+    return benchmark_tool_workdir, which_benchmark_tool
+
+
 def which_local(benchmark_tool, executable, full_path, which_benchmark_tool):
     if which_benchmark_tool:
         return which_benchmark_tool

redisbench_admin/run_local/run_local.py

@@ -13,7 +13,7 @@ def create_run_remote_arguments(parser):
     parser.add_argument(
         "--allowed-tools",
         type=str,
-        default="redis-benchmark,redisgraph-benchmark-go",
+        default="redis-benchmark,redisgraph-benchmark-go,ycsb",
         help="comma separated list of allowed tools for this module. By default all the supported are allowed.",
     )
     parser.add_argument(