diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index 520863ef..e74ee61b 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md @@ -500,7 +500,7 @@ table with `sketch_bench_e2e/sketch_bench_e2e_configs.py`. - **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` +- **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. In `sketchdb` mode the engine ingests over remote write (`streaming.remote_write.base_port`/`path`) instead of `dataset.local_data_file`, and the feed goes to ASAPQuery only, so it forwards no unsupported queries to ClickHouse (whose copy of the data would load the measured node); it waits for no `flow.steady_state_wait` and needs `controller.prebuilt_output_dir` (a fed dataset has no `dataset.precompute` to build the planner's input from). `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 diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 877e7e4b..e9d7bc35 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -91,14 +91,18 @@ 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. +experiment_params.dataset.paced_feeder is set. Each mode then recreates the +tables and starts its own feed. sketchdb's feed goes only to the engine's +remote-write ingest, so it forwards no queries to ClickHouse, and it needs +controller.prebuilt_output_dir. """ import atexit +import functools import json import os import time -from typing import Callable, List, Optional +from typing import Callable, Dict, List, Optional from urllib.parse import urlparse import hydra @@ -116,6 +120,7 @@ from experiment_utils.services.misc import ControllerService, DiscoveryBackend from experiment_utils.services.paced_feeder import ( SINK_CLICKHOUSE, + SINK_QUERY_ENGINE, PacedFeederService, aligned_interval_lcm, check_paced_feeder_feed, @@ -331,14 +336,18 @@ def _check_clickhouse_paced_feeder( experiment_modes: List[str], client_config: dict, minimum_experiment_running_time: int, + controller_prebuilt_output_dir: Optional[str] = None, ) -> 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: + if ( + constants.SKETCHDB_EXPERIMENT_NAME in experiment_modes + and not controller_prebuilt_output_dir + ): 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" + "sketchdb on dataset.paced_feeder needs controller.prebuilt_output_dir: " + "the planner's input is built from dataset.precompute, which a fed " + "dataset has none of" ) check_paced_feeder_feed( feeder_config, @@ -349,7 +358,18 @@ def _check_clickhouse_paced_feeder( return aligned_interval_lcm(client_config) -def _start_clickhouse_feed( +def _feed_sinks( + experiment_mode: str, clickhouse_url: str, query_engine_url: str +) -> Dict[str, str]: + """The sinks a mode's feed writes to. sketchdb feeds only ASAPQuery: it + forwards no queries, and ClickHouse ingesting a copy would load the node + its CPU and latency are measured on. Every other mode reads ClickHouse.""" + if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: + return {SINK_QUERY_ENGINE: query_engine_url} + return {SINK_CLICKHOUSE: clickhouse_url} + + +def _start_feed( *, provider, node_offset: int, @@ -359,14 +379,14 @@ def _start_clickhouse_feed( feeder_align_secs: int, dataset_name: str, init_sql_file: str, - clickhouse_http_port: int, + sinks: Dict[str, str], 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.""" + """Recreate the metric tables, start one feeder per (sink, metric), and + point the mode's aligned query groups at the feed's start; returns that + start in epoch ms; `sinks` from _feed_sinks.""" # 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() @@ -379,11 +399,7 @@ def _start_clickhouse_feed( data_loader.run_init_sql(dataset_name, init_sql_file) start_ms = feeder_service.start( config=feeder_config, - sinks={ - # The feeder runs on the next node (see main()). - SINK_CLICKHOUSE: f"http://{provider.get_node_ip(node_offset)}:" - f"{clickhouse_http_port}/?database={CLICKHOUSE_DATABASE}" - }, + sinks=sinks, 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, @@ -632,12 +648,17 @@ def main(cfg: DictConfig) -> None: paced_feeder_config["workload_table"] = os.path.abspath( paced_feeder_config["workload_table"] ) - # Every mode's client config has the same query timing. + # Every mode but sketchdb runs every query group (sketchdb may skip + # trailing ones), so any other mode's client config has them all. + all_groups_mode = next( + (m for m in experiment_modes if m != constants.SKETCHDB_EXPERIMENT_NAME), + experiment_modes[0], + ) with open( os.path.join( local_experiment_root_dir, "controller_client_configs", - f"{experiment_modes[0]}.yaml", + f"{all_groups_mode}.yaml", ) ) as f: paced_feeder_align_secs = _check_clickhouse_paced_feeder( @@ -645,6 +666,7 @@ def main(cfg: DictConfig) -> None: experiment_modes, yaml.safe_load(f), minimum_experiment_running_time, + controller_prebuilt_output_dir, ) # On its own node, as in experiment_run_e2e.py, so the feeder neither # takes ClickHouse's cores nor shows up as its CPU. @@ -688,7 +710,31 @@ def main(cfg: DictConfig) -> None: clickhouse_http_port=clickhouse_http_port, ) remote_monitor_service = RemoteMonitorService(provider, node_offset=node_offset) - if paced_feeder_service is None: + if paced_feeder_service is not None: + start_feed = functools.partial( + _start_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, + experiment_root_output_dir=experiment_root_output_dir, + local_experiment_root_dir=local_experiment_root_dir, + ) + # The feeder runs on the next node, so it reaches both by node1's IP. + feed_target_ip = provider.get_node_ip(node_offset) + clickhouse_feed_url = ( + f"http://{feed_target_ip}:{clickhouse_http_port}/" + f"?database={CLICKHOUSE_DATABASE}" + ) + query_engine_url = ( + f"http://{feed_target_ip}:{int(cfg.streaming.remote_write.base_port)}" + f"{cfg.streaming.remote_write.path}" + ) + else: ingest_mode = _ingest_mode(experiment_modes) ingest_client_config = os.path.join( experiment_root_output_dir, @@ -742,7 +788,8 @@ def main(cfg: DictConfig) -> None: os.makedirs(local_experiment_dir, exist_ok=True) if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: - # --- sketchdb mode: precompute engine + JSON ingest + ClickHouse fallback --- + # --- sketchdb mode: precompute engine + ClickHouse fallback, ingesting + # the paced feed (remote write) or local_data_file (JSON) --- asap_http_port = urlparse(mode_server_urls[experiment_mode]).port # Kill any leftover query_engine_rust from a previous run (mirrors # query_engine_service.stop() at the top of the e2e mode loop). @@ -798,13 +845,27 @@ def generate_controller_output() -> None: generate_controller_output, ) - # Start query engine (precompute + JSON ingest + ClickHouse fallback) + # Start query engine (precompute + ClickHouse fallback), ingesting + # the paced feed over remote write or else local_data_file as JSON + remote_write_port = int(cfg.streaming.remote_write.base_port) + if paced_feeder_service is not None: + ingest_json_config = None + else: + dataset_precompute_cfg = dataset_cfg.precompute + ingest_json_config = { + "path": remote_data_file, + "metric_name": str(dataset_cfg.metric_name), + "value_col": str(dataset_precompute_cfg.value_col), + "label_cols": list(dataset_precompute_cfg.label_cols), + "timestamp_col": str(dataset_precompute_cfg.timestamp_col), + "timestamp_unit": "seconds", + "batch_size": 1000, + } query_engine_service = QueryEngineRustService( provider=provider, use_container=False, node_offset=node_offset, ) - dataset_precompute_cfg = dataset_cfg.precompute query_engine_service.start( experiment_output_dir=experiment_output_dir, local_experiment_dir=local_experiment_dir, @@ -819,26 +880,35 @@ def generate_controller_output() -> None: "type": "clickhouse", "url": clickhouse_url, "database": CLICKHOUSE_DATABASE, - "forward_unsupported_queries": True, + # A fed run feeds ClickHouse only in the baseline modes, + # so it would answer a forwarded query from no data. + "forward_unsupported_queries": paced_feeder_service is None, }, http_port=asap_http_port, spill=cfg.query_engine.spill, retention_slack_ms=config.get_retention_slack_ms( ep.query_groups, data_ingestion_interval_ms ), - ingest_json_config={ - "path": remote_data_file, - "metric_name": str(dataset_cfg.metric_name), - "value_col": str(dataset_precompute_cfg.value_col), - "label_cols": list(dataset_precompute_cfg.label_cols), - "timestamp_col": str(dataset_precompute_cfg.timestamp_col), - "timestamp_unit": "seconds", - "batch_size": 1000, - }, + remote_write_port=remote_write_port, + ingest_json_config=ingest_json_config, ) query_engine_service.wait_until_ready() + if paced_feeder_service is not None: + # The engine accepts writes now; the aligned query groups' + # starting delay, not steady_state_wait, gives data time to + # arrive. + start_feed( + experiment_mode=experiment_mode, + sinks=_feed_sinks( + experiment_mode, clickhouse_feed_url, query_engine_url + ), + ) + pre_query_wait_seconds = 0 + else: + pre_query_wait_seconds = int(cfg.flow.steady_state_wait) + controller_client_config = os.path.join( experiment_root_output_dir, "controller_client_configs", @@ -846,13 +916,12 @@ def generate_controller_output() -> None: ) # Single continuous monitor spanning ingest + query: the monitor - # starts now (engine is ingesting) and waits steady_state_wait + # starts now (engine is ingesting) and waits pre_query_wait_seconds # before the query client runs, so one monitor_output.json captures # precompute CPU (pc-worker threads) during ingest and query CPU # during the query phase, with per-thread attribution. - steady_state_wait = int(cfg.flow.steady_state_wait) print( - f"Monitoring precompute+query CPU; waiting {steady_state_wait}s " + f"Monitoring precompute+query CPU; waiting {pre_query_wait_seconds}s " "for ingest before queries..." ) _run_query_workload( @@ -868,9 +937,11 @@ def generate_controller_output() -> None: query_engine_service=query_engine_service, profile_query_engine=profile_query_engine, profile_prometheus_time=profile_prometheus_time, - pre_query_wait_seconds=steady_state_wait, + pre_query_wait_seconds=pre_query_wait_seconds, monitor_interval_seconds=monitor_interval_seconds, ) + if paced_feeder_service is not None: + _finish_clickhouse_feed(paced_feeder_service) spill_error = query_engine_service.record_spill() sync.rsync_experiment_data( @@ -890,19 +961,11 @@ def generate_controller_output() -> None: else: # --- baseline mode --- if paced_feeder_service is not None: - _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, + start_feed( experiment_mode=experiment_mode, - experiment_root_output_dir=experiment_root_output_dir, - local_experiment_root_dir=local_experiment_root_dir, + sinks=_feed_sinks( + experiment_mode, clickhouse_feed_url, query_engine_url + ), ) controller_client_config = os.path.join( experiment_root_output_dir, diff --git a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py index d0098802..5728165c 100644 --- a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py +++ b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py @@ -18,7 +18,8 @@ _prepare_clickhouse_controller_output, _start_clickhouse_controller, _finish_clickhouse_feed, - _start_clickhouse_feed, + _feed_sinks, + _start_feed, ) from experiment_utils.config import validate_experiment_config from experiment_utils.services.paced_feeder import PACED_FEEDER_STARTUP_CHECK_SECS @@ -123,12 +124,13 @@ def setUp(self): def tearDown(self): self.tmp.cleanup() - def check(self, modes, **overrides): + def check(self, modes, prebuilt_output_dir=None, **overrides): return _check_clickhouse_paced_feeder( feeder_config(self.table, **overrides), modes, aligned_client_config(10, 60), minimum_experiment_running_time=300, + controller_prebuilt_output_dir=prebuilt_output_dir, ) def test_aligns_to_the_query_intervals(self): @@ -138,11 +140,14 @@ 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"): + def test_sketchdb_without_a_prebuilt_plan_refused(self): + # The planner's input needs dataset.precompute, which a feed has none of. + with self.assertRaisesRegex(ValueError, "prebuilt_output_dir"): self.check(["baseline", "sketchdb"]) + def test_sketchdb_with_a_prebuilt_plan_is_fed(self): + self.check(["baseline", "sketchdb"], prebuilt_output_dir="/plans/saved") + def test_feed_ending_before_the_queries_refused(self): with self.assertRaisesRegex(ValueError, "ends before the queries"): self.check(["baseline"], duration_secs=300) @@ -159,8 +164,29 @@ def test_feed_must_outlast_the_query_client_startup(self): ) -class StartClickHouseFeedTest(unittest.TestCase): - def start(self, calls, exited=True): +CLICKHOUSE_FEED_URL = "http://10.10.1.1:8123/?database=default" +ENGINE_FEED_URL = "http://10.10.1.1:8080/api/v1/write" + + +class FeedSinksTest(unittest.TestCase): + def test_baseline_modes_feed_clickhouse(self): + for mode in ("baseline", "baseline_mv"): + self.assertEqual( + _feed_sinks(mode, CLICKHOUSE_FEED_URL, ENGINE_FEED_URL), + {"clickhouse": CLICKHOUSE_FEED_URL}, + ) + + def test_sketchdb_feeds_only_the_engine(self): + # It forwards no queries, and ClickHouse ingesting a copy would load + # the node its CPU and latency are measured on. + self.assertEqual( + _feed_sinks("sketchdb", CLICKHOUSE_FEED_URL, ENGINE_FEED_URL), + {"query_engine": ENGINE_FEED_URL}, + ) + + +class StartFeedTest(unittest.TestCase): + def start(self, calls, exited=True, mode="baseline"): calls.feeder.start.return_value = 1700000060000 calls.feeder.wait_for_exit.return_value = exited with tempfile.TemporaryDirectory() as root, patch( @@ -168,13 +194,13 @@ def start(self, calls, exited=True): ) 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") + config_path = os.path.join(config_dir, f"{mode}.yaml") with open(config_path, "w") as f: yaml.safe_dump(aligned_client_config(60), f) provider = Mock() provider.get_node_ip.return_value = "10.10.1.1" - start_ms = _start_clickhouse_feed( + start_ms = _start_feed( provider=provider, node_offset=0, data_loader=calls.loader, @@ -183,8 +209,8 @@ def start(self, calls, exited=True): feeder_align_secs=60, dataset_name="custom", init_sql_file="init.sql", - clickhouse_http_port=8123, - experiment_mode="baseline", + sinks={"clickhouse": CLICKHOUSE_FEED_URL}, + experiment_mode=mode, experiment_root_output_dir="/exp", local_experiment_root_dir=root, ) @@ -221,6 +247,16 @@ def test_tables_are_created_before_the_feed_and_queries_follow_it(self): ) self.rsync.assert_called_once() + def test_sketchdb_queries_follow_its_own_feed(self): + calls = Mock() + self.start(calls, mode="sketchdb") + start_kwargs = calls.feeder.start.call_args.kwargs + self.assertEqual(start_kwargs["experiment_output_dir"], "/exp/sketchdb") + self.assertEqual( + self.client_config["query_groups"][0]["client_options"]["data_start_time"], + 1700000060, + ) + 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.