fix(asap-tools): complete ClickHouse runs and clean CPU for random E2E pilots

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],

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,14 +455,31 @@ 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<String> = Vec::new();
     let mut failed_requests = 0;
 
     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 {

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

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"),

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,

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 "

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"
         )
 
 

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):

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"])

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__":