diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index 2a4855ed..7bdd370d 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: @@ -150,78 +152,46 @@ def _expand_groups(groups): ) -def _run_query_client( - provider, - node_offset: int, - config_file: str, - output_dir: str, +def _run_query_workload( + experiment_mode: str, + experiment_output_dir: str, + controller_client_config: str, + controller_remote_output_dir: str, use_container: bool, parallel: 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: - """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" + """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, + 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, + backend_protocol="clickhouse", ) - - 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" + 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, ) - else: - cmd = ( - f"python3 -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, - ) @hydra.main(version_base=None, config_path="config", config_name="config") @@ -243,6 +213,12 @@ def main(cfg: DictConfig) -> None: no_teardown = cfg.flow.no_teardown use_container = cfg.use_container.prometheus_client parallel = cfg.prometheus_client.parallel + 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 +321,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 +330,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 +450,19 @@ def main(cfg: DictConfig) -> None: "controller_client_configs", f"{experiment_mode}.yaml", ) - _run_query_client( - provider=provider, - node_offset=node_offset, - config_file=controller_client_config, - output_dir=os.path.join( - experiment_output_dir, "prometheus_client_output" - ), + _run_query_workload( + 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_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 +482,19 @@ def main(cfg: DictConfig) -> None: "controller_client_configs", f"{experiment_mode}.yaml", ) - _run_query_client( - provider=provider, - node_offset=node_offset, - config_file=controller_client_config, - output_dir=os.path.join( - experiment_output_dir, "prometheus_client_output" - ), + _run_query_workload( + 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_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_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index 3779c2cb..6edbf074 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -624,7 +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_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 773f1cbe..28986df0 100644 --- a/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py +++ b/asap-tools/experiments/experiment_utils/services/remote_monitor_service.py @@ -43,11 +43,12 @@ 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, + backend_protocol: str, + backend_tool: Optional[str] = None, timed_duration: Optional[int] = None, ) -> None: """ @@ -56,13 +57,15 @@ 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") + 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 - if monitoring_tool == "victoriametrics": + # Determine which config file to look for based on monitoring tool. + # 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 @@ -73,9 +76,21 @@ def start( else: config_keywords = [constants.PROMETHEUS_CONFIG_FILE] - if use_timed_mode: - # Build command for timed mode (skip_querying) - keywords = config_keywords + # Build the list of process keywords to monitor. + 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 + # ``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 experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: if query_engine_service is not None: @@ -95,6 +110,8 @@ def start( else: keywords.append("arroyo.*worker") + if use_timed_mode: + # Build command for timed mode (skip_querying) cmd = ( "python3 -u remote_monitor.py " "--execution_mode timed " @@ -148,26 +165,6 @@ def start( # Original prometheus_client mode logic assert controller_remote_output_dir is not None - keywords = config_keywords - - if 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 -u remote_monitor.py " "--execution_mode prometheus_client " @@ -221,6 +218,8 @@ def start( if profile_prometheus_time is not None: cmd += " --profile_prometheus_time {}".format(profile_prometheus_time) + 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 a92a50d2..6e802d76 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_protocol=args.backend_protocol, ) 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_protocol", + type=str, + required=True, + help="Query backend for prometheus-client (prometheus or clickhouse)", + ) args = parser.parse_args() args.keywords = args.keywords.strip().split(",") if args.profile_flink_pids: