diff --git a/asap-tools/experiments/classes/process_monitor.py b/asap-tools/experiments/classes/process_monitor.py index 4449f53..e42aa48 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,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 diff --git a/asap-tools/experiments/config/config.yaml b/asap-tools/experiments/config/config.yaml index 7b69745..1da9f41 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 2358a3a..7492cf3 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 7bdd370..ae3d607 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,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( 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 28986df..f08a78f 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 @@ -50,6 +52,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 +63,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 +132,7 @@ def start( "--time_to_run {} " "--node_offset {} " "--streaming_engine {} " + "--monitor_interval_seconds {} " ).format( experiment_mode, ",".join(keywords), @@ -136,6 +146,7 @@ def start( timed_duration, self.node_offset, streaming_engine, + monitor_interval_seconds, ) cmd_dir = os.path.join( @@ -176,6 +187,7 @@ def start( "--prometheus_client_output_file {} " "--node_offset {} " "--streaming_engine {} " + "--monitor_interval_seconds {} " ).format( experiment_mode, ",".join(keywords), @@ -189,8 +201,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 +268,193 @@ 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 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"] + 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 {shlex.quote(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, + ) + + @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, + 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, + execution_mode: Optional[str] = None, + experiment_output_dir: Optional[str] = None, + ) -> None: + """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( + execution_mode=execution_mode, + experiment_output_dir=experiment_output_dir, + ): + return + time.sleep(polling_interval) + mode_desc = execution_mode or "any" + raise RuntimeError( + 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, + 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()" + ) + print( + "Waiting for remote monitor to exit " + 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 {shlex.quote(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 {shlex.quote(stop_file)}", + cmd_dir="", + nohup=False, + popen=False, + ignore_errors=True, + ) + def stop(self, **kwargs) -> None: """ Stop remote monitor processes. @@ -261,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, @@ -277,6 +488,9 @@ def wait_for_remote_monitor_to_finish( self, 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. @@ -284,6 +498,9 @@ 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 """ print( "Waiting for {} seconds for remote monitor to finish".format( @@ -293,18 +510,28 @@ 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( - 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 == "": + 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 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 6e802d7..52c9c1e 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,36 @@ 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. + # 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( 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 +403,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 +447,31 @@ 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/SIGINT", + constants.INGEST_MONITOR_STOP_FILE, + ) + 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) @@ -442,8 +499,16 @@ def main(args): 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") @@ -454,8 +519,12 @@ 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 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( @@ -526,6 +595,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: