From 79024a93fb204f384094dbb2edf02201b145ac43 Mon Sep 17 00:00:00 2001 From: Akanksha Akkihal Date: Mon, 29 Jun 2026 19:00:54 -0400 Subject: [PATCH 1/7] add remote_monitor (CPU/memory) support to experiment_run_clickhouse.py --- asap-tools/experiments/config/config.yaml | 1 + .../experiments/experiment_run_clickhouse.py | 122 ++++++++++++++++-- .../services/remote_monitor_service.py | 26 +++- asap-tools/experiments/remote_monitor.py | 14 +- 4 files changed, 143 insertions(+), 20 deletions(-) diff --git a/asap-tools/experiments/config/config.yaml b/asap-tools/experiments/config/config.yaml index 7b697450..f85289a7 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 + remote_monitor: false # ClickHouse runner: collect CPU/memory via remote_monitor.py skip_copy_prometheus_data: false # Skip copying Prometheus data back to local machine diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 2a4855ed..6e56d243 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -109,8 +109,10 @@ ) from experiment_utils.services.misc import ControllerService, DiscoveryBackend from experiment_utils.services.query_engine import QueryEngineRustService +from experiment_utils.services.remote_monitor_service import RemoteMonitorService CLICKHOUSE_DATABASE = "default" +REMOTE_PROCESS_POLLING_INTERVAL = 10 def _inline_sql_queries_in_experiment_config(local_experiment_root_dir: str) -> None: @@ -207,7 +209,7 @@ def _run_query_client( ) else: cmd = ( - f"python3 -u main_prometheus_client.py" + f"python3.11 -u main_prometheus_client.py" f" --config_file {config_file}" f" --output_dir {output_dir}" f" --output_file prometheus_client_output.txt" @@ -224,6 +226,81 @@ def _run_query_client( ) +def _clickhouse_monitor_keywords(experiment_mode: str) -> list[str]: + """Process keywords for remote_monitor on ClickHouse experiments. + + Use ps-friendly patterns (not Docker container names): host-network + ClickHouse does not show up as ``clickhouse-server`` in ``ps``, and + ``docker inspect`` can fail in background SSH shells. + """ + if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: + return [constants.QUERY_ENGINE_RS_PROCESS_KEYWORD] + return ["clickhouse"] + + +def _run_query_workload( + *, + provider, + node_offset: int, + experiment_mode: str, + experiment_output_dir: str, + controller_client_config: str, + controller_remote_output_dir: str, + use_container: bool, + parallel: bool, + remote_monitor_enabled: bool, + remote_monitor_service: RemoteMonitorService, + minimum_experiment_running_time: int, + manual_remote_monitor: bool, + query_engine_service: QueryEngineRustService | None, + profile_query_engine: bool, + profile_prometheus_time, +) -> None: + """Run the SQL query workload, optionally wrapped in remote_monitor.""" + prometheus_client_output_dir = os.path.join( + experiment_output_dir, "prometheus_client_output" + ) + + if not remote_monitor_enabled: + _run_query_client( + provider=provider, + node_offset=node_offset, + config_file=controller_client_config, + output_dir=prometheus_client_output_dir, + use_container=use_container, + parallel=parallel, + ) + return + + remote_monitor_service.start( + controller_client_config=controller_client_config, + experiment_output_dir=experiment_output_dir, + experiment_mode=experiment_mode, + profile_query_engine=profile_query_engine, + profile_prometheus_time=profile_prometheus_time, + profile_flink=False, + flink_pids=None, + profile_arroyo=False, + arroyo_pids=None, + manual_mode=manual_remote_monitor, + do_local_flink=False, + streaming_engine="precompute", + query_engine_service=query_engine_service, + arroyo_service=None, + controller_remote_output_dir=controller_remote_output_dir, + use_container_prometheus_client=use_container, + prometheus_client_parallel=parallel, + monitoring_tool="prometheus", + backend_type="clickhouse", + monitor_keywords=_clickhouse_monitor_keywords(experiment_mode), + ) + 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, + polling_interval=REMOTE_PROCESS_POLLING_INTERVAL, + ) + + @hydra.main(version_base=None, config_path="config", config_name="config") def main(cfg: DictConfig) -> None: config.validate_basic_config( @@ -243,6 +320,13 @@ def main(cfg: DictConfig) -> None: no_teardown = cfg.flow.no_teardown use_container = cfg.use_container.prometheus_client parallel = cfg.prometheus_client.parallel + remote_monitor_enabled = bool(cfg.flow.get("remote_monitor", False)) + manual_remote_monitor = bool(cfg.manual.remote_monitor) + profile_query_engine = bool(cfg.profiling.query_engine) + profile_prometheus_time = cfg.profiling.prometheus_time + minimum_experiment_running_time = config.get_minimum_experiment_running_time( + cfg.experiment_params + ) if provider.is_remote(): local_experiment_root_dir = os.path.join( @@ -345,6 +429,7 @@ def main(cfg: DictConfig) -> None: 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}") @@ -353,6 +438,7 @@ def main(cfg: DictConfig) -> None: # mirroring the prometheus_client_service.stop() call at the top of the # e2e mode loop. prometheus_client_service.stop() + remote_monitor_service.stop() experiment_output_dir = os.path.join( experiment_root_output_dir, experiment_mode @@ -472,15 +558,22 @@ def main(cfg: DictConfig) -> None: "controller_client_configs", f"{experiment_mode}.yaml", ) - _run_query_client( + _run_query_workload( provider=provider, node_offset=node_offset, - config_file=controller_client_config, - output_dir=os.path.join( - experiment_output_dir, "prometheus_client_output" - ), + experiment_mode=experiment_mode, + experiment_output_dir=experiment_output_dir, + controller_client_config=controller_client_config, + controller_remote_output_dir=remote_controller_dir, use_container=use_container, parallel=parallel, + remote_monitor_enabled=remote_monitor_enabled, + remote_monitor_service=remote_monitor_service, + minimum_experiment_running_time=minimum_experiment_running_time, + manual_remote_monitor=manual_remote_monitor, + query_engine_service=query_engine_service, + profile_query_engine=profile_query_engine, + profile_prometheus_time=profile_prometheus_time, ) sync.rsync_experiment_data( @@ -500,15 +593,22 @@ def main(cfg: DictConfig) -> None: "controller_client_configs", f"{experiment_mode}.yaml", ) - _run_query_client( + _run_query_workload( provider=provider, node_offset=node_offset, - config_file=controller_client_config, - output_dir=os.path.join( - experiment_output_dir, "prometheus_client_output" - ), + experiment_mode=experiment_mode, + experiment_output_dir=experiment_output_dir, + controller_client_config=controller_client_config, + controller_remote_output_dir=experiment_root_output_dir, use_container=use_container, parallel=parallel, + remote_monitor_enabled=remote_monitor_enabled, + remote_monitor_service=remote_monitor_service, + minimum_experiment_running_time=minimum_experiment_running_time, + manual_remote_monitor=manual_remote_monitor, + query_engine_service=None, + profile_query_engine=profile_query_engine, + profile_prometheus_time=profile_prometheus_time, ) 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 773f1cbe..5546a3dc 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -43,12 +43,14 @@ def start( do_local_flink: bool, streaming_engine: str, query_engine_service: "BaseQueryEngineService", - arroyo_service: "ArroyoService", + arroyo_service: Optional["ArroyoService"], controller_remote_output_dir: str, use_container_prometheus_client: bool, prometheus_client_parallel: bool, monitoring_tool: str, timed_duration: Optional[int] = None, + backend_type: str = "prometheus", + monitor_keywords: Optional[List[str]] = None, ) -> None: """ Start remote monitor processes. @@ -75,9 +77,13 @@ def start( if use_timed_mode: # Build command for timed mode (skip_querying) - keywords = config_keywords + keywords = ( + list(monitor_keywords) + if monitor_keywords is not None + else list(config_keywords) + ) - if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: + if monitor_keywords is None and experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: if query_engine_service is not None: keywords.append(query_engine_service.get_monitoring_keyword()) else: @@ -96,7 +102,7 @@ def start( keywords.append("arroyo.*worker") cmd = ( - "python3 -u remote_monitor.py " + "python3.11 -u remote_monitor.py " "--execution_mode timed " "--experiment_mode {} " r"--keywords \"{}\" " @@ -148,9 +154,13 @@ def start( # Original prometheus_client mode logic assert controller_remote_output_dir is not None - keywords = config_keywords + keywords = ( + list(monitor_keywords) + if monitor_keywords is not None + else list(config_keywords) + ) - if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: + if monitor_keywords is None and experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: if query_engine_service is not None: keywords.append(query_engine_service.get_monitoring_keyword()) else: @@ -169,7 +179,7 @@ def start( keywords.append("arroyo.*worker") cmd = ( - "python3 -u remote_monitor.py " + "python3.11 -u remote_monitor.py " "--execution_mode prometheus_client " "--experiment_mode {} " r"--keywords \"{}\" " @@ -221,6 +231,8 @@ def start( if profile_prometheus_time is not None: cmd += " --profile_prometheus_time {}".format(profile_prometheus_time) + cmd += " --backend_type {}".format(backend_type) + cmd_dir = os.path.join( self.provider.get_home_dir(), "code", "asap-tools", "experiments" ) diff --git a/asap-tools/experiments/remote_monitor.py b/asap-tools/experiments/remote_monitor.py index a92a50d2..4b59032e 100644 --- a/asap-tools/experiments/remote_monitor.py +++ b/asap-tools/experiments/remote_monitor.py @@ -295,7 +295,11 @@ def main(args): pids = [] keywords_expanded = [] for keyword in args.keywords: - keyword_pids = get_pids(keyword) + try: + keyword_pids = get_pids(keyword) + except ValueError as exc: + logger.warning("Skipping keyword {}: {}", keyword, exc) + continue pids.extend(keyword_pids) keywords_expanded.extend([keyword] * len(keyword_pids)) # pid_keyword_map[pid] = keyword @@ -395,7 +399,7 @@ def main(args): profile_query_engine_pid, args.profile_prometheus_time, args.prometheus_client_parallel, - backend_type="prometheus", + backend_type=args.backend_type, ) if prometheus_client_service.use_container: @@ -516,6 +520,12 @@ def main(args): required=True, help="Streaming engine type (e.g. precompute, arroyo)", ) + parser.add_argument( + "--backend_type", + type=str, + default="prometheus", + help="Query backend for prometheus-client (prometheus or clickhouse)", + ) args = parser.parse_args() args.keywords = args.keywords.strip().split(",") if args.profile_flink_pids: From aed589e541eae40bd67501680d92a6111d82888d Mon Sep 17 00:00:00 2001 From: Akanksha Akkihal Date: Mon, 29 Jun 2026 19:08:16 -0400 Subject: [PATCH 2/7] style(tools): black formatting for remote_monitor_service.py --- .../services/remote_monitor_service.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) 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 5546a3dc..201e763e 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -83,7 +83,10 @@ def start( else list(config_keywords) ) - if monitor_keywords is None and experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: + if ( + monitor_keywords is None + and experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME + ): if query_engine_service is not None: keywords.append(query_engine_service.get_monitoring_keyword()) else: @@ -160,7 +163,10 @@ def start( else list(config_keywords) ) - if monitor_keywords is None and experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: + if ( + monitor_keywords is None + and experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME + ): if query_engine_service is not None: keywords.append(query_engine_service.get_monitoring_keyword()) else: From a67d2be3736fe365f9a166585244e8b099a8fe6b Mon Sep 17 00:00:00 2001 From: Akanksha Akkihal Date: Tue, 30 Jun 2026 10:32:25 -0400 Subject: [PATCH 3/7] refactor(tools): address review on remote_monitor for clickhouse experiments - remote_monitor.py: make --backend_type required (drop default). - remote_monitor_service.py: make backend_type required; consolidate process keyword selection in one place (clickhouse vs prometheus), dropping the monitor_keywords argument; revert remote_monitor invocation to python3. - experiment_run_clickhouse.py: remove flow.remote_monitor flag (monitoring is always on); drop now-dead _run_query_client and _clickhouse_monitor_keywords. - config.yaml: remove flow.remote_monitor flag. - experiment_run_e2e.py: pass backend_type=prometheus explicitly. Co-authored-by: Cursor --- asap-tools/experiments/config/config.yaml | 1 - .../experiments/experiment_run_clickhouse.py | 114 +----------------- asap-tools/experiments/experiment_run_e2e.py | 1 + .../services/remote_monitor_service.py | 63 ++++------ asap-tools/experiments/remote_monitor.py | 2 +- 5 files changed, 24 insertions(+), 157 deletions(-) diff --git a/asap-tools/experiments/config/config.yaml b/asap-tools/experiments/config/config.yaml index f85289a7..7b697450 100644 --- a/asap-tools/experiments/config/config.yaml +++ b/asap-tools/experiments/config/config.yaml @@ -47,7 +47,6 @@ flow: no_teardown: false replace_query_engine_with_dumb_consumer: false steady_state_wait: 60 - remote_monitor: false # ClickHouse runner: collect CPU/memory via remote_monitor.py skip_copy_prometheus_data: false # Skip copying Prometheus data back to local machine diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 6e56d243..87b6443d 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -152,103 +152,14 @@ def _expand_groups(groups): ) -def _run_query_client( - provider, - node_offset: int, - config_file: str, - output_dir: str, - use_container: bool, - parallel: bool, -) -> None: - """SSH to the node and run the prometheus-client, blocking until done. - - For bare-metal: runs main_prometheus_client.py directly. - For containerized: generates a docker-compose file then runs - `docker compose up --no-build` (foreground, exits when container exits). - """ - home_dir = provider.get_home_dir() - prometheus_client_dir = os.path.join( - home_dir, "code", "asap-tools", "queriers", "prometheus-client" - ) - - if use_container: - helper_script = os.path.join( - home_dir, - "code", - "asap-tools", - "experiments", - "generate_prometheus_client_compose.py", - ) - template_path = os.path.join(prometheus_client_dir, "docker-compose.yml.j2") - remote_compose_file = os.path.join( - output_dir, "prometheus-client-docker-compose.yml" - ) - node_ip = provider.get_node_ip(node_offset) - - gen_compose_cmd = ( - f"python3 {helper_script}" - f" --template-path {template_path}" - f" --compose-output-path {remote_compose_file}" - f" --prometheusclient-dir {prometheus_client_dir}" - f" --container-name sketchdb-prometheusclient" - f" --experiment-output-dir {output_dir}" - f" --config-file {config_file}" - f" --client-output-dir {output_dir}" - f" --client-output-file prometheus_client_output.txt" - f" --prometheus-host {node_ip}" - f" --sketchdb-host {node_ip}" - ) - if parallel: - gen_compose_cmd += " --parallel" - - # docker compose up without -d: foreground, blocks until container exits - cmd = ( - f"mkdir -p {output_dir}; " - f"{gen_compose_cmd}; " - f"docker compose -f {remote_compose_file} up --no-build" - ) - else: - cmd = ( - f"python3.11 -u main_prometheus_client.py" - f" --config_file {config_file}" - f" --output_dir {output_dir}" - f" --output_file prometheus_client_output.txt" - ) - if parallel: - cmd += " --parallel" - - provider.execute_command( - node_idx=node_offset, - cmd=cmd, - cmd_dir=prometheus_client_dir, - nohup=False, - popen=False, - ) - - -def _clickhouse_monitor_keywords(experiment_mode: str) -> list[str]: - """Process keywords for remote_monitor on ClickHouse experiments. - - Use ps-friendly patterns (not Docker container names): host-network - ClickHouse does not show up as ``clickhouse-server`` in ``ps``, and - ``docker inspect`` can fail in background SSH shells. - """ - if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: - return [constants.QUERY_ENGINE_RS_PROCESS_KEYWORD] - return ["clickhouse"] - - def _run_query_workload( *, - provider, - node_offset: int, experiment_mode: str, experiment_output_dir: str, controller_client_config: str, controller_remote_output_dir: str, use_container: bool, parallel: bool, - remote_monitor_enabled: bool, remote_monitor_service: RemoteMonitorService, minimum_experiment_running_time: int, manual_remote_monitor: bool, @@ -256,22 +167,7 @@ def _run_query_workload( profile_query_engine: bool, profile_prometheus_time, ) -> None: - """Run the SQL query workload, optionally wrapped in remote_monitor.""" - prometheus_client_output_dir = os.path.join( - experiment_output_dir, "prometheus_client_output" - ) - - if not remote_monitor_enabled: - _run_query_client( - provider=provider, - node_offset=node_offset, - config_file=controller_client_config, - output_dir=prometheus_client_output_dir, - use_container=use_container, - parallel=parallel, - ) - return - + """Run the SQL query workload wrapped in remote_monitor (CPU/memory).""" remote_monitor_service.start( controller_client_config=controller_client_config, experiment_output_dir=experiment_output_dir, @@ -292,7 +188,6 @@ def _run_query_workload( prometheus_client_parallel=parallel, monitoring_tool="prometheus", backend_type="clickhouse", - monitor_keywords=_clickhouse_monitor_keywords(experiment_mode), ) if not manual_remote_monitor and constants.AVOID_REMOTE_MONITOR_LONG_SSH: remote_monitor_service.wait_for_remote_monitor_to_finish( @@ -320,7 +215,6 @@ def main(cfg: DictConfig) -> None: no_teardown = cfg.flow.no_teardown use_container = cfg.use_container.prometheus_client parallel = cfg.prometheus_client.parallel - remote_monitor_enabled = bool(cfg.flow.get("remote_monitor", False)) manual_remote_monitor = bool(cfg.manual.remote_monitor) profile_query_engine = bool(cfg.profiling.query_engine) profile_prometheus_time = cfg.profiling.prometheus_time @@ -559,15 +453,12 @@ def main(cfg: DictConfig) -> None: f"{experiment_mode}.yaml", ) _run_query_workload( - provider=provider, - node_offset=node_offset, experiment_mode=experiment_mode, experiment_output_dir=experiment_output_dir, controller_client_config=controller_client_config, controller_remote_output_dir=remote_controller_dir, use_container=use_container, parallel=parallel, - remote_monitor_enabled=remote_monitor_enabled, remote_monitor_service=remote_monitor_service, minimum_experiment_running_time=minimum_experiment_running_time, manual_remote_monitor=manual_remote_monitor, @@ -594,15 +485,12 @@ def main(cfg: DictConfig) -> None: f"{experiment_mode}.yaml", ) _run_query_workload( - provider=provider, - node_offset=node_offset, experiment_mode=experiment_mode, experiment_output_dir=experiment_output_dir, controller_client_config=controller_client_config, controller_remote_output_dir=experiment_root_output_dir, use_container=use_container, parallel=parallel, - remote_monitor_enabled=remote_monitor_enabled, remote_monitor_service=remote_monitor_service, minimum_experiment_running_time=minimum_experiment_running_time, manual_remote_monitor=manual_remote_monitor, diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index 3779c2cb..f4391447 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -625,6 +625,7 @@ def main(cfg: DictConfig): use_container_prometheus_client=args.use_container_prometheus_client, prometheus_client_parallel=args.prometheus_client_parallel, monitoring_tool=cfg.experiment_params.monitoring.tool, + backend_type="prometheus", 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 201e763e..228f2b64 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -48,9 +48,8 @@ def start( use_container_prometheus_client: bool, prometheus_client_parallel: bool, monitoring_tool: str, + backend_type: str, timed_duration: Optional[int] = None, - backend_type: str = "prometheus", - monitor_keywords: Optional[List[str]] = None, ) -> None: """ Start remote monitor processes. @@ -75,18 +74,23 @@ def start( else: config_keywords = [constants.PROMETHEUS_CONFIG_FILE] - if use_timed_mode: - # Build command for timed mode (skip_querying) - keywords = ( - list(monitor_keywords) - if monitor_keywords is not None - else list(config_keywords) - ) + # Build the list of process keywords to monitor. + if backend_type == "clickhouse": + # ClickHouse/SQL experiments: use ps-friendly process names rather + # than Docker container names. Host-network bare-metal ClickHouse + # does not show up as ``clickhouse-server`` in ``ps``, and + # ``docker inspect`` can fail in background SSH shells. + if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: + if query_engine_service is not None: + keywords = [query_engine_service.get_monitoring_keyword()] + else: + keywords = [constants.QUERY_ENGINE_RS_PROCESS_KEYWORD] + else: + keywords = ["clickhouse"] + else: + keywords = list(config_keywords) - if ( - monitor_keywords is None - and experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME - ): + if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: if query_engine_service is not None: keywords.append(query_engine_service.get_monitoring_keyword()) else: @@ -104,8 +108,10 @@ def start( else: keywords.append("arroyo.*worker") + if use_timed_mode: + # Build command for timed mode (skip_querying) cmd = ( - "python3.11 -u remote_monitor.py " + "python3 -u remote_monitor.py " "--execution_mode timed " "--experiment_mode {} " r"--keywords \"{}\" " @@ -157,35 +163,8 @@ def start( # Original prometheus_client mode logic assert controller_remote_output_dir is not None - keywords = ( - list(monitor_keywords) - if monitor_keywords is not None - else list(config_keywords) - ) - - if ( - monitor_keywords is None - and experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME - ): - if query_engine_service is not None: - keywords.append(query_engine_service.get_monitoring_keyword()) - else: - keywords.append(constants.QUERY_ENGINE_RS_PROCESS_KEYWORD) - - if streaming_engine == "flink": - keywords.append("sketch-0.1.jar") # flinksketch jar - if not do_local_flink: - keywords.append( - "org.apache.flink.runtime.taskexecutor.TaskManagerRunner" - ) - elif streaming_engine == "arroyo": - if arroyo_service is not None: - keywords.append(arroyo_service.get_monitoring_keyword()) - else: - keywords.append("arroyo.*worker") - cmd = ( - "python3.11 -u remote_monitor.py " + "python3 -u remote_monitor.py " "--execution_mode prometheus_client " "--experiment_mode {} " r"--keywords \"{}\" " diff --git a/asap-tools/experiments/remote_monitor.py b/asap-tools/experiments/remote_monitor.py index 4b59032e..7dbd46c5 100644 --- a/asap-tools/experiments/remote_monitor.py +++ b/asap-tools/experiments/remote_monitor.py @@ -523,7 +523,7 @@ def main(args): parser.add_argument( "--backend_type", type=str, - default="prometheus", + required=True, help="Query backend for prometheus-client (prometheus or clickhouse)", ) args = parser.parse_args() From 9bb1b8d6c2422f52311662b6c7de4b14c4c8b915 Mon Sep 17 00:00:00 2001 From: Akanksha Akkihal Date: Tue, 30 Jun 2026 11:57:53 -0400 Subject: [PATCH 4/7] Review changes --- asap-tools/experiments/experiment_run_clickhouse.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 87b6443d..26527ad0 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -153,7 +153,6 @@ def _expand_groups(groups): def _run_query_workload( - *, experiment_mode: str, experiment_output_dir: str, controller_client_config: str, @@ -186,6 +185,9 @@ def _run_query_workload( controller_remote_output_dir=controller_remote_output_dir, use_container_prometheus_client=use_container, prometheus_client_parallel=parallel, + # monitoring_tool only selects the Prometheus/VictoriaMetrics scrape-config + # keyword, which is unused on the ClickHouse path: process keywords are + # chosen from backend_type="clickhouse" below. Harmless placeholder. monitoring_tool="prometheus", backend_type="clickhouse", ) From f09872d9438504d8500fb9f0089e4f8edc3ebaac Mon Sep 17 00:00:00 2001 From: Akanksha Akkihal Date: Tue, 30 Jun 2026 12:02:51 -0400 Subject: [PATCH 5/7] Review changes --- asap-tools/experiments/experiment_run_clickhouse.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 26527ad0..ed70ab2c 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -187,8 +187,8 @@ def _run_query_workload( prometheus_client_parallel=parallel, # monitoring_tool only selects the Prometheus/VictoriaMetrics scrape-config # keyword, which is unused on the ClickHouse path: process keywords are - # chosen from backend_type="clickhouse" below. Harmless placeholder. - monitoring_tool="prometheus", + # chosen from backend_type="clickhouse" below. + monitoring_tool="clickhouse", backend_type="clickhouse", ) if not manual_remote_monitor and constants.AVOID_REMOTE_MONITOR_LONG_SSH: From bb4c5404df174394379b63374286505b9c7236ab Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 30 Jun 2026 21:04:50 -0400 Subject: [PATCH 6/7] simplified logic --- asap-tools/experiments/experiment_run_clickhouse.py | 4 ---- .../experiment_utils/services/remote_monitor_service.py | 8 +++++--- 2 files changed, 5 insertions(+), 7 deletions(-) diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index ed70ab2c..c72d23a5 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -185,10 +185,6 @@ def _run_query_workload( controller_remote_output_dir=controller_remote_output_dir, use_container_prometheus_client=use_container, prometheus_client_parallel=parallel, - # monitoring_tool only selects the Prometheus/VictoriaMetrics scrape-config - # keyword, which is unused on the ClickHouse path: process keywords are - # chosen from backend_type="clickhouse" below. - monitoring_tool="clickhouse", backend_type="clickhouse", ) if not manual_remote_monitor and constants.AVOID_REMOTE_MONITOR_LONG_SSH: 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 228f2b64..70f599e7 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -47,8 +47,8 @@ def start( controller_remote_output_dir: str, use_container_prometheus_client: bool, prometheus_client_parallel: bool, - monitoring_tool: str, backend_type: str, + monitoring_tool: Optional[str] = None, timed_duration: Optional[int] = None, ) -> None: """ @@ -57,12 +57,14 @@ def start( Args: **kwargs: Additional configuration (currently unused) timed_duration: If provided, use timed mode instead of prometheus_client mode - monitoring_tool: Monitoring tool being used ("prometheus" or "victoriametrics") + monitoring_tool: TSDB running on the node ("prometheus" or "victoriametrics"). + Not used when backend_type="clickhouse". """ # Determine execution mode use_timed_mode = timed_duration is not None - # Determine which config file to look for based on monitoring tool + # Determine which config file to look for based on monitoring tool. + # Only relevant when backend_type != "clickhouse". if monitoring_tool == "victoriametrics": # first one is for vmagent # second one is for vmsingle From 684a60b24cb7dd079b56de58929f685923c45cd6 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 30 Jun 2026 21:08:38 -0400 Subject: [PATCH 7/7] renamed variables --- .../experiments/experiment_run_clickhouse.py | 2 +- asap-tools/experiments/experiment_run_e2e.py | 4 ++-- .../experiments/experiment_run_grafana_demo.py | 2 +- .../services/prometheus_client_service.py | 14 +++++++------- .../services/remote_monitor_service.py | 16 ++++++++-------- asap-tools/experiments/remote_monitor.py | 4 ++-- 6 files changed, 21 insertions(+), 21 deletions(-) diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index c72d23a5..7bdd370d 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -185,7 +185,7 @@ def _run_query_workload( controller_remote_output_dir=controller_remote_output_dir, use_container_prometheus_client=use_container, prometheus_client_parallel=parallel, - backend_type="clickhouse", + backend_protocol="clickhouse", ) if not manual_remote_monitor and constants.AVOID_REMOTE_MONITOR_LONG_SSH: remote_monitor_service.wait_for_remote_monitor_to_finish( diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index f4391447..6edbf074 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -624,8 +624,8 @@ 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, - monitoring_tool=cfg.experiment_params.monitoring.tool, - backend_type="prometheus", + backend_tool=cfg.experiment_params.monitoring.tool, + backend_protocol="prometheus", timed_duration=minimum_experiment_running_time if skip_querying else None, ) diff --git a/asap-tools/experiments/experiment_run_grafana_demo.py b/asap-tools/experiments/experiment_run_grafana_demo.py index 906fa3bd..b7383bad 100644 --- a/asap-tools/experiments/experiment_run_grafana_demo.py +++ b/asap-tools/experiments/experiment_run_grafana_demo.py @@ -554,7 +554,7 @@ 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, - # monitoring_tool=cfg.experiment_params.monitoring.tool, + # backend_tool=cfg.experiment_params.monitoring.tool, # timed_duration=None, # ) diff --git a/asap-tools/experiments/experiment_utils/services/prometheus_client_service.py b/asap-tools/experiments/experiment_utils/services/prometheus_client_service.py index 9d763429..6ec08351 100644 --- a/asap-tools/experiments/experiment_utils/services/prometheus_client_service.py +++ b/asap-tools/experiments/experiment_utils/services/prometheus_client_service.py @@ -39,7 +39,7 @@ def start( profile_query_engine_pid: Optional[int], profile_prometheus_time: Optional[int], parallel: bool, - backend_type: str, + backend_protocol: str, **kwargs, ): if self.use_container: @@ -53,7 +53,7 @@ def start( profile_query_engine_pid, profile_prometheus_time, parallel, - backend_type, + backend_protocol, ) else: return self._start_bare_metal( @@ -66,7 +66,7 @@ def start( profile_query_engine_pid, profile_prometheus_time, parallel, - backend_type, + backend_protocol, ) def _start_containerized( @@ -80,7 +80,7 @@ def _start_containerized( profile_query_engine_pid: Optional[int], profile_prometheus_time: Optional[int], parallel: bool, - backend_type: str, + backend_protocol: str, ): prometheus_client_dir = os.path.join( self.provider.get_home_dir(), @@ -122,7 +122,7 @@ def _start_containerized( if ( experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME - and backend_type != "clickhouse" + and backend_protocol != "clickhouse" ): assert query_engine_config_file is not None gen_compose_cmd += f" --align-query-time --server-for-alignment sketchdb --query-engine-config-file {query_engine_config_file}" @@ -157,7 +157,7 @@ def _start_bare_metal( profile_query_engine_pid: Optional[int], profile_prometheus_time: Optional[int], parallel: bool, - backend_type: str, + backend_protocol: str, ): cmd = "python3 -u main_prometheus_client.py --config_file {} --output_dir {} --output_file {}{}".format( config_file, output_dir, output_file, " --parallel" if parallel else "" @@ -165,7 +165,7 @@ def _start_bare_metal( if ( experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME - and backend_type != "clickhouse" + and backend_protocol != "clickhouse" ): assert query_engine_config_file is not None cmd += " --align_query_time --server_for_alignment sketchdb --query_engine_config_file {}".format( 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 70f599e7..28986df0 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -47,8 +47,8 @@ def start( controller_remote_output_dir: str, use_container_prometheus_client: bool, prometheus_client_parallel: bool, - backend_type: str, - monitoring_tool: Optional[str] = None, + backend_protocol: str, + backend_tool: Optional[str] = None, timed_duration: Optional[int] = None, ) -> None: """ @@ -57,15 +57,15 @@ def start( Args: **kwargs: Additional configuration (currently unused) timed_duration: If provided, use timed mode instead of prometheus_client mode - monitoring_tool: TSDB running on the node ("prometheus" or "victoriametrics"). - Not used when backend_type="clickhouse". + backend_tool: TSDB running on the node ("prometheus" or "victoriametrics"). + Not used when backend_protocol="clickhouse". """ # Determine execution mode use_timed_mode = timed_duration is not None # Determine which config file to look for based on monitoring tool. - # Only relevant when backend_type != "clickhouse". - if monitoring_tool == "victoriametrics": + # Only relevant when backend_protocol != "clickhouse". + if backend_tool == "victoriametrics": # first one is for vmagent # second one is for vmsingle # TODO: remove this hardcoding and instead query the service to get this @@ -77,7 +77,7 @@ def start( config_keywords = [constants.PROMETHEUS_CONFIG_FILE] # Build the list of process keywords to monitor. - if backend_type == "clickhouse": + if backend_protocol == "clickhouse": # ClickHouse/SQL experiments: use ps-friendly process names rather # than Docker container names. Host-network bare-metal ClickHouse # does not show up as ``clickhouse-server`` in ``ps``, and @@ -218,7 +218,7 @@ def start( if profile_prometheus_time is not None: cmd += " --profile_prometheus_time {}".format(profile_prometheus_time) - cmd += " --backend_type {}".format(backend_type) + cmd += " --backend_protocol {}".format(backend_protocol) cmd_dir = os.path.join( self.provider.get_home_dir(), "code", "asap-tools", "experiments" diff --git a/asap-tools/experiments/remote_monitor.py b/asap-tools/experiments/remote_monitor.py index 7dbd46c5..6e802d76 100644 --- a/asap-tools/experiments/remote_monitor.py +++ b/asap-tools/experiments/remote_monitor.py @@ -399,7 +399,7 @@ def main(args): profile_query_engine_pid, args.profile_prometheus_time, args.prometheus_client_parallel, - backend_type=args.backend_type, + backend_protocol=args.backend_protocol, ) if prometheus_client_service.use_container: @@ -521,7 +521,7 @@ def main(args): help="Streaming engine type (e.g. precompute, arroyo)", ) parser.add_argument( - "--backend_type", + "--backend_protocol", type=str, required=True, help="Query backend for prometheus-client (prometheus or clickhouse)",