Added memtier_benchmark-playbook-session-storage

Author: fcostaoliveira

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "redis-benchmarks-specification"
-version = "0.1.298"
+version = "0.1.302"
 description = "The Redis benchmarks specification describes the cross-language/tools requirements and expectations to foster performance and observability standards around redis related technologies. Members from both industry and academia, including organizations and individuals are encouraged to contribute."
 authors = ["filipecosta90 <[email protected]>","Redis Performance Group <[email protected]>"]
 readme = "Readme.md"

redis_benchmarks_specification/__common__/runner.py

@@ -3,23 +3,26 @@
 import os
 import pathlib
 import re
-
+import datetime as dt
 import redis
-from redisbench_admin.run.metrics import collect_redis_metrics
-from redisbench_admin.run_remote.run_remote import export_redis_metrics
 
 from redis_benchmarks_specification.__common__.timeseries import (
     timeseries_test_sucess_flow,
+    push_data_to_redistimeseries,
+    get_project_ts_tags,
 )
 
 
 def execute_init_commands(benchmark_config, r, dbconfig_keyname="dbconfig"):
     cmds = None
+    lua_scripts = None
     res = 0
     if dbconfig_keyname in benchmark_config:
         for k, v in benchmark_config[dbconfig_keyname].items():
             if "init_commands" in k:
                 cmds = v
+            elif "init_lua" in k:
+                lua_scripts = v
 
     if type(cmds) == str:
         cmds = [cmds]
@@ -37,7 +40,7 @@ def execute_init_commands(benchmark_config, r, dbconfig_keyname="dbconfig"):
                     quoting=csv.QUOTE_ALL,
                     skipinitialspace=True,
                 ):
-                    if lines[0] != " " and len(lines[0]) > 0:
+                    if len(lines) > 0 and lines[0] != " " and len(lines[0]) > 0:
                         cols.append(lines[0])
                 cmd = cols
                 is_array = True
@@ -57,6 +60,21 @@ def execute_init_commands(benchmark_config, r, dbconfig_keyname="dbconfig"):
                     )
                 )
 
+    # Process init_lua scripts
+    if lua_scripts is not None:
+        if type(lua_scripts) == str:
+            lua_scripts = [lua_scripts]
+
+        for lua_script in lua_scripts:
+            try:
+                logging.info("Executing Lua script (length: {} chars)".format(len(lua_script)))
+                # Execute the Lua script using EVAL command with 0 keys
+                stdout = r.execute_command("EVAL", lua_script, 0)
+                logging.info("Lua script result: {}".format(stdout))
+                res = res + 1
+            except Exception as e:
+                logging.error("Lua script execution failed: {}".format(e))
+
     return res
 
 
@@ -119,6 +137,210 @@ def extract_testsuites(args):
     return testsuite_spec_files
 
 
+
+def commandstats_latencystats_process_name(
+    metric_name, prefix, setup_name, variant_labels_dict
+):
+    if prefix in metric_name:
+        command_and_metric_and_shard = metric_name[len(prefix) :]
+        command = (
+            command_and_metric_and_shard[0]
+            + command_and_metric_and_shard[1:].split("_", 1)[0]
+        )
+        metric_and_shard = command_and_metric_and_shard[1:].split("_", 1)[1]
+        metric = metric_and_shard
+        shard = "1"
+        if "_shard_" in metric_and_shard:
+            metric = metric_and_shard.split("_shard_")[0]
+            shard = metric_and_shard.split("_shard_")[1]
+        variant_labels_dict["metric"] = metric
+        variant_labels_dict["command"] = command
+        variant_labels_dict["command_and_metric"] = "{} - {}".format(command, metric)
+        variant_labels_dict["command_and_metric_and_setup"] = "{} - {} - {}".format(
+            command, metric, setup_name
+        )
+        variant_labels_dict["command_and_setup"] = "{} - {}".format(command, setup_name)
+        variant_labels_dict["shard"] = shard
+        variant_labels_dict["metric_and_shard"] = metric_and_shard
+
+        version = None
+        branch = None
+        if "version" in variant_labels_dict:
+            version = variant_labels_dict["version"]
+        if "branch" in variant_labels_dict:
+            branch = variant_labels_dict["branch"]
+
+        if version is not None:
+            variant_labels_dict["command_and_metric_and_version"] = (
+                "{} - {} - {}".format(command, metric, version)
+            )
+            variant_labels_dict["command_and_metric_and_setup_and_version"] = (
+                "{} - {} - {} - {}".format(command, metric, setup_name, version)
+            )
+
+        if branch is not None:
+            variant_labels_dict["command_and_metric_and_branch"] = (
+                "{} - {} - {}".format(command, metric, branch)
+            )
+            variant_labels_dict["command_and_metric_and_setup_and_branch"] = (
+                "{} - {} - {} - {}".format(command, metric, setup_name, branch)
+            )
+
+
+def collect_redis_metrics(
+    redis_conns,
+    sections=["memory", "cpu", "commandstats", "latencystats"],
+    section_filter=None,
+):
+    start_time = dt.datetime.utcnow()
+    start_time_ms = int((start_time - dt.datetime(1970, 1, 1)).total_seconds() * 1000)
+    res = []
+    overall = {}
+    multi_shard = False
+    if len(redis_conns) > 1:
+        multi_shard = True
+    for conn_n, conn in enumerate(redis_conns):
+        conn_res = {}
+        for section in sections:
+            info = conn.info(section)
+            conn_res[section] = info
+            if section not in overall:
+                overall[section] = {}
+            for k, v in info.items():
+                collect = True
+                if section_filter is not None:
+                    if section in section_filter:
+                        if k not in section_filter[section]:
+                            collect = False
+                if collect and type(v) is float or type(v) is int:
+                    if k not in overall[section]:
+                        overall[section][k] = 0
+                    overall[section][k] += v
+                if collect and type(v) is dict:
+                    for inner_k, inner_v in v.items():
+                        if type(inner_v) is float or type(inner_v) is int:
+                            final_str_k = "{}_{}".format(k, inner_k)
+                            if multi_shard:
+                                final_str_k += "_shard_{}".format(conn_n + 1)
+                            if final_str_k not in overall[section]:
+                                overall[section][final_str_k] = inner_v
+
+        res.append(conn_res)
+
+    kv_overall = {}
+    for sec, kv_detail in overall.items():
+        for k, metric_value in kv_detail.items():
+            metric_name = "{}_{}".format(sec, k)
+            kv_overall[metric_name] = metric_value
+
+    return start_time_ms, res, kv_overall
+
+def export_redis_metrics(
+    artifact_version,
+    end_time_ms,
+    overall_end_time_metrics,
+    rts,
+    setup_name,
+    setup_type,
+    test_name,
+    tf_github_branch,
+    tf_github_org,
+    tf_github_repo,
+    tf_triggering_env,
+    metadata_dict=None,
+    expire_ms=0,
+    git_hash=None,
+    running_platform=None,
+):
+    datapoint_errors = 0
+    datapoint_inserts = 0
+    sprefix = (
+        "ci.benchmarks.redis/"
+        + "{triggering_env}/{github_org}/{github_repo}".format(
+            triggering_env=tf_triggering_env,
+            github_org=tf_github_org,
+            github_repo=tf_github_repo,
+        )
+    )
+    logging.info(
+        "Adding a total of {} server side metrics collected at the end of benchmark".format(
+            len(list(overall_end_time_metrics.items()))
+        )
+    )
+    timeseries_dict = {}
+    by_variants = {}
+    if tf_github_branch is not None and tf_github_branch != "":
+        by_variants["by.branch/{}".format(tf_github_branch)] = {
+            "branch": tf_github_branch
+        }
+    if git_hash is not None and git_hash != "":
+        by_variants["by.hash/{}".format(git_hash)] = {
+            "hash": git_hash
+        }
+    if artifact_version is not None and artifact_version != "":
+        by_variants["by.version/{}".format(artifact_version)] = {
+            "version": artifact_version
+        }
+    for (
+        by_variant,
+        variant_labels_dict,
+    ) in by_variants.items():
+        for (
+            metric_name,
+            metric_value,
+        ) in overall_end_time_metrics.items():
+            tsname_metric = "{}/{}/{}/benchmark_end/{}/{}".format(
+                sprefix,
+                test_name,
+                by_variant,
+                setup_name,
+                metric_name,
+            )
+
+            logging.debug(
+                "Adding a redis server side metric collected at the end of benchmark."
+                + " metric_name={} metric_value={} time-series name: {}".format(
+                    metric_name,
+                    metric_value,
+                    tsname_metric,
+                )
+            )
+            variant_labels_dict["metric"] = metric_name
+            commandstats_latencystats_process_name(
+                metric_name, "commandstats_cmdstat_", setup_name, variant_labels_dict
+            )
+            commandstats_latencystats_process_name(
+                metric_name,
+                "latencystats_latency_percentiles_usec_",
+                setup_name,
+                variant_labels_dict,
+            )
+
+            variant_labels_dict["test_name"] = test_name
+            if metadata_dict is not None:
+                variant_labels_dict.update(metadata_dict)
+
+            timeseries_dict[tsname_metric] = {
+                "labels": get_project_ts_tags(
+                    tf_github_org,
+                    tf_github_repo,
+                    setup_name,
+                    setup_type,
+                    tf_triggering_env,
+                    variant_labels_dict,
+                    None,
+                    running_platform,
+                ),
+                "data": {end_time_ms: metric_value},
+            }
+    i_errors, i_inserts = push_data_to_redistimeseries(rts, timeseries_dict, expire_ms)
+    datapoint_errors = datapoint_errors + i_errors
+    datapoint_inserts = datapoint_inserts + i_inserts
+    return datapoint_errors, datapoint_inserts
+
+
+
+
 def reset_commandstats(redis_conns):
     for pos, redis_conn in enumerate(redis_conns):
         logging.info("Resetting commmandstats for shard {}".format(pos))
@@ -248,13 +470,16 @@ def exporter_datasink_common(
     deployment_type_and_name = f"{setup_type}_AND_{setup_name}"
     deployment_type_and_name_and_version = f"{setup_type}_AND_{setup_name}_AND_{git_version}"
 
-    # Add to deployment-specific set
-    deployment_set_key = f"ci.benchmarks.redis/{tf_triggering_env}/{deployment_type_and_name_and_version}:set"
-    datasink_conn.sadd(deployment_set_key, test_name)
+    # Add to deployment-specific set (only if datasink connection is available)
+    if datasink_conn is not None:
+        deployment_set_key = f"ci.benchmarks.redis/{tf_triggering_env}/{deployment_type_and_name_and_version}:set"
+        datasink_conn.sadd(deployment_set_key, test_name)
 
-    # Add to testcases set
-    testcases_set_key = f"ci.benchmarks.redis/{tf_triggering_env}/testcases:set"
-    datasink_conn.sadd(testcases_set_key, test_name)
+        # Add to testcases set
+        testcases_set_key = f"ci.benchmarks.redis/{tf_triggering_env}/testcases:set"
+        datasink_conn.sadd(testcases_set_key, test_name)
+    else:
+        logging.debug("Datasink connection not available, skipping set operations")
 
     # Add metadata fields to timeseries metadata
     metadata["deployment_type_AND_deployment_name"] = deployment_type_and_name

redis_benchmarks_specification/__runner__/runner.py

@@ -28,7 +28,10 @@
     prepare_benchmark_parameters,
     dbconfig_keyspacelen_check,
 )
-from redisbench_admin.run_remote.run_remote import export_redis_metrics
+
+from redis_benchmarks_specification.__common__.runner import (
+    export_redis_metrics,
+)
 
 from redisbench_admin.run.metrics import extract_results_table
 from redisbench_admin.run.run import calculate_client_tool_duration_and_check
@@ -56,6 +59,7 @@
     exporter_datasink_common,
     reset_commandstats,
     execute_init_commands,
+    export_redis_metrics,
 )
 from redis_benchmarks_specification.__common__.spec import (
     extract_client_container_image,
@@ -1935,8 +1939,10 @@ def delete_temporary_files(
                                 tf_github_org,
                                 tf_github_repo,
                                 tf_triggering_env,
-                                {"metric-type": "memory-stats"},
+                                metadata,
                                 expire_redis_metrics_ms,
+                                git_hash,
+                                running_platform,
                             )
 
                             exporter_datasink_common(
@@ -1962,8 +1968,8 @@ def delete_temporary_files(
                                 topology_spec_name,
                                 default_metrics,
                                 git_hash,
-                                collect_commandstats=False,
-                                collect_memory_metrics=True,
+                                False,
+                                True,
                             )
 
                         # Send MEMORY PURGE after memory comparison (if FLUSHALL at test end is not enabled)