feat(asap-tools): separate clickhouse ingest and query cpu/memory monitoring in experiment run clickhouse

asap-tools/experiments/classes/process_monitor.py

@@ -189,13 +189,20 @@ def run(self):
 
         try:
             while True:
+                if self.pipe.poll(0):
+                    break
+
                 if self.thread_attribution_keyword is not None:
                     now = time.monotonic()
                     elapsed = now - self._prev_poll_monotonic
                     self._prev_poll_monotonic = now
 
                 iteration_info = []
+                stop_requested = False
                 for pid, p in self.psutil_handles.items():
+                    if self.pipe.poll(0):
+                        stop_requested = True
+                        break
                     iteration_info += self.update_pid_monitor_map(p)
                     if (
                         self.thread_attribution_keyword is not None
@@ -205,6 +212,9 @@ def run(self):
                         self._compute_thread_group_cpu(pid, elapsed)
                     if self.include_children:
                         for child in p.children(recursive=True):
+                            if self.pipe.poll(0):
+                                stop_requested = True
+                                break
                             if child.pid not in self.pid_monitor_map:
                                 self.add_child_pid_to_map(pid, child.pid)
                                 self.child_handles[child.pid] = child
@@ -216,6 +226,11 @@ def run(self):
                                 == self.thread_attribution_keyword
                             ):
                                 self._compute_thread_group_cpu(handle.pid, elapsed)
+                        if stop_requested:
+                            break
+
+                if stop_requested:
+                    break
 
                 self.update_hooks(iteration_info)
                 stop = self.pipe.poll(self.interval)
@@ -257,14 +272,19 @@ def start_monitor(
     return monitor, control_pipe, monitor_pipe
 
 
-def stop_monitor(monitor, control_pipe, monitor_pipe):
+def stop_monitor(monitor, control_pipe, monitor_pipe, timeout=30):
     control_pipe.send("stop")
-    can_read = control_pipe.poll(10)
+    can_read = control_pipe.poll(timeout)
     if can_read:
         monitor_info = control_pipe.recv()
-        monitor.join()
+        monitor.join(timeout=10)
     else:
         monitor_info = None
         monitor.terminate()
-        monitor.join()
+        monitor.join(timeout=10)
+    # terminate/join can still leave a stuck sampler; escalate so callers
+    # do not leak orphaned monitor processes across experiment phases.
+    if monitor.is_alive():
+        monitor.kill()
+        monitor.join(timeout=5)
     return monitor_info

asap-tools/experiments/config/config.yaml

@@ -47,6 +47,7 @@ flow:
   no_teardown: false
   replace_query_engine_with_dumb_consumer: false
   steady_state_wait: 60
+  monitor_interval_seconds: 1.0  # remote_monitor CPU/memory sample interval (0.1 = 100ms)
   skip_copy_prometheus_data: false  # Skip copying Prometheus data back to local machine
 
 

asap-tools/experiments/constants.py

@@ -25,6 +25,7 @@
 
 SKETCHDB_EXPERIMENT_NAME = "sketchdb"
 BASELINE_EXPERIMENT_NAME = "baseline"
+INGEST_MONITOR_STOP_FILE = ".ingest_monitor_stop"
 AVOID_REMOTE_MONITOR_LONG_SSH = True
 AVOID_RUN_ARROYOSKETCH_LONG_SSH = True
 

asap-tools/experiments/experiment_run_clickhouse.py

@@ -113,6 +113,8 @@
 
 CLICKHOUSE_DATABASE = "default"
 REMOTE_PROCESS_POLLING_INTERVAL = 10
+CLICKHOUSE_INGEST_MONITOR_OUTPUT_FILE = "monitor_output_ingest.json"
+DEFAULT_MONITOR_INTERVAL_SECONDS = 1.0
 
 
 def _inline_sql_queries_in_experiment_config(local_experiment_root_dir: str) -> None:
@@ -165,8 +167,16 @@ def _run_query_workload(
     query_engine_service: QueryEngineRustService | None,
     profile_query_engine: bool,
     profile_prometheus_time,
+    pre_query_wait_seconds: int = 0,
+    monitor_interval_seconds: float = DEFAULT_MONITOR_INTERVAL_SECONDS,
 ) -> None:
-    """Run the SQL query workload wrapped in remote_monitor (CPU/memory)."""
+    """Run the SQL query workload wrapped in remote_monitor (CPU/memory).
+
+    When ``pre_query_wait_seconds`` > 0, the monitor starts first and waits that
+    long before the query client runs, so a single monitor_output.json captures
+    the precompute/ingest phase (pc-worker threads) and the query phase in one
+    continuous, per-thread-attributed session.
+    """
     remote_monitor_service.start(
         controller_client_config=controller_client_config,
         experiment_output_dir=experiment_output_dir,
@@ -186,12 +196,136 @@ def _run_query_workload(
         use_container_prometheus_client=use_container,
         prometheus_client_parallel=parallel,
         backend_protocol="clickhouse",
+        pre_query_wait_seconds=pre_query_wait_seconds,
+        monitor_interval_seconds=monitor_interval_seconds,
     )
     if not manual_remote_monitor and constants.AVOID_REMOTE_MONITOR_LONG_SSH:
         remote_monitor_service.wait_for_remote_monitor_to_finish(
-            minimum_experiment_running_time=minimum_experiment_running_time,
+            minimum_experiment_running_time=(
+                minimum_experiment_running_time + pre_query_wait_seconds
+            ),
             polling_interval=REMOTE_PROCESS_POLLING_INTERVAL,
+            execution_mode="prometheus_client",
+            experiment_output_dir=experiment_output_dir,
+        )
+
+
+def _run_clickhouse_ingest_with_monitor(
+    provider,
+    node_offset: int,
+    remote_monitor_service: RemoteMonitorService,
+    baseline_output_dir: str,
+    local_baseline_dir: str,
+    controller_client_config: str,
+    data_loader: ClickHouseDataLoaderService,
+    dataset_name: str,
+    remote_data_file: str,
+    table: str,
+    init_sql_file: str,
+    max_rows: int,
+    manual_remote_monitor: bool,
+    monitor_interval_seconds: float = DEFAULT_MONITOR_INTERVAL_SECONDS,
+) -> None:
+    """Bulk-load netflow into ClickHouse while sampling ``clickhouse`` CPU/memory."""
+    os.makedirs(local_baseline_dir, exist_ok=True)
+    provider.execute_command(
+        node_idx=node_offset,
+        cmd=f"mkdir -p {baseline_output_dir}",
+        cmd_dir="",
+        nohup=False,
+        popen=False,
+    )
+
+    print(
+        f"Starting ClickHouse ingest monitor "
+        f"({monitor_interval_seconds}s samples)..."
+    )
+    ingest_monitor_remote = os.path.join(
+        baseline_output_dir,
+        "remote_monitor_output",
+        CLICKHOUSE_INGEST_MONITOR_OUTPUT_FILE,
+    )
+    provider.execute_command(
+        node_idx=node_offset,
+        cmd=f"rm -f {ingest_monitor_remote}",
+        cmd_dir="",
+        nohup=False,
+        popen=False,
+        ignore_errors=True,
+    )
+    remote_monitor_service.start_clickhouse_ingest_monitor(
+        controller_client_config=controller_client_config,
+        experiment_output_dir=baseline_output_dir,
+        monitor_interval_seconds=monitor_interval_seconds,
+        monitor_output_file=CLICKHOUSE_INGEST_MONITOR_OUTPUT_FILE,
+        manual_mode=manual_remote_monitor,
+    )
+    remote_monitor_service.wait_for_remote_monitor_start(
+        timeout=30,
+        execution_mode="ingest",
+        experiment_output_dir=baseline_output_dir,
+    )
+    time.sleep(1)
+
+    try:
+        print("Loading data into ClickHouse...")
+        data_loader.start(
+            dataset_name=dataset_name,
+            remote_data_file=remote_data_file,
+            table=table,
+            init_sql_file=init_sql_file,
+            max_rows=max_rows,
+        )
+    finally:
+        print("Stopping ClickHouse ingest monitor...")
+        remote_monitor_service.signal_ingest_monitor_stop(baseline_output_dir)
+        try:
+            remote_monitor_service.wait_for_remote_monitor_process_exit(
+                timeout=60,
+                execution_mode="ingest",
+                experiment_output_dir=baseline_output_dir,
+            )
+        except RuntimeError:
+            print("Ingest monitor did not exit gracefully; forcing kill...")
+            remote_monitor_service.kill_remote_monitor(
+                execution_mode="ingest",
+                experiment_output_dir=baseline_output_dir,
+            )
+            remote_monitor_service.wait_for_remote_monitor_process_exit(
+                timeout=30,
+                execution_mode="ingest",
+                experiment_output_dir=baseline_output_dir,
+            )
+        finally:
+            remote_monitor_service.cleanup_ingest_monitor_stop_file(baseline_output_dir)
+
+    sync.rsync_experiment_data(
+        provider,
+        baseline_output_dir,
+        local_baseline_dir,
+        node_offset=node_offset,
+    )
+    ingest_monitor_path = os.path.join(
+        local_baseline_dir,
+        "remote_monitor_output",
+        CLICKHOUSE_INGEST_MONITOR_OUTPUT_FILE,
+    )
+    if os.path.isfile(ingest_monitor_path):
+        with open(ingest_monitor_path) as f:
+            ingest_data = json.load(f)
+        if not ingest_data:
+            raise RuntimeError(
+                f"ClickHouse ingest monitor wrote empty/null data to "
+                f"{ingest_monitor_path}. Check baseline/remote_monitor.out and "
+                "baseline/remote_monitor_output/remote_monitor.log on the node."
+            )
+    elif not manual_remote_monitor:
+        raise RuntimeError(
+            "ClickHouse ingest monitor output file was not created at "
+            f"{ingest_monitor_path} (absent after stop timeout or write skip). "
+            "Check baseline/remote_monitor.out on the node."
         )
+    print(f"ClickHouse ingest monitor output: {ingest_monitor_path}")
 
 
 @hydra.main(version_base=None, config_path="config", config_name="config")
@@ -216,6 +350,7 @@ def main(cfg: DictConfig) -> None:
     manual_remote_monitor = bool(cfg.manual.remote_monitor)
     profile_query_engine = bool(cfg.profiling.query_engine)
     profile_prometheus_time = cfg.profiling.prometheus_time
+    monitor_interval_seconds = float(cfg.flow.monitor_interval_seconds)
     minimum_experiment_running_time = config.get_minimum_experiment_running_time(
         cfg.experiment_params
     )
@@ -302,26 +437,46 @@ def main(cfg: DictConfig) -> None:
         database=CLICKHOUSE_DATABASE,
     )
 
-    # --- load data once before the mode loop (DROP + reload) ---
+    # --- load data once before the mode loop (DROP + reload), with ingest monitor ---
     data_loader = ClickHouseDataLoaderService(
         provider,
         num_nodes=num_nodes,
         node_offset=node_offset,
         clickhouse_http_port=clickhouse_http_port,
     )
-    data_loader.start(
+    baseline_client_config = os.path.join(
+        experiment_root_output_dir,
+        "controller_client_configs",
+        f"{constants.BASELINE_EXPERIMENT_NAME}.yaml",
+    )
+    baseline_output_dir = os.path.join(
+        experiment_root_output_dir, constants.BASELINE_EXPERIMENT_NAME
+    )
+    local_baseline_dir = os.path.join(
+        local_experiment_root_dir, constants.BASELINE_EXPERIMENT_NAME
+    )
+    remote_monitor_service = RemoteMonitorService(provider, node_offset=node_offset)
+    _run_clickhouse_ingest_with_monitor(
+        provider=provider,
+        node_offset=node_offset,
+        remote_monitor_service=remote_monitor_service,
+        baseline_output_dir=baseline_output_dir,
+        local_baseline_dir=local_baseline_dir,
+        controller_client_config=baseline_client_config,
+        data_loader=data_loader,
         dataset_name=dataset_name,
         remote_data_file=remote_data_file,
         table=table,
         init_sql_file=init_sql_file,
         max_rows=max_rows,
+        manual_remote_monitor=manual_remote_monitor,
+        monitor_interval_seconds=monitor_interval_seconds,
     )
 
     # --- mode loop ---
     prometheus_client_service = PrometheusClientService(
         provider, use_container=use_container, node_offset=node_offset
     )
-    remote_monitor_service = RemoteMonitorService(provider, node_offset=node_offset)
 
     for experiment_mode in experiment_modes:
         print(f"Running experiment mode: {experiment_mode}")
@@ -441,15 +596,22 @@ def main(cfg: DictConfig) -> None:
 
             query_engine_service.wait_until_ready()
 
-            steady_state_wait = int(cfg.flow.steady_state_wait)
-            print(f"Waiting {steady_state_wait}s for precompute ingest to complete...")
-            time.sleep(steady_state_wait)
-
             controller_client_config = os.path.join(
                 experiment_root_output_dir,
                 "controller_client_configs",
                 f"{experiment_mode}.yaml",
             )
+
+            # Single continuous monitor spanning ingest + query: the monitor
+            # starts now (engine is ingesting) and waits steady_state_wait
+            # 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 "
+                "for ingest before queries..."
+            )
             _run_query_workload(
                 experiment_mode=experiment_mode,
                 experiment_output_dir=experiment_output_dir,
@@ -463,6 +625,8 @@ def main(cfg: DictConfig) -> 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,
+                monitor_interval_seconds=monitor_interval_seconds,
             )
 
             sync.rsync_experiment_data(
@@ -495,6 +659,7 @@ def main(cfg: DictConfig) -> None:
                 query_engine_service=None,
                 profile_query_engine=profile_query_engine,
                 profile_prometheus_time=profile_prometheus_time,
+                monitor_interval_seconds=monitor_interval_seconds,
             )
 
             sync.rsync_experiment_data(