Fixed timeout issues when pushing benchmark timeseries data (#448)

* architecture aware benchmark client link downloads

* Fixed timeout issues when pushing benchmark timeseries data

Author: filipecosta90

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "redisbench-admin"
-version = "0.11.26"
+version = "0.11.32"
 description = "Redis benchmark run helper. A wrapper around Redis and Redis Modules benchmark tools ( ftsb_redisearch, memtier_benchmark, redis-benchmark, aibench, etc... )."
 authors = ["filipecosta90 <filipecosta.90@gmail.com>","Redis Performance Group <performance@redis.com>"]
 readme = "README.md"

redisbench_admin/compare/compare.py

@@ -49,6 +49,7 @@ def compare_command_logic(args, project_name, project_version):
         port=args.redistimeseries_port,
         password=args.redistimeseries_pass,
         username=args.redistimeseries_user,
+        retry_on_timeout=True,
     )
     rts.ping()
     default_baseline_branch = None

redisbench_admin/export/export.py

@@ -172,6 +172,7 @@ def export_command_logic(args, project_name, project_version):
         host=args.redistimeseries_host,
         port=args.redistimeseries_port,
         password=args.redistimeseries_pass,
+        retry_on_timeout=True,
     )
     try:
         rts.ping()

redisbench_admin/run/ftsb/ftsb.py

@@ -55,12 +55,16 @@ def prepare_ftsb_benchmark_command(
 
 
 def extract_ftsb_extra_links(
-    benchmark_config, benchmark_tool, config_key="clientconfig"
+    benchmark_config, benchmark_tool, config_key="clientconfig", architecture="x86_64"
 ):
     remote_tool_link = "/tmp/{}".format(benchmark_tool)
+    arch_txt = "amd64"
+    if architecture == "aarch64":
+        arch_txt = "arm64"
+
     tool_link = (
         "https://s3.amazonaws.com/benchmarks.redislabs/"
-        + "redisearch/tools/ftsb/{}_linux_amd64".format(benchmark_tool)
+        + f"redisearch/tools/ftsb/{benchmark_tool}_linux_{arch_txt}"
     )
     queries_file_link = None
     for entry in benchmark_config[config_key]:

redisbench_admin/run/run.py

@@ -124,7 +124,7 @@ def define_benchmark_plan(benchmark_definitions, default_specs):
                     logging.info(f"FINAL DB CONFIG: {final_db_config}")
                     test_benchmark_config["dbconfig"] = final_db_config
 
-                logging.info(
+                logging.debug(
                     f"final benchmark config for setup: {setup_name} and test: {test_name}. {test_benchmark_config}"
                 )
                 # add benchmark

redisbench_admin/run_local/run_local.py

@@ -129,6 +129,7 @@ def run_local_command_logic(args, project_name, project_version):
             host=args.redistimeseries_host,
             port=args.redistimeseries_port,
             password=args.redistimeseries_pass,
+            retry_on_timeout=True,
         )
         rts.ping()
 

redisbench_admin/run_remote/remote_client.py

@@ -53,6 +53,7 @@ def run_remote_client_tool(
     redis_conns=[],
     do_post_process=True,
     redis_password=None,
+    architecture="x86_64",
 ):
     (
         benchmark_min_tool_version,
@@ -83,6 +84,7 @@ def run_remote_client_tool(
         arch_str,
         client_ssh_port,
         private_key,
+        architecture,
     )
     if "ann-benchmarks" in benchmark_tool:
         logging.info(

redisbench_admin/run_remote/remote_db.py

@@ -104,6 +104,7 @@ def remote_db_spin(
     ignore_keyspace_errors=False,
     continue_on_module_check_error=False,
     keyspace_check_timeout=60,
+    architecture="x86_64",
 ):
     (
         _,
@@ -135,6 +136,7 @@ def remote_db_spin(
             username,
             continue_on_module_check_error,
         )
+        logging.info(f"final remote module files {remote_module_files}...")
     # setup Redis
     redis_setup_result = True
     redis_conns = []
@@ -337,6 +339,7 @@ def remote_db_spin(
             [],
             False,
             redis_password,
+            architecture,
         )
         logging.info(
             "Finished loading the data via client tool. Took {} seconds. Result={}".format(

redisbench_admin/run_remote/remote_helpers.py

@@ -66,8 +66,11 @@ def remote_tool_pre_bench_step(
     arch_str,
     client_ssh_port,
     private_key,
+    architecture="x86_64",
 ):
-    logging.info("Settting up remote tool {} requirements".format(benchmark_tool))
+    logging.info(
+        f"Settting up remote tool {benchmark_tool} requirements. architecture ={architecture}"
+    )
     if benchmark_tool == "redisgraph-benchmark-go":
         setup_remote_benchmark_tool_redisgraph_benchmark_go(
             client_public_ip,
@@ -90,7 +93,9 @@ def remote_tool_pre_bench_step(
             queries_file_link,
             remote_tool_link,
             tool_link,
-        ) = extract_ftsb_extra_links(benchmark_config, benchmark_tool, config_key)
+        ) = extract_ftsb_extra_links(
+            benchmark_config, benchmark_tool, config_key, architecture
+        )
         logging.info(
             "FTSB Extracted:\nremote tool input: {}\nremote tool link: {}\ntool path: {}".format(
                 queries_file_link, remote_tool_link, tool_link

redisbench_admin/run_remote/run_remote.py

@@ -276,6 +276,7 @@ def run_remote_command_logic(args, project_name, project_version):
             host=args.redistimeseries_host,
             port=args.redistimeseries_port,
             password=args.redistimeseries_pass,
+            retry_on_timeout=True,
         )
         rts.ping()
 
@@ -553,6 +554,8 @@ def run_remote_command_logic(args, project_name, project_version):
                                             flushall_on_every_test_start,
                                             ignore_keyspace_errors,
                                             continue_on_module_check_error,
+                                            60,
+                                            architecture,
                                         )
                                         if benchmark_type == "read-only":
                                             ro_benchmark_set(
@@ -698,6 +701,7 @@ def run_remote_command_logic(args, project_name, project_version):
                                         redis_conns,
                                         True,
                                         redis_password,
+                                        architecture,
                                     )
 
                                     if profilers_enabled:

redisbench_admin/run_remote/standalone.py

@@ -75,7 +75,9 @@ def remote_module_files_cp(
     continue_on_module_check_error=False,
 ):
     remote_module_files = []
+
     if local_module_files is not None:
+        logging.info(f"there is a total of  {len(local_module_files)} modules")
         for local_module_file in local_module_files:
             splitted_module_and_plugins = []
             if type(local_module_file) is str:
@@ -135,7 +137,7 @@ def remote_module_files_cp(
                 if pos > 1:
                     remote_module_files_in = remote_module_files_in + " "
                 remote_module_files_in = remote_module_files_in + remote_module_file
-        remote_module_files.append(remote_module_files_in)
+            remote_module_files.append(remote_module_files_in)
     logging.info(
         "There are a total of {} remote files {}".format(
             len(remote_module_files), remote_module_files

redisbench_admin/utils/remote.py

@@ -9,7 +9,7 @@
 import os
 import sys
 import tempfile
-
+import time
 import git
 import paramiko
 import pysftp
@@ -93,13 +93,22 @@ def copy_file_to_remote_setup(
             cnopts=cnopts,
             port=port,
         )
-        srv.put(full_local_path, remote_file, callback=view_bar_simple)
-        srv.close()
+        if srv.exists(remote_file):
+            print(f"Remote file {remote_file} already exists. Skipping the copy.")
+        else:
+            uploaded_file = srv.put(
+                full_local_path, remote_file, callback=view_bar_simple
+            )
+            print(f"File uploaded to: {uploaded_file}")
         logging.info(
             "Finished Copying file {} to remote server {} ".format(
                 full_local_path, remote_file
             )
         )
+        file_info = srv.stat(remote_file)
+        print(f"Remote file size: {file_info.st_size} bytes")
+        print(f"Remote file permissions: {oct(file_info.st_mode)[-3:]}")
+        srv.close()
         res = True
     else:
         if continue_on_module_check_error:
@@ -314,6 +323,9 @@ def setup_remote_environment(
         },
         raise_on_error=True,
     )
+    infra_wait_secs = 60
+    logging.warning(f"Infra ready wait... for {infra_wait_secs} secs")
+    time.sleep(infra_wait_secs)
     return retrieve_tf_connection_vars(return_code, tf)
 
 
@@ -616,42 +628,50 @@ def push_data_to_redistimeseries(rts, time_series_dict: dict, expire_msecs=0):
             unit="benchmark time-series", total=len(time_series_dict.values())
         )
         for timeseries_name, time_series in time_series_dict.items():
-            exporter_create_ts(rts, time_series, timeseries_name)
-            for timestamp, value in time_series["data"].items():
-                try:
-                    if timestamp is None:
-                        logging.warning("The provided timestamp is null. Using auto-ts")
-                        rts.ts().add(
-                            timeseries_name,
-                            value,
-                            duplicate_policy="last",
-                        )
-                    else:
-                        rts.ts().add(
-                            timeseries_name,
-                            timestamp,
-                            value,
-                            duplicate_policy="last",
+            try:
+                exporter_create_ts(rts, time_series, timeseries_name)
+                for timestamp, value in time_series["data"].items():
+                    try:
+                        if timestamp is None:
+                            logging.warning(
+                                "The provided timestamp is null. Using auto-ts"
+                            )
+                            rts.ts().add(
+                                timeseries_name,
+                                value,
+                                duplicate_policy="last",
+                            )
+                        else:
+                            rts.ts().add(
+                                timeseries_name,
+                                timestamp,
+                                value,
+                                duplicate_policy="last",
+                            )
+                        datapoint_inserts += 1
+                    except redis.exceptions.DataError:
+                        logging.warning(
+                            "Error while inserting datapoint ({} : {}) in timeseries named {}. ".format(
+                                timestamp, value, timeseries_name
+                            )
                         )
-                    datapoint_inserts += 1
-                except redis.exceptions.DataError:
-                    logging.warning(
-                        "Error while inserting datapoint ({} : {}) in timeseries named {}. ".format(
-                            timestamp, value, timeseries_name
+                        datapoint_errors += 1
+                        pass
+                    except redis.exceptions.ResponseError:
+                        logging.warning(
+                            "Error while inserting datapoint ({} : {}) in timeseries named {}. ".format(
+                                timestamp, value, timeseries_name
+                            )
                         )
-                    )
-                    datapoint_errors += 1
-                    pass
-                except redis.exceptions.ResponseError:
-                    logging.warning(
-                        "Error while inserting datapoint ({} : {}) in timeseries named {}. ".format(
-                            timestamp, value, timeseries_name
-                        )
-                    )
-                    datapoint_errors += 1
-                    pass
-            if expire_msecs > 0:
-                rts.pexpire(timeseries_name, expire_msecs)
+                        datapoint_errors += 1
+                        pass
+                if expire_msecs > 0:
+                    rts.pexpire(timeseries_name, expire_msecs)
+            except redis.exceptions.TimeoutError:
+                logging.error(
+                    f"Error while working in timeseries named {timeseries_name}. "
+                )
+                datapoint_errors += 1
             progress.update()
     return datapoint_errors, datapoint_inserts
 

redisbench_admin/watchdog/watchdog.py

@@ -153,6 +153,7 @@ def watchdog_command_logic(args, project_name, project_version):
         host=args.redistimeseries_host,
         port=args.redistimeseries_port,
         password=args.redistimeseries_pass,
+        retry_on_timeout=True,
     )
     rts.ping()
     ec2_client = boto3.client("ec2")