[add] Enabled --csv format in redisbench-admin export tool (#228)

* [add] Enabled --csv format in redisbench-admin export tool

* [add] included test json for memtier

* Bumping version from 0.5.9 to 0.5.10

Author: filipecosta90

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "redisbench-admin"
-version = "0.5.9"
+version = "0.5.10"
 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/export/args.py

@@ -22,7 +22,6 @@ def create_export_arguments(parser):
     parser.add_argument(
         "--exporter-spec-file",
         type=str,
-        required=True,
         help="Exporter definition file, containing info of the metrics to extract",
     )
     parser.add_argument(
@@ -46,7 +45,7 @@ def create_export_arguments(parser):
     parser.add_argument(
         "--test-name",
         type=str,
-        required=True,
+        default=None,
         help="Test name",
     )
     parser.add_argument("--github_actor", type=str, default=None, nargs="?", const="")
@@ -65,7 +64,7 @@ def create_export_arguments(parser):
         type=str,
         default="json",
         help="results format of the the benchmark results files to read "
-        "results from ( either json, redis-benchmark-txt )",
+        "results from ( either csv, json, redis-benchmark-txt )",
     )
     parser.add_argument(
         "--use-result",

redisbench_admin/export/export.py

@@ -3,8 +3,7 @@
 #  Copyright (c) 2021., Redis Labs Modules
 #  All rights reserved.
 #
-
-
+import csv
 import json
 import logging
 import datetime
@@ -20,6 +19,7 @@
     get_defaults,
     parse_exporter_timemetric,
 )
+from redisbench_admin.utils.remote import get_ts_tags_and_name
 
 
 def export_command_logic(args, project_name, project_version):
@@ -28,47 +28,47 @@ def export_command_logic(args, project_name, project_version):
             project_name=project_name, project_version=project_version
         )
     )
+    timeseries_dict = None
     benchmark_file = args.benchmark_result_file
+    test_name = args.test_name
+    deployment_version = args.deployment_version
+    triggering_env = args.triggering_env
+    deployment_name = args.deployment_name
+    deployment_type = args.deployment_type
     results_format = args.results_format
-    (_, tf_github_branch, tf_github_org, tf_github_repo, _,) = git_vars_crosscheck(
+    (_, github_branch, github_org, github_repo, _,) = git_vars_crosscheck(
         None, args.github_branch, args.github_org, args.github_repo, None
     )
-    results_dict = {}
-    if results_format == "json":
-        with open(benchmark_file, "r") as json_file:
-            results_dict = json.load(json_file)
-    extra_tags_dict = split_tags_string(args.extra_tags)
-    logging.info("Using the following extra tags: {}".format(extra_tags_dict))
-
-    logging.info(
-        "Checking connection to RedisTimeSeries to host: {}:{}".format(
-            args.redistimeseries_host, args.redistimeseries_port
-        )
-    )
-    rts = Client(
-        host=args.redistimeseries_host,
-        port=args.redistimeseries_port,
-        password=args.redistimeseries_pass,
-    )
-    try:
-        rts.redis.ping()
-    except redis.exceptions.ConnectionError as e:
+    exporter_timemetric_path = None
+    metrics = []
+    exporter_spec_file = args.exporter_spec_file
+    if github_branch is None and deployment_version is None:
         logging.error(
-            "Error while connecting to RedisTimeSeries data sink at: {}:{}. Error: {}".format(
-                args.redistimeseries_host, args.redistimeseries_port, e.__str__()
-            )
+            "You need to specify at least one (or more) of --deployment-version --github_branch arguments"
         )
         exit(1)
+    if results_format != "csv":
+        if exporter_spec_file is None:
+            logging.error(
+                "--exporter-spec-file is required for all formats with exception of csv"
+            )
+            exit(1)
+        else:
+            (
+                _,
+                metrics,
+                exporter_timemetric_path,
+                _,
+                _,
+            ) = get_defaults(exporter_spec_file)
 
-    benchmark_duration_seconds = None
-    exporter_spec_file = args.exporter_spec_file
-    (
-        _,
-        metrics,
-        exporter_timemetric_path,
-        _,
-        _,
-    ) = get_defaults(exporter_spec_file)
+    extra_tags_dict = split_tags_string(args.extra_tags)
+    logging.info("Using the following extra tags: {}".format(extra_tags_dict))
+
+    results_dict = {}
+    if results_format == "json":
+        with open(benchmark_file, "r") as json_file:
+            results_dict = json.load(json_file)
     if args.override_test_time:
         datapoints_timestamp = int(args.override_test_time.timestamp() * 1000.0)
         logging.info(
@@ -94,26 +94,120 @@ def export_command_logic(args, project_name, project_version):
                     datapoints_timestamp
                 )
             )
+    break_by_dict = {}
+    if deployment_version is not None:
+        break_by_dict.update({"version": deployment_version})
+    if github_branch is not None:
+        break_by_dict.update({"branch": github_branch})
+    if results_format == "csv":
+        logging.info("Parsing CSV format from {}".format(benchmark_file))
+        timeseries_dict = export_opereto_csv_to_timeseries_dict(
+            benchmark_file,
+            break_by_dict,
+            datapoints_timestamp,
+            deployment_name,
+            deployment_type,
+            extra_tags_dict,
+            github_org,
+            github_repo,
+            triggering_env,
+        )
+        logging.info("Parsed a total of {} metrics".format(len(timeseries_dict.keys())))
+    logging.info(
+        "Checking connection to RedisTimeSeries to host: {}:{}".format(
+            args.redistimeseries_host, args.redistimeseries_port
+        )
+    )
+    rts = Client(
+        host=args.redistimeseries_host,
+        port=args.redistimeseries_port,
+        password=args.redistimeseries_pass,
+    )
+    try:
+        rts.redis.ping()
+    except redis.exceptions.ConnectionError as e:
+        logging.error(
+            "Error while connecting to RedisTimeSeries data sink at: {}:{}. Error: {}".format(
+                args.redistimeseries_host, args.redistimeseries_port, e.__str__()
+            )
+        )
+        exit(1)
+
+    benchmark_duration_seconds = None
 
     timeseries_test_sucess_flow(
         True,
-        args.deployment_version,
+        deployment_version,
         None,
         benchmark_duration_seconds,
         None,
         metrics,
-        args.deployment_name,
-        args.deployment_type,
+        deployment_name,
+        deployment_type,
         exporter_timemetric_path,
         results_dict,
         rts,
         datapoints_timestamp,
-        args.test_name,
-        tf_github_branch,
-        tf_github_org,
-        tf_github_repo,
-        args.triggering_env,
+        test_name,
+        github_branch,
+        github_org,
+        github_repo,
+        triggering_env,
         extra_tags_dict,
         None,
         None,
+        timeseries_dict,
     )
+
+
+def export_opereto_csv_to_timeseries_dict(
+    benchmark_file,
+    break_by_dict,
+    datapoints_timestamp,
+    deployment_name,
+    deployment_type,
+    extra_tags_dict,
+    tf_github_org,
+    tf_github_repo,
+    triggering_env,
+):
+    results_dict = {}
+    with open(benchmark_file, "r") as csv_file:
+        full_csv = list(csv.reader(csv_file, delimiter=",", quoting=csv.QUOTE_ALL))
+        if len(full_csv) >= 2:
+            header = full_csv[0]
+            metrics = header[1:]
+            for row in full_csv[1:]:
+                assert len(row) == len(header)
+                test_name = row[0]
+                for metric_pos, metric_name in enumerate(metrics):
+                    metric_v_pos = 1 + metric_pos
+                    if metric_v_pos < len(row):
+                        metric_value = row[metric_v_pos]
+
+                        for break_by_key, break_by_value in break_by_dict.items():
+                            break_by_str = "by.{}".format(break_by_key)
+                            timeserie_tags, ts_name = get_ts_tags_and_name(
+                                break_by_key,
+                                break_by_str,
+                                break_by_value,
+                                None,
+                                deployment_name,
+                                deployment_type,
+                                extra_tags_dict,
+                                metric_name,
+                                metric_name,
+                                metric_name,
+                                triggering_env,
+                                test_name,
+                                metric_name,
+                                tf_github_org,
+                                tf_github_repo,
+                                triggering_env,
+                                False,
+                            )
+                            results_dict[ts_name] = {
+                                "labels": timeserie_tags.copy(),
+                                "data": {datapoints_timestamp: metric_value},
+                            }
+    return results_dict

redisbench_admin/run/common.py

@@ -39,7 +39,6 @@
 )
 from redisbench_admin.utils.remote import (
     extract_perversion_timeseries_from_results,
-    push_data_to_redistimeseries,
     extract_perbranch_timeseries_from_results,
 )
 
@@ -214,7 +213,6 @@ def common_exporter_logic(
     exporter_timemetric_path,
     metrics,
     results_dict,
-    rts,
     test_name,
     tf_github_branch,
     tf_github_org,
@@ -226,8 +224,8 @@ def common_exporter_logic(
     running_platform=None,
     datapoints_timestamp=None,
 ):
-    per_version_time_series_dict = None
-    per_branch_time_series_dict = None
+    per_version_time_series_dict = {}
+    per_branch_time_series_dict = {}
     testcase_metric_context_paths = []
 
     if exporter_timemetric_path is not None and len(metrics) > 0:
@@ -243,7 +241,7 @@ def common_exporter_logic(
         ):
             # extract per-version datapoints
             (
-                ok,
+                _,
                 per_version_time_series_dict,
             ) = extract_perversion_timeseries_from_results(
                 datapoints_timestamp,
@@ -261,12 +259,9 @@ def common_exporter_logic(
                 running_platform,
                 testcase_metric_context_paths,
             )
-            if ok:
-                # push per-version data
-                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, per_branch_time_series_dict = extract_perbranch_timeseries_from_results(
+            _, per_branch_time_series_dict = extract_perbranch_timeseries_from_results(
                 datapoints_timestamp,
                 metrics,
                 results_dict,
@@ -282,9 +277,6 @@ def common_exporter_logic(
                 running_platform,
                 testcase_metric_context_paths,
             )
-            if ok:
-                # push per-branch data
-                push_data_to_redistimeseries(rts, per_branch_time_series_dict)
         else:
             logging.warning(
                 "Requested to push data to RedisTimeSeries but no git"