Skip to content

Commit eee2cec

Browse files
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 <noreply@anthropic.com>
1 parent a33ebe5 commit eee2cec

3 files changed

Lines changed: 72 additions & 51 deletions

File tree

‎asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -500,7 +500,7 @@ table with `sketch_bench_e2e/sketch_bench_e2e_configs.py`.
500500
- **Example**: `"--concurrency 32"`
501501

502502
#### `experiment_params.dataset.paced_feeder` (dict, optional; `experiment_run_clickhouse.py`)
503-
- **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`
503+
- **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`
504504
- **Example**: `{workload_table: /path/to/table.json, metrics: [data, data_total], seed: 42, duration_secs: 4000}`
505505

506506
### Query Group Configuration

‎asap-tools/experiments/experiment_run_clickhouse.py‎

Lines changed: 42 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -92,17 +92,17 @@
9292
A sketch-bench workload table (sketch_bench_workloads.py configs) is fed in
9393
real time instead, one paced feeder per metric table, when
9494
experiment_params.dataset.paced_feeder is set. Each mode then recreates the
95-
tables and starts its own feed; sketchdb's feed goes to the engine's
96-
remote-write ingest as well as ClickHouse (which answers forwarded queries),
97-
from one start, and needs controller.prebuilt_output_dir.
95+
tables and starts its own feed. sketchdb's feed goes only to the engine's
96+
remote-write ingest, so it forwards no queries to ClickHouse, and it needs
97+
controller.prebuilt_output_dir.
9898
"""
9999

100100
import atexit
101101
import functools
102102
import json
103103
import os
104104
import time
105-
from typing import Callable, List, Optional
105+
from typing import Callable, Dict, List, Optional
106106
from urllib.parse import urlparse
107107

108108
import hydra
@@ -336,7 +336,7 @@ def _check_clickhouse_paced_feeder(
336336
experiment_modes: List[str],
337337
client_config: dict,
338338
minimum_experiment_running_time: int,
339-
controller_prebuilt_output_dir: Optional[str],
339+
controller_prebuilt_output_dir: Optional[str] = None,
340340
) -> int:
341341
"""Fail before ClickHouse starts on a feeder config the run can't serve;
342342
returns the seconds the feeders' start must align to."""
@@ -358,7 +358,18 @@ def _check_clickhouse_paced_feeder(
358358
return aligned_interval_lcm(client_config)
359359

360360

361-
def _start_clickhouse_feed(
361+
def _feed_sinks(
362+
experiment_mode: str, clickhouse_url: str, query_engine_url: str
363+
) -> Dict[str, str]:
364+
"""The sinks a mode's feed writes to. sketchdb feeds only ASAPQuery: it
365+
forwards no queries, and ClickHouse ingesting a copy would load the node
366+
its CPU and latency are measured on. Every other mode reads ClickHouse."""
367+
if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME:
368+
return {SINK_QUERY_ENGINE: query_engine_url}
369+
return {SINK_CLICKHOUSE: clickhouse_url}
370+
371+
372+
def _start_feed(
362373
*,
363374
provider,
364375
node_offset: int,
@@ -368,20 +379,14 @@ def _start_clickhouse_feed(
368379
feeder_align_secs: int,
369380
dataset_name: str,
370381
init_sql_file: str,
371-
clickhouse_http_port: int,
382+
sinks: Dict[str, str],
372383
experiment_mode: str,
373384
experiment_root_output_dir: str,
374385
local_experiment_root_dir: str,
375-
query_engine_url: Optional[str] = None,
376386
) -> int:
377387
"""Recreate the metric tables, start one feeder per (sink, metric), and
378388
point the mode's aligned query groups at the feed's start; returns that
379-
start in epoch ms.
380-
381-
ClickHouse is always fed; with `query_engine_url` (sketchdb) so is
382-
ASAPQuery's remote-write ingest, from the same start and seed, so the
383-
queries it forwards to ClickHouse read the same data.
384-
"""
389+
start in epoch ms; `sinks` from _feed_sinks."""
385390
# An earlier mode's feeders (left running by no_teardown, or with an
386391
# INSERT in flight) would write into the recreated tables.
387392
feeder_service.stop()
@@ -392,13 +397,6 @@ def _start_clickhouse_feed(
392397
)
393398
# Before any data arrives: views only see rows inserted after they exist.
394399
data_loader.run_init_sql(dataset_name, init_sql_file)
395-
sinks = {
396-
# The feeder runs on the next node (see main()).
397-
SINK_CLICKHOUSE: f"http://{provider.get_node_ip(node_offset)}:"
398-
f"{clickhouse_http_port}/?database={CLICKHOUSE_DATABASE}"
399-
}
400-
if query_engine_url is not None:
401-
sinks[SINK_QUERY_ENGINE] = query_engine_url
402400
start_ms = feeder_service.start(
403401
config=feeder_config,
404402
sinks=sinks,
@@ -714,7 +712,7 @@ def main(cfg: DictConfig) -> None:
714712
remote_monitor_service = RemoteMonitorService(provider, node_offset=node_offset)
715713
if paced_feeder_service is not None:
716714
start_feed = functools.partial(
717-
_start_clickhouse_feed,
715+
_start_feed,
718716
provider=provider,
719717
node_offset=node_offset,
720718
data_loader=data_loader,
@@ -723,10 +721,19 @@ def main(cfg: DictConfig) -> None:
723721
feeder_align_secs=paced_feeder_align_secs,
724722
dataset_name=dataset_name,
725723
init_sql_file=init_sql_file,
726-
clickhouse_http_port=clickhouse_http_port,
727724
experiment_root_output_dir=experiment_root_output_dir,
728725
local_experiment_root_dir=local_experiment_root_dir,
729726
)
727+
# The feeder runs on the next node, so it reaches both by node1's IP.
728+
feed_target_ip = provider.get_node_ip(node_offset)
729+
clickhouse_feed_url = (
730+
f"http://{feed_target_ip}:{clickhouse_http_port}/"
731+
f"?database={CLICKHOUSE_DATABASE}"
732+
)
733+
query_engine_url = (
734+
f"http://{feed_target_ip}:{int(cfg.streaming.remote_write.base_port)}"
735+
f"{cfg.streaming.remote_write.path}"
736+
)
730737
else:
731738
ingest_mode = _ingest_mode(experiment_modes)
732739
ingest_client_config = os.path.join(
@@ -873,7 +880,9 @@ def generate_controller_output() -> None:
873880
"type": "clickhouse",
874881
"url": clickhouse_url,
875882
"database": CLICKHOUSE_DATABASE,
876-
"forward_unsupported_queries": True,
883+
# A fed run feeds ClickHouse only in the baseline modes,
884+
# so it would answer a forwarded query from no data.
885+
"forward_unsupported_queries": paced_feeder_service is None,
877886
},
878887
http_port=asap_http_port,
879888
spill=cfg.query_engine.spill,
@@ -892,9 +901,9 @@ def generate_controller_output() -> None:
892901
# arrive.
893902
start_feed(
894903
experiment_mode=experiment_mode,
895-
# The feeder runs on the next node (see main()).
896-
query_engine_url=f"http://{provider.get_node_ip(node_offset)}:"
897-
f"{remote_write_port}{cfg.streaming.remote_write.path}",
904+
sinks=_feed_sinks(
905+
experiment_mode, clickhouse_feed_url, query_engine_url
906+
),
898907
)
899908
pre_query_wait_seconds = 0
900909
else:
@@ -952,7 +961,12 @@ def generate_controller_output() -> None:
952961
else:
953962
# --- baseline mode ---
954963
if paced_feeder_service is not None:
955-
start_feed(experiment_mode=experiment_mode)
964+
start_feed(
965+
experiment_mode=experiment_mode,
966+
sinks=_feed_sinks(
967+
experiment_mode, clickhouse_feed_url, query_engine_url
968+
),
969+
)
956970
controller_client_config = os.path.join(
957971
experiment_root_output_dir,
958972
"controller_client_configs",

‎asap-tools/experiments/tests/test_experiment_run_clickhouse.py‎

Lines changed: 29 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,8 @@
1818
_prepare_clickhouse_controller_output,
1919
_start_clickhouse_controller,
2020
_finish_clickhouse_feed,
21-
_start_clickhouse_feed,
21+
_feed_sinks,
22+
_start_feed,
2223
)
2324
from experiment_utils.config import validate_experiment_config
2425
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):
163164
)
164165

165166

166-
class StartClickHouseFeedTest(unittest.TestCase):
167-
def start(self, calls, exited=True, mode="baseline", query_engine_url=None):
167+
CLICKHOUSE_FEED_URL = "http://10.10.1.1:8123/?database=default"
168+
ENGINE_FEED_URL = "http://10.10.1.1:8080/api/v1/write"
169+
170+
171+
class FeedSinksTest(unittest.TestCase):
172+
def test_baseline_modes_feed_clickhouse(self):
173+
for mode in ("baseline", "baseline_mv"):
174+
self.assertEqual(
175+
_feed_sinks(mode, CLICKHOUSE_FEED_URL, ENGINE_FEED_URL),
176+
{"clickhouse": CLICKHOUSE_FEED_URL},
177+
)
178+
179+
def test_sketchdb_feeds_only_the_engine(self):
180+
# It forwards no queries, and ClickHouse ingesting a copy would load
181+
# the node its CPU and latency are measured on.
182+
self.assertEqual(
183+
_feed_sinks("sketchdb", CLICKHOUSE_FEED_URL, ENGINE_FEED_URL),
184+
{"query_engine": ENGINE_FEED_URL},
185+
)
186+
187+
188+
class StartFeedTest(unittest.TestCase):
189+
def start(self, calls, exited=True, mode="baseline"):
168190
calls.feeder.start.return_value = 1700000060000
169191
calls.feeder.wait_for_exit.return_value = exited
170192
with tempfile.TemporaryDirectory() as root, patch(
@@ -178,7 +200,7 @@ def start(self, calls, exited=True, mode="baseline", query_engine_url=None):
178200

179201
provider = Mock()
180202
provider.get_node_ip.return_value = "10.10.1.1"
181-
start_ms = _start_clickhouse_feed(
203+
start_ms = _start_feed(
182204
provider=provider,
183205
node_offset=0,
184206
data_loader=calls.loader,
@@ -187,11 +209,10 @@ def start(self, calls, exited=True, mode="baseline", query_engine_url=None):
187209
feeder_align_secs=60,
188210
dataset_name="custom",
189211
init_sql_file="init.sql",
190-
clickhouse_http_port=8123,
212+
sinks={"clickhouse": CLICKHOUSE_FEED_URL},
191213
experiment_mode=mode,
192214
experiment_root_output_dir="/exp",
193215
local_experiment_root_dir=root,
194-
query_engine_url=query_engine_url,
195216
)
196217
with open(config_path) as f:
197218
self.client_config = yaml.safe_load(f)
@@ -226,24 +247,10 @@ def test_tables_are_created_before_the_feed_and_queries_follow_it(self):
226247
)
227248
self.rsync.assert_called_once()
228249

229-
def test_sketchdb_feeds_the_engine_and_clickhouse_from_one_start(self):
230-
# ClickHouse answers the queries ASAPQuery forwards, so it gets the
231-
# same data from the same start.
250+
def test_sketchdb_queries_follow_its_own_feed(self):
232251
calls = Mock()
233-
self.start(
234-
calls,
235-
mode="sketchdb",
236-
query_engine_url="http://10.10.1.1:8080/api/v1/write",
237-
)
238-
calls.feeder.start.assert_called_once()
252+
self.start(calls, mode="sketchdb")
239253
start_kwargs = calls.feeder.start.call_args.kwargs
240-
self.assertEqual(
241-
start_kwargs["sinks"],
242-
{
243-
"clickhouse": "http://10.10.1.1:8123/?database=default",
244-
"query_engine": "http://10.10.1.1:8080/api/v1/write",
245-
},
246-
)
247254
self.assertEqual(start_kwargs["experiment_output_dir"], "/exp/sketchdb")
248255
self.assertEqual(
249256
self.client_config["query_groups"][0]["client_options"]["data_start_time"],

0 commit comments

Comments
 (0)