[fix] local oss-cluster runner/profiler running as expected. [add] Printing summary result at the end of each benchmark (#232)

Author: filipecosta90

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "redisbench-admin"
-version = "0.5.10"
+version = "0.5.11"
 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]>","Redis Performance Group <[email protected]>"]
 readme = "README.md"

redisbench_admin/environments/oss_cluster.py

@@ -29,7 +29,7 @@ def spin_up_local_redis_cluster(
         shard_port = master_shard_id + start_port - 1
 
         command = generate_cluster_redis_server_args(
-            dbdir, local_module_file, ip, shard_port, configuration_parameters
+            dbdir, local_module_file, ip, shard_port, configuration_parameters, "no"
         )
 
         logging.info(
@@ -41,7 +41,7 @@ def spin_up_local_redis_cluster(
         r = redis.StrictRedis(port=shard_port)
         result = wait_for_conn(r, dataset_load_timeout_secs)
         if result is True:
-            logging.info("Redis available")
+            logging.info("Redis available. pid={}".format(redis_process.pid))
         redis_conns.append(r)
         redis_processes.append(redis_process)
     return redis_processes, redis_conns
@@ -131,6 +131,7 @@ def generate_cluster_redis_server_args(
     ip,
     port,
     configuration_parameters=None,
+    daemonize="yes",
 ):
     # start redis-server
     command = [
@@ -142,7 +143,7 @@ def generate_cluster_redis_server_args(
         "--cluster-enabled",
         "yes",
         "--daemonize",
-        "yes",
+        daemonize,
         "--dbfilename",
         get_cluster_dbfilename(port),
         "--protected-mode",

redisbench_admin/profilers/perf.py

@@ -170,8 +170,10 @@ def _is_alive(self, process):
             return False
         # Check if child process has terminated. Set and return returncode
         # attribute
-        if process.poll() is None:
+        ret = process.poll()
+        if ret is None:
             return True
+        self.profiler_process_exit_code = ret
         return False
 
     def stop_profile(self, **kwargs):
@@ -182,8 +184,16 @@ def stop_profile(self, **kwargs):
         self.profile_end_time = time.time()
         if not self._is_alive(self.profiler_process):
             logging.error(
-                "Profiler process is not alive, might have crash during test execution, "
+                "Profiler process is not alive, might have crash during test execution.  Exit code: {}".format(
+                    self.profiler_process_exit_code
+                )
             )
+            (
+                self.profiler_process_stdout,
+                self.profiler_process_stderr,
+            ) = self.profiler_process.communicate()
+            logging.error("Profiler stderr: {}".format(self.profiler_process_stderr))
+            logging.error("Profiler stdout: {}".format(self.profiler_process_stdout))
             return result
         try:
             self.profiler_process.terminate()

redisbench_admin/run/args.py

@@ -22,6 +22,7 @@
 DEFAULT_TRIGGERING_ENV = socket.gethostname()
 TRIGGERING_ENV = os.getenv("TRIGGERING_ENV", DEFAULT_TRIGGERING_ENV)
 ENV = os.getenv("ENV", "oss-standalone,oss-cluster")
+SETUP = os.getenv("SETUP", "")
 PUSH_S3 = bool(os.getenv("PUSH_S3", False))
 PROFILERS_DSO = os.getenv("PROFILERS_DSO", None)
 PROFILERS_ENABLED = bool(os.getenv("PROFILE", 0))
@@ -121,6 +122,12 @@ def common_run_args(parser):
         "--allowed-envs",
         type=str,
         default=ENV,
-        help="Comma delimited allowed setups: 'oss-standalone','oss-cluster'",
+        help="Comma delimited allowed topologies: 'oss-standalone','oss-cluster'",
+    )
+    parser.add_argument(
+        "--allowed-setups",
+        type=str,
+        default=SETUP,
+        help="Comma delimited allowed setups. By default all setups are allowed.",
     )
     return parser

redisbench_admin/run/common.py

@@ -10,6 +10,7 @@
 import os
 import time
 import redis
+from pytablewriter import MarkdownTableWriter
 
 from redisbench_admin.run.aibench_run_inference_redisai_vision.aibench_run_inference_redisai_vision import (
     prepare_aibench_benchmark_command,
@@ -18,6 +19,7 @@
 from redisbench_admin.run.memtier_benchmark.memtier_benchmark import (
     prepare_memtier_benchmark_command,
 )
+from redisbench_admin.run.metrics import extract_results_table
 from redisbench_admin.run.redis_benchmark.redis_benchmark import (
     prepare_redis_benchmark_command,
 )
@@ -570,3 +572,27 @@ def common_properties_log(
     logging.info("\tgithub_sha: {}".format(tf_github_sha))
     logging.info("\ttriggering env: {}".format(tf_triggering_env))
     logging.info("\tsetup_name sufix: {}".format(tf_setup_name_sufix))
+
+
+def print_results_table_stdout(
+    benchmark_config, default_metrics, results_dict, setup_name, test_name
+):
+    # check which metrics to extract
+    (_, metrics,) = merge_default_and_config_metrics(
+        benchmark_config,
+        default_metrics,
+        None,
+    )
+    table_name = "Results for {} test-case on {} topology".format(test_name, setup_name)
+    results_matrix_headers = [
+        "Metric JSON Path",
+        "Metric Value",
+    ]
+    results_matrix = extract_results_table(metrics, results_dict)
+    results_matrix = [[x[0], "{:.3f}".format(x[3])] for x in results_matrix]
+    writer = MarkdownTableWriter(
+        table_name=table_name,
+        headers=results_matrix_headers,
+        value_matrix=results_matrix,
+    )
+    writer.write_table()

redisbench_admin/run/metrics.py

@@ -0,0 +1,83 @@
+#  BSD 3-Clause License
+#
+#  Copyright (c) 2021., Redis Labs Modules
+#  All rights reserved.
+#
+import logging
+
+from jsonpath_ng import parse
+
+
+def extract_results_table(
+    metrics,
+    results_dict,
+):
+    results_matrix = []
+    cleaned_metrics = []
+    already_present_metrics = []
+    # insert first the dict metrics
+    for jsonpath in metrics:
+        if type(jsonpath) == dict:
+            cleaned_metrics.append(jsonpath)
+            metric_jsonpath = list(jsonpath.keys())[0]
+            already_present_metrics.append(metric_jsonpath)
+    for jsonpath in metrics:
+        if type(jsonpath) == str:
+            if jsonpath not in already_present_metrics:
+                already_present_metrics.append(jsonpath)
+                cleaned_metrics.append(jsonpath)
+
+    for jsonpath in cleaned_metrics:
+        test_case_targets_dict = {}
+        metric_jsonpath = jsonpath
+        find_res = None
+        try:
+            if type(jsonpath) == str:
+                jsonpath_expr = parse(jsonpath)
+            if type(jsonpath) == dict:
+                metric_jsonpath = list(jsonpath.keys())[0]
+                test_case_targets_dict = jsonpath[metric_jsonpath]
+                jsonpath_expr = parse(metric_jsonpath)
+            find_res = jsonpath_expr.find(results_dict)
+        except Exception:
+            pass
+        finally:
+            if find_res is not None:
+                use_metric_context_path = False
+                if len(find_res) > 1:
+                    use_metric_context_path = True
+                for metric in find_res:
+                    metric_name = str(metric.path)
+                    metric_value = float(metric.value)
+                    metric_context_path = str(metric.context.path)
+                    if metric_jsonpath[0] == "$":
+                        metric_jsonpath = metric_jsonpath[1:]
+                    if metric_jsonpath[0] == ".":
+                        metric_jsonpath = metric_jsonpath[1:]
+
+                    # retro-compatible naming
+                    if use_metric_context_path is False:
+                        metric_name = metric_jsonpath
+
+                    metric_name = metric_name.replace("'", "")
+                    metric_name = metric_name.replace('"', "")
+                    metric_name = metric_name.replace("(", "")
+                    metric_name = metric_name.replace(")", "")
+                    metric_name = metric_name.replace(" ", "_")
+
+                    results_matrix.append(
+                        [
+                            metric_jsonpath,
+                            metric_context_path,
+                            metric_name,
+                            metric_value,
+                            test_case_targets_dict,
+                            use_metric_context_path,
+                        ]
+                    )
+
+            else:
+                logging.warning(
+                    "Unable to find metric path {} in result dict".format(jsonpath)
+                )
+    return results_matrix

redisbench_admin/run_local/local_db.py

@@ -5,16 +5,28 @@
 #
 import logging
 import tempfile
+import datetime
 
 import redis
 
+from redisbench_admin.run.run import calculate_client_tool_duration_and_check
+from redisbench_admin.run_local.local_helpers import (
+    check_benchmark_binaries_local_requirements,
+    run_local_benchmark,
+)
+
 from redisbench_admin.environments.oss_cluster import (
     spin_up_local_redis_cluster,
     setup_redis_cluster_from_conns,
 )
 from redisbench_admin.environments.oss_standalone import spin_up_local_redis
 from redisbench_admin.run.cluster import cluster_init_steps
-from redisbench_admin.run.common import run_redis_pre_steps
+from redisbench_admin.run.common import (
+    run_redis_pre_steps,
+    check_dbconfig_tool_requirement,
+    prepare_benchmark_parameters,
+    dbconfig_keyspacelen_check,
+)
 from redisbench_admin.utils.benchmark_config import extract_redis_dbconfig_parameters
 from redisbench_admin.utils.local import (
     check_dataset_local_requirements,
@@ -101,15 +113,64 @@ def local_db_spin(
             dataset_load_timeout_secs,
         )
 
-        for redis_process in redis_processes:
-            if is_process_alive(redis_process) is False:
-                raise Exception("Redis process is not alive. Failing test.")
-
         r = redis.StrictRedis(port=args.port)
         redis_conns.append(r)
+
+    for shardn, redis_process in enumerate(redis_processes):
+        logging.info(
+            "Checking if shard #{} process with pid={} is alive".format(
+                shardn + 1, redis_process.pid
+            )
+        )
+        if is_process_alive(redis_process) is False:
+            raise Exception("Redis process is not alive. Failing test.")
+
     if setup_type == "oss-cluster":
 
         cluster_init_steps(clusterconfig, redis_conns, local_module_file)
 
+    if check_dbconfig_tool_requirement(benchmark_config):
+        logging.info("Detected the requirements to load data via client tool")
+        local_benchmark_output_filename = "{}/load-data.txt".format(temporary_dir)
+        (
+            benchmark_tool,
+            full_benchmark_path,
+            benchmark_tool_workdir,
+        ) = check_benchmark_binaries_local_requirements(
+            benchmark_config, args.allowed_tools, "./binaries", "dbconfig"
+        )
+
+        # prepare the benchmark command
+        command, command_str = prepare_benchmark_parameters(
+            benchmark_config,
+            full_benchmark_path,
+            args.port,
+            "localhost",
+            local_benchmark_output_filename,
+            False,
+            benchmark_tool_workdir,
+            cluster_api_enabled,
+            "dbconfig",
+        )
+
+        # run the benchmark
+        load_via_benchmark_start_time = datetime.datetime.now()
+        run_local_benchmark(benchmark_tool, command)
+        load_via_benchmark_end_time = datetime.datetime.now()
+        load_via_benchmark_duration_seconds = calculate_client_tool_duration_and_check(
+            load_via_benchmark_end_time, load_via_benchmark_start_time
+        )
+        logging.info(
+            "Loading data via benchmark tool took {} secs.".format(
+                load_via_benchmark_duration_seconds
+            )
+        )
+
+    dbconfig_keyspacelen_check(
+        benchmark_config,
+        redis_conns,
+    )
+
     run_redis_pre_steps(benchmark_config, redis_conns[0], required_modules)
+
     return cluster_api_enabled, redis_conns, redis_processes

redisbench_admin/run_local/local_helpers.py

@@ -29,7 +29,10 @@ def run_local_benchmark(benchmark_tool, command):
 
 
 def check_benchmark_binaries_local_requirements(
-    benchmark_config, allowed_tools, binaries_localtemp_dir="./binaries"
+    benchmark_config,
+    allowed_tools,
+    binaries_localtemp_dir="./binaries",
+    config_key="clientconfig",
 ):
     (
         benchmark_min_tool_version,
@@ -40,7 +43,7 @@ def check_benchmark_binaries_local_requirements(
         tool_source,
         tool_source_bin_path,
         _,
-    ) = extract_benchmark_tool_settings(benchmark_config)
+    ) = extract_benchmark_tool_settings(benchmark_config, config_key)
     which_benchmark_tool = None
     if benchmark_tool is not None:
         logging.info("Detected benchmark config tool {}".format(benchmark_tool))

redisbench_admin/run_local/run_local.py

@@ -19,6 +19,7 @@
     BENCHMARK_REPETITIONS,
     get_setup_type_and_primaries_count,
     dso_check,
+    print_results_table_stdout,
 )
 from redisbench_admin.run.redistimeseries import datasink_profile_tabular_data
 from redisbench_admin.run.run import (
@@ -108,7 +109,7 @@ def run_local_command_logic(args, project_name, project_version):
 
     (
         benchmark_definitions,
-        _,
+        default_metrics,
         _,
         default_specs,
         clusterconfig,
@@ -141,6 +142,15 @@ def run_local_command_logic(args, project_name, project_version):
                             setup_type,
                             shard_count,
                         ) = get_setup_type_and_primaries_count(setup_settings)
+                        if args.allowed_setups != "":
+                            allowed_setups = args.allowed_setups.split()
+                            if setup_name not in allowed_setups:
+                                logging.warning(
+                                    "SKIPPING setup named {} of topology type {}.".format(
+                                        setup_name, setup_type
+                                    )
+                                )
+                                continue
                         if setup_type in args.allowed_envs:
                             redis_processes = []
                             # after we've spinned Redis, even on error we should always teardown
@@ -314,6 +324,13 @@ def run_local_command_logic(args, project_name, project_version):
                                     local_benchmark_output_filename, "r"
                                 ) as json_file:
                                     results_dict = json.load(json_file)
+                                    print_results_table_stdout(
+                                        benchmark_config,
+                                        default_metrics,
+                                        results_dict,
+                                        setup_name,
+                                        test_name,
+                                    )
 
                                     # check KPIs
                                     return_code = results_dict_kpi_check(