diff --git a/asap-query-engine/tests/e2e_precompute_equivalence.rs b/asap-query-engine/tests/e2e_precompute_equivalence.rs index b0a1f3b0..1a595dc7 100644 --- a/asap-query-engine/tests/e2e_precompute_equivalence.rs +++ b/asap-query-engine/tests/e2e_precompute_equivalence.rs @@ -2070,7 +2070,7 @@ async fn e2e_avg_of_per_series_quantile_over_time() { }) .collect(); let (engine, query) = NativeDagScenario { - port: 19429, + port: 19437, metric, query, aggregation_configs: vec![config], diff --git a/asap-tools/data-sources/paced-feeder/src/main.rs b/asap-tools/data-sources/paced-feeder/src/main.rs index 1a106ba2..d60a41a6 100644 --- a/asap-tools/data-sources/paced-feeder/src/main.rs +++ b/asap-tools/data-sources/paced-feeder/src/main.rs @@ -139,7 +139,7 @@ struct RunStats { rows_sent: usize, bytes_sent: usize, /// Rows sent divided by wall-clock time from the first tick's schedule to - /// the later of the last tick's end and the run's nominal end. + /// the later of the last tick's end and the end of its second. achieved_rows_per_sec: f64, /// Ticks whose work took longer than one second. overrun_ticks: usize, @@ -455,6 +455,17 @@ async fn main() -> Result<()> { workload.rows_per_tick(), ); + // SIGTERM ends the feed at the next tick, still writing its stats: a run + // stops its feeders when its queries are done, before their duration. + let stop = Arc::new(tokio::sync::Notify::new()); + let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; + let on_sigterm = Arc::clone(&stop); + tokio::spawn(async move { + sigterm.recv().await; + // Stores a permit, so a tick already sending stops at the next one. + on_sigterm.notify_one(); + }); + let base = wait_for_wall_clock(start_ms).await; let mut ticks = Vec::with_capacity(duration_secs as usize); let mut errors: Vec = Vec::new(); @@ -462,7 +473,13 @@ async fn main() -> Result<()> { for tick in 0..duration_secs { let scheduled = base + Duration::from_secs(tick); - tokio::time::sleep_until(scheduled.into()).await; + tokio::select! { + _ = tokio::time::sleep_until(scheduled.into()) => {} + _ = stop.notified() => { + eprintln!("SIGTERM: stopping after {tick} ticks"); + break; + } + } let started = Instant::now(); let (mut rows, mut bytes) = (0, 0); @@ -532,9 +549,10 @@ async fn main() -> Result<()> { } } - // A run that keeps up still spans the full duration, even though the last - // tick's work ends before its second does. - let wall_secs = base.elapsed().as_secs_f64().max(duration_secs as f64); + // A run that keeps up still spans its ticks' seconds, even though the last + // tick's work ends before its second does. A SIGTERM'd run spans the + // ticks it sent, not its duration. + let wall_secs = base.elapsed().as_secs_f64().max(ticks.len() as f64); let rows_sent: usize = ticks.iter().map(|t| t.rows).sum(); let grouped = args.workload_table.is_none(); let run = RunStats { diff --git a/asap-tools/experiments/constants.py b/asap-tools/experiments/constants.py index 6c7ee31a..ded60da9 100644 --- a/asap-tools/experiments/constants.py +++ b/asap-tools/experiments/constants.py @@ -14,6 +14,10 @@ QUERY_ENGINE_RS_BINARY_NAME = "query_engine_rust" QUERY_ENGINE_RS_FP_BINARY_NAME = "query_engine_rust_fp" QUERY_ENGINE_RS_CONTAINER_NAME = "sketchdb-queryengine-rust" +# remote_monitor's grep -E pattern for the ClickHouse server: `clickhouse-server` +# in Docker, `clickhouse server` bare metal. Plain "clickhouse" also matches a +# paced feeder's `--sink clickhouse`, whose exit would end the run. +CLICKHOUSE_SERVER_PROCESS_KEYWORD = "clickhouse.server" PROMETHEUS_THROUGHPUT_POLLING_INTERVAL = 5 PROMETHEUS_HEALTH_POLLING_INTERVAL = 5 diff --git a/asap-tools/experiments/datasets/fake_metric/random_workload_artifacts.py b/asap-tools/experiments/datasets/fake_metric/random_workload_artifacts.py index 1ec9b934..75181522 100644 --- a/asap-tools/experiments/datasets/fake_metric/random_workload_artifacts.py +++ b/asap-tools/experiments/datasets/fake_metric/random_workload_artifacts.py @@ -27,6 +27,10 @@ ACCURACY_SLA = 0.95 PILOT_SUPPORT = "common-promql" PILOT_DURATION_MS = 10 * 60 * 1000 +# Runs stop their feeders once the queries are done, so a feed's duration +# only bounds a run that hangs: this many times the scheduled query phase, +# past the slowest pilot arm (raw ClickHouse, ~4x). +FEED_BOUND_QUERY_PHASES = 10 QUERY_FRESHNESS_DELAY_SECS = 10 # Relative to EXPERIMENTS_DIR, where the runners read configs and SQL. CONFIG_DIR = "config/experiment_type/random_workload" @@ -139,6 +143,7 @@ def pilot_configs(workload, pilot_dir): lookback_secs = max(q["lookback_ms"] for q in queries) // 1000 starting_delay = lookback_secs + QUERY_FRESHNESS_DELAY_SECS duration = PILOT_DURATION_MS // 1000 + feed_secs = starting_delay + FEED_BOUND_QUERY_PHASES * duration seed = workload["data_facts"]["feeder_seed"] table = pilot_dir / "eval_table.json" out_dir = Path(CONFIG_DIR) / workload["id"] @@ -149,6 +154,7 @@ def pilot_configs(workload, pilot_dir): starting_delay, duration, QUERY_FRESHNESS_DELAY_SECS, + feeder_duration_secs=feed_secs, prebuilt_output_dir=str(pilot_dir), ) # One monitoring tool per run, so VictoriaMetrics is its own baseline run. @@ -205,7 +211,7 @@ def pilot_configs(workload, pilot_dir): "workload_table": str(table.resolve()), "metrics": list(metrics), "seed": seed, - "duration_secs": starting_delay + duration + sbw.FEEDER_SLACK_SECS, + "duration_secs": feed_secs, } for mode, init, name in ( (sbw.BASELINE, "init.sql", "clickhouse.yaml"), diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 8934b04d..041e431d 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -380,8 +380,9 @@ def _start_clickhouse_feed( start_ms = feeder_service.start( config=feeder_config, sinks={ - SINK_CLICKHOUSE: f"http://localhost:{clickhouse_http_port}/" - f"?database={CLICKHOUSE_DATABASE}" + # 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}" }, experiment_output_dir=os.path.join(experiment_root_output_dir, experiment_mode), local_experiment_dir=os.path.join(local_experiment_root_dir, experiment_mode), @@ -406,16 +407,16 @@ def _start_clickhouse_feed( 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 _finish_clickhouse_feed(feeder_service: PacedFeederService) -> None: + """Stop the feeders once the queries are done: the run ends with its + queries, however slow, and its feed's duration is only an upper bound. + SIGTERM lets a feeder finish its tick and write its stats.""" + feeder_service.stop() + if not feeder_service.wait_for_exit(PACED_FEEDER_STOP_TIMEOUT_SECS): + print( + f"Warning: paced_feeder still running {PACED_FEEDER_STOP_TIMEOUT_SECS} s " + "after SIGTERM; the next mode's feed start refuses to run over it" + ) def _run_clickhouse_ingest_with_monitor( @@ -645,8 +646,14 @@ def main(cfg: DictConfig) -> None: yaml.safe_load(f), minimum_experiment_running_time, ) - # Next to ClickHouse, so the feed crosses no network. - paced_feeder_service = PacedFeederService(provider, [node_offset]) + # On its own node, as in experiment_run_e2e.py, so the feeder neither + # takes ClickHouse's cores nor shows up as its CPU. + if num_nodes < 1: + raise ValueError( + "dataset.paced_feeder runs the feeder on its own node: " + "providers.*.num_nodes must be at least 1" + ) + paced_feeder_service = PacedFeederService(provider, [node_offset + 1]) if not no_teardown: # Otherwise a failed run leaves feeders inserting into the next # run's tables. @@ -883,7 +890,7 @@ def generate_controller_output() -> None: else: # --- baseline mode --- if paced_feeder_service is not None: - feed_start_ms = _start_clickhouse_feed( + _start_clickhouse_feed( provider=provider, node_offset=node_offset, data_loader=data_loader, @@ -918,11 +925,7 @@ def generate_controller_output() -> None: 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"], - ) + _finish_clickhouse_feed(paced_feeder_service) sync.rsync_experiment_data( provider, diff --git a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py index d054d896..219acd70 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -90,7 +90,7 @@ def start( else: keywords = [constants.QUERY_ENGINE_RS_PROCESS_KEYWORD] else: - keywords = ["clickhouse"] + keywords = [constants.CLICKHOUSE_SERVER_PROCESS_KEYWORD] else: keywords = list(config_keywords) @@ -258,7 +258,7 @@ def start_clickhouse_ingest_monitor( (``constants.INGEST_MONITOR_STOP_FILE``) is created, or SIGTERM/SIGINT, then writes ``monitor_output_file`` under ``remote_monitor_output/``. """ - keywords = ["clickhouse"] + keywords = [constants.CLICKHOUSE_SERVER_PROCESS_KEYWORD] cmd = ( "python3 -u remote_monitor.py " "--execution_mode ingest " diff --git a/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py b/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py index dc7bf986..7578016f 100644 --- a/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py +++ b/asap-tools/experiments/post_experiment/multi_experiment/compare_arms.py @@ -26,8 +26,17 @@ def read_jsonl(path): - with gzip.open(path, "rt") as f: - return [json.loads(line) for line in f if line.strip()] + """The complete lines of a gzipped JSONL file. A client stopped mid-run + leaves the gzip stream unterminated: keep what precedes the cut.""" + rows = [] + try: + with gzip.open(path, "rt") as f: + for line in f: + if line.endswith("\n"): + rows.append(json.loads(line)) + except EOFError: + pass + return rows def sql_labels(sql): @@ -76,13 +85,17 @@ def __init__(self, mode_dir): client = mode_dir.parent / "controller_client_configs" / f"{mode_dir.name}.yaml" groups = yaml.safe_load(client.read_text())["query_groups"] self.interval_secs = [g["repetition_delay_ms"] / 1000 for g in groups] + # Queries start after this; until then the systems only ingest. + self.warmup_secs = min(g["client_options"]["starting_delay"] for g in groups) self.monitor = mode_dir / "remote_monitor_output" / "monitor_output.json" hydra = yaml.safe_load((mode_dir.parent / "hydra_config.yaml").read_text()) self.monitor_interval_secs = float(hydra["flow"]["monitor_interval_seconds"]) def resources(self): - """(total CPU-seconds, peak memory bytes), summed over monitored - processes at each sample.""" + """CPU-seconds over the whole monitored run, its warm-up and the rest + (queries and any lag), and peak memory bytes; each sample sums the + monitored processes. The monitor starts with the query client, so + the warm-up is its first `warmup_secs`.""" processes = json.loads(self.monitor.read_text()).values() n = max(len(p["cpu_percent"]) for p in processes) cpu, mem = [0.0] * n, [0.0] * n @@ -91,7 +104,15 @@ def resources(self): cpu[t] += c for t, m in enumerate(p["memory_info"]): mem[t] += m - return sum(cpu) / 100 * self.monitor_interval_secs, max(mem) + secs = self.monitor_interval_secs + warm = int(self.warmup_secs / secs) + return { + "monitored_secs": n * secs, + "cpu_secs": sum(cpu) / 100 * secs, + "warmup_cpu_secs": sum(cpu[:warm]) / 100 * secs, + "query_cpu_secs": sum(cpu[warm:]) / 100 * secs, + "peak_memory_bytes": max(mem), + } def percentile(xs, p): @@ -155,11 +176,14 @@ def report(arms, reference): f"missing series {missing}, empty reference windows {empty}" ) - print("\n== Resources (whole monitored run)") + print("\n== Resources: CPU-seconds (whole run = warm-up + queries), peak memory") for arm_name, arm in arms.items(): - cpu_secs, peak = arm.resources() + r = arm.resources() print( - f" {arm_name:16s} CPU {cpu_secs:9.0f} s peak memory {peak / 2**30:5.2f} GiB" + f" {arm_name:16s} {r['monitored_secs']:6.0f} s monitored CPU " + f"{r['cpu_secs']:7.0f} = {r['warmup_cpu_secs']:6.0f} + " + f"{r['query_cpu_secs']:6.0f} peak memory " + f"{r['peak_memory_bytes'] / 2**30:5.2f} GiB" ) diff --git a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py index 743bdb01..65ab509f 100644 --- a/asap-tools/experiments/tests/test_experiment_run_clickhouse.py +++ b/asap-tools/experiments/tests/test_experiment_run_clickhouse.py @@ -172,8 +172,10 @@ def start(self, calls, exited=True): 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( - provider=Mock(), + provider=provider, node_offset=0, data_loader=calls.loader, feeder_service=calls.feeder, @@ -209,7 +211,8 @@ def test_tables_are_created_before_the_feed_and_queries_follow_it(self): self.assertEqual(start_kwargs["align_secs"], 60) self.assertEqual( start_kwargs["sinks"], - {"clickhouse": "http://localhost:8123/?database=default"}, + # From the feeder's own node to ClickHouse's. + {"clickhouse": "http://10.10.1.1:8123/?database=default"}, ) self.assertEqual(start_kwargs["experiment_output_dir"], "/exp/baseline") self.assertEqual( @@ -229,23 +232,14 @@ def test_feeders_that_outlive_pkill_stop_the_run(self): 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() + def test_stops_the_feed_once_the_queries_are_done(self): + # Waiting out the feed's duration let it, not the queries, end the run; + # SIGTERM still lets feeders write their stats. + calls = Mock() + calls.wait_for_exit.return_value = True + _finish_clickhouse_feed(calls) + self.assertEqual([c[0] for c in calls.mock_calls], ["stop", "wait_for_exit"]) + calls.wait_for_exit.assert_called_once_with(PACED_FEEDER_STOP_TIMEOUT_SECS) class FedClickHouseConfigValidationTest(unittest.TestCase): diff --git a/asap-tools/experiments/tests/test_random_workload_artifacts.py b/asap-tools/experiments/tests/test_random_workload_artifacts.py index 35ef2be6..9f192828 100644 --- a/asap-tools/experiments/tests/test_random_workload_artifacts.py +++ b/asap-tools/experiments/tests/test_random_workload_artifacts.py @@ -157,6 +157,9 @@ def test_every_arm_feeds_the_same_data_and_schedule(self): (feeder["workload_table"], feeder["seed"], feeder["metrics"]), (str((PILOT_DIR / "eval_table.json").resolve()), 862, ["data"]), ) + # Runs stop feeders when their queries are done; the duration + # only bounds a hung run, past a slow arm's lagging queries. + self.assertEqual(feeder["duration_secs"], 1210 + 10 * 600) # The longest lookback (20 min) plus the freshness delay, then 10 min. schedules = [ (g["repetition_delay_ms"], g["client_options"]) diff --git a/asap-tools/experiments/tests/test_remote_monitor_keywords.py b/asap-tools/experiments/tests/test_remote_monitor_keywords.py index 0cd080cd..71d070b3 100644 --- a/asap-tools/experiments/tests/test_remote_monitor_keywords.py +++ b/asap-tools/experiments/tests/test_remote_monitor_keywords.py @@ -1,8 +1,11 @@ """Tests that remote_monitor.py receives its keyword list as one clean argument.""" +import re import shlex import unittest +import constants + from experiment_utils.services.remote_monitor_service import RemoteMonitorService @@ -71,13 +74,33 @@ def test_remote_provider_keeps_escaped_quotes_for_ssh(self): def test_clickhouse_ingest_local_provider_gets_unescaped_keywords(self): self.assertEqual( - clickhouse_ingest_keywords_seen_by_monitor(remote=False), "clickhouse" + clickhouse_ingest_keywords_seen_by_monitor(remote=False), + "clickhouse.server", ) def test_clickhouse_ingest_remote_provider_keeps_escaped_quotes_for_ssh(self): self.assertEqual( - clickhouse_ingest_keywords_seen_by_monitor(remote=True), "clickhouse" + clickhouse_ingest_keywords_seen_by_monitor(remote=True), + "clickhouse.server", + ) + + +class ClickHouseKeywordTest(unittest.TestCase): + def test_matches_the_server_but_not_a_feeder_into_it(self): + # "clickhouse" also matched the paced feeder's `--sink clickhouse`; + # when the feed ended, the monitor took it for ClickHouse exiting and + # stopped the query client mid-run. + pattern = re.compile(constants.CLICKHOUSE_SERVER_PROCESS_KEYWORD) + for server in ( + "/usr/bin/clickhouse-server --config-file=/etc/clickhouse-server/config.xml", + "/usr/bin/clickhouse server --config-file=/etc/clickhouse-server/config.xml", + ): + self.assertTrue(pattern.search(server), server) + feeder = ( + "./target/release/paced_feeder --sink clickhouse " + "--url http://localhost:8123/?database=default --metric data" ) + self.assertIsNone(pattern.search(feeder)) if __name__ == "__main__":