feat(asap-tools): add remote_monitor (CPU/memory) support to experiment_run_clickhouse.py

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(

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,
         )
 

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,
     # )
 

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,15 +157,15 @@ 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 ""
         )
 
         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(

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"
         )