Implement query tests

Author: swoehrl-mw

README.md

@@ -22,7 +22,7 @@ All tests were run on an empty database.
 
 Upon execution the helm chart found in `deployment/` is installed on the cluster. By default we execute 3 runs and average out the results to compensate for fluctuations in performance. A run consists of the workers and one collector instance being spawned in individual pods inside the same k3s cluster. Due to node selectors workers could not run on the same nodes as DB instances to avoid interference.
 Each worker generates and writes the configured amount of events into the database. Event schemata and API usage are found in `simulator/modules/{database_name}.py` (note: for cockroachdb and yugabytedb we use the postgres module), event generation logic may be reviewed under `simulator/modules/event_generator.py`.
-After each run the workers reports statistics to the collector instance. The database is wiped inbetween separate runs to have a reproducible baseline.
+After each run the workers report statistics to the collector instance. The database is wiped inbetween separate runs to have a reproducible baseline.
 
 For generating primary keys for the events we have two modes: Calculating it on the client side based on some of the fields that from a functional perspective guarantee uniqueness, or having the database increment a `SERIAL` primary key field. The exception here is Cassandra as using serial primary keys for rows opposes the main concepts of Cassandra we omitted this step and always relied on a db-generated unique partition key (device_id, timestamp).
 
@@ -83,11 +83,11 @@ Note: The provided values are for a k3s cluster. If you use another distribution
 
 ### Run the test
 
-To run the test use `python run.py`. You can use the following options:
+To run the test use `python run.py insert`. You can use the following options:
 
 * `--target`: The target database to use (name must correspond to the target name in `config.yaml`). This is required
 * `--workers`: Set of worker counts to try, default is `1,4,8,12,16` meaning the test will try with 1 concurrent worker, then with 4, then 8, then 12 and finally 16
-* `--runs`: How often should the test be repeated for each worker count, default is `3`.
+* `--runs`: How often should the test be repeated for each worker count, default is `3`
 * `--primary-key`: Defines how the primary key should be generated, see below for choices. Defaults to `db`
 * `--tables`: To simulate how the databases behave if inserts are done to several tables this option can be changed from `single` to `multiple` to have the test write into four instead of just one table
 * `--num-inserts`: The number of inserts each worker should do, by default 10000 to get a quick result. Increase this to see how the databases behave under constant load. Also increase the timout option accordingly
@@ -100,6 +100,15 @@ To run the test use `python run.py`. You can use the following options:
 
 If the test takes too long and the timeout is reached or the script runs into any problems it will crash. To clean up you must then manually uninstall the simulator by running `helm uninstall dbtest`.
 
+The query test can be run using `python run.py query`. You can use the following options:
+
+* `--target`: The target database to use (name must correspond to the target name in `config.yaml`). This is required
+* `--workers`: Number of concurrent workers that should issue queries
+* `--runs`: How often should each query be issued, default is `3`
+* `--timeout`: How long should the script wait for the insert test to complete in seconds. Default is `0`. Increase accordingly if you increase the number of inserts or disable by stting to `0`
+
+Before running the query test use the insert test to provide an appropriate amount of data.
+
 ### Primary key
 
 There are several options on how the primary key for the database table can be generated, defined by the `--primary-key` option:

cli/__init__.py

@@ -0,0 +1,5 @@
+import click
+
+@click.group()
+def commands():
+    pass

cli/insert.py

@@ -0,0 +1,94 @@
+import base64
+import json
+import sys
+import click
+import yaml
+from . import commands
+from .run import one_run
+
+
+
+@commands.command()
+@click.option('-t', '--target', required=True, help="Name of the target")
+@click.option('-c', '--config', default="config.yaml", help="Name of the config file to use")
+@click.option('-w', '--workers', default="1,4,8,12,16", help="Sets of worker counts to use, separate by comma without space, default='1,4,8,12,16'")
+@click.option('-r', '--runs', default=3, help='Number of runs per worker count, default=3')
+@click.option("--primary-key", default="db", type=click.Choice(['sql', 'db', 'client', 'uuid'], case_sensitive=False))
+@click.option("--tables", default="single", type=click.Choice(['single', 'multiple'], case_sensitive=False))
+@click.option("--num-inserts", default=10000, help="Number of inserts per worker, default=10000")
+@click.option("--prefill", default=0, help="Insert this number of events into the table before starting the test run, default=0")
+@click.option("--extra-option", multiple=True, help="Extra options for the database module")
+@click.option("--timeout", default=0, help="Timeout in seconds to wait for one run to complete. Increase this if you use higher number of inserts, or set to 0 to disable timeout. default=0")
+@click.option("--batch", default=0, help="Number of events to insert in one batch, default 0 disables batch mode")
+@click.option('--clean/--no-clean', default=True, help="Clean up the database before each run, enabled by default")
+@click.option('--steps', default=0, help="TODO")
+def insert(target, config, workers, runs, primary_key, tables, num_inserts, prefill, extra_option, timeout, batch, clean, steps):
+    worker_counts = list(map(lambda el: int(el), workers.split(",")))
+    config = _read_config(config)
+    target_config = config["targets"][target]
+    namespace = config.get("namespace", "default")
+    if steps and len(worker_counts) > 1:
+        print("ERROR: If using the --steps option only one worker count can be used")
+        sys.exit(1)
+    if steps and runs > 1:
+        print("ERROR: If using the --steps option only one run is allowed")
+        sys.exit(1)
+    if steps and prefill:
+        print("ERROR: --steps and --prefill cannot be used at the same time")
+        sys.exit(1)
+
+    if steps:
+        _steps_test(target_config, worker_counts[0], namespace, primary_key, tables, num_inserts, extra_option, timeout, batch, clean, steps)
+    else:
+        _normal_test(target_config, worker_counts, namespace, runs, primary_key, tables, num_inserts, prefill, extra_option, timeout, batch, clean)
+
+
+def _normal_test(target_config, worker_counts, namespace, runs, primary_key, tables, num_inserts, prefill, extra_option, timeout, batch, clean):
+    run_config, target_module = _prepare_run_config(target_config, primary_key, tables, num_inserts, prefill, int(batch) if batch else None, clean, extra_option)
+
+    print(f"Workers\tMin\tMax\tAvg")
+    for worker_count in worker_counts:
+        run_results = [one_run(worker_count, run_config, target_module, timeout, namespace)["sum"]["ops_per_second"] for _ in range(runs)]
+        result_min = round(min(run_results))
+        result_max = round(max(run_results))
+        result_avg = round(int(sum(run_results)/len(run_results)))
+        print(f"{worker_count:2}\t{result_min:6}\t{result_max:6}\t{result_avg:6}")
+
+
+def _steps_test(target_config, workers, namespace, primary_key, tables, num_inserts, extra_option, timeout, batch, clean, steps):
+    run_config, target_module = _prepare_run_config(target_config, primary_key, tables, num_inserts, 0, int(batch) if batch else None, clean, extra_option)
+    run_config_continued, _ = _prepare_run_config(target_config, primary_key, tables, num_inserts, 0, int(batch) if batch else None, False, extra_option)
+    stepsize = workers*num_inserts
+    width = len(f"{stepsize*steps}")
+    print(f"Stepsize: {stepsize}")
+    print(f"Level".rjust(width)+"\tInserts/s")
+    for step in range(steps):
+        fill = f"{step*stepsize}".rjust(width)
+        inserts = int(round(one_run(workers, run_config, target_module, timeout, namespace)["sum"]["ops_per_second"], -1))
+        run_config = run_config_continued
+        print(f"{fill}\t{inserts:6}")
+
+
+def _prepare_run_config(target_config, primary_key, tables, num_inserts, prefill, batch, clean, extra_options):
+    config = target_config
+    config.update({
+        "task": "insert",
+        "num_inserts": num_inserts,
+        "prefill": int(prefill),
+        "primary_key": primary_key,
+        "use_multiple_tables": tables=="multiple",
+        "clean_database": clean,
+    })
+    if batch:
+        config["batch_mode"] = True
+        config["batch_size"] = batch
+    for option in extra_options:
+        k, v = option.split("=", 1)
+        config[k] = v
+    return base64.b64encode(json.dumps(config).encode("utf-8")).decode("utf-8"), config["module"]
+    
+
+def _read_config(config_file):
+    with open(config_file) as f:
+        config = yaml.safe_load(f)
+    return config

cli/query.py

@@ -0,0 +1,43 @@
+import base64
+import json
+import click
+import yaml
+from .run import one_run
+from . import commands
+
+
+@commands.command()
+@click.option('-t', '--target', required=True, help="Name of the target")
+@click.option('-c', '--config', default="config.yaml", help="Name of the config file to use")
+@click.option('-w', '--workers', default=1, help="Number of workers to use")
+@click.option('-r', '--runs', default=3, help='Number of times each query should be executed, default=3')
+@click.option("--timeout", default=0, help="Timeout in seconds to wait for one run to complete. Increase this if you use higher number of inserts, or set to 0 to disable timeout. default=0")
+def query(target, config, workers, runs, timeout):
+    config = _read_config(config)
+    target_config = config["targets"][target]
+    namespace = config.get("namespace", "default")
+    run_config, target_module = _prepare_run_config(target_config, runs)
+    results = one_run(workers, run_config, target_module, timeout, namespace, endpoint="/report/queries")
+    max_name_len = max([len(name) for name in results["queries"].keys()])
+    spacing = " " * (max_name_len - len("Query"))
+    print(f"Query{spacing}\tMin  \tMax  \tAvg")
+    for name, stats in results["queries"].items():
+        spacing = " " * (max_name_len - len(name))
+        result_min = round(stats['min'], 2)
+        result_max = round(stats['max'], 2)
+        result_avg = round(stats['avg'], 2)
+        print(f"{name}{spacing}\t{result_min:>5.2f}\t{result_max:>5.2f}\t{result_avg:>5.2f}")
+
+def _prepare_run_config(target_config, runs):
+    config = target_config
+    config.update({
+        "task": "query",
+        "runs": runs,
+    })
+    return base64.b64encode(json.dumps(config).encode("utf-8")).decode("utf-8"), config["module"]
+    
+
+def _read_config(config_file):
+    with open(config_file) as f:
+        config = yaml.safe_load(f)
+    return config

cli/test_run.py renamed to cli/run.py

@@ -17,7 +17,7 @@ def http_request(url):
     raise Exception("http_request failed with retry")
 
 
-def one_run(num_workers, run_config, target_module, timeout, namespace):
+def one_run(num_workers, run_config, target_module, timeout, namespace, endpoint="/report/insert"):
     kube = Kubernetes()
     res = subprocess.run(["helm", "install", "-n", namespace, "dbtest", ".", "--set", f"workers={num_workers}", "--set", f"run_config={run_config}", 
                             "--set", f"target_module={target_module}", "--set", f"namespace={namespace}"], cwd="deployment", stdout=subprocess.DEVNULL)
@@ -34,8 +34,8 @@ def one_run(num_workers, run_config, target_module, timeout, namespace):
         time.sleep(10)
     collector_pod_name = kube.find_pod(namespace, "app", "dbtest-collector")
     kube.patch_socket()
-    results = json.loads(http_request(f"http://{collector_pod_name}.pod.{namespace}.kubernetes:5000/report"))
+    results = json.loads(http_request(f"http://{collector_pod_name}.pod.{namespace}.kubernetes:5000{endpoint}"))
     res = subprocess.run(f"helm uninstall -n {namespace} dbtest".split(" "), stdout=subprocess.DEVNULL)
     res.check_returncode()
     kube.wait_for_pods_terminated(namespace, "app", "dbtest-worker")
-    return results["sum"]["ops_per_second"]
+    return results

run.py

@@ -1,95 +1,7 @@
-import base64
-import json
-import sys
-import click
-import yaml
-from cli.test_run import one_run
-
-
-@click.command()
-@click.option('-t', '--target', required=True, help="Name of the target")
-@click.option('-c', '--config', default="config.yaml", help="Name of the config file to use")
-@click.option('-w', '--workers', default="1,4,8,12,16", help="Sets of worker counts to use, separate by comma without space, default='1,4,8,12,16'")
-@click.option('-r', '--runs', default=3, help='Number of runs per worker count, default=3')
-@click.option("--primary-key", default="db", type=click.Choice(['sql', 'db', 'client', 'uuid'], case_sensitive=False))
-@click.option("--tables", default="single", type=click.Choice(['single', 'multiple'], case_sensitive=False))
-@click.option("--num-inserts", default=10000, help="Number of inserts per worker, default=10000")
-@click.option("--prefill", default=0, help="Insert this number of events into the table before starting the test run, default=0")
-@click.option("--extra-option", multiple=True, help="Extra options for the database module")
-@click.option("--timeout", default=0, help="Timeout in seconds to wait for one run to complete. Increase this if you use higher number of inserts, or set to 0 to disable timeout. default=0")
-@click.option("--batch", default=0, help="Number of events to insert in one batch, default 0 disables batch mode")
-@click.option('--clean/--no-clean', default=True, help="Clean up the database before each run, enabled by default")
-@click.option('--steps', default=0, help="TODO")
-def main(target, config, workers, runs, primary_key, tables, num_inserts, prefill, extra_option, timeout, batch, clean, steps):
-    worker_counts = list(map(lambda el: int(el), workers.split(",")))
-    config = _read_config(config)
-    target_config = config["targets"][target]
-    namespace = config.get("namespace", "default")
-    if steps and len(worker_counts) > 1:
-        print("ERROR: If using the --steps option only one worker count can be used")
-        sys.exit(1)
-    if steps and runs > 1:
-        print("ERROR: If using the --steps option only one run is allowed")
-        sys.exit(1)
-    if steps and prefill:
-        print("ERROR: --steps and --prefill cannot be used at the same time")
-        sys.exit(1)
-
-    if steps:
-        _steps_test(target_config, worker_counts[0], namespace, primary_key, tables, num_inserts, extra_option, timeout, batch, clean, steps)
-    else:
-        _normal_test(target_config, worker_counts, namespace, runs, primary_key, tables, num_inserts, prefill, extra_option, timeout, batch, clean)
-
-
-def _normal_test(target_config, worker_counts, namespace, runs, primary_key, tables, num_inserts, prefill, extra_option, timeout, batch, clean):
-    run_config, target_module = _prepare_run_config(target_config, primary_key, tables, num_inserts, prefill, int(batch) if batch else None, clean, extra_option)
-
-    print(f"Workers\tMin\tMax\tAvg")
-    for worker_count in worker_counts:
-        run_results = [one_run(worker_count, run_config, target_module, timeout, namespace) for _ in range(runs)]
-        result_min = round(min(run_results))
-        result_max = round(max(run_results))
-        result_avg = round(int(sum(run_results)/len(run_results)))
-        print(f"{worker_count:2}\t{result_min:6}\t{result_max:6}\t{result_avg:6}")
-
-
-def _steps_test(target_config, workers, namespace, primary_key, tables, num_inserts, extra_option, timeout, batch, clean, steps):
-    run_config, target_module = _prepare_run_config(target_config, primary_key, tables, num_inserts, 0, int(batch) if batch else None, clean, extra_option)
-    run_config_continued, _ = _prepare_run_config(target_config, primary_key, tables, num_inserts, 0, int(batch) if batch else None, False, extra_option)
-    stepsize = workers*num_inserts
-    width = len(f"{stepsize*steps}")
-    print(f"Stepsize: {stepsize}")
-    print(f"Level".rjust(width)+"\tInserts/s")
-    for step in range(steps):
-        fill = f"{step*stepsize}".rjust(width)
-        inserts = int(round(one_run(workers, run_config, target_module, timeout, namespace), -1))
-        run_config = run_config_continued
-        print(f"{fill}\t{inserts:6}")
-
-
-def _prepare_run_config(target_config, primary_key, tables, num_inserts, prefill, batch, clean, extra_options):
-    config = target_config
-    config.update({
-        "num_inserts": num_inserts,
-        "prefill": int(prefill),
-        "primary_key": primary_key,
-        "use_multiple_tables": tables=="multiple",
-        "clean_database": clean,
-    })
-    if batch:
-        config["batch_mode"] = True
-        config["batch_size"] = batch
-    for option in extra_options:
-        k, v = option.split("=", 1)
-        config[k] = v
-    return base64.b64encode(json.dumps(config).encode("utf-8")).decode("utf-8"), config["module"]
-    
-
-def _read_config(config_file):
-    with open(config_file) as f:
-        config = yaml.safe_load(f)
-    return config
+from cli import commands
+from cli.query import query
+from cli.insert import insert
 
 
 if __name__ == '__main__':
-    main()
+    commands()

simulator/collector.py

@@ -2,6 +2,7 @@
 import time
 from flask import Flask, request, jsonify, make_response
 from modules import select_module
+from modules.config import config
 
 
 app = Flask(__name__)
@@ -37,8 +38,8 @@ def report_result():
     return "OK"
 
 
-@app.route("/report")
-def collect_results():
+@app.route("/report/insert")
+def collect_results_insert():
     report = dict()
     report["workers"] = results
     sum_ops, sum_duration = 0, 0
@@ -50,8 +51,30 @@ def collect_results():
     return jsonify(report)
 
 
+@app.route("/report/queries")
+def collect_results_queries():
+    report = dict()
+    queries = dict([(name, []) for name in list(results.values())[0]["results"].keys()])
+    report["workers"] = results
+    report["queries"] = dict()
+    
+    for worker in results.values():
+        for name, values in worker["results"].items():
+            queries[name].extend(values)
+
+    for name, values in queries.items():
+        report["queries"][name] = {
+            "min": min(values),
+            "max": max(values),
+            "avg": sum(values)/len(values),
+        }
+    
+    return jsonify(report)
+
 def run():
-    select_module().init()
+    module = select_module()
+    if config.get("task", "insert") == "insert":
+        module.init()
     # It looks like in some cases for yugabytedb the created table is not instantly available for all workers so wait a few seconds
     time.sleep(10) 
     app.run(host='0.0.0.0', port=5000, debug=False, threaded=False)

simulator/modules/cassandra.py

@@ -21,10 +21,10 @@ def init():
     if config["clean_database"]:
         for table_name in ["events0", "events1", "events2", "events3", "events"]:
             session.execute(f"DROP TABLE IF EXISTS {KEYSPACE}.{table_name}")
-    session.execute(f"""DROP KEYSPACE IF EXISTS {KEYSPACE}""")
-    session.execute(f"""CREATE KEYSPACE IF NOT EXISTS {KEYSPACE}
-        WITH replication = {{ 'class': 'SimpleStrategy', 'replication_factor':{config["replication_factor"]}}}
-        """)
+        session.execute(f"""DROP KEYSPACE IF EXISTS {KEYSPACE}""")
+        session.execute(f"""CREATE KEYSPACE IF NOT EXISTS {KEYSPACE}
+            WITH replication = {{ 'class': 'SimpleStrategy', 'replication_factor':{config["replication_factor"]}}}
+            """)
 
     if config["use_multiple_tables"]:
         table_names = ["events0", "events1", "events2", "events3"]