feat(asap-tools): feed every metric table of a sketch-bench workload into ClickHouse

asap-tools/components.conf

@@ -9,5 +9,6 @@ asap-query-engine
 asap-planner-rs
 asap-quickstart
 asap-tools/data-sources/prometheus-exporters
+asap-tools/data-sources/paced-feeder
 asap-tools/queriers/prometheus-client
 asap-tools/execution-utilities

asap-tools/data-sources/paced-feeder/installation/install.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+
+set -e
+
+THIS_DIR=$(dirname "$(readlink -f "${BASH_SOURCE[0]}")")
+
+echo "Building paced_feeder..."
+source "$HOME/.cargo/env"
+cd "$(dirname "$THIS_DIR")"
+cargo build --release
+
+echo "paced_feeder built: $(pwd)/target/release/paced_feeder"

asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md

@@ -460,7 +460,7 @@ These parameters come from the `experiment_type` config group and are prefixed w
 #### Paced Feeder Parameters
 
 `paced_feeder` runs `asap-tools/data-sources/paced-feeder` (built by
-`installation/common/setup_dependencies.sh`) bare-metal on the worker nodes,
+its `installation/install.sh`) bare-metal on the worker nodes,
 one process per (sink, metric). It pushes over remote write instead of being
 scraped: in `sketchdb` mode to the query engine's ingest
 (`streaming.remote_write`, which must have `parallelism: 1`), plus Prometheus
@@ -499,6 +499,10 @@ table with `sketch_bench_e2e/sketch_bench_e2e_configs.py`.
 - **Description**: Appended to every feeder command line
 - **Example**: `"--concurrency 32"`
 
+#### `experiment_params.dataset.paced_feeder` (dict, optional; `experiment_run_clickhouse.py`)
+- **Description**: The same keys as `exporters.exporter_list.paced_feeder`, for the ClickHouse runner. In place of the bulk load of `dataset.local_data_file`, each mode runs `dataset.init_sql_file`, then feeds every listed metric (records metrics too) from the ClickHouse node into the table named after it while its queries run. `sketchdb` mode is refused: its engine ingests `dataset.local_data_file`, and feeding it live is not supported yet (#874). `duration_secs` must cover the startup check, the query client's startup (30 s) and the query groups' running time; after the queries each mode waits for the feeders to finish, so they write their `--stats-out`. Requires `dataset.name: custom` and `dataset.init_sql_file`
+- **Example**: `{workload_table: /path/to/table.json, metrics: [data, data_total], seed: 42, duration_secs: 4000}`
+
 ### Query Group Configuration
 
 #### `experiment_params.query_groups` (list, required)

asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py

@@ -18,8 +18,9 @@
   sketch-bench's exporter writes it) with NOW() replaced by the floored eval
   time;
 - <stem>_clickhouse.yaml, <stem>_clickhouse_mv.yaml: one
-  experiment_run_clickhouse.py config per arm, each loading data through its
-  own init SQL, with one query group per T, repeating every T seconds.
+  experiment_run_clickhouse.py config per arm, each creating the tables
+  through its own init SQL and feeding every metric with the paced feeder,
+  with one query group per T, repeating every T seconds.
 
 baseline_mv keeps exact mergeable states, no sketches: one view per (metric,
 value, statistic, T) holds per T-second bucket and group the state of sums,
@@ -56,7 +57,7 @@
 
 Run from asap-tools/experiments:
     python3 datasets/fake_metric/sketch_bench_workloads.py \
-        --table <sketch-bench table.json> --data-file <path> \
+        --table <sketch-bench table.json> --seed <n> \
         --starting-delay-secs <s> --duration-secs <s>
 """
 import argparse
@@ -92,6 +93,9 @@
 # Seconds after an eval time the client sends it, so data up to the eval
 # time has arrived.
 QUERY_TIME_OFFSET_SECS = 10
+# Feed time beyond the query phase, for the feeder's startup check; the
+# runner checks the total.
+FEEDER_SLACK_SECS = 60
 VALUE_COLUMN = "value"
 # Every view lives here, so dropping it removes views any earlier workload
 # attached to the metric tables; they would keep firing on inserts.
@@ -589,7 +593,12 @@ def statements_text(statements: List[str]) -> str:
 
 
 def convert(
-    table_path: str, data_file: str, starting_delay_secs: int, duration_secs: int
+    table_path: str,
+    seed: int,
+    starting_delay_secs: int,
+    duration_secs: int,
+    samples_per_sec: Optional[int] = None,
+    records_per_sec: Optional[int] = None,
 ) -> List[str]:
     """Writes the SQL files and one config per arm; returns the config paths.
     Everything is rendered before anything is written, so a table that fails
@@ -679,6 +688,17 @@ def convert(
             }
         )
 
+    feeder = {
+        "workload_table": os.path.abspath(table_path),
+        "metrics": list(metrics),
+        "seed": seed,
+        "duration_secs": starting_delay_secs + duration_secs + FEEDER_SLACK_SECS,
+    }
+    if samples_per_sec is not None:
+        feeder["samples_per_sec"] = samples_per_sec
+    if records_per_sec is not None:
+        feeder["records_per_sec"] = records_per_sec
+
     out_paths = []
     for mode in modes:
         suffix = "_mv" if mode == BASELINE_MV else ""
@@ -687,12 +707,8 @@ def convert(
             "servers": [{"name": "clickhouse", "url": CLICKHOUSE_URL}],
             "dataset": {
                 "name": "custom",
-                "local_data_file": data_file,
-                # The runner's loader fills only this table; nothing fills the
-                # other metrics' tables yet.
-                "table": next(iter(labels)),
                 "init_sql_file": init_paths[mode],
-                "max_rows": 0,
+                "paced_feeder": feeder,
             },
             # Both arms' SQL, so the saved config labels queries by baseline's text.
             "query_groups": query_groups,
@@ -721,16 +737,21 @@ def convert(
 def main() -> None:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--table", required=True, help="sketch-bench table .json")
-    parser.add_argument(
-        "--data-file", required=True, help="dataset path as seen from experiments/"
-    )
+    parser.add_argument("--seed", type=int, required=True)
     parser.add_argument("--starting-delay-secs", type=int, required=True)
     parser.add_argument(
         "--duration-secs", type=int, required=True, help="query phase length"
     )
+    parser.add_argument("--samples-per-sec", type=int, help="gauges and counters")
+    parser.add_argument("--records-per-sec", type=int, help="records metrics")
     args = parser.parse_args()
     for path in convert(
-        args.table, args.data_file, args.starting_delay_secs, args.duration_secs
+        args.table,
+        args.seed,
+        args.starting_delay_secs,
+        args.duration_secs,
+        args.samples_per_sec,
+        args.records_per_sec,
     ):
         print(path)