From 9e79be95238c02da659ea246bf2087cea4aec39b Mon Sep 17 00:00:00 2001 From: Akanksha Akkihal Date: Sat, 11 Jul 2026 21:39:02 -0400 Subject: [PATCH 1/4] feat(asap-tools): separate ClickHouse ingest and query resource monitoring Baseline bulk load now runs under a dedicated ingest monitor that writes monitor_output_ingest.json and stops via a stop file, so ingest CPU/memory is not mixed with the query-phase monitor_output.json. Co-authored-by: Cursor --- .../experiments/classes/process_monitor.py | 23 ++- asap-tools/experiments/config/config.yaml | 1 + asap-tools/experiments/constants.py | 1 + .../experiments/experiment_run_clickhouse.py | 167 +++++++++++++++- .../services/remote_monitor_service.py | 180 +++++++++++++++++- asap-tools/experiments/remote_monitor.py | 87 ++++++++- 6 files changed, 440 insertions(+), 19 deletions(-) diff --git a/asap-tools/experiments/classes/process_monitor.py b/asap-tools/experiments/classes/process_monitor.py index 4449f532..58295395 100644 --- a/asap-tools/experiments/classes/process_monitor.py +++ b/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,14 @@ 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) return monitor_info diff --git a/asap-tools/experiments/config/config.yaml b/asap-tools/experiments/config/config.yaml index 7b697450..1da9f414 100644 --- a/asap-tools/experiments/config/config.yaml +++ b/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 diff --git a/asap-tools/experiments/constants.py b/asap-tools/experiments/constants.py index 2358a3ae..7492cf3d 100644 --- a/asap-tools/experiments/constants.py +++ b/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 diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 7bdd370d..a13766e5 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/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,14 +196,122 @@ 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, ) +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) + 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) + except RuntimeError: + print("Ingest monitor did not exit gracefully; forcing kill...") + remote_monitor_service.kill_remote_monitor() + remote_monitor_service.wait_for_remote_monitor_process_exit(timeout=30) + 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 ingest_data is None: + raise RuntimeError( + f"ClickHouse ingest monitor wrote no data to {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}. 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") def main(cfg: DictConfig) -> None: config.validate_basic_config( @@ -216,6 +334,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 +421,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 +580,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 +609,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 +643,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( 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 28986df0..305c7483 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -50,6 +50,8 @@ def start( backend_protocol: str, backend_tool: Optional[str] = None, timed_duration: Optional[int] = None, + pre_query_wait_seconds: int = 0, + monitor_interval_seconds: float = 1.0, ) -> None: """ Start remote monitor processes. @@ -59,6 +61,11 @@ def start( timed_duration: If provided, use timed mode instead of prometheus_client mode backend_tool: TSDB running on the node ("prometheus" or "victoriametrics"). Not used when backend_protocol="clickhouse". + pre_query_wait_seconds: In prometheus_client mode, seconds to wait + (with the process monitor already running) before launching the + query client. Used to capture the precompute/ingest phase in the + same continuous monitor session as the query phase. + monitor_interval_seconds: Seconds between CPU/memory samples. """ # Determine execution mode use_timed_mode = timed_duration is not None @@ -123,6 +130,7 @@ def start( "--time_to_run {} " "--node_offset {} " "--streaming_engine {} " + "--monitor_interval_seconds {} " ).format( experiment_mode, ",".join(keywords), @@ -136,6 +144,7 @@ def start( timed_duration, self.node_offset, streaming_engine, + monitor_interval_seconds, ) cmd_dir = os.path.join( @@ -176,6 +185,7 @@ def start( "--prometheus_client_output_file {} " "--node_offset {} " "--streaming_engine {} " + "--monitor_interval_seconds {} " ).format( experiment_mode, ",".join(keywords), @@ -189,8 +199,12 @@ def start( "prometheus_client_output.txt", self.node_offset, streaming_engine, + monitor_interval_seconds, ) + if pre_query_wait_seconds > 0: + cmd += " --pre_query_wait_seconds {}".format(pre_query_wait_seconds) + # Add container flag if enabled if use_container_prometheus_client: cmd += " --use_container_prometheus_client" @@ -252,6 +266,158 @@ def start( popen=False, ) + def start_clickhouse_ingest_monitor( + self, + controller_client_config: str, + experiment_output_dir: str, + monitor_interval_seconds: float = 1.0, + monitor_output_file: str = "monitor_output.json", + manual_mode: bool = False, + ) -> None: + """Monitor ClickHouse CPU/memory during bulk JSONEachRow load. + + Runs ``remote_monitor.py`` in ``ingest`` mode (samples until SIGTERM), + then writes ``monitor_output_file`` under ``remote_monitor_output/``. + """ + keywords = ["clickhouse"] + cmd = ( + "python3 -u remote_monitor.py " + "--execution_mode ingest " + "--experiment_mode {} " + r"--keywords \"{}\" " + "--config_file {} " + "--experiment_output_dir {} " + "--monitor_output_file {} " + "--node_offset {} " + "--streaming_engine {} " + "--monitor_interval_seconds {} " + "--backend_protocol clickhouse " + ).format( + constants.BASELINE_EXPERIMENT_NAME, + ",".join(keywords), + controller_client_config, + experiment_output_dir, + monitor_output_file, + self.node_offset, + "precompute", + monitor_interval_seconds, + ) + + cmd_dir = os.path.join( + self.provider.get_home_dir(), "code", "asap-tools", "experiments" + ) + stop_file = os.path.join( + experiment_output_dir, constants.INGEST_MONITOR_STOP_FILE + ) + self.provider.execute_command( + node_idx=self.node_offset, + cmd=f"rm -f {stop_file}", + cmd_dir="", + nohup=False, + popen=False, + ignore_errors=True, + ) + cmd += " > {}/remote_monitor.out 2>&1".format(experiment_output_dir) + + if manual_mode: + input( + "In manual mode. ClickHouse ingest monitor will not start. Press Enter" + ) + print(cmd_dir) + print(cmd) + return + + cmd += " < /dev/null &" + self.provider.execute_command( + node_idx=self.node_offset, + cmd=cmd, + cmd_dir=cmd_dir, + nohup=True, + popen=False, + ) + + def is_remote_monitor_running(self) -> bool: + cmd = "pgrep -f remote_monitor.py" + result = self.provider.execute_command( + node_idx=self.node_offset, + cmd=cmd, + cmd_dir=None, + nohup=False, + popen=False, + ignore_errors=True, + ) + assert isinstance(result, subprocess.CompletedProcess) + return bool(result.stdout.strip()) + + def wait_for_remote_monitor_start( + self, timeout: int = 30, polling_interval: int = 1 + ) -> None: + """Poll until ``remote_monitor.py`` is running on the node.""" + start = time.time() + while time.time() - start < timeout: + if self.is_remote_monitor_running(): + return + time.sleep(polling_interval) + raise RuntimeError( + "remote_monitor.py did not start within " + f"{timeout}s. Check remote_monitor.out under the experiment output dir." + ) + + def wait_for_remote_monitor_process_exit( + self, polling_interval: int = 2, timeout: int = 300 + ) -> None: + """Poll until ``remote_monitor.py`` is no longer running on the node.""" + start = time.time() + while True: + cmd = "pgrep -f remote_monitor.py" + result = self.provider.execute_command( + node_idx=self.node_offset, + cmd=cmd, + cmd_dir=None, + nohup=False, + popen=False, + ignore_errors=True, + ) + assert isinstance(result, subprocess.CompletedProcess) + if result.stdout.strip() == "": + return + if time.time() - start > timeout: + raise RuntimeError( + f"remote_monitor.py did not exit within {timeout}s after stop()" + ) + print( + "Waiting for remote monitor to exit after ingest " + f"(checking again in {polling_interval}s)..." + ) + time.sleep(polling_interval) + + def signal_ingest_monitor_stop(self, experiment_output_dir: str) -> None: + """Ask an ingest-mode ``remote_monitor.py`` to shut down gracefully.""" + stop_file = os.path.join( + experiment_output_dir, constants.INGEST_MONITOR_STOP_FILE + ) + self.provider.execute_command( + node_idx=self.node_offset, + cmd=f"touch {stop_file}", + cmd_dir="", + nohup=False, + popen=False, + ignore_errors=True, + ) + + def cleanup_ingest_monitor_stop_file(self, experiment_output_dir: str) -> None: + stop_file = os.path.join( + experiment_output_dir, constants.INGEST_MONITOR_STOP_FILE + ) + self.provider.execute_command( + node_idx=self.node_offset, + cmd=f"rm -f {stop_file}", + cmd_dir="", + nohup=False, + popen=False, + ignore_errors=True, + ) + def stop(self, **kwargs) -> None: """ Stop remote monitor processes. @@ -277,6 +443,7 @@ def wait_for_remote_monitor_to_finish( self, minimum_experiment_running_time: int, polling_interval: int = 10, + timeout: int = 600, ) -> None: """ Wait for remote monitor process to finish. @@ -284,6 +451,7 @@ def wait_for_remote_monitor_to_finish( Args: minimum_experiment_running_time: Minimum time to wait before polling polling_interval: Interval between polling checks + timeout: Max seconds to poll after the minimum wait before force-kill """ print( "Waiting for {} seconds for remote monitor to finish".format( @@ -293,6 +461,7 @@ def wait_for_remote_monitor_to_finish( time.sleep(minimum_experiment_running_time) print("Done waiting for remote monitor to finish. Will start polling") + poll_start = time.time() while True: cmd = "pgrep -f remote_monitor.py" result = self.provider.execute_command( @@ -304,7 +473,16 @@ def wait_for_remote_monitor_to_finish( ignore_errors=True, ) assert isinstance(result, subprocess.CompletedProcess) - if result.stdout == "": + if result.stdout.strip() == "": + break + if time.time() - poll_start > timeout: + print( + "Remote monitor still running after {}s; forcing kill".format( + timeout + ) + ) + self.kill_remote_monitor() + self.wait_for_remote_monitor_process_exit(timeout=60) break print( "Remote monitor is still running. Will check again in {} seconds".format( diff --git a/asap-tools/experiments/remote_monitor.py b/asap-tools/experiments/remote_monitor.py index 6e802d76..3dd57ecb 100644 --- a/asap-tools/experiments/remote_monitor.py +++ b/asap-tools/experiments/remote_monitor.py @@ -3,6 +3,7 @@ import time import argparse import subprocess +import threading import yaml import signal from loguru import logger @@ -348,15 +349,34 @@ def main(args): node_offset=args.node_offset, ) + # Bulk ClickHouse ingest can spawn many short-lived children; walking them + # recursively blocks the monitor loop long enough to miss the stop signal. + include_children = args.execution_mode != "ingest" + monitor, control_pipe, monitor_pipe = process_monitor.start_monitor( pids, keywords_expanded, - 1, + args.monitor_interval_seconds, ["memory_info", "cpu_percent"], - include_children=True, + include_children=include_children, hooks=monitor_hooks, thread_attribution_keyword=( - constants.QUERY_ENGINE_RS_CONTAINER_NAME + # Attribute pc-worker (precompute) vs other (query) threads. The + # monitored keyword differs by deployment: containerized runs use + # the container name, bare-metal runs use the process keyword. Pick + # whichever is actually being monitored so attribution turns on for + # both (bare-metal ClickHouse runs use QUERY_ENGINE_RS_PROCESS_KEYWORD). + next( + ( + kw + for kw in ( + constants.QUERY_ENGINE_RS_CONTAINER_NAME, + constants.QUERY_ENGINE_RS_PROCESS_KEYWORD, + ) + if kw in args.keywords + ), + None, + ) if args.streaming_engine == "precompute" and args.experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME else None @@ -381,6 +401,16 @@ def main(args): ) if args.execution_mode == "prometheus_client": + # Wait out the precompute/ingest phase while the monitor is already + # sampling, so a single monitor_output.json captures precompute CPU + # (pc-worker threads) during ingest and query CPU during the client run. + if args.pre_query_wait_seconds > 0: + logger.debug( + "Waiting {} seconds for precompute/ingest before query client", + args.pre_query_wait_seconds, + ) + time.sleep(args.pre_query_wait_seconds) + logger.debug("Starting prometheus client") # Create CloudLab local provider for local execution with CloudLab paths provider = CloudLabLocalProvider(username="user", use_cloudlab_ips=False) @@ -415,6 +445,28 @@ def main(args): elif args.execution_mode == "interactive": logger.debug("Waiting for user input to stop monitoring") input("Press Enter to stop monitoring...") + elif args.execution_mode == "ingest": + stop_file = os.path.join( + args.experiment_output_dir, constants.INGEST_MONITOR_STOP_FILE + ) + if os.path.exists(stop_file): + os.remove(stop_file) + logger.debug("Monitoring ingest until stop file or SIGTERM") + shutdown = threading.Event() + + def _request_shutdown(signum, _frame): + logger.debug("Ingest monitor received signal {}, shutting down", signum) + shutdown.set() + + signal.signal(signal.SIGTERM, _request_shutdown) + signal.signal(signal.SIGINT, _request_shutdown) + while not shutdown.is_set(): + if os.path.exists(stop_file): + logger.debug("Ingest monitor stop file detected, shutting down") + os.remove(stop_file) + shutdown.set() + break + shutdown.wait(timeout=0.5) elif args.execution_mode == "timed": logger.debug(f"Running for {args.time_to_run} seconds") time.sleep(args.time_to_run) @@ -438,6 +490,11 @@ def main(args): logger.debug("Stopping process monitors") monitor_info = process_monitor.stop_monitor(monitor, control_pipe, monitor_pipe) + if monitor_info is None: + logger.error( + "Process monitor did not return data before timeout; " + f"{args.monitor_output_file} will be empty" + ) monitor_output_file = os.path.join( remote_monitor_output_dir, args.monitor_output_file @@ -454,8 +511,8 @@ def main(args): "--execution_mode", type=str, required=True, - choices=["interactive", "timed", "prometheus_client"], - help="Execution mode: interactive, timed, or prometheus_client", + choices=["interactive", "timed", "ingest", "prometheus_client"], + help="Execution mode: interactive, timed, ingest (until SIGTERM), or prometheus_client", ) parser.add_argument("--experiment_mode", type=str, required=True) parser.add_argument( @@ -526,6 +583,26 @@ def main(args): required=True, help="Query backend for prometheus-client (prometheus or clickhouse)", ) + parser.add_argument( + "--monitor_interval_seconds", + type=float, + default=1.0, + help=( + "Seconds between process/thread CPU+memory samples. Smaller values " + "(e.g. 0.1) capture short precompute/query CPU spikes that 1s " + "sampling rounds to zero, at the cost of a larger monitor_output.json." + ), + ) + parser.add_argument( + "--pre_query_wait_seconds", + type=int, + default=0, + help=( + "Seconds to wait (monitor already running) before launching the " + "query client, to capture the precompute/ingest phase in the same " + "monitor session as the query phase." + ), + ) args = parser.parse_args() args.keywords = args.keywords.strip().split(",") if args.profile_flink_pids: From 686981e03f749ada00e4ba94accf8032bb2fef9b Mon Sep 17 00:00:00 2001 From: Akanksha Akkihal Date: Sat, 11 Jul 2026 21:49:02 -0400 Subject: [PATCH 2/4] fix(asap-tools): harden ingest-monitor stop and process matching Match remote_monitor by execution mode and experiment dir, avoid writing null monitor JSON, escalate stuck sampler with kill(), and clarify ingest stop-file semantics. Co-authored-by: Cursor --- .../experiments/classes/process_monitor.py | 5 + .../experiments/experiment_run_clickhouse.py | 34 +++-- .../services/remote_monitor_service.py | 131 ++++++++++++------ asap-tools/experiments/remote_monitor.py | 30 ++-- 4 files changed, 142 insertions(+), 58 deletions(-) diff --git a/asap-tools/experiments/classes/process_monitor.py b/asap-tools/experiments/classes/process_monitor.py index 58295395..e42aa485 100644 --- a/asap-tools/experiments/classes/process_monitor.py +++ b/asap-tools/experiments/classes/process_monitor.py @@ -282,4 +282,9 @@ def stop_monitor(monitor, control_pipe, monitor_pipe, timeout=30): monitor_info = None monitor.terminate() 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 diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index a13766e5..43b2aa44 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -205,6 +205,8 @@ def _run_query_workload( minimum_experiment_running_time + pre_query_wait_seconds ), polling_interval=REMOTE_PROCESS_POLLING_INTERVAL, + execution_mode="prometheus_client", + experiment_output_dir=experiment_output_dir, ) @@ -258,7 +260,11 @@ def _run_clickhouse_ingest_with_monitor( monitor_output_file=CLICKHOUSE_INGEST_MONITOR_OUTPUT_FILE, manual_mode=manual_remote_monitor, ) - remote_monitor_service.wait_for_remote_monitor_start(timeout=30) + remote_monitor_service.wait_for_remote_monitor_start( + timeout=30, + execution_mode="ingest", + experiment_output_dir=baseline_output_dir, + ) time.sleep(1) try: @@ -274,11 +280,22 @@ def _run_clickhouse_ingest_with_monitor( 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) + 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() - remote_monitor_service.wait_for_remote_monitor_process_exit(timeout=30) + 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 @@ -298,16 +315,17 @@ def _run_clickhouse_ingest_with_monitor( if os.path.isfile(ingest_monitor_path): with open(ingest_monitor_path) as f: ingest_data = json.load(f) - if ingest_data is None: + if not ingest_data: raise RuntimeError( - f"ClickHouse ingest monitor wrote no data to {ingest_monitor_path}. " - "Check baseline/remote_monitor.out and " + 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}. Check baseline/remote_monitor.out on the node." + 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}") 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 305c7483..f08a78ff 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -3,6 +3,8 @@ """ import os +import re +import shlex import time import subprocess from typing import List, Optional @@ -276,7 +278,8 @@ def start_clickhouse_ingest_monitor( ) -> None: """Monitor ClickHouse CPU/memory during bulk JSONEachRow load. - Runs ``remote_monitor.py`` in ``ingest`` mode (samples until SIGTERM), + Runs ``remote_monitor.py`` in ``ingest`` mode until the stop file + (``constants.INGEST_MONITOR_STOP_FILE``) is created, or SIGTERM/SIGINT, then writes ``monitor_output_file`` under ``remote_monitor_output/``. """ keywords = ["clickhouse"] @@ -311,7 +314,7 @@ def start_clickhouse_ingest_monitor( ) self.provider.execute_command( node_idx=self.node_offset, - cmd=f"rm -f {stop_file}", + cmd=f"rm -f {shlex.quote(stop_file)}", cmd_dir="", nohup=False, popen=False, @@ -336,8 +339,35 @@ def start_clickhouse_ingest_monitor( popen=False, ) - def is_remote_monitor_running(self) -> bool: - cmd = "pgrep -f remote_monitor.py" + @staticmethod + def _remote_monitor_pgrep_pattern( + execution_mode: Optional[str] = None, + experiment_output_dir: Optional[str] = None, + ) -> str: + """Build a ``pgrep -f`` regex that prefers mode/dir over a bare script name. + + Matching only ``remote_monitor.py`` can hit leftover processes from a + prior crashed run. Prefer ``--execution_mode`` and the experiment output + directory when known. + """ + parts = ["remote_monitor\\.py"] + if execution_mode: + parts.append(f"--execution_mode {execution_mode}") + if experiment_output_dir: + # Escape regex metacharacters in the path (e.g. dots in hostnames). + parts.append(re.escape(experiment_output_dir)) + return ".*".join(parts) + + def is_remote_monitor_running( + self, + execution_mode: Optional[str] = None, + experiment_output_dir: Optional[str] = None, + ) -> bool: + pattern = self._remote_monitor_pgrep_pattern( + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ) + cmd = f"pgrep -f {shlex.quote(pattern)}" result = self.provider.execute_command( node_idx=self.node_offset, cmd=cmd, @@ -350,43 +380,50 @@ def is_remote_monitor_running(self) -> bool: return bool(result.stdout.strip()) def wait_for_remote_monitor_start( - self, timeout: int = 30, polling_interval: int = 1 + self, + timeout: int = 30, + polling_interval: int = 1, + execution_mode: Optional[str] = None, + experiment_output_dir: Optional[str] = None, ) -> None: - """Poll until ``remote_monitor.py`` is running on the node.""" + """Poll until a matching ``remote_monitor.py`` is running on the node.""" start = time.time() while time.time() - start < timeout: - if self.is_remote_monitor_running(): + if self.is_remote_monitor_running( + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ): return time.sleep(polling_interval) + mode_desc = execution_mode or "any" raise RuntimeError( - "remote_monitor.py did not start within " + f"remote_monitor.py ({mode_desc}) did not start within " f"{timeout}s. Check remote_monitor.out under the experiment output dir." ) def wait_for_remote_monitor_process_exit( - self, polling_interval: int = 2, timeout: int = 300 + self, + polling_interval: int = 2, + timeout: int = 300, + execution_mode: Optional[str] = None, + experiment_output_dir: Optional[str] = None, ) -> None: - """Poll until ``remote_monitor.py`` is no longer running on the node.""" + """Poll until matching ``remote_monitor.py`` processes have exited.""" start = time.time() while True: - cmd = "pgrep -f remote_monitor.py" - result = self.provider.execute_command( - node_idx=self.node_offset, - cmd=cmd, - cmd_dir=None, - nohup=False, - popen=False, - ignore_errors=True, - ) - assert isinstance(result, subprocess.CompletedProcess) - if result.stdout.strip() == "": + if not self.is_remote_monitor_running( + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ): return if time.time() - start > timeout: + mode_desc = execution_mode or "any" raise RuntimeError( - f"remote_monitor.py did not exit within {timeout}s after stop()" + f"remote_monitor.py ({mode_desc}) did not exit within " + f"{timeout}s after stop()" ) print( - "Waiting for remote monitor to exit after ingest " + "Waiting for remote monitor to exit " f"(checking again in {polling_interval}s)..." ) time.sleep(polling_interval) @@ -398,7 +435,7 @@ def signal_ingest_monitor_stop(self, experiment_output_dir: str) -> None: ) self.provider.execute_command( node_idx=self.node_offset, - cmd=f"touch {stop_file}", + cmd=f"touch {shlex.quote(stop_file)}", cmd_dir="", nohup=False, popen=False, @@ -411,7 +448,7 @@ def cleanup_ingest_monitor_stop_file(self, experiment_output_dir: str) -> None: ) self.provider.execute_command( node_idx=self.node_offset, - cmd=f"rm -f {stop_file}", + cmd=f"rm -f {shlex.quote(stop_file)}", cmd_dir="", nohup=False, popen=False, @@ -427,9 +464,17 @@ def stop(self, **kwargs) -> None: """ self.kill_remote_monitor() - def kill_remote_monitor(self) -> None: - """Kill remote monitor processes.""" - cmd = "pkill -f remote_monitor.py" + def kill_remote_monitor( + self, + execution_mode: Optional[str] = None, + experiment_output_dir: Optional[str] = None, + ) -> None: + """Kill matching remote monitor processes.""" + pattern = self._remote_monitor_pgrep_pattern( + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ) + cmd = f"pkill -f {shlex.quote(pattern)}" self.provider.execute_command( node_idx=self.node_offset, cmd=cmd, @@ -444,6 +489,8 @@ def wait_for_remote_monitor_to_finish( minimum_experiment_running_time: int, polling_interval: int = 10, timeout: int = 600, + execution_mode: Optional[str] = None, + experiment_output_dir: Optional[str] = None, ) -> None: """ Wait for remote monitor process to finish. @@ -452,6 +499,8 @@ def wait_for_remote_monitor_to_finish( minimum_experiment_running_time: Minimum time to wait before polling polling_interval: Interval between polling checks timeout: Max seconds to poll after the minimum wait before force-kill + execution_mode: If set, only match this ``--execution_mode`` + experiment_output_dir: Optional path to further disambiguate the process """ print( "Waiting for {} seconds for remote monitor to finish".format( @@ -463,17 +512,10 @@ def wait_for_remote_monitor_to_finish( poll_start = time.time() while True: - cmd = "pgrep -f remote_monitor.py" - result = self.provider.execute_command( - node_idx=self.node_offset, - cmd=cmd, - cmd_dir=None, - nohup=False, - popen=False, - ignore_errors=True, - ) - assert isinstance(result, subprocess.CompletedProcess) - if result.stdout.strip() == "": + if not self.is_remote_monitor_running( + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ): break if time.time() - poll_start > timeout: print( @@ -481,8 +523,15 @@ def wait_for_remote_monitor_to_finish( timeout ) ) - self.kill_remote_monitor() - self.wait_for_remote_monitor_process_exit(timeout=60) + self.kill_remote_monitor( + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ) + self.wait_for_remote_monitor_process_exit( + timeout=60, + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ) break print( "Remote monitor is still running. Will check again in {} seconds".format( diff --git a/asap-tools/experiments/remote_monitor.py b/asap-tools/experiments/remote_monitor.py index 3dd57ecb..52c9c1ed 100644 --- a/asap-tools/experiments/remote_monitor.py +++ b/asap-tools/experiments/remote_monitor.py @@ -351,6 +351,8 @@ def main(args): # Bulk ClickHouse ingest can spawn many short-lived children; walking them # recursively blocks the monitor loop long enough to miss the stop signal. + # Trade-off: ingest samples only the main matched process (e.g. clickhouse + # server), not helper children. include_children = args.execution_mode != "ingest" monitor, control_pipe, monitor_pipe = process_monitor.start_monitor( @@ -451,7 +453,10 @@ def main(args): ) if os.path.exists(stop_file): os.remove(stop_file) - logger.debug("Monitoring ingest until stop file or SIGTERM") + logger.debug( + "Monitoring ingest until stop file ({}) or SIGTERM/SIGINT", + constants.INGEST_MONITOR_STOP_FILE, + ) shutdown = threading.Event() def _request_shutdown(signum, _frame): @@ -490,17 +495,20 @@ def _request_shutdown(signum, _frame): logger.debug("Stopping process monitors") monitor_info = process_monitor.stop_monitor(monitor, control_pipe, monitor_pipe) - if monitor_info is None: - logger.error( - "Process monitor did not return data before timeout; " - f"{args.monitor_output_file} will be empty" - ) monitor_output_file = os.path.join( remote_monitor_output_dir, args.monitor_output_file ) - with open(monitor_output_file, "w") as f: - json.dump(monitor_info, f) + if monitor_info is None: + logger.error( + "Process monitor did not return data before timeout; " + f"not writing {args.monitor_output_file} (leave file absent)" + ) + if os.path.exists(monitor_output_file): + os.remove(monitor_output_file) + else: + with open(monitor_output_file, "w") as f: + json.dump(monitor_info, f) logger.debug("Done") @@ -512,7 +520,11 @@ def _request_shutdown(signum, _frame): type=str, required=True, choices=["interactive", "timed", "ingest", "prometheus_client"], - help="Execution mode: interactive, timed, ingest (until SIGTERM), or prometheus_client", + help=( + "Execution mode: interactive, timed, ingest (until stop file " + f"{constants.INGEST_MONITOR_STOP_FILE} or SIGTERM/SIGINT), " + "or prometheus_client" + ), ) parser.add_argument("--experiment_mode", type=str, required=True) parser.add_argument( From f5ee516db6f7989305d7bb386346967409c06579 Mon Sep 17 00:00:00 2001 From: Akanksha Akkihal Date: Sat, 11 Jul 2026 22:25:56 -0400 Subject: [PATCH 3/4] style(asap-tools): satisfy black formatting in experiment_run_clickhouse Co-authored-by: Cursor --- asap-tools/experiments/experiment_run_clickhouse.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 43b2aa44..ae3d6076 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -297,9 +297,7 @@ def _run_clickhouse_ingest_with_monitor( experiment_output_dir=baseline_output_dir, ) finally: - remote_monitor_service.cleanup_ingest_monitor_stop_file( - baseline_output_dir - ) + remote_monitor_service.cleanup_ingest_monitor_stop_file(baseline_output_dir) sync.rsync_experiment_data( provider, From 84c77dc65a9dfdf27d0c362dcb01f6e907123b68 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 17 Aug 2026 20:38:05 -0400 Subject: [PATCH 4/4] fix(asap-tools): address PR 522 review comments on remote monitor lifecycle - Remove the timeout-driven force-kill escalation from wait_for_remote_monitor_to_finish() and the ingest monitor's finally block; both now poll forever like the pre-PR behavior. This was unscoped for experiment_run_e2e.py's call site and, in the ingest case, pkill could hit the forked sampler child directly (same cmdline as the parent, no exec), losing partial data instead of letting the graceful pipe handoff complete. kill_remote_monitor() is still used, just only from the existing .stop() teardown path. - Fix cfg.manual.remote_monitor=true always crashing in the ClickHouse ingest path: wait_for_remote_monitor_start() assumed the monitor was already running, but manual mode never starts it automatically. Guard it like the existing pattern in _run_query_workload. - Replace scattered timeout/poll-interval literals with named constants in constants.py, matching the file's existing pattern. - Drop now-redundant default arguments (timeout/polling_interval on the wait_for_remote_monitor_* helpers, pre_query_wait_seconds/ monitor_interval_seconds on RemoteMonitorService.start(), and --monitor_interval_seconds on remote_monitor.py's CLI) now that every real call site passes them explicitly from config.yaml. Co-Authored-By: Claude Sonnet 5 --- .../experiments/classes/process_monitor.py | 10 ++-- asap-tools/experiments/constants.py | 8 +++ .../experiments/experiment_run_clickhouse.py | 27 +++------ asap-tools/experiments/experiment_run_e2e.py | 4 +- .../services/remote_monitor_service.py | 59 +++++-------------- asap-tools/experiments/remote_monitor.py | 13 +++- 6 files changed, 50 insertions(+), 71 deletions(-) diff --git a/asap-tools/experiments/classes/process_monitor.py b/asap-tools/experiments/classes/process_monitor.py index e42aa485..80ec79fe 100644 --- a/asap-tools/experiments/classes/process_monitor.py +++ b/asap-tools/experiments/classes/process_monitor.py @@ -5,6 +5,8 @@ from typing import List, Any, Optional from classes.ProcessMonitorHook import ProcessMonitorHook, ProcessMetricSnapshot +import constants + _PRECOMPUTE_THREAD_PREFIX = "pc-worker" @@ -272,19 +274,19 @@ def start_monitor( return monitor, control_pipe, monitor_pipe -def stop_monitor(monitor, control_pipe, monitor_pipe, timeout=30): +def stop_monitor(monitor, control_pipe, monitor_pipe, timeout: float): control_pipe.send("stop") can_read = control_pipe.poll(timeout) if can_read: monitor_info = control_pipe.recv() - monitor.join(timeout=10) + monitor.join(timeout=constants.PROCESS_MONITOR_JOIN_TIMEOUT_SECONDS) else: monitor_info = None monitor.terminate() - monitor.join(timeout=10) + monitor.join(timeout=constants.PROCESS_MONITOR_JOIN_TIMEOUT_SECONDS) # 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) + monitor.join(timeout=constants.PROCESS_MONITOR_JOIN_TIMEOUT_SECONDS) return monitor_info diff --git a/asap-tools/experiments/constants.py b/asap-tools/experiments/constants.py index 7492cf3d..d8173c96 100644 --- a/asap-tools/experiments/constants.py +++ b/asap-tools/experiments/constants.py @@ -29,6 +29,14 @@ AVOID_REMOTE_MONITOR_LONG_SSH = True AVOID_RUN_ARROYOSKETCH_LONG_SSH = True +# remote_monitor.py process lifecycle timing (start/exit polling) +REMOTE_MONITOR_START_TIMEOUT_SECONDS = 30 +REMOTE_MONITOR_START_POLL_INTERVAL_SECONDS = 1 +REMOTE_MONITOR_EXIT_POLL_INTERVAL_SECONDS = 2 +INGEST_MONITOR_SHUTDOWN_POLL_INTERVAL_SECONDS = 0.5 +PROCESS_MONITOR_STOP_TIMEOUT_SECONDS = 30 +PROCESS_MONITOR_JOIN_TIMEOUT_SECONDS = 10 + PROMETHEUS_CONFIG_DIR = "prometheus_config" PROMETHEUS_CONFIG_FILE = "prometheus.yml" diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index ae3d6076..4752f450 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -260,12 +260,14 @@ def _run_clickhouse_ingest_with_monitor( 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) + if not manual_remote_monitor: + remote_monitor_service.wait_for_remote_monitor_start( + timeout=constants.REMOTE_MONITOR_START_TIMEOUT_SECONDS, + polling_interval=constants.REMOTE_MONITOR_START_POLL_INTERVAL_SECONDS, + execution_mode="ingest", + experiment_output_dir=baseline_output_dir, + ) + time.sleep(1) try: print("Loading data into ClickHouse...") @@ -281,18 +283,7 @@ def _run_clickhouse_ingest_with_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, + polling_interval=constants.REMOTE_MONITOR_EXIT_POLL_INTERVAL_SECONDS, execution_mode="ingest", experiment_output_dir=baseline_output_dir, ) diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index 6edbf074..20632169 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -624,8 +624,10 @@ def main(cfg: DictConfig): controller_remote_output_dir=CONTROLLER_REMOTE_OUTPUT_DIR, use_container_prometheus_client=args.use_container_prometheus_client, prometheus_client_parallel=args.prometheus_client_parallel, - backend_tool=cfg.experiment_params.monitoring.tool, backend_protocol="prometheus", + pre_query_wait_seconds=0, + monitor_interval_seconds=float(cfg.flow.monitor_interval_seconds), + backend_tool=cfg.experiment_params.monitoring.tool, timed_duration=minimum_experiment_running_time if skip_querying else None, ) 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 f08a78ff..304c7979 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -50,10 +50,10 @@ def start( use_container_prometheus_client: bool, prometheus_client_parallel: bool, backend_protocol: str, + pre_query_wait_seconds: int, + monitor_interval_seconds: float, backend_tool: Optional[str] = None, timed_duration: Optional[int] = None, - pre_query_wait_seconds: int = 0, - monitor_interval_seconds: float = 1.0, ) -> None: """ Start remote monitor processes. @@ -381,8 +381,8 @@ def is_remote_monitor_running( def wait_for_remote_monitor_start( self, - timeout: int = 30, - polling_interval: int = 1, + timeout: int, + polling_interval: int, execution_mode: Optional[str] = None, experiment_output_dir: Optional[str] = None, ) -> None: @@ -403,25 +403,15 @@ def wait_for_remote_monitor_start( def wait_for_remote_monitor_process_exit( self, - polling_interval: int = 2, - timeout: int = 300, + polling_interval: int, execution_mode: Optional[str] = None, experiment_output_dir: Optional[str] = None, ) -> None: """Poll until matching ``remote_monitor.py`` processes have exited.""" - start = time.time() - while True: - if not self.is_remote_monitor_running( - execution_mode=execution_mode, - experiment_output_dir=experiment_output_dir, - ): - return - if time.time() - start > timeout: - mode_desc = execution_mode or "any" - raise RuntimeError( - f"remote_monitor.py ({mode_desc}) did not exit within " - f"{timeout}s after stop()" - ) + while self.is_remote_monitor_running( + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ): print( "Waiting for remote monitor to exit " f"(checking again in {polling_interval}s)..." @@ -487,8 +477,7 @@ def kill_remote_monitor( def wait_for_remote_monitor_to_finish( self, minimum_experiment_running_time: int, - polling_interval: int = 10, - timeout: int = 600, + polling_interval: int, execution_mode: Optional[str] = None, experiment_output_dir: Optional[str] = None, ) -> None: @@ -498,7 +487,6 @@ def wait_for_remote_monitor_to_finish( Args: minimum_experiment_running_time: Minimum time to wait before polling polling_interval: Interval between polling checks - timeout: Max seconds to poll after the minimum wait before force-kill execution_mode: If set, only match this ``--execution_mode`` experiment_output_dir: Optional path to further disambiguate the process """ @@ -510,29 +498,10 @@ def wait_for_remote_monitor_to_finish( time.sleep(minimum_experiment_running_time) print("Done waiting for remote monitor to finish. Will start polling") - poll_start = time.time() - while True: - if not self.is_remote_monitor_running( - execution_mode=execution_mode, - experiment_output_dir=experiment_output_dir, - ): - break - if time.time() - poll_start > timeout: - print( - "Remote monitor still running after {}s; forcing kill".format( - timeout - ) - ) - self.kill_remote_monitor( - execution_mode=execution_mode, - experiment_output_dir=experiment_output_dir, - ) - self.wait_for_remote_monitor_process_exit( - timeout=60, - execution_mode=execution_mode, - experiment_output_dir=experiment_output_dir, - ) - break + while self.is_remote_monitor_running( + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ): print( "Remote monitor is still running. Will check again in {} seconds".format( polling_interval diff --git a/asap-tools/experiments/remote_monitor.py b/asap-tools/experiments/remote_monitor.py index 52c9c1ed..41c58bfd 100644 --- a/asap-tools/experiments/remote_monitor.py +++ b/asap-tools/experiments/remote_monitor.py @@ -471,7 +471,9 @@ def _request_shutdown(signum, _frame): os.remove(stop_file) shutdown.set() break - shutdown.wait(timeout=0.5) + shutdown.wait( + timeout=constants.INGEST_MONITOR_SHUTDOWN_POLL_INTERVAL_SECONDS + ) elif args.execution_mode == "timed": logger.debug(f"Running for {args.time_to_run} seconds") time.sleep(args.time_to_run) @@ -494,7 +496,12 @@ def _request_shutdown(signum, _frame): convert_query_engine_perf_data(args.experiment_output_dir) logger.debug("Stopping process monitors") - monitor_info = process_monitor.stop_monitor(monitor, control_pipe, monitor_pipe) + monitor_info = process_monitor.stop_monitor( + monitor, + control_pipe, + monitor_pipe, + timeout=constants.PROCESS_MONITOR_STOP_TIMEOUT_SECONDS, + ) monitor_output_file = os.path.join( remote_monitor_output_dir, args.monitor_output_file @@ -598,7 +605,7 @@ def _request_shutdown(signum, _frame): parser.add_argument( "--monitor_interval_seconds", type=float, - default=1.0, + required=True, help=( "Seconds between process/thread CPU+memory samples. Smaller values " "(e.g. 0.1) capture short precompute/query CPU spikes that 1s "