[add] Refactored run-local and run-remote to use more common flows. Preparing run-remote to allow oss-cluster benchmarks (#156)

Author: filipecosta90

redisbench_admin/__init__.py

@@ -4,4 +4,11 @@
 #  All rights reserved.
 #
 
-__version__ = "0.1.30"
+# This attribute is the only one place that the version number is written down,
+# so there is only one place to change it when the version number changes.
+import pkg_resources
+
+try:
+    __version__ = pkg_resources.get_distribution("redisbench-admin").version
+except (pkg_resources.DistributionNotFound, AttributeError):
+    __version__ = "99.99.99"  # like redis modules

redisbench_admin/cli.py

@@ -123,17 +123,17 @@ def main():
         )
 
     if requested_tool == "run-local":
-        run_local_command_logic(args)
+        run_local_command_logic(args, project_name, project_version)
     if requested_tool == "run-remote":
-        run_remote_command_logic(args)
+        run_remote_command_logic(args, project_name, project_version)
     if requested_tool == "export":
-        export_command_logic(args)
+        export_command_logic(args, project_name, project_version)
     if requested_tool == "extract":
-        extract_command_logic(args)
+        extract_command_logic(args, project_name, project_version)
     if requested_tool == "watchdog":
-        watchdog_command_logic(args)
+        watchdog_command_logic(args, project_name, project_version)
     if requested_tool == "compare":
-        compare_command_logic(args)
+        compare_command_logic(args, project_name, project_version)
 
 
 def print_invalid_tool_option(requested_tool, valid_tool_options):

redisbench_admin/compare/args.py

@@ -24,7 +24,7 @@
 ) = extract_git_vars()
 
 _, NOW_UTC, _ = get_start_time_vars()
-LAST_24_HOURS_UTC = NOW_UTC - (24 * 60 * 60 * 1000)
+LAST_WEEK_UTC = NOW_UTC - (24 * 90 * 60 * 60 * 1000)
 
 
 def create_compare_arguments(parser):
@@ -58,6 +58,6 @@ def create_compare_arguments(parser):
     parser.add_argument(
         "--redistimeseries_pass", type=str, default=PERFORMANCE_RTS_AUTH
     )
-    parser.add_argument("--from_timestamp", type=int, default=LAST_24_HOURS_UTC)
+    parser.add_argument("--from_timestamp", type=int, default=LAST_WEEK_UTC)
     parser.add_argument("--to_timestamp", type=int, default=NOW_UTC)
     return parser

redisbench_admin/compare/compare.py

@@ -14,7 +14,12 @@
 from redisbench_admin.utils.utils import get_ts_metric_name
 
 
-def compare_command_logic(args):
+def compare_command_logic(args, project_name, project_version):
+    logging.info(
+        "Using: {project_name} {project_version}".format(
+            project_name=project_name, project_version=project_version
+        )
+    )
     logging.info("Checking connection to RedisTimeSeries.")
     rts = Client(
         host=args.redistimeseries_host,
@@ -59,6 +64,9 @@ def compare_command_logic(args):
     )
     profilers_artifacts_matrix = []
     detected_regressions = []
+    total_improvements = 0
+    total_stable = 0
+    total_regressions = 0
     for test_name in test_names:
 
         test_name = test_name.decode()
@@ -113,12 +121,24 @@ def compare_command_logic(args):
                 ) * 100.0
         if baseline_v != "N/A" or comparison_v != "N/A":
             detected_regression = False
+            detected_improvement = False
             if (
                 percentage_change < 0.0
                 and percentage_change < -args.regressions_percent_lower_limit
             ):
                 detected_regression = True
+                total_regressions = total_regressions + 1
                 detected_regressions.append(test_name)
+            if (
+                percentage_change > 0.0
+                and percentage_change > args.regressions_percent_lower_limit
+            ):
+                detected_improvement = True
+                total_improvements = total_improvements + 1
+
+            if detected_improvement is False and detected_regression is False:
+                total_stable = total_stable + 1
+
             if args.print_regressions_only is False or detected_regression:
                 profilers_artifacts_matrix.append(
                     [
@@ -144,13 +164,25 @@ def compare_command_logic(args):
         value_matrix=profilers_artifacts_matrix,
     )
     writer.write_table()
-    total_regressions = len(detected_regressions)
+    if total_stable > 0:
+        logging.info(
+            "Detected a total of {} stable tests between versions.".format(
+                total_stable,
+            )
+        )
+    if total_improvements > 0:
+        logging.info(
+            "Detected a total of {} improvements above the improvement water line (> {} %%)".format(
+                total_improvements, args.regressions_percent_lower_limit
+            )
+        )
     if total_regressions > 0:
         logging.warning(
-            "Detected a total of {} regressions. Printing BENCHMARK env var compatible list".format(
-                total_regressions
+            "Detected a total of {} regressions bellow the regression water line (< -{} %%)".format(
+                total_regressions, args.regressions_percent_lower_limit
             )
         )
+        logging.warning("Printing BENCHMARK env var compatible list")
         logging.warning(
             "BENCHMARK={}".format(
                 ",".join(["{}.yml".format(x) for x in detected_regressions])

redisbench_admin/export/export.py

@@ -6,6 +6,7 @@
 
 
 import json
+import logging
 import os
 
 import redis
@@ -19,7 +20,12 @@
 from redisbench_admin.utils.utils import retrieve_local_or_remote_input_json
 
 
-def export_command_logic(args):
+def export_command_logic(args, project_name, project_version):
+    logging.info(
+        "Using: {project_name} {project_version}".format(
+            project_name=project_name, project_version=project_version
+        )
+    )
     benchmark_files = args.benchmark_result_files
     local_path = os.path.abspath(args.local_dir)
     results_format = args.results_format

redisbench_admin/extract/extract.py

@@ -5,6 +5,7 @@
 #
 
 import json
+import logging
 import time
 
 import redis
@@ -14,7 +15,12 @@ def current_milli_time():
     return round(time.time() * 1000)
 
 
-def extract_command_logic(args):
+def extract_command_logic(args, project_name, project_version):
+    logging.info(
+        "Using: {project_name} {project_version}".format(
+            project_name=project_name, project_version=project_version
+        )
+    )
     redis_url = args.redis_url
     output_json = args.output_tags_json
     redis_client = redis.from_url(redis_url)

redisbench_admin/run/aibench_run_inference_redisai_vision/aibench_run_inference_redisai_vision.py

@@ -37,3 +37,18 @@ def prepare_aibench_benchmark_command(
 
     command_str = " ".join(command_arr)
     return command_arr, command_str
+
+
+def extract_aibench_extra_links(benchmark_config, benchmark_tool):
+    remote_tool_link = "/tmp/{}".format(benchmark_tool)
+    tool_link = (
+        "https://s3.amazonaws.com/benchmarks.redislabs/"
+        + "tools/redisai/aibench/{}_linux_amd64".format(benchmark_tool)
+    )
+    queries_file_link = None
+    for entry in benchmark_config["clientconfig"]:
+        if "parameters" in entry:
+            for parameter in entry["parameters"]:
+                if "file" in parameter:
+                    queries_file_link = parameter["file"]
+    return queries_file_link, remote_tool_link, tool_link

redisbench_admin/run/cluster.py

@@ -0,0 +1,84 @@
+#  BSD 3-Clause License
+#
+#  Copyright (c) 2021., Redis Labs Modules
+#  All rights reserved.
+#
+import logging
+
+import redis
+
+
+def cluster_init_steps(args, clusterconfig, local_module_file, r_conns, shard_count):
+    startup_nodes = []
+    for p in range(args.port, args.port + shard_count):
+        primary_conn = redis.StrictRedis(port=p)
+        primary_conn.execute_command("DEBUG RELOAD NOSAVE")
+        r_conns.append(primary_conn)
+        startup_nodes.append({"host": "127.0.0.1", "port": "{}".format(p)})
+    if clusterconfig is not None:
+        if "init_commands" in clusterconfig:
+            for command_group in clusterconfig["init_commands"]:
+                skip = False
+                if "when_modules_present" in command_group:
+                    m_found = False
+                    for module_required in command_group["when_modules_present"]:
+                        if type(local_module_file) == list:
+                            for local_m in local_module_file:
+                                if module_required in local_m:
+                                    m_found = True
+                                    logging.info(
+                                        "Required module {}  found in {}".format(
+                                            module_required,
+                                            local_m,
+                                        )
+                                    )
+                        else:
+                            if module_required in local_module_file:
+                                m_found = True
+                                logging.info(
+                                    "Required module {}  found in {}".format(
+                                        module_required,
+                                        local_module_file,
+                                    )
+                                )
+                    skip = not (m_found)
+                if skip is False:
+                    for command in command_group["commands"]:
+                        for conn_n, rc in enumerate(r_conns):
+                            rc.execute_command(command)
+                            logging.info(
+                                "Cluster node {}: sent command {}".format(
+                                    conn_n + 1, command
+                                )
+                            )
+                else:
+                    logging.info(
+                        "Skipping to send the command group: {}.".format(
+                            command_group["commands"],
+                        )
+                        + "Given the when_modules_present condition ({}) is not true.".format(
+                            command_group["when_modules_present"],
+                        )
+                    )
+    return startup_nodes
+
+
+# noinspection PyBroadException
+def spin_up_redis_cluster_remote_redis(
+    benchmark_config,
+    server_public_ip,
+    username,
+    private_key,
+    local_module_file,
+    remote_module_file,
+    remote_dataset_file,
+    logname,
+    dirname,
+    redis_configuration_parameters,
+    dbdir_folder,
+    shard_count,
+    port,
+):
+    raise Exception(
+        "Remote cluster is still not implemented =(. We're working hard to get it ASAP =)!!"
+    )

redisbench_admin/run/common.py

@@ -4,6 +4,7 @@
 #  All rights reserved.
 #
 import csv
+import datetime
 import datetime as dt
 import logging
 import os
@@ -23,8 +24,14 @@
     prepare_tsbs_benchmark_command,
 )
 from redisbench_admin.run.ycsb.ycsb import prepare_ycsb_benchmark_command
+from redisbench_admin.run_remote.remote_helpers import (
+    extract_module_semver_from_info_modules_cmd,
+)
 from redisbench_admin.utils.benchmark_config import (
     parse_exporter_timemetric,
+    parse_exporter_metrics_definition,
+    parse_exporter_timemetric_definition,
+    check_required_modules,
 )
 from redisbench_admin.utils.remote import (
     execute_remote_commands,
@@ -285,7 +292,9 @@ def execute_init_commands(benchmark_config, r, dbconfig_keyname="dbconfig"):
                 )
 
 
-def extract_test_feasible_setups(benchmark_config, param, default_specs):
+def extract_test_feasible_setups(
+    benchmark_config, param, default_specs, backwards_compatible=True
+):
     feasible_setups_map = {}
     if param in benchmark_config:
         feasible_setups_list = benchmark_config[param]
@@ -295,10 +304,82 @@ def extract_test_feasible_setups(benchmark_config, param, default_specs):
                 for setup in default_specs["setups"]:
                     if setup_name == setup["name"]:
                         feasible_setups_map[setup_name] = setup
+    if len(feasible_setups_map.keys()) == 0 and backwards_compatible:
+        feasible_setups_map["oss-standalone"] = {
+            "name": "oss-standalone",
+            "type": "oss-standalone",
+            "redis_topology": {"primaries": 1, "replicas": 0},
+            "resources": {"requests": {"cpu": "1000m"}, "limits": {"cpu": "2000m"}},
+        }
+        logging.info(
+            "Using a backwards compatible 'oss-standalone' setup, with settings: {}".format(
+                feasible_setups_map["oss-standalone"]
+            )
+        )
+
     return feasible_setups_map
 
 
 def get_setup_type_and_primaries_count(setup_settings):
     setup_type = setup_settings["type"]
     shard_count = setup_settings["redis_topology"]["primaries"]
     return setup_type, shard_count
+
+
+def merge_default_and_config_metrics(
+    benchmark_config, default_metrics, exporter_timemetric_path
+):
+    metrics = default_metrics
+    if "exporter" in benchmark_config:
+        extra_metrics = parse_exporter_metrics_definition(benchmark_config["exporter"])
+        metrics.extend(extra_metrics)
+        extra_timemetric_path = parse_exporter_timemetric_definition(
+            benchmark_config["exporter"]
+        )
+        if extra_timemetric_path is not None:
+            exporter_timemetric_path = extra_timemetric_path
+    return exporter_timemetric_path, metrics
+
+
+def run_redis_pre_steps(benchmark_config, r, required_modules):
+    stdout = r.execute_command("info modules")
+    (
+        module_names,
+        artifact_versions,
+    ) = extract_module_semver_from_info_modules_cmd(stdout)
+    check_required_modules(module_names, required_modules)
+    # run initialization commands before benchmark starts
+    logging.info("Running initialization commands before benchmark starts.")
+    execute_init_commands_start_time = datetime.datetime.now()
+    execute_init_commands(benchmark_config, r)
+    execute_init_commands_duration_seconds = (
+        datetime.datetime.now() - execute_init_commands_start_time
+    ).seconds
+    logging.info(
+        "Running initialization commands took {} secs.".format(
+            execute_init_commands_duration_seconds
+        )
+    )
+    return artifact_versions[0]
+
+
+def dso_check(dso, local_module_file):
+    if dso is None:
+        logging.warning("No dso specified for perf analysis {}".format(dso))
+        if local_module_file is not None:
+
+            if type(local_module_file) == str:
+                dso = local_module_file
+                logging.warning(
+                    "Using provided module = {} to specify dso".format(
+                        local_module_file
+                    )
+                )
+            if type(local_module_file) == list:
+                dso = local_module_file[0]
+                logging.warning(
+                    "Using first module = {} to specify dso".format(
+                        local_module_file[0]
+                    )
+                )
+    return dso