cpu status properly summarizing even on high client overhead (#317)

Author: filipecosta90

poetry.lock

@@ -14,7 +14,7 @@ python = "^3.6.1"
 humanize = "^2.4.0"
 requests = "^2.23.0"
 py_cpuinfo = "^5.0.0"
-redis = "^4.1.4"
+redis = { git = "https://github.com/redis/redis-py.git", tag = "v4.2.0rc3" }
 boto3 = "^1.13.24"
 tqdm = "^4.46.1"
 toml = "^0.10.1"
@@ -43,6 +43,7 @@ psutil = "^5.6.6"
 scipy = "^1.3.3"
 scikit-learn = "^0.22.2"
 Jinja2 = "^3.0.3"
+watchdog = "^2.1.6"
 
 [tool.poetry.dev-dependencies]
 pytest = "^4.6"

pyproject.toml

@@ -132,32 +132,43 @@ def collect_redis_metrics(
 
 
 def from_info_to_overall_shard_cpu(benchmark_cpu_stats):
+    import numpy as np
+
     total_avg_cpu_pct = 0.0
     res = {}
     for shard_n, cpu_stats_arr in benchmark_cpu_stats.items():
         avg_cpu_pct = None
+        shards_cpu_arr = []
         # we need at least 2 elements to compute the cpu usage
         if len(cpu_stats_arr) >= 2:
-            stats_start_pos = cpu_stats_arr[0]
-            stats_end_pos = cpu_stats_arr[len(cpu_stats_arr) - 1]
-            if (
-                "server_time_usec" in stats_end_pos
-                and "server_time_usec" in stats_start_pos
-            ):
-                start_ts_micros = stats_start_pos["server_time_usec"]
-                end_ts_micros = stats_end_pos["server_time_usec"]
-                start_total_cpu = get_total_cpu(stats_start_pos)
-                end_total_cpu = get_total_cpu(stats_end_pos)
-                total_secs = (end_ts_micros - start_ts_micros) / 1000000
-                total_cpu_usage = end_total_cpu - start_total_cpu
-                avg_cpu_pct = 100.0 * (total_cpu_usage / total_secs)
+            for start_pos in range(0, len(cpu_stats_arr) - 2):
+                avg_cpu_pct = get_avg_cpu_pct(
+                    avg_cpu_pct, cpu_stats_arr[start_pos], cpu_stats_arr[start_pos + 1]
+                )
+                if avg_cpu_pct is not None:
+                    shards_cpu_arr.append(avg_cpu_pct)
+
+            avg_cpu_pct = np.percentile(shards_cpu_arr, 75)
 
         res[shard_n] = avg_cpu_pct
         if avg_cpu_pct is not None:
             total_avg_cpu_pct += avg_cpu_pct
     return total_avg_cpu_pct, res
 
 
+def get_avg_cpu_pct(avg_cpu_pct, stats_start_pos, stats_end_pos):
+    avg_cpu_pct = None
+    if "server_time_usec" in stats_end_pos and "server_time_usec" in stats_start_pos:
+        start_ts_micros = stats_start_pos["server_time_usec"]
+        end_ts_micros = stats_end_pos["server_time_usec"]
+        start_total_cpu = get_total_cpu(stats_start_pos)
+        end_total_cpu = get_total_cpu(stats_end_pos)
+        total_secs = (end_ts_micros - start_ts_micros) / 1000000
+        total_cpu_usage = end_total_cpu - start_total_cpu
+        avg_cpu_pct = 100.0 * (total_cpu_usage / total_secs)
+    return avg_cpu_pct
+
+
 def get_total_cpu(info_data):
     total_cpu = 0.0
     total_cpu = total_cpu + info_data["used_cpu_sys"]

redisbench_admin/run/metrics.py

@@ -306,6 +306,11 @@ def run_local_command_logic(args, project_name, project_version):
                                         total_shards_cpu_usage
                                     )
                                 )
+                                logging.info(
+                                    "CPU MAP: {}".format(
+                                        json.dumps(cpu_usage_map, indent=2)
+                                    )
+                                )
                                 benchmark_duration_seconds = (
                                     calculate_client_tool_duration_and_check(
                                         benchmark_end_time, benchmark_start_time

redisbench_admin/run_local/run_local.py

@@ -28,7 +28,16 @@ def check_dataset_local_requirements(
     full_path = None
     tmp_path = None
     if dbconfig_keyname in benchmark_config:
-        for k in benchmark_config[dbconfig_keyname]:
+        entry_type = type(benchmark_config[dbconfig_keyname])
+        if entry_type == list:
+            for k in benchmark_config[dbconfig_keyname]:
+                if "dataset" in k:
+                    dataset = k["dataset"]
+                    full_path = k["dataset"]
+                if "dataset_name" in k:
+                    dataset_name = k["dataset_name"]
+        if entry_type == dict:
+            k = benchmark_config[dbconfig_keyname]
             if "dataset" in k:
                 dataset = k["dataset"]
                 full_path = k["dataset"]