From 750aaeb79df6890467e5a320bbf1959c98f1b96b Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 22:55:19 -0400 Subject: [PATCH 1/3] feat(asap-tools): feed the sketchdb arm in the ClickHouse runner With dataset.paced_feeder, sketchdb mode now starts the engine on remote-write ingest and feeds it and ClickHouse (which answers forwarded queries) from one start and seed, then points the aligned query groups at that start, as baseline does. A fed dataset has no dataset.precompute for the planner's input, so fed sketchdb needs controller.prebuilt_output_dir. Co-Authored-By: Claude Opus 5.5 --- .../CONFIG_PARAMETERS_REFERENCE.md | 2 +- .../experiments/experiment_run_clickhouse.py | 129 ++++++++++++------ .../tests/test_experiment_run_clickhouse.py | 43 +++++- 3 files changed, 123 insertions(+), 51 deletions(-) diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index 520863ef..d19b77f5 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 same feed (one start, one seed) also goes to ClickHouse, which answers the queries ASAPQuery forwards; 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..fa8a96b0 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -91,10 +91,14 @@ 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 to the engine's +remote-write ingest as well as ClickHouse (which answers forwarded queries), +from one start, and needs controller.prebuilt_output_dir. """ import atexit +import functools import json import os import time @@ -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], ) -> 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, @@ -363,10 +372,16 @@ def _start_clickhouse_feed( experiment_mode: str, experiment_root_output_dir: str, local_experiment_root_dir: str, + query_engine_url: Optional[str] = None, ) -> 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. + + ClickHouse is always fed; with `query_engine_url` (sketchdb) so is + ASAPQuery's remote-write ingest, from the same start and seed, so the + queries it forwards to ClickHouse read the same data. + """ # 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() @@ -377,13 +392,16 @@ def _start_clickhouse_feed( ) # Before any data arrives: views only see rows inserted after they exist. data_loader.run_init_sql(dataset_name, init_sql_file) + 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}" + } + if query_engine_url is not None: + sinks[SINK_QUERY_ENGINE] = query_engine_url 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, @@ -645,6 +663,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 +707,22 @@ 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_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_root_output_dir=experiment_root_output_dir, + local_experiment_root_dir=local_experiment_root_dir, + ) + else: ingest_mode = _ingest_mode(experiment_modes) ingest_client_config = os.path.join( experiment_root_output_dir, @@ -798,13 +832,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, @@ -826,19 +874,26 @@ def generate_controller_output() -> None: 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, + # The feeder runs on the next node (see main()). + query_engine_url=f"http://{provider.get_node_ip(node_offset)}:" + f"{remote_write_port}{cfg.streaming.remote_write.path}", + ) + 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 +901,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 +922,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,20 +946,7 @@ 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, - experiment_mode=experiment_mode, - experiment_root_output_dir=experiment_root_output_dir, - local_experiment_root_dir=local_experiment_root_dir, - ) + start_feed(experiment_mode=experiment_mode) controller_client_config = os.path.join( experiment_root_output_dir, "controller_client_configs", diff --git a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py index d0098802..3e737283 100644 --- a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py +++ b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py @@ -123,12 +123,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 +139,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) @@ -160,7 +164,7 @@ def test_feed_must_outlast_the_query_client_startup(self): class StartClickHouseFeedTest(unittest.TestCase): - def start(self, calls, exited=True): + def start(self, calls, exited=True, mode="baseline", query_engine_url=None): calls.feeder.start.return_value = 1700000060000 calls.feeder.wait_for_exit.return_value = exited with tempfile.TemporaryDirectory() as root, patch( @@ -168,7 +172,7 @@ 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) @@ -184,9 +188,10 @@ def start(self, calls, exited=True): dataset_name="custom", init_sql_file="init.sql", clickhouse_http_port=8123, - experiment_mode="baseline", + experiment_mode=mode, experiment_root_output_dir="/exp", local_experiment_root_dir=root, + query_engine_url=query_engine_url, ) with open(config_path) as f: self.client_config = yaml.safe_load(f) @@ -221,6 +226,30 @@ def test_tables_are_created_before_the_feed_and_queries_follow_it(self): ) self.rsync.assert_called_once() + def test_sketchdb_feeds_the_engine_and_clickhouse_from_one_start(self): + # ClickHouse answers the queries ASAPQuery forwards, so it gets the + # same data from the same start. + calls = Mock() + self.start( + calls, + mode="sketchdb", + query_engine_url="http://10.10.1.1:8080/api/v1/write", + ) + calls.feeder.start.assert_called_once() + start_kwargs = calls.feeder.start.call_args.kwargs + self.assertEqual( + start_kwargs["sinks"], + { + "clickhouse": "http://10.10.1.1:8123/?database=default", + "query_engine": "http://10.10.1.1:8080/api/v1/write", + }, + ) + 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. From 44dc19e37214592540552eccf9a8494814581174 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 15 Oct 2026 00:04:28 -0400 Subject: [PATCH 2/3] fix(asap-tools): align the feed to every query group's interval The feed's alignment and the aligned groups' checks read the first mode's client config, which for sketchdb may lack trailing groups; they now read a mode that runs every group. Co-Authored-By: Claude Opus 5.5 --- asap-tools/experiments/experiment_run_clickhouse.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index fa8a96b0..a0851792 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -650,12 +650,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( @@ -776,7 +781,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). From 7202240d91f693b7de211b072854810f61d25b2a Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 15 Oct 2026 00:09:47 -0400 Subject: [PATCH 3/3] feat(asap-tools): feed only ASAPQuery in fed sketchdb runs A ClickHouse copy of the feed during sketchdb only served queries ASAPQuery forwards, and loaded the node sketchdb's CPU and latency are measured on. Fed sketchdb now feeds the engine alone and forwards no unsupported queries, so an unplanned query fails instead of being answered by ClickHouse. Co-Authored-By: Claude Opus 5.5 --- .../CONFIG_PARAMETERS_REFERENCE.md | 2 +- .../experiments/experiment_run_clickhouse.py | 70 +++++++++++-------- .../tests/test_experiment_run_clickhouse.py | 51 ++++++++------ 3 files changed, 72 insertions(+), 51 deletions(-) diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index d19b77f5..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. In `sketchdb` mode the engine ingests over remote write (`streaming.remote_write.base_port`/`path`) instead of `dataset.local_data_file`, and the same feed (one start, one seed) also goes to ClickHouse, which answers the queries ASAPQuery forwards; 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` +- **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 a0851792..e9d7bc35 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -92,9 +92,9 @@ 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. Each mode then recreates the -tables and starts its own feed; sketchdb's feed goes to the engine's -remote-write ingest as well as ClickHouse (which answers forwarded queries), -from one start, and needs controller.prebuilt_output_dir. +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 @@ -102,7 +102,7 @@ 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 @@ -336,7 +336,7 @@ def _check_clickhouse_paced_feeder( experiment_modes: List[str], client_config: dict, minimum_experiment_running_time: int, - controller_prebuilt_output_dir: Optional[str], + 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.""" @@ -358,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, @@ -368,20 +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, - query_engine_url: Optional[str] = None, ) -> int: """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. - - ClickHouse is always fed; with `query_engine_url` (sketchdb) so is - ASAPQuery's remote-write ingest, from the same start and seed, so the - queries it forwards to ClickHouse read the same data. - """ + 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() @@ -392,13 +397,6 @@ def _start_clickhouse_feed( ) # Before any data arrives: views only see rows inserted after they exist. data_loader.run_init_sql(dataset_name, init_sql_file) - 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}" - } - if query_engine_url is not None: - sinks[SINK_QUERY_ENGINE] = query_engine_url start_ms = feeder_service.start( config=feeder_config, sinks=sinks, @@ -714,7 +712,7 @@ def main(cfg: DictConfig) -> None: remote_monitor_service = RemoteMonitorService(provider, node_offset=node_offset) if paced_feeder_service is not None: start_feed = functools.partial( - _start_clickhouse_feed, + _start_feed, provider=provider, node_offset=node_offset, data_loader=data_loader, @@ -723,10 +721,19 @@ def main(cfg: DictConfig) -> None: feeder_align_secs=paced_feeder_align_secs, dataset_name=dataset_name, init_sql_file=init_sql_file, - clickhouse_http_port=clickhouse_http_port, 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( @@ -873,7 +880,9 @@ 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, @@ -892,9 +901,9 @@ def generate_controller_output() -> None: # arrive. start_feed( experiment_mode=experiment_mode, - # The feeder runs on the next node (see main()). - query_engine_url=f"http://{provider.get_node_ip(node_offset)}:" - f"{remote_write_port}{cfg.streaming.remote_write.path}", + sinks=_feed_sinks( + experiment_mode, clickhouse_feed_url, query_engine_url + ), ) pre_query_wait_seconds = 0 else: @@ -952,7 +961,12 @@ def generate_controller_output() -> None: else: # --- baseline mode --- if paced_feeder_service is not None: - start_feed(experiment_mode=experiment_mode) + start_feed( + experiment_mode=experiment_mode, + sinks=_feed_sinks( + experiment_mode, clickhouse_feed_url, query_engine_url + ), + ) controller_client_config = os.path.join( experiment_root_output_dir, "controller_client_configs", diff --git a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py index 3e737283..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 @@ -163,8 +164,29 @@ def test_feed_must_outlast_the_query_client_startup(self): ) -class StartClickHouseFeedTest(unittest.TestCase): - def start(self, calls, exited=True, mode="baseline", query_engine_url=None): +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( @@ -178,7 +200,7 @@ def start(self, calls, exited=True, mode="baseline", query_engine_url=None): 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, @@ -187,11 +209,10 @@ def start(self, calls, exited=True, mode="baseline", query_engine_url=None): feeder_align_secs=60, dataset_name="custom", init_sql_file="init.sql", - clickhouse_http_port=8123, + sinks={"clickhouse": CLICKHOUSE_FEED_URL}, experiment_mode=mode, experiment_root_output_dir="/exp", local_experiment_root_dir=root, - query_engine_url=query_engine_url, ) with open(config_path) as f: self.client_config = yaml.safe_load(f) @@ -226,24 +247,10 @@ def test_tables_are_created_before_the_feed_and_queries_follow_it(self): ) self.rsync.assert_called_once() - def test_sketchdb_feeds_the_engine_and_clickhouse_from_one_start(self): - # ClickHouse answers the queries ASAPQuery forwards, so it gets the - # same data from the same start. + def test_sketchdb_queries_follow_its_own_feed(self): calls = Mock() - self.start( - calls, - mode="sketchdb", - query_engine_url="http://10.10.1.1:8080/api/v1/write", - ) - calls.feeder.start.assert_called_once() + self.start(calls, mode="sketchdb") start_kwargs = calls.feeder.start.call_args.kwargs - self.assertEqual( - start_kwargs["sinks"], - { - "clickhouse": "http://10.10.1.1:8123/?database=default", - "query_engine": "http://10.10.1.1:8080/api/v1/write", - }, - ) self.assertEqual(start_kwargs["experiment_output_dir"], "/exp/sketchdb") self.assertEqual( self.client_config["query_groups"][0]["client_options"]["data_start_time"],