diff --git a/asap-tools/components.conf b/asap-tools/components.conf index 70e60321..7e1c7b52 100644 --- a/asap-tools/components.conf +++ b/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 diff --git a/asap-tools/data-sources/paced-feeder/installation/install.sh b/asap-tools/data-sources/paced-feeder/installation/install.sh new file mode 100755 index 00000000..63acf2cd --- /dev/null +++ b/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" diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index 53bc78a8..eed807a5 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/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) diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py index d814aa9c..adeb011b 100644 --- a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py +++ b/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; - _clickhouse.yaml, _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 --data-file \ + --table --seed \ --starting-delay-secs --duration-secs """ 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) diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 7d2ee54d..8934b04d 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -88,8 +88,13 @@ Sketchdb mode runs when experiment_params.experiment contains an entry with mode: sketchdb (mirrors the Prometheus experiment config structure). Remove that entry to run baseline only. + +A sketch-bench workload table (sketch_bench_workloads.py configs) is fed in +real time instead, one paced feeder per metric table, when +experiment_params.dataset.paced_feeder is set. """ +import atexit import json import os import time @@ -109,6 +114,14 @@ PrometheusClientService, ) from experiment_utils.services.misc import ControllerService, DiscoveryBackend +from experiment_utils.services.paced_feeder import ( + SINK_CLICKHOUSE, + PacedFeederService, + aligned_interval_lcm, + check_paced_feeder_feed, + paced_feeder_min_duration_secs, + set_data_start, +) from experiment_utils.services.query_engine import QueryEngineRustService from experiment_utils.services.remote_monitor_service import RemoteMonitorService @@ -116,6 +129,12 @@ REMOTE_PROCESS_POLLING_INTERVAL = 10 CLICKHOUSE_INGEST_MONITOR_OUTPUT_FILE = "monitor_output_ingest.json" DEFAULT_MONITOR_INTERVAL_SECONDS = 1.0 +# remote_monitor and the query client's container start after the feed +# (10-20 s on CloudLab), so the feed must outlast the queries by this much. +QUERY_CLIENT_STARTUP_SECS = 30 +# For feeders to exit once their feed ends or after pkill, in-flight INSERTs +# included. +PACED_FEEDER_STOP_TIMEOUT_SECS = 30 def _start_clickhouse_controller(controller_service, cfg, **kwargs): @@ -307,6 +326,98 @@ def _ingest_mode(experiment_modes: List[str]) -> str: return experiment_modes[0] +def _check_clickhouse_paced_feeder( + feeder_config: dict, + experiment_modes: List[str], + client_config: dict, + minimum_experiment_running_time: int, +) -> int: + """Fail before ClickHouse starts on a feeder config the run can't serve; + returns the seconds the feeders' start must align to.""" + if constants.SKETCHDB_EXPERIMENT_NAME in experiment_modes: + raise ValueError( + "dataset.paced_feeder feeds ClickHouse only, and sketchdb mode's " + "engine ingests dataset.local_data_file; sketchdb on a paced feed " + "is not supported yet" + ) + check_paced_feeder_feed( + feeder_config, + paced_feeder_min_duration_secs( + QUERY_CLIENT_STARTUP_SECS, minimum_experiment_running_time + ), + ) + return aligned_interval_lcm(client_config) + + +def _start_clickhouse_feed( + *, + provider, + node_offset: int, + data_loader: ClickHouseDataLoaderService, + feeder_service: PacedFeederService, + feeder_config: dict, + feeder_align_secs: int, + dataset_name: str, + init_sql_file: str, + clickhouse_http_port: int, + experiment_mode: str, + experiment_root_output_dir: str, + local_experiment_root_dir: str, +) -> int: + """Recreate the metric tables, start one feeder per metric, and point the + mode's aligned query groups at the feed's start; returns that start in + epoch ms.""" + # An earlier mode's feeders (left running by no_teardown, or with an + # INSERT in flight) would write into the recreated tables. + feeder_service.stop() + if not feeder_service.wait_for_exit(PACED_FEEDER_STOP_TIMEOUT_SECS): + raise RuntimeError( + f"paced_feeder: still running {PACED_FEEDER_STOP_TIMEOUT_SECS} s " + "after pkill" + ) + # Before any data arrives: views only see rows inserted after they exist. + data_loader.run_init_sql(dataset_name, init_sql_file) + start_ms = feeder_service.start( + config=feeder_config, + sinks={ + SINK_CLICKHOUSE: f"http://localhost:{clickhouse_http_port}/" + f"?database={CLICKHOUSE_DATABASE}" + }, + experiment_output_dir=os.path.join(experiment_root_output_dir, experiment_mode), + local_experiment_dir=os.path.join(local_experiment_root_dir, experiment_mode), + align_secs=feeder_align_secs, + ) + local_client_config = os.path.join( + local_experiment_root_dir, + "controller_client_configs", + f"{experiment_mode}.yaml", + ) + with open(local_client_config) as f: + client_config = yaml.safe_load(f) + set_data_start(client_config, start_ms) + with open(local_client_config, "w") as f: + yaml.safe_dump(client_config, f, sort_keys=False) + sync.rsync_controller_client_configs( + provider, + experiment_root_output_dir, + local_experiment_root_dir, + node_offset=node_offset, + ) + return start_ms + + +def _finish_clickhouse_feed( + feeder_service: PacedFeederService, start_ms: int, duration_secs: int +) -> None: + """Let the feeders run to the end of their feed, so they write their + stats (they write none when killed), then stop any that overstay.""" + remaining_secs = start_ms / 1000 + duration_secs - time.time() + if not feeder_service.wait_for_exit( + max(remaining_secs, 0) + PACED_FEEDER_STOP_TIMEOUT_SECS + ): + feeder_service.stop() + + def _run_clickhouse_ingest_with_monitor( provider, node_offset: int, @@ -484,10 +595,10 @@ def main(cfg: DictConfig) -> None: ep = cfg.experiment_params dataset_cfg = ep.dataset dataset_name = str(dataset_cfg.name) - local_data_file = str(dataset_cfg.local_data_file) - table = dataset_cfg.table + # Either a feed of every metric table in real time or a bulk load of + # local_data_file into `table` before the modes run. + paced_feeder_config = dataset_cfg.get("paced_feeder") init_sql_file = dataset_cfg.init_sql_file - max_rows = int(dataset_cfg.max_rows) # --- experiment modes and server URLs from config (mirrors Prometheus structure) --- experiment_cfg = OmegaConf.to_container(ep.experiment, resolve=True) @@ -513,11 +624,39 @@ def main(cfg: DictConfig) -> None: node_offset=node_offset, ) - # --- rsync dataset file to node --- - remote_data_dir = os.path.join(experiment_root_output_dir, "data") - remote_data_file = sync.rsync_dataset_file( - provider, local_data_file, remote_data_dir, node_offset - ) + paced_feeder_service = None + if paced_feeder_config is not None: + paced_feeder_config = OmegaConf.to_container(paced_feeder_config, resolve=True) + # The feeder runs from its own directory. + paced_feeder_config["workload_table"] = os.path.abspath( + paced_feeder_config["workload_table"] + ) + # Every mode's client config has the same query timing. + with open( + os.path.join( + local_experiment_root_dir, + "controller_client_configs", + f"{experiment_modes[0]}.yaml", + ) + ) as f: + paced_feeder_align_secs = _check_clickhouse_paced_feeder( + paced_feeder_config, + experiment_modes, + yaml.safe_load(f), + minimum_experiment_running_time, + ) + # Next to ClickHouse, so the feed crosses no network. + paced_feeder_service = PacedFeederService(provider, [node_offset]) + if not no_teardown: + # Otherwise a failed run leaves feeders inserting into the next + # run's tables. + atexit.register(paced_feeder_service.stop) + else: + # --- rsync dataset file to node --- + remote_data_dir = os.path.join(experiment_root_output_dir, "data") + remote_data_file = sync.rsync_dataset_file( + provider, str(dataset_cfg.local_data_file), remote_data_dir, node_offset + ) # --- start ClickHouse (persists across all modes) --- clickhouse_service = ClickHouseService( @@ -533,38 +672,40 @@ def main(cfg: DictConfig) -> None: image_tag=dataset_cfg.get("clickhouse_image_tag", "latest"), ) - # --- load data once before the mode loop (DROP + reload), with ingest monitor --- + # --- bulk load once before the mode loop (DROP + reload), with ingest + # monitor; a paced feed instead starts in each mode --- data_loader = ClickHouseDataLoaderService( provider, num_nodes=num_nodes, node_offset=node_offset, clickhouse_http_port=clickhouse_http_port, ) - ingest_mode = _ingest_mode(experiment_modes) - ingest_client_config = os.path.join( - experiment_root_output_dir, - "controller_client_configs", - f"{ingest_mode}.yaml", - ) - ingest_output_dir = os.path.join(experiment_root_output_dir, ingest_mode) - local_ingest_dir = os.path.join(local_experiment_root_dir, ingest_mode) remote_monitor_service = RemoteMonitorService(provider, node_offset=node_offset) - _run_clickhouse_ingest_with_monitor( - provider=provider, - node_offset=node_offset, - remote_monitor_service=remote_monitor_service, - ingest_output_dir=ingest_output_dir, - local_ingest_dir=local_ingest_dir, - controller_client_config=ingest_client_config, - data_loader=data_loader, - dataset_name=dataset_name, - remote_data_file=remote_data_file, - table=table, - init_sql_file=init_sql_file, - max_rows=max_rows, - manual_remote_monitor=manual_remote_monitor, - monitor_interval_seconds=monitor_interval_seconds, - ) + if paced_feeder_service is None: + ingest_mode = _ingest_mode(experiment_modes) + ingest_client_config = os.path.join( + experiment_root_output_dir, + "controller_client_configs", + f"{ingest_mode}.yaml", + ) + ingest_output_dir = os.path.join(experiment_root_output_dir, ingest_mode) + local_ingest_dir = os.path.join(local_experiment_root_dir, ingest_mode) + _run_clickhouse_ingest_with_monitor( + provider=provider, + node_offset=node_offset, + remote_monitor_service=remote_monitor_service, + ingest_output_dir=ingest_output_dir, + local_ingest_dir=local_ingest_dir, + controller_client_config=ingest_client_config, + data_loader=data_loader, + dataset_name=dataset_name, + remote_data_file=remote_data_file, + table=dataset_cfg.table, + init_sql_file=init_sql_file, + max_rows=int(dataset_cfg.max_rows), + manual_remote_monitor=manual_remote_monitor, + monitor_interval_seconds=monitor_interval_seconds, + ) # --- mode loop --- prometheus_client_service = PrometheusClientService( @@ -741,6 +882,21 @@ def generate_controller_output() -> None: else: # --- baseline mode --- + if paced_feeder_service is not None: + feed_start_ms = _start_clickhouse_feed( + provider=provider, + node_offset=node_offset, + data_loader=data_loader, + feeder_service=paced_feeder_service, + feeder_config=paced_feeder_config, + feeder_align_secs=paced_feeder_align_secs, + dataset_name=dataset_name, + init_sql_file=init_sql_file, + clickhouse_http_port=clickhouse_http_port, + experiment_mode=experiment_mode, + experiment_root_output_dir=experiment_root_output_dir, + local_experiment_root_dir=local_experiment_root_dir, + ) controller_client_config = os.path.join( experiment_root_output_dir, "controller_client_configs", @@ -761,6 +917,12 @@ def generate_controller_output() -> None: profile_prometheus_time=profile_prometheus_time, monitor_interval_seconds=monitor_interval_seconds, ) + if paced_feeder_service is not None: + _finish_clickhouse_feed( + paced_feeder_service, + feed_start_ms, + paced_feeder_config["duration_secs"], + ) sync.rsync_experiment_data( provider, diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index ca388bcf..6c139544 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -35,6 +35,7 @@ paced_feeder_sinks, set_data_start, ) +from experiment_utils.services.fake_exporters import get_exporter_target_nodes from experiment_utils.services.misc import DiscoveryBackend CONTROLLER_LOCAL_OUTPUT_DIR = None @@ -224,7 +225,9 @@ def main(cfg: DictConfig): paced_feeder_align_secs = aligned_interval_lcm( OmegaConf.to_container(cfg.experiment_params, resolve=True) ) - paced_feeder_service = PacedFeederService(provider, args) + paced_feeder_service = PacedFeederService( + provider, get_exporter_target_nodes(args) + ) if not args.no_teardown: # Otherwise a failed run leaves feeders pushing into the next # run's sinks. diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py index 5c64beb8..197232d0 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/asap-tools/experiments/experiment_utils/config.py @@ -132,13 +132,33 @@ def _validate_clickhouse_experiment_config(experiment_params: DictConfig) -> Non f"Valid choices: {valid_dataset_names}" ) + feeder = dataset.get("paced_feeder") local_data_file = dataset.get("local_data_file") - if not local_data_file or local_data_file == "???": + if feeder is not None: + # Fed live instead of bulk-loaded: the feeder generates sketch-bench + # tables, which only the init SQL creates. + if dataset_name != "custom": + raise ValueError( + f"dataset.paced_feeder needs dataset.name='custom', got {dataset_name!r}" + ) + workload_table = feeder.get("workload_table") + if not workload_table or not os.path.exists(workload_table): + raise ValueError( + f"dataset.paced_feeder.workload_table={workload_table!r} " + "does not exist." + ) + init_sql_file = dataset.get("init_sql_file") + if not init_sql_file or not os.path.exists(init_sql_file): + raise ValueError( + f"dataset.init_sql_file={init_sql_file!r} does not exist; with " + "dataset.paced_feeder it creates the metric tables." + ) + elif not local_data_file or local_data_file == "???": raise ValueError( "dataset.local_data_file is required. " "Provide the path to the JSON-lines data file on this machine." ) - if not os.path.exists(local_data_file): + elif not os.path.exists(local_data_file): raise ValueError( f"dataset.local_data_file={local_data_file!r} does not exist. " "Run benchmark/prepare_data.py first to produce the JSON-lines file." diff --git a/asap-tools/experiments/experiment_utils/services/clickhouse_service.py b/asap-tools/experiments/experiment_utils/services/clickhouse_service.py index 93e28eae..4b08174b 100644 --- a/asap-tools/experiments/experiment_utils/services/clickhouse_service.py +++ b/asap-tools/experiments/experiment_utils/services/clickhouse_service.py @@ -316,12 +316,7 @@ def start( # resetting dependent materialized-view tables. self._exec_sql(f"TRUNCATE TABLE IF EXISTS {table}", url) print(f"Running init SQL from {init_sql_file!r}...") - remote_ddl = f"/tmp/{dataset_name}_init_{os.getpid()}.sql" - self._rsync_to_remote(init_sql_file, remote_ddl) - try: - self._exec_sql_file(remote_ddl, url) - finally: - self._remote_rm(remote_ddl) + self._exec_local_sql_file(dataset_name, init_sql_file, url) elif dataset_name in self.BUILTIN_DDL_FILES: print(f"Dropping table {table!r}...") self._exec_sql(f"DROP TABLE IF EXISTS {table}", url) @@ -329,12 +324,7 @@ def start( self._ASSETS_DIR, self.BUILTIN_DDL_FILES[dataset_name] ) print(f"Initializing schema for {dataset_name!r} from {local_ddl!r}...") - remote_ddl = f"/tmp/{dataset_name}_init_{os.getpid()}.sql" - self._rsync_to_remote(local_ddl, remote_ddl) - try: - self._exec_sql_file(remote_ddl, url) - finally: - self._remote_rm(remote_ddl) + self._exec_local_sql_file(dataset_name, local_ddl, url) elif dataset_name == "custom": print(f"Dropping table {table!r}...") self._exec_sql(f"DROP TABLE IF EXISTS {table}", url) @@ -358,9 +348,25 @@ def start( count = self._check_row_count(table, url) print(f"Loaded {count:,} rows into ClickHouse ({table!r})") + def run_init_sql(self, dataset_name: str, init_sql_file: str) -> None: + """Run a local DDL SQL file on the node, loading no data.""" + url = f"http://localhost:{self.clickhouse_http_port}/" + self._ensure_clickhouse_http_ready() + print(f"Running init SQL from {init_sql_file!r}...") + self._exec_local_sql_file(dataset_name, init_sql_file, url) + def stop(self, **kwargs) -> None: pass + def _exec_local_sql_file(self, dataset_name: str, local_file: str, url: str): + """Copy a local SQL file to the node and execute it there.""" + remote_ddl = f"/tmp/{dataset_name}_init_{os.getpid()}.sql" + self._rsync_to_remote(local_file, remote_ddl) + try: + self._exec_sql_file(remote_ddl, url) + finally: + self._remote_rm(remote_ddl) + # ------------------------------------------------------------------ # # Internal helpers # # ------------------------------------------------------------------ # diff --git a/asap-tools/experiments/experiment_utils/services/paced_feeder.py b/asap-tools/experiments/experiment_utils/services/paced_feeder.py index cc9da43c..038d2473 100644 --- a/asap-tools/experiments/experiment_utils/services/paced_feeder.py +++ b/asap-tools/experiments/experiment_utils/services/paced_feeder.py @@ -1,9 +1,10 @@ """ Paced feeder service: pushes a sketch-bench workload table's metrics in real -time over Prometheus remote write, one feeder process per (sink, metric). +time, one feeder process per (sink, metric). Unlike the fake exporters, nothing is scraped: the feeder writes straight to -ASAPQuery's ingest and/or Prometheus' remote-write receiver. Every process gets +ASAPQuery's ingest and/or Prometheus' remote-write receiver over remote write, +or to ClickHouse's HTTP interface, one table per metric. Every process gets the same seed and table, so each system receives identical data. """ @@ -16,7 +17,6 @@ import constants from .base import BaseService -from .fake_exporters import get_exporter_target_nodes from experiment_utils import sync from experiment_utils.providers.base import InfrastructureProvider @@ -39,6 +39,7 @@ # A feeder stops after its first tick if every request fails (e.g. a wrong # URL), so check the processes one tick after the shared start. PACED_FEEDER_STARTUP_CHECK_SECS = 2 +PACED_FEEDER_EXIT_POLL_SECS = 5 # Feeders wait for the next multiple of the aligned intervals' LCM before # sending; refuse intervals that would make that wait longer than this. PACED_FEEDER_MAX_ALIGN_SECS = 600 @@ -48,6 +49,8 @@ SINK_QUERY_ENGINE = "query_engine" SINK_PROMETHEUS = "prometheus" +# Takes every metric type, each into the table named after it. +SINK_CLICKHOUSE = "clickhouse" def paced_feeder_metric_names(experiment_params) -> List[str]: @@ -159,6 +162,25 @@ def check_paced_feeder_config( "controller.auto_discover_labels reads labels from Prometheus before " "the paced_feeder starts; list the labels under experiment_params.metrics" ) + table_metrics = check_paced_feeder_feed(config, min_feed_secs) + for metric in config["metrics"]: + if table_metrics[metric] == RECORDS_METRIC_TYPE and prometheus_modes: + raise ValueError( + f"paced_feeder metric {metric!r} is a records metric, which " + f"only ASAPQuery takes, but modes {prometheus_modes} feed " + "Prometheus (baseline or query_prometheus_too)" + ) + if sketchdb_modes and parallelism != 1: + raise ValueError( + "paced_feeder sends to one ingest port; set streaming.parallelism=1" + ) + + +def check_paced_feeder_feed( + config: Dict[str, Any], min_feed_secs: float +) -> Dict[str, str]: + """Fail on a metric the table lacks or one of unknown type, or a feed + that stops before the queries end; returns the table's metric types.""" table_metrics = table_metric_types(config["workload_table"]) for metric in config["metrics"]: if metric not in table_metrics: @@ -167,21 +189,10 @@ def check_paced_feeder_config( f"{config['workload_table']} (has {sorted(table_metrics)})" ) metric_type = table_metrics[metric] - if metric_type == RECORDS_METRIC_TYPE: - if prometheus_modes: - raise ValueError( - f"paced_feeder metric {metric!r} is a records metric, which " - f"only ASAPQuery takes, but modes {prometheus_modes} feed " - "Prometheus (baseline or query_prometheus_too)" - ) - elif metric_type not in PROMETHEUS_METRIC_TYPES: + if metric_type not in PROMETHEUS_METRIC_TYPES + (RECORDS_METRIC_TYPE,): raise ValueError( f"paced_feeder metric {metric!r} has unknown type {metric_type!r}" ) - if sketchdb_modes and parallelism != 1: - raise ValueError( - "paced_feeder sends to one ingest port; set streaming.parallelism=1" - ) if config["duration_secs"] < min_feed_secs: raise ValueError( f"paced_feeder.duration_secs={config['duration_secs']} ends before the " @@ -189,6 +200,7 @@ def check_paced_feeder_config( "(feeder startup + flow.steady_state_wait + the query groups' " "running time)" ) + return table_metrics def paced_feeder_sinks( @@ -215,9 +227,10 @@ def paced_feeder_command( metric_type: str, start_ms: int, stats_out: str, + sink_kind: str = "remote-write", ) -> str: cmd = ( - f"{PACED_FEEDER_BINARY} --sink remote-write --url {url}" + f"{PACED_FEEDER_BINARY} --sink {sink_kind} --url {url}" f" --workload-table {workload_table} --metric {metric}" f" --seed {config['seed']} --duration-secs {config['duration_secs']}" f" --start-ms {start_ms} --stats-out {stats_out}" @@ -234,12 +247,11 @@ def paced_feeder_command( class PacedFeederService(BaseService): - """Runs paced_feeder processes bare-metal, spread over the worker nodes - (the coordinator in local mode).""" + """Runs paced_feeder processes bare-metal, spread over `target_nodes`.""" - def __init__(self, provider: InfrastructureProvider, args): + def __init__(self, provider: InfrastructureProvider, target_nodes: List[int]): super().__init__(provider) - self.target_nodes = get_exporter_target_nodes(args) + self.target_nodes = target_nodes self.cmd_dir = os.path.join( provider.get_home_dir(), "code", @@ -301,6 +313,7 @@ def start( metric_types[metric], start_ms, os.path.join(output_dir, f"{name}_stats.json"), + "clickhouse" if sink == SINK_CLICKHOUSE else "remote-write", ) node_cmds.append( ( @@ -351,6 +364,29 @@ def start( ) return start_ms + def wait_for_exit(self, timeout_secs: float) -> bool: + """Wait until no feeder runs on the target nodes; False on timeout.""" + deadline = time.time() + timeout_secs + while True: + running = [ + node_idx + for node_idx in self.target_nodes + if self.provider.execute_command( + node_idx=node_idx, + cmd=f"pgrep -f '{PACED_FEEDER_PROCESS_PATTERN}'", + cmd_dir=None, + nohup=False, + popen=False, + ignore_errors=True, + ).returncode + == 0 + ] + if not running: + return True + if time.time() >= deadline: + return False + time.sleep(PACED_FEEDER_EXIT_POLL_SECS) + def stop(self, **kwargs) -> None: self.provider.execute_command_parallel( node_idxs=self.target_nodes, diff --git a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py index e73370c3..743bdb01 100644 --- a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py +++ b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py @@ -1,16 +1,28 @@ """Regression tests for ClickHouse experiment controller wiring.""" +import os +import tempfile import unittest from unittest.mock import Mock, patch +import yaml + from omegaconf import OmegaConf from experiment_run_clickhouse import ( + PACED_FEEDER_STOP_TIMEOUT_SECS, + QUERY_CLIENT_STARTUP_SECS, + _check_clickhouse_paced_feeder, _deploy_prebuilt_clickhouse_controller_output, _ingest_mode, _prepare_clickhouse_controller_output, _start_clickhouse_controller, + _finish_clickhouse_feed, + _start_clickhouse_feed, ) +from experiment_utils.config import validate_experiment_config +from experiment_utils.services.paced_feeder import PACED_FEEDER_STARTUP_CHECK_SECS +from test_paced_feeder import feeder_config, write_table class ClickHouseControllerConfigTest(unittest.TestCase): @@ -90,5 +102,199 @@ def test_baseline_when_present_else_first_mode(self): self.assertEqual(_ingest_mode(["baseline_mv"]), "baseline_mv") +def aligned_client_config(*intervals_secs): + return { + "query_groups": [ + { + "id": i, + "repetition_delay_ms": secs * 1000, + "client_options": {"align_to_interval": True, "query_time_offset": 10}, + } + for i, secs in enumerate(intervals_secs) + ] + } + + +class ClickHousePacedFeederCheckTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.table = write_table(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def check(self, modes, **overrides): + return _check_clickhouse_paced_feeder( + feeder_config(self.table, **overrides), + modes, + aligned_client_config(10, 60), + minimum_experiment_running_time=300, + ) + + def test_aligns_to_the_query_intervals(self): + self.assertEqual(self.check(["baseline", "baseline_mv"]), 60) + + def test_records_metrics_go_to_clickhouse(self): + # Unlike Prometheus, ClickHouse keeps every record of a slot. + self.check(["baseline"], metrics=["data", "flows"]) + + def test_sketchdb_mode_refused(self): + # sketchdb's engine ingests local_data_file, which a feed has none of. + with self.assertRaisesRegex(ValueError, "sketchdb"): + self.check(["baseline", "sketchdb"]) + + def test_feed_ending_before_the_queries_refused(self): + with self.assertRaisesRegex(ValueError, "ends before the queries"): + self.check(["baseline"], duration_secs=300) + + def test_feed_must_outlast_the_query_client_startup(self): + # The client starts after remote_monitor and its container, not right + # after the feeders' startup check. + startup_check_only = 300 + PACED_FEEDER_STARTUP_CHECK_SECS + with self.assertRaisesRegex(ValueError, "ends before the queries"): + self.check(["baseline"], duration_secs=startup_check_only) + self.check( + ["baseline"], + duration_secs=startup_check_only + QUERY_CLIENT_STARTUP_SECS, + ) + + +class StartClickHouseFeedTest(unittest.TestCase): + def start(self, calls, exited=True): + calls.feeder.start.return_value = 1700000060000 + calls.feeder.wait_for_exit.return_value = exited + with tempfile.TemporaryDirectory() as root, patch( + "experiment_run_clickhouse.sync.rsync_controller_client_configs" + ) as self.rsync: + config_dir = os.path.join(root, "controller_client_configs") + os.makedirs(config_dir) + config_path = os.path.join(config_dir, "baseline.yaml") + with open(config_path, "w") as f: + yaml.safe_dump(aligned_client_config(60), f) + + start_ms = _start_clickhouse_feed( + provider=Mock(), + node_offset=0, + data_loader=calls.loader, + feeder_service=calls.feeder, + feeder_config={"metrics": ["data"]}, + feeder_align_secs=60, + dataset_name="custom", + init_sql_file="init.sql", + clickhouse_http_port=8123, + experiment_mode="baseline", + experiment_root_output_dir="/exp", + local_experiment_root_dir=root, + ) + with open(config_path) as f: + self.client_config = yaml.safe_load(f) + return start_ms + + def test_tables_are_created_before_the_feed_and_queries_follow_it(self): + calls = Mock() + self.assertEqual(self.start(calls), 1700000060000) + # Old feeders gone before the tables are recreated; a view sees only + # rows inserted after it exists. + self.assertEqual( + [c[0] for c in calls.mock_calls], + [ + "feeder.stop", + "feeder.wait_for_exit", + "loader.run_init_sql", + "feeder.start", + ], + ) + # The feed starts on a boundary of the aligned intervals' LCM. + start_kwargs = calls.feeder.start.call_args.kwargs + self.assertEqual(start_kwargs["align_secs"], 60) + self.assertEqual( + start_kwargs["sinks"], + {"clickhouse": "http://localhost:8123/?database=default"}, + ) + self.assertEqual(start_kwargs["experiment_output_dir"], "/exp/baseline") + self.assertEqual( + self.client_config["query_groups"][0]["client_options"]["data_start_time"], + 1700000060, + ) + self.rsync.assert_called_once() + + def test_feeders_that_outlive_pkill_stop_the_run(self): + # With no_teardown, or an INSERT in flight, they would write into the + # next mode's recreated tables. + calls = Mock() + with self.assertRaisesRegex(RuntimeError, "still running"): + self.start(calls, exited=False) + calls.loader.run_init_sql.assert_not_called() + calls.feeder.start.assert_not_called() + + +class FinishClickHouseFeedTest(unittest.TestCase): + def finish(self, exited): + feeder = Mock() + feeder.wait_for_exit.return_value = exited + with patch("experiment_run_clickhouse.time.time", return_value=1700000600): + _finish_clickhouse_feed(feeder, 1700000000000, duration_secs=720) + return feeder + + def test_waits_out_the_feed_so_feeders_write_their_stats(self): + # A killed feeder writes no stats; 120 s of feed remain. + feeder = self.finish(exited=True) + feeder.wait_for_exit.assert_called_once_with( + 120 + PACED_FEEDER_STOP_TIMEOUT_SECS + ) + feeder.stop.assert_not_called() + + def test_stops_feeders_that_overstay(self): + self.finish(exited=False).stop.assert_called_once() + + +class FedClickHouseConfigValidationTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.table = write_table(self.tmp.name) + self.init_sql = os.path.join(self.tmp.name, "init.sql") + self.sql = os.path.join(self.tmp.name, "q.sql") + for path in (self.init_sql, self.sql): + with open(path, "w") as f: + f.write("SELECT 1;\n") + + def tearDown(self): + self.tmp.cleanup() + + def validate(self, feeder=None, **dataset): + validate_experiment_config( + OmegaConf.create( + { + "experiment": [{"mode": "baseline", "server": "clickhouse"}], + "servers": [{"name": "clickhouse", "url": "http://localhost:8123"}], + "dataset": { + "name": "custom", + "init_sql_file": self.init_sql, + "paced_feeder": feeder_config(self.table, **(feeder or {})), + **dataset, + }, + "query_groups": [{"sql_file": {"baseline": self.sql}}], + } + ) + ) + + def test_fed_config_needs_no_data_file(self): + self.validate() + + def test_missing_init_sql_refused(self): + # It is the only thing that creates the metric tables. + with self.assertRaisesRegex(ValueError, "init_sql_file"): + self.validate(init_sql_file=None) + + def test_missing_workload_table_refused(self): + with self.assertRaisesRegex(ValueError, "workload_table"): + self.validate(feeder={"workload_table": None}) + + def test_builtin_dataset_refused(self): + # The feeder generates sketch-bench tables, not clickbench's or h2o's. + with self.assertRaisesRegex(ValueError, "custom"): + self.validate(name="clickbench") + + if __name__ == "__main__": unittest.main() diff --git a/asap-tools/experiments/tests/test_paced_feeder.py b/asap-tools/experiments/tests/test_paced_feeder.py index c297519a..2ecd12bc 100644 --- a/asap-tools/experiments/tests/test_paced_feeder.py +++ b/asap-tools/experiments/tests/test_paced_feeder.py @@ -33,6 +33,7 @@ BASELINE = {"mode": "baseline", "server": "prometheus"} BOTH_MODES = [SKETCHDB, BASELINE] PROM_URL = "http://10.10.1.1:9090/api/v1/write" +CH_URL = "http://localhost:8123/?database=default" # workloads[0].e2e.metrics of synthetic-templatesclassic-shared1.json, plus a # records metric from the multi-grouping tables. @@ -377,14 +378,6 @@ def execute_command(self, node_idx, cmd, **kwargs): return subprocess.CompletedProcess(cmd, returncode) -class Args: - def get_node_range(self, include_coordinator=True): - return [0, 1, 2] if include_coordinator else [1, 2] - - def get_coordinator_node(self): - return 0 - - class PacedFeederServiceStartTest(unittest.TestCase): def start( self, @@ -396,7 +389,7 @@ def start( ): provider = RecordingProvider(running_count, launch_secs) self.provider = provider - service = PacedFeederService(provider, Args()) + service = PacedFeederService(provider, [1, 2]) def rsync(*args): provider.clock += PREP_SECS @@ -482,6 +475,21 @@ def test_records_and_gauge_processes_get_their_rate_flags(self): {"data": ["--samples-per-sec"], "flows": ["--records-per-sec"]}, ) + def test_clickhouse_sink_sends_rowbinary_inserts(self): + commands = self.start( + running_count=1, + sinks={"clickhouse": CH_URL}, + metrics=["data", "flows"], + ) + self.assertEqual( + [ + (c.split(" --sink ")[1].split()[0], c.split(" --url ")[1].split()[0]) + for _, c in commands + if "--sink" in c + ], + [("clickhouse", CH_URL), ("clickhouse", CH_URL)], + ) + def test_one_process_per_sink_and_metric_spread_over_workers(self): commands = self.start(running_count=2) feeders = [(n, c) for n, c in commands if "--sink" in c] @@ -532,6 +540,47 @@ def execute_command(self, node_idx, cmd, **kwargs): return subprocess.CompletedProcess(cmd, 1) +class PgrepProvider: + """pgrep finds feeders on a node for its first `polls[node]` checks.""" + + def __init__(self, polls): + self.polls = dict(polls) + self.clock = 0.0 + + def get_home_dir(self): + return "/home/u" + + def execute_command(self, node_idx, cmd, **kwargs): + assert cmd.startswith("pgrep -f "), cmd + running = self.polls.get(node_idx, 0) > 0 + self.polls[node_idx] = self.polls.get(node_idx, 0) - 1 + return subprocess.CompletedProcess(cmd, 0 if running else 1) + + +class WaitForExitTest(unittest.TestCase): + def wait(self, polls, timeout_secs): + provider = PgrepProvider(polls) + service = PacedFeederService(provider, [1, 2]) + + def sleep(secs): + provider.clock += secs + + with mock.patch( + "experiment_utils.services.paced_feeder.time.sleep", side_effect=sleep + ), mock.patch( + "experiment_utils.services.paced_feeder.time.time", + lambda: provider.clock, + ): + return service.wait_for_exit(timeout_secs) + + def test_returns_once_no_node_runs_a_feeder(self): + # Node 2's feeder exits after two polls. + self.assertTrue(self.wait({2: 2}, timeout_secs=60)) + + def test_times_out_while_a_feeder_runs(self): + self.assertFalse(self.wait({1: 100}, timeout_secs=10)) + + class PrometheusRemoteWriteReceiverTest(unittest.TestCase): def bare_metal_command(self, receiver): provider = StartCommandProvider() diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py index 5f1ad016..8b3540b3 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_workloads.py +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -30,6 +30,8 @@ ) import sketch_bench_workloads as sbw # noqa: E402 +from experiment_utils.config import validate_experiment_config # noqa: E402 +from omegaconf import OmegaConf # noqa: E402 SERIES_LABELS = [{"name": "job"}, {"name": "label_0", "child_of": "job"}] HTTP_LABELS = ["region", "service", "endpoint", "status"] @@ -300,7 +302,7 @@ def test_writes_config_grouped_by_interval(self): try: # 30 s < offset + the 15m windows: early windows predate data. with self.assertWarns(UserWarning): - out_paths = sbw.convert(table, "data.json", 30, 600) + out_paths = sbw.convert(table, 42, 30, 600) cfgs = {} for path in out_paths: with open(path) as f: @@ -310,6 +312,19 @@ def test_writes_config_grouped_by_interval(self): cfgs[arm["mode"]] = (cfg, f.read()) self.assertEqual(sorted(cfgs), ["baseline", "baseline_mv"]) cfg, init_mv = cfgs["baseline_mv"] + # Every metric table gets a feed, not just the first. + self.assertEqual( + cfg["dataset"]["paced_feeder"], + { + "workload_table": os.path.abspath(table), + "metrics": list(METRICS), + "seed": 42, + "duration_secs": 30 + 600 + sbw.FEEDER_SLACK_SECS, + }, + ) + self.assertNotIn("local_data_file", cfg["dataset"]) + # The runner's first check: a fed config has no data file. + validate_experiment_config(OmegaConf.create(cfg)) _, init = cfgs["baseline"] groups = {g["name"]: g for g in cfg["query_groups"]} self.assertEqual(sorted(groups), ["interval_1s", "interval_60s"]) @@ -392,12 +407,12 @@ def test_unservable_query_skips_mv_arm(self): } json.dump({"workloads": [{"e2e": e2e}]}, f) with self.assertWarns(UserWarning): - sbw.convert(table, "data.json", 30, 600) + sbw.convert(table, 42, 30, 600) with open(table, "w") as f: e2e = {"metrics": METRICS, "queries": [instant, servable]} json.dump({"workloads": [{"e2e": e2e}]}, f) with contextlib.redirect_stderr(io.StringIO()) as err: - (path,) = sbw.convert(table, "data.json", 30, 600) + (path,) = sbw.convert(table, 42, 30, 600) self.assertIn("no baseline_mv arm", err.getvalue()) self.assertIn("sum by (job) (data)", err.getvalue()) with open(path) as f: @@ -438,7 +453,7 @@ def convert(queries): with open(table, "w") as f: e2e = {"metrics": METRICS, "queries": queries} json.dump({"workloads": [{"e2e": e2e}]}, f) - return sbw.convert(table, "data.json", 310, 600) + return sbw.convert(table, 42, 310, 600) self.assertEqual(len(convert([good])), 2) before = snapshot(sbw.OUT_CONFIG_DIR) @@ -459,16 +474,16 @@ def test_checks_workloads_duration_and_delay(self): with open(table, "w") as f: json.dump({"workloads": [{"e2e": e2e}, {"e2e": e2e}]}, f) with self.assertRaises(ValueError): - sbw.convert(table, "data.json", 310, 600) + sbw.convert(table, 42, 310, 600) with open(table, "w") as f: json.dump({"workloads": [{"e2e": e2e}]}, f) # Interval 60 s, so 30 s would yield 0 repetitions. with self.assertRaises(ValueError): - sbw.convert(table, "data.json", 310, 30) + sbw.convert(table, 42, 310, 30) # Delay covers offset + window: no warning. with warnings.catch_warnings(): warnings.simplefilter("error") - sbw.convert(table, "data.json", 310, 600) + sbw.convert(table, 42, 310, 600) finally: os.chdir(cwd) diff --git a/asap-tools/installation/common/setup_dependencies.sh b/asap-tools/installation/common/setup_dependencies.sh index 2e3930a4..6120a78a 100755 --- a/asap-tools/installation/common/setup_dependencies.sh +++ b/asap-tools/installation/common/setup_dependencies.sh @@ -24,7 +24,6 @@ else fi # (cd "$THIS_DIR/../../experiments/fake_exporter_rust/fake_exporter"; cargo build --release) (cd "$THIS_DIR/../../../asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter" && cargo build --release) -(cd "$THIS_DIR/../../../asap-tools/data-sources/paced-feeder" && cargo build --release) DOCKER_DATA_DIR=/scratch/var_lib_docker