[fix] Fixed metric extraction expections on missing metrics. Made RTS results pushing more resilient (#109)

Author: filipecosta90

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "redisbench-admin"
-version = "0.1.76"
+version = "0.1.77"
 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"

redisbench_admin/run/common.py

@@ -145,6 +145,9 @@ def common_exporter_logic(
     tf_triggering_env,
     artifact_version="N/A",
 ):
+    per_version_time_series_dict = None
+    per_branch_time_series_dict = None
+
     if exporter_timemetric_path is not None and len(metrics) > 0:
         # extract timestamp
         datapoints_timestamp = parse_exporter_timemetric(
@@ -171,7 +174,7 @@ def common_exporter_logic(
         push_data_to_redistimeseries(rts, per_version_time_series_dict)
         if tf_github_branch is not None and tf_github_branch != "":
             # extract per branch datapoints
-            ok, branch_time_series_dict = extract_perbranch_timeseries_from_results(
+            ok, per_branch_time_series_dict = extract_perbranch_timeseries_from_results(
                 datapoints_timestamp,
                 metrics,
                 results_dict,
@@ -183,7 +186,7 @@ def common_exporter_logic(
                 tf_triggering_env,
             )
             # push per-branch data
-            push_data_to_redistimeseries(rts, branch_time_series_dict)
+            push_data_to_redistimeseries(rts, per_branch_time_series_dict)
         else:
             logging.warning(
                 "Requested to push data to RedisTimeSeries but no git"
@@ -196,6 +199,7 @@ def common_exporter_logic(
             "Requested to push data to RedisTimeSeries but "
             'no exporter definition was found. Missing "exporter" config.'
         )
+    return per_version_time_series_dict, per_branch_time_series_dict
 
 
 def get_start_time_vars(start_time=None):

redisbench_admin/run/tsbs_run_queries_redistimeseries/tsbs_run_queries_redistimeseries.py

@@ -28,17 +28,17 @@ def prepare_tsbs_benchmark_command(
     command_arr.extend(
         ["--host", "{}:{}".format(server_private_ip, server_plaintext_port)]
     )
-
-    for k in benchmark_config["parameters"]:
-        if "file" in k:
-            input_file = k["file"]
-            input_file = check_if_needs_remote_fetch(
-                input_file, "/tmp", None, remote_queries_file, is_remote
-            )
-            command_arr.extend(["--file", input_file])
-        else:
-            for kk in k.keys():
-                command_arr.extend(["--{}".format(kk), str(k[kk])])
+    if "parameters" in benchmark_config:
+        for k in benchmark_config["parameters"]:
+            if "file" in k:
+                input_file = k["file"]
+                input_file = check_if_needs_remote_fetch(
+                    input_file, "/tmp", None, remote_queries_file, is_remote
+                )
+                command_arr.extend(["--file", input_file])
+            else:
+                for kk in k.keys():
+                    command_arr.extend(["--{}".format(kk), str(k[kk])])
 
     command_arr.extend(["--results-file", result_file])
 

redisbench_admin/run_remote/run_remote.py

@@ -4,6 +4,7 @@
 import sys
 import traceback
 
+import redis
 from python_terraform import Terraform
 from redistimeseries.client import Client
 
@@ -525,31 +526,47 @@ def run_remote_command_logic(args):
                         tf_github_repo,
                         tf_triggering_env,
                     )
-                    rts.redis.sadd(testcases_setname, test_name)
-                    rts.incrby(
-                        tsname_project_total_success,
-                        1,
-                        timestamp=start_time_ms,
-                        labels=get_project_ts_tags(
-                            tf_github_org,
-                            tf_github_repo,
-                            deployment_type,
-                            tf_triggering_env,
-                        ),
-                    )
+                    try:
+                        rts.redis.sadd(testcases_setname, test_name)
+                        rts.incrby(
+                            tsname_project_total_success,
+                            1,
+                            timestamp=start_time_ms,
+                            labels=get_project_ts_tags(
+                                tf_github_org,
+                                tf_github_repo,
+                                deployment_type,
+                                tf_triggering_env,
+                            ),
+                        )
+                    except redis.exceptions.ResponseError as e:
+                        logging.warning(
+                            "Error while updating secondary data structures {}. ".format(
+                                e.__str__()
+                            )
+                        )
+                        pass
             except:
                 if args.push_results_redistimeseries:
-                    rts.incrby(
-                        tsname_project_total_failures,
-                        1,
-                        timestamp=start_time_ms,
-                        labels=get_project_ts_tags(
-                            tf_github_org,
-                            tf_github_repo,
-                            deployment_type,
-                            tf_triggering_env,
-                        ),
-                    )
+                    try:
+                        rts.incrby(
+                            tsname_project_total_failures,
+                            1,
+                            timestamp=start_time_ms,
+                            labels=get_project_ts_tags(
+                                tf_github_org,
+                                tf_github_repo,
+                                deployment_type,
+                                tf_triggering_env,
+                            ),
+                        )
+                    except redis.exceptions.ResponseError as e:
+                        logging.warning(
+                            "Error while updating secondary data structures {}. ".format(
+                                e.__str__()
+                            )
+                        )
+                        pass
                 return_code |= 1
                 logging.critical(
                     "Some unexpected exception was caught "
@@ -596,7 +613,7 @@ def redistimeseries_results_logic(
     exporter_timemetric_path, metrics = merge_default_and_config_metrics(
         benchmark_config, default_metrics, exporter_timemetric_path
     )
-    common_exporter_logic(
+    per_version_time_series_dict, per_branch_time_series_dict = common_exporter_logic(
         deployment_type,
         exporter_timemetric_path,
         metrics,
@@ -609,6 +626,7 @@ def redistimeseries_results_logic(
         tf_triggering_env,
         artifact_version,
     )
+    return per_version_time_series_dict, per_branch_time_series_dict
 
 
 def merge_default_and_config_metrics(

redisbench_admin/utils/remote.py

@@ -414,36 +414,37 @@ def fetch_remote_setup_from_config(
 def push_data_to_redistimeseries(rts: client, branch_time_series_dict: dict):
     datapoint_errors = 0
     datapoint_inserts = 0
-    for timeseries_name, time_series in branch_time_series_dict.items():
-        try:
-            logging.info(
-                "Creating timeseries named {} with labels {}".format(
-                    timeseries_name, time_series["labels"]
-                )
-            )
-            rts.create(timeseries_name, labels=time_series["labels"])
-        except redis.exceptions.ResponseError:
-            logging.warning(
-                "Timeseries named {} already exists".format(timeseries_name)
-            )
-            pass
-        for timestamp, value in time_series["data"].items():
+    if rts is not None:
+        for timeseries_name, time_series in branch_time_series_dict.items():
             try:
-                rts.add(
-                    timeseries_name,
-                    timestamp,
-                    value,
-                    duplicate_policy="last",
+                logging.info(
+                    "Creating timeseries named {} with labels {}".format(
+                        timeseries_name, time_series["labels"]
+                    )
                 )
-                datapoint_inserts += 1
+                rts.create(timeseries_name, labels=time_series["labels"])
             except redis.exceptions.ResponseError:
                 logging.warning(
-                    "Error while inserting datapoint ({} : {}) in timeseries named {}. ".format(
-                        timestamp, value, timeseries_name
-                    )
+                    "Timeseries named {} already exists".format(timeseries_name)
                 )
-                datapoint_errors += 1
                 pass
+            for timestamp, value in time_series["data"].items():
+                try:
+                    rts.add(
+                        timeseries_name,
+                        timestamp,
+                        value,
+                        duplicate_policy="last",
+                    )
+                    datapoint_inserts += 1
+                except redis.exceptions.ResponseError:
+                    logging.warning(
+                        "Error while inserting datapoint ({} : {}) in timeseries named {}. ".format(
+                            timestamp, value, timeseries_name
+                        )
+                    )
+                    datapoint_errors += 1
+                    pass
     return datapoint_errors, datapoint_inserts
 
 
@@ -470,34 +471,40 @@ def extract_perversion_timeseries_from_results(
     for jsonpath in metrics:
         jsonpath_expr = parse(jsonpath)
         metric_name = jsonpath[2:]
-        metric_value = float(jsonpath_expr.find(results_dict)[0].value)
-        # prepare tags
-        # branch tags
-        version_tags = get_project_ts_tags(
-            tf_github_org, tf_github_repo, deployment_type, tf_triggering_env
-        )
-        version_tags["version"] = project_version
-        version_tags["test_name"] = str(test_name)
-        version_tags["metric"] = str(metric_name)
-
-        ts_name = (
-            "ci.benchmarks.redislabs/by.version/"
-            "{triggering_env}/{github_org}/{github_repo}/"
-            "{test_name}/{deployment_type}/{version}/{metric}".format(
-                version=project_version,
-                github_org=tf_github_org,
-                github_repo=tf_github_repo,
-                deployment_type=deployment_type,
-                test_name=test_name,
-                triggering_env=tf_triggering_env,
-                metric=metric_name,
+        find_res = jsonpath_expr.find(results_dict)
+        if find_res is not None and len(find_res) > 0:
+            metric_value = float(find_res[0].value)
+            # prepare tags
+            # branch tags
+            version_tags = get_project_ts_tags(
+                tf_github_org, tf_github_repo, deployment_type, tf_triggering_env
+            )
+            version_tags["version"] = project_version
+            version_tags["test_name"] = str(test_name)
+            version_tags["metric"] = str(metric_name)
+
+            ts_name = (
+                "ci.benchmarks.redislabs/by.version/"
+                "{triggering_env}/{github_org}/{github_repo}/"
+                "{test_name}/{deployment_type}/{version}/{metric}".format(
+                    version=project_version,
+                    github_org=tf_github_org,
+                    github_repo=tf_github_repo,
+                    deployment_type=deployment_type,
+                    test_name=test_name,
+                    triggering_env=tf_triggering_env,
+                    metric=metric_name,
+                )
             )
-        )
 
-        branch_time_series_dict[ts_name] = {
-            "labels": version_tags.copy(),
-            "data": {datapoints_timestamp: metric_value},
-        }
+            branch_time_series_dict[ts_name] = {
+                "labels": version_tags.copy(),
+                "data": {datapoints_timestamp: metric_value},
+            }
+        else:
+            logging.warning(
+                "Unable to find metric path {} in {}".format(jsonpath, results_dict)
+            )
     return True, branch_time_series_dict
 
 
@@ -532,34 +539,38 @@ def extract_perbranch_timeseries_from_results(
     for jsonpath in metrics:
         jsonpath_expr = parse(jsonpath)
         metric_name = jsonpath[2:]
-        metric_value = float(jsonpath_expr.find(results_dict)[0].value)
-        # prepare tags
-        # branch tags
+        find_res = jsonpath_expr.find(results_dict)
+        if find_res is not None and len(find_res) > 0:
+            metric_value = float(find_res[0].value)
 
-        branch_tags = get_project_ts_tags(
-            tf_github_org, tf_github_repo, deployment_type, tf_triggering_env
-        )
-        branch_tags["branch"] = str(tf_github_branch)
-        branch_tags["test_name"] = str(test_name)
-        branch_tags["metric"] = str(metric_name)
-        ts_name = (
-            "ci.benchmarks.redislabs/by.branch/"
-            "{triggering_env}/{github_org}/{github_repo}/"
-            "{test_name}/{deployment_type}/{branch}/{metric}".format(
-                branch=str(tf_github_branch),
-                github_org=tf_github_org,
-                github_repo=tf_github_repo,
-                deployment_type=deployment_type,
-                test_name=test_name,
-                triggering_env=tf_triggering_env,
-                metric=metric_name,
+            branch_tags = get_project_ts_tags(
+                tf_github_org, tf_github_repo, deployment_type, tf_triggering_env
+            )
+            branch_tags["branch"] = str(tf_github_branch)
+            branch_tags["test_name"] = str(test_name)
+            branch_tags["metric"] = str(metric_name)
+            ts_name = (
+                "ci.benchmarks.redislabs/by.branch/"
+                "{triggering_env}/{github_org}/{github_repo}/"
+                "{test_name}/{deployment_type}/{branch}/{metric}".format(
+                    branch=str(tf_github_branch),
+                    github_org=tf_github_org,
+                    github_repo=tf_github_repo,
+                    deployment_type=deployment_type,
+                    test_name=test_name,
+                    triggering_env=tf_triggering_env,
+                    metric=metric_name,
+                )
             )
-        )
 
-        branch_time_series_dict[ts_name] = {
-            "labels": branch_tags.copy(),
-            "data": {datapoints_timestamp: metric_value},
-        }
+            branch_time_series_dict[ts_name] = {
+                "labels": branch_tags.copy(),
+                "data": {datapoints_timestamp: metric_value},
+            }
+        else:
+            logging.warning(
+                "Unable to find metric path {} in {}".format(jsonpath, results_dict)
+            )
     return True, branch_time_series_dict
 
 

tests/test_data/tsbs-devops-ingestion-scale100-4days.yml

@@ -0,0 +1,16 @@
+name: "tsbs-devops-ingestion-scale100-4days"
+remote:
+  - type: oss-standalone
+  - setup: redistimeseries-m5
+clientconfig:
+  - tool: tsbs_load_redistimeseries
+  - parameters:
+    - workers: 64
+    - reporting-period: 1s
+    - file: "https://s3.amazonaws.com/benchmarks.redislabs/redistimeseries/tsbs/datasets/devops/scale100/data_redistimeseries_cpu-only_100.dat"
+exporter:
+  redistimeseries:
+    timemetric: "$.StartTime"
+    metrics:
+      - "$.Totals.metricRate"
+      - "$.Totals.rowRate"