From 290aad9676a721e7b4e54cedbda8bb42a121d66a Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 24 Jun 2026 16:32:25 -0400 Subject: [PATCH 1/2] feat(tools): added support in asap-tools for running benchmarks experiments locally (instead of Cloudlab) --- .../experiments/experiment_run_clickhouse.py | 32 +++-- asap-tools/experiments/experiment_run_e2e.py | 34 +++-- .../experiments/experiment_utils/config.py | 38 ++++-- .../experiment_utils/providers/base.py | 13 ++ .../experiment_utils/providers/factory.py | 21 ++-- .../experiment_utils/providers/local.py | 119 ++++++++++++++++++ .../experiment_utils/services/arroyo.py | 15 ++- .../experiment_utils/services/base.py | 6 - .../services/clickhouse_service.py | 41 ++++-- .../services/docker_victoriametrics.py | 19 +-- .../services/fake_exporters.py | 5 +- .../experiment_utils/services/prometheus.py | 33 +++-- .../experiment_utils/services/query_engine.py | 5 + .../experiments/experiment_utils/sync.py | 22 +++- .../experiment_utils/test_local_provider.py | 88 +++++++++++++ 15 files changed, 400 insertions(+), 91 deletions(-) create mode 100644 asap-tools/experiments/experiment_utils/providers/local.py create mode 100644 asap-tools/experiments/experiment_utils/test_local_provider.py diff --git a/asap-tools/experiments/experiment_run_clickhouse.py b/asap-tools/experiments/experiment_run_clickhouse.py index e49b3fed..d452e062 100644 --- a/asap-tools/experiments/experiment_run_clickhouse.py +++ b/asap-tools/experiments/experiment_run_clickhouse.py @@ -233,10 +233,8 @@ def main(cfg: DictConfig) -> None: cfg, required_params=[ ("experiment.name", "Human-readable experiment name"), - ("cloudlab.num_nodes", "Number of CloudLab nodes to use"), - ("cloudlab.username", "Your CloudLab username"), - ("cloudlab.hostname_suffix", "CloudLab experiment hostname suffix"), - ], + ] + + config.required_cloudlab_params(cfg), script_name="experiment_run_clickhouse", ) config.validate_experiment_config(cfg.experiment_params) @@ -244,14 +242,25 @@ def main(cfg: DictConfig) -> None: provider = create_provider(cfg) experiment_name = cfg.experiment.name - node_offset = cfg.cloudlab.node_offset + num_nodes, node_offset = config.get_node_params(cfg) no_teardown = cfg.flow.no_teardown use_container = cfg.use_container.prometheus_client parallel = cfg.prometheus_client.parallel - local_experiment_root_dir = os.path.join( - constants.LOCAL_EXPERIMENT_DIR, experiment_name - ) + if provider.is_remote(): + local_experiment_root_dir = os.path.join( + constants.LOCAL_EXPERIMENT_DIR, experiment_name + ) + experiment_root_output_dir = os.path.join( + provider.get_home_dir(), "experiment_outputs", experiment_name + ) + else: + # Local mode: "remote" and "local" roots are the same filesystem path, + # so every rsync/scp call downstream becomes a no-op. + local_experiment_root_dir = os.path.join( + provider.get_home_dir(), "experiment_outputs", experiment_name + ) + experiment_root_output_dir = local_experiment_root_dir os.makedirs(local_experiment_root_dir, exist_ok=True) with open(os.path.join(local_experiment_root_dir, "hydra_config.yaml"), "w") as f: @@ -259,9 +268,6 @@ def main(cfg: DictConfig) -> None: with open(os.path.join(local_experiment_root_dir, "cmdline_args.txt"), "w") as f: json.dump({"experiment_name": experiment_name, "node_offset": node_offset}, f) - experiment_root_output_dir = ( - f"{constants.CLOUDLAB_HOME_DIR}/experiment_outputs/{experiment_name}" - ) provider.execute_command( node_idx=node_offset, cmd=f"mkdir -p {experiment_root_output_dir}", @@ -314,7 +320,7 @@ def main(cfg: DictConfig) -> None: # --- start ClickHouse (persists across all modes) --- clickhouse_service = ClickHouseService( - provider, num_nodes=cfg.cloudlab.num_nodes, node_offset=node_offset + provider, num_nodes=num_nodes, node_offset=node_offset ) clickhouse_service.start( experiment_output_dir=experiment_root_output_dir, @@ -326,7 +332,7 @@ def main(cfg: DictConfig) -> None: # --- load data once before the mode loop (DROP + reload) --- data_loader = ClickHouseDataLoaderService( provider, - num_nodes=cfg.cloudlab.num_nodes, + num_nodes=num_nodes, node_offset=node_offset, clickhouse_http_port=clickhouse_http_port, ) diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index 1e470073..4e841d97 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -59,15 +59,35 @@ def main(cfg: DictConfig): config.validate_config(cfg) # Validate experiment configuration config.validate_experiment_config(cfg.experiment_params) - # Convert config to args-like object for backward compatibility - args = config.Args(cfg) # Create infrastructure provider provider = create_provider(cfg) - local_experiment_root_dir = os.path.join( - constants.LOCAL_EXPERIMENT_DIR, args.experiment_name + # Route the remote-write IP through the provider (127.0.0.1 for local mode, + # CloudLab's 10.10.1.x scheme otherwise) before anything resolves + # cfg.streaming.remote_write.ip (Args does, below). + OmegaConf.register_new_resolver( + "remote_write_ip", provider.get_node_ip, replace=True ) + + # Convert config to args-like object for backward compatibility + args = config.Args(cfg) + + if provider.is_remote(): + local_experiment_root_dir = os.path.join( + constants.LOCAL_EXPERIMENT_DIR, args.experiment_name + ) + experiment_root_output_dir = os.path.join( + provider.get_home_dir(), "experiment_outputs", args.experiment_name + ) + else: + # Local mode: the "remote" and "local" roots are the same filesystem + # path by construction, so every rsync/scp call downstream becomes a + # no-op rather than something to opportunistically skip. + local_experiment_root_dir = os.path.join( + provider.get_home_dir(), "experiment_outputs", args.experiment_name + ) + experiment_root_output_dir = local_experiment_root_dir os.makedirs(local_experiment_root_dir, exist_ok=True) # dump config to a file @@ -78,10 +98,6 @@ def main(cfg: DictConfig): with open(os.path.join(local_experiment_root_dir, "cmdline_args.txt"), "w") as f: json.dump(vars(args), f) - experiment_root_output_dir = ( - f"{constants.CLOUDLAB_HOME_DIR}/experiment_outputs/{args.experiment_name}" - ) - global CONTROLLER_REMOTE_OUTPUT_DIR, CONTROLLER_LOCAL_OUTPUT_DIR CONTROLLER_LOCAL_OUTPUT_DIR = os.path.join( local_experiment_root_dir, "controller_output" @@ -93,7 +109,7 @@ def main(cfg: DictConfig): provider.execute_command( node_idx=args.get_coordinator_node(), cmd="mkdir -p {} {}".format( - os.path.dirname(constants.CLOUDLAB_QUERY_LOG_FILE), + os.path.dirname(provider.get_query_log_file()), experiment_root_output_dir, ), cmd_dir="", diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py index cf0a8347..89cc8e9e 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/asap-tools/experiments/experiment_utils/config.py @@ -55,6 +55,26 @@ def validate_basic_config( raise ValueError(error_msg) +def get_node_params(cfg: DictConfig) -> Tuple[int, int]: + """Return (num_nodes, node_offset). Local mode is a single node and never + touches cfg.cloudlab (whose username/hostname_suffix are mandatory-missing + markers that raise on mere attribute access if not provided).""" + if hasattr(cfg, "local"): + return 0, 0 + return cfg.cloudlab.num_nodes, cfg.cloudlab.node_offset + + +def required_cloudlab_params(cfg: DictConfig) -> List[Tuple[str, str]]: + """Return cloudlab.* required-param tuples, or [] when running in local mode.""" + if hasattr(cfg, "local"): + return [] + return [ + ("cloudlab.num_nodes", "Number of CloudLab nodes to use"), + ("cloudlab.username", "Your CloudLab username"), + ("cloudlab.hostname_suffix", "CloudLab experiment hostname suffix"), + ] + + def _is_clickhouse_experiment(experiment_params: DictConfig) -> bool: """Return True if experiment_params describes a ClickHouse (SQL) experiment.""" return "dataset" in experiment_params @@ -469,11 +489,14 @@ def __init__(self, cfg: DictConfig): # Experiment configuration self.experiment_name = cfg.experiment.name - # CloudLab configuration - self.num_nodes = cfg.cloudlab.num_nodes - self.node_offset = cfg.cloudlab.node_offset - self.cloudlab_username = cfg.cloudlab.username - self.hostname_suffix = cfg.cloudlab.hostname_suffix + # CloudLab configuration (or local-mode equivalents) + self.num_nodes, self.node_offset = get_node_params(cfg) + if hasattr(cfg, "local"): + self.cloudlab_username = None + self.hostname_suffix = None + else: + self.cloudlab_username = cfg.cloudlab.username + self.hostname_suffix = cfg.cloudlab.hostname_suffix # Logging and debugging self.log_level = cfg.logging.level @@ -577,10 +600,7 @@ def validate_config(cfg: DictConfig, script_name: str = "experiment_run_e2e"): # Check for required parameters that must be provided via command line required_params = [ ("experiment.name", "Human-readable experiment name"), - ("cloudlab.num_nodes", "Number of CloudLab nodes to use"), - ("cloudlab.username", "Your CloudLab username"), - ("cloudlab.hostname_suffix", "CloudLab experiment hostname suffix"), - ] + ] + required_cloudlab_params(cfg) # Use the existing validate_basic_config function validate_basic_config(cfg, required_params, script_name) diff --git a/asap-tools/experiments/experiment_utils/providers/base.py b/asap-tools/experiments/experiment_utils/providers/base.py index 50bfb8df..13a0f4e0 100644 --- a/asap-tools/experiments/experiment_utils/providers/base.py +++ b/asap-tools/experiments/experiment_utils/providers/base.py @@ -132,6 +132,19 @@ def get_provider_type(self) -> str: """ return self.__class__.__name__.replace("Provider", "").lower() + def is_remote(self) -> bool: + """ + Whether this provider executes on a separate machine reachable over SSH. + + Local providers override this to False so that callers can skip + rsync/scp file-transfer steps entirely (source and destination are + the same filesystem). + + Returns: + True by default (e.g. CloudLab); False for local execution. + """ + return True + def __str__(self) -> str: """String representation of the provider.""" return f"{self.__class__.__name__}()" diff --git a/asap-tools/experiments/experiment_utils/providers/factory.py b/asap-tools/experiments/experiment_utils/providers/factory.py index ea07d28a..34773445 100644 --- a/asap-tools/experiments/experiment_utils/providers/factory.py +++ b/asap-tools/experiments/experiment_utils/providers/factory.py @@ -9,6 +9,7 @@ from .base import InfrastructureProvider from .cloudlab import CloudLabProvider +from .local import LocalProvider def create_provider(cfg: DictConfig) -> InfrastructureProvider: @@ -28,8 +29,13 @@ def create_provider(cfg: DictConfig) -> InfrastructureProvider: ValueError: If the configuration doesn't contain required parameters or specifies an unsupported provider type """ - # Phase 1: Always return CloudLab provider for backward compatibility - # Future phases will add detection logic for other providers + # Local mode (e.g. `+local.home_dir=...`) takes priority and must be checked + # before any `cfg.cloudlab.*` access — those fields are Hydra `???` mandatory- + # missing markers, so merely reading them raises if not provided via CLI. + if hasattr(cfg, "local"): + if not hasattr(cfg.local, "home_dir") or not cfg.local.home_dir: + raise ValueError("Missing 'local.home_dir' configuration parameter") + return LocalProvider(home_dir=cfg.local.home_dir) # Validate that we have the required CloudLab parameters if not hasattr(cfg, "cloudlab"): @@ -62,11 +68,13 @@ def detect_provider_type(cfg: DictConfig) -> str: Returns: String identifier for the provider type ('cloudlab', 'aws', 'local', etc.) """ - # Phase 1: Always detect as CloudLab + if hasattr(cfg, "local"): + return "local" + # Future phases will add logic to detect other provider types # based on configuration parameters like: # - cfg.infrastructure.provider - # - presence of aws/kubernetes/local configuration sections + # - presence of aws/kubernetes configuration sections # - environment variables if ( @@ -83,10 +91,9 @@ def detect_provider_type(cfg: DictConfig) -> str: # return "aws" # if hasattr(cfg, "kubernetes"): # return "kubernetes" - # if cfg.get("local", False): - # return "local" raise ValueError( "Unable to detect infrastructure provider type from configuration. " - "Currently supported: CloudLab (requires cloudlab.username and cloudlab.hostname_suffix)" + "Currently supported: CloudLab (requires cloudlab.username and cloudlab.hostname_suffix) " + "or Local (requires local.home_dir)" ) diff --git a/asap-tools/experiments/experiment_utils/providers/local.py b/asap-tools/experiments/experiment_utils/providers/local.py new file mode 100644 index 00000000..d2c320dc --- /dev/null +++ b/asap-tools/experiments/experiment_utils/providers/local.py @@ -0,0 +1,119 @@ +""" +Local infrastructure provider implementation. + +This module provides a single-machine execution provider: no SSH, no +CloudLab paths/usernames. Everything runs as a local subprocess against a +user-configured home directory that mirrors CloudLab's deployment layout. +""" + +import os +from typing import Union, Optional, List +import subprocess + +from .base import InfrastructureProvider + + +class LocalProvider(InfrastructureProvider): + """ + Local infrastructure provider for running experiments on a single dev + machine, with no CloudLab account/cluster involved. + """ + + def __init__(self, home_dir: str): + """ + Initialize Local provider. + + Args: + home_dir: Local directory mirroring CloudLab's deployment layout + (prometheus/, code/arroyo/..., experiment_outputs/, etc.) + """ + self.home_dir = home_dir + + def execute_command( + self, + node_idx: int, + cmd: str, + cmd_dir: Optional[str] = None, + nohup: bool = False, + popen: bool = False, + ignore_errors: bool = False, + manual: bool = False, + ) -> Union[subprocess.Popen, subprocess.CompletedProcess]: + """Execute a command locally (node_idx is ignored: there is only one node).""" + if manual: + print(f"Please run manually: {cmd}") + if cmd_dir: + print(f"In directory: {cmd_dir}") + return subprocess.CompletedProcess([], 0, "", "") + + if nohup: + cmd = f"nohup {cmd}" + + if popen: + return subprocess.Popen( + cmd, + shell=True, + cwd=cmd_dir, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + + try: + return subprocess.run( + cmd, + shell=True, + cwd=cmd_dir, + capture_output=True, + text=True, + check=not ignore_errors, + ) + except subprocess.CalledProcessError as e: + if ignore_errors: + return e + raise + + def execute_command_parallel( + self, + node_idxs: List[int], + cmd: str, + cmd_dir: Optional[str] = None, + nohup: bool = False, + popen: bool = True, + redirect: bool = False, + wait: bool = True, + ) -> List[subprocess.Popen]: + """Execute a command once locally (there is only one local node).""" + if redirect: + cmd += " > /dev/null 2>&1" + + process = self.execute_command( + node_idx=node_idxs[0] if node_idxs else 0, + cmd=cmd, + cmd_dir=cmd_dir, + nohup=nohup, + popen=True, + ) + processes = [process] + if wait: + for p in processes: + p.wait() + return processes + + def is_remote(self) -> bool: + return False + + def get_node_address(self, node_idx: int) -> str: + return "localhost" + + def get_node_ip(self, node_idx: int) -> str: + return "127.0.0.1" + + def get_home_dir(self) -> str: + return self.home_dir + + def get_query_log_file(self) -> str: + return os.path.join(self.home_dir, "prometheus", "queries.log") + + def __repr__(self) -> str: + return f"LocalProvider(home_dir='{self.home_dir}')" diff --git a/asap-tools/experiments/experiment_utils/services/arroyo.py b/asap-tools/experiments/experiment_utils/services/arroyo.py index e329a80b..a59d7691 100644 --- a/asap-tools/experiments/experiment_utils/services/arroyo.py +++ b/asap-tools/experiments/experiment_utils/services/arroyo.py @@ -160,7 +160,7 @@ def run_arroyosketch( if enable_optimized_remote_write: cmd += " --prometheus_remote_write_source optimized" cmd_dir = os.path.join( - constants.CLOUDLAB_HOME_DIR, "code", "asap-summary-ingest" + self.provider.get_home_dir(), "code", "asap-summary-ingest" ) if avoid_long_ssh: @@ -216,7 +216,7 @@ def stop_arroyosketch(self, pipeline_id: str) -> None: """ cmd = "python3 delete_pipeline.py --pipeline_id {}".format(pipeline_id) cmd_dir = os.path.join( - constants.CLOUDLAB_HOME_DIR, "code", "asap-summary-ingest" + self.provider.get_home_dir(), "code", "asap-summary-ingest" ) self.provider.execute_command( node_idx=self.node_offset, @@ -309,10 +309,15 @@ def is_healthy(self) -> bool: def _start_bare_metal(self, experiment_output_dir: str, **kwargs) -> None: """Start Arroyo cluster using bare metal deployment (original implementation).""" arroyo_config_file_path = os.path.join( - constants.CLOUDLAB_HOME_DIR, "code", "asap-summary-ingest", "config.yaml" + self.provider.get_home_dir(), "code", "asap-summary-ingest", "config.yaml" ) arroyo_bin_path = os.path.join( - constants.CLOUDLAB_HOME_DIR, "code", "arroyo", "target", "release", "arroyo" + self.provider.get_home_dir(), + "code", + "arroyo", + "target", + "release", + "arroyo", ) arroyo_output_file = os.path.join(experiment_output_dir, "arroyo_cluster.out") @@ -332,7 +337,7 @@ def _start_bare_metal(self, experiment_output_dir: str, **kwargs) -> None: def _start_containerized(self, experiment_output_dir: str, **kwargs) -> None: """Start Arroyo cluster using Docker container deployment.""" arroyo_config_file_path = os.path.join( - constants.CLOUDLAB_HOME_DIR, "code", "asap-summary-ingest", "config.yaml" + self.provider.get_home_dir(), "code", "asap-summary-ingest", "config.yaml" ) arroyo_output_file = os.path.join(experiment_output_dir, "arroyo_cluster.out") diff --git a/asap-tools/experiments/experiment_utils/services/base.py b/asap-tools/experiments/experiment_utils/services/base.py index 82c9d25e..3c781bc3 100644 --- a/asap-tools/experiments/experiment_utils/services/base.py +++ b/asap-tools/experiments/experiment_utils/services/base.py @@ -22,12 +22,6 @@ def __init__(self, provider: InfrastructureProvider): """ self.provider = provider - # Maintain backward compatibility properties - if hasattr(provider, "username"): - self.username = provider.username - if hasattr(provider, "hostname_suffix"): - self.hostname_suffix = provider.hostname_suffix - @abstractmethod def start(self, **kwargs) -> None: """ diff --git a/asap-tools/experiments/experiment_utils/services/clickhouse_service.py b/asap-tools/experiments/experiment_utils/services/clickhouse_service.py index 36112eb0..a423c944 100644 --- a/asap-tools/experiments/experiment_utils/services/clickhouse_service.py +++ b/asap-tools/experiments/experiment_utils/services/clickhouse_service.py @@ -124,15 +124,16 @@ def start( ) self.compose_file = remote_compose_file - hostname = f"node{self.node_offset}.{self.provider.hostname_suffix}" - rsync_cmd = 'rsync -azh -e "ssh {}" {} {}@{}:{}'.format( - constants.SSH_OPTIONS, - local_compose_file, - self.provider.username, - hostname, - remote_compose_file, - ) - utils.run_cmd_with_retry(rsync_cmd, popen=False, ignore_errors=False) + if self.provider.is_remote(): + hostname = f"node{self.node_offset}.{self.provider.hostname_suffix}" + rsync_cmd = 'rsync -azh -e "ssh {}" {} {}@{}:{}'.format( + constants.SSH_OPTIONS, + local_compose_file, + self.provider.username, + hostname, + remote_compose_file, + ) + utils.run_cmd_with_retry(rsync_cmd, popen=False, ignore_errors=False) self.provider.execute_command( node_idx=self.node_offset, @@ -229,6 +230,11 @@ def prepare(self, local_data_file: str, remote_dir: str) -> str: nohup=False, popen=False, ) + if not self.provider.is_remote(): + # Same filesystem: no copy needed, the file is already at this path. + self.remote_data_file = local_data_file + return self.remote_data_file + hostname = f"node{self.node_offset}.{self.provider.hostname_suffix}" rsync_cmd = 'rsync -azh --progress -e "ssh {}" {} {}@{}:{}/'.format( constants.SSH_OPTIONS, @@ -326,7 +332,22 @@ def stop(self, **kwargs) -> None: # ------------------------------------------------------------------ # def _rsync_to_remote(self, local_path: str, remote_path: str) -> None: - """Rsync a single local file to an exact remote path.""" + """Copy a single local file to an exact path on the node. + + remote_path is an ad hoc /tmp path distinct from local_path (unlike the + experiment_output_dir tree, it isn't collapsed onto local_path in local + mode), so local mode still needs an actual copy — just a local one + instead of rsync-over-SSH. + """ + if not self.provider.is_remote(): + self.provider.execute_command( + node_idx=self.node_offset, + cmd=f"cp {shlex.quote(local_path)} {shlex.quote(remote_path)}", + cmd_dir=None, + nohup=False, + popen=False, + ) + return hostname = f"node{self.node_offset}.{self.provider.hostname_suffix}" cmd = 'rsync -azh -e "ssh {}" {} {}@{}:{}'.format( constants.SSH_OPTIONS, diff --git a/asap-tools/experiments/experiment_utils/services/docker_victoriametrics.py b/asap-tools/experiments/experiment_utils/services/docker_victoriametrics.py index e9690bbd..019e37d8 100644 --- a/asap-tools/experiments/experiment_utils/services/docker_victoriametrics.py +++ b/asap-tools/experiments/experiment_utils/services/docker_victoriametrics.py @@ -193,15 +193,16 @@ def start( ) self.compose_file = remote_compose_file - hostname = f"node{self.node_offset}.{self.provider.hostname_suffix}" - rsync_cmd = 'rsync -azh -e "ssh {}" {} {}@{}:{}'.format( - constants.SSH_OPTIONS, - local_compose_file, - self.provider.username, - hostname, - remote_compose_file, - ) - utils.run_cmd_with_retry(rsync_cmd, popen=False, ignore_errors=False) + if self.provider.is_remote(): + hostname = f"node{self.node_offset}.{self.provider.hostname_suffix}" + rsync_cmd = 'rsync -azh -e "ssh {}" {} {}@{}:{}'.format( + constants.SSH_OPTIONS, + local_compose_file, + self.provider.username, + hostname, + remote_compose_file, + ) + utils.run_cmd_with_retry(rsync_cmd, popen=False, ignore_errors=False) # Start containers using docker-compose self.provider.execute_command( diff --git a/asap-tools/experiments/experiment_utils/services/fake_exporters.py b/asap-tools/experiments/experiment_utils/services/fake_exporters.py index d6b78c36..1ae533af 100644 --- a/asap-tools/experiments/experiment_utils/services/fake_exporters.py +++ b/asap-tools/experiments/experiment_utils/services/fake_exporters.py @@ -6,7 +6,6 @@ from abc import abstractmethod from typing import Tuple, List, Dict, Any -import constants from .base import BaseService from experiment_utils.providers.base import InfrastructureProvider @@ -133,7 +132,7 @@ def _start_bare_metal( cmds.append(cmd) cmd_dir = os.path.join( - constants.CLOUDLAB_HOME_DIR, + self.provider.get_home_dir(), "code", "asap-tools", "data-sources", @@ -375,7 +374,7 @@ def _start_bare_metal( ) cmds.append(cmd) - cmd_dir = f"{constants.CLOUDLAB_HOME_DIR}/code/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter" + cmd_dir = f"{self.provider.get_home_dir()}/code/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter" # Dump workload configuration to a file os.makedirs( diff --git a/asap-tools/experiments/experiment_utils/services/prometheus.py b/asap-tools/experiments/experiment_utils/services/prometheus.py index 1a5d0b73..fb86d79b 100644 --- a/asap-tools/experiments/experiment_utils/services/prometheus.py +++ b/asap-tools/experiments/experiment_utils/services/prometheus.py @@ -119,24 +119,21 @@ def _stop_prometheus(self) -> None: pass def reset(self) -> None: - """Reset Prometheus data across nodes.""" - # For provider-based architecture, we need to handle reset differently - # This maintains backward compatibility for CloudLab while allowing future provider extensions - if hasattr(self.provider, "username") and hasattr( - self.provider, "hostname_suffix" - ): - cmd = "python3 reset_prometheus.py --num_nodes {} --cloudlab_username {} --hostname_suffix {} --node_offset {}".format( - self.num_nodes, - self.provider.username, - self.provider.hostname_suffix, - self.node_offset, - ) - subprocess.run(cmd, shell=True, check=True) - else: - # For non-CloudLab providers, implement provider-specific reset logic - raise NotImplementedError( - "Reset functionality not yet implemented for this provider type" - ) + """Reset Prometheus data. + + reset_prometheus.py (the historical CloudLab-only path) just runs + `rm -rf data; rm -f queries.log` over SSH via the same + utils.run_on_cloudlab_node that CloudLabProvider.execute_command + already calls with identical args — so there's nothing to branch on. + """ + self.provider.execute_command( + node_idx=self.node_offset, + cmd="rm -rf data; rm -f queries.log", + cmd_dir=os.path.join(self.provider.get_home_dir(), "prometheus"), + nohup=False, + popen=False, + ignore_errors=True, + ) def is_healthy(self) -> bool: """ diff --git a/asap-tools/experiments/experiment_utils/services/query_engine.py b/asap-tools/experiments/experiment_utils/services/query_engine.py index 57210369..001cc52b 100644 --- a/asap-tools/experiments/experiment_utils/services/query_engine.py +++ b/asap-tools/experiments/experiment_utils/services/query_engine.py @@ -176,6 +176,11 @@ def _write_engine_config_to_remote( with open(local_path, "w") as f: f.write(config_yaml) + if not self.provider.is_remote(): + # Same filesystem: local_path and remote_path resolve to the same + # location once experiment_output_dir/local_experiment_dir collapse. + return + hostname = f"node{self.node_offset}.{self.provider.hostname_suffix}" # Ensure the remote directory exists before rsyncing self.provider.execute_command( diff --git a/asap-tools/experiments/experiment_utils/sync.py b/asap-tools/experiments/experiment_utils/sync.py index 3a44a35b..d0dc4983 100644 --- a/asap-tools/experiments/experiment_utils/sync.py +++ b/asap-tools/experiments/experiment_utils/sync.py @@ -14,6 +14,9 @@ def copy_prometheus_data( provider: InfrastructureProvider, experiment_name: str, node_offset: int ): """Copy Prometheus data from remote to local machine.""" + if not provider.is_remote(): + return + remote_prometheus_home_dir = os.path.join(provider.get_home_dir(), "prometheus") data_to_copy = [ f"{remote_prometheus_home_dir}/data", @@ -36,6 +39,8 @@ def rsync_experiment_data( node_offset: int, ): """Sync experiment data from remote to local machine.""" + if not provider.is_remote(): + return cmd = 'rsync -azh -e "ssh {}" {}@node{}.{}:{}/ {}/'.format( constants.SSH_OPTIONS, provider.username, @@ -69,6 +74,9 @@ def rsync_prometheus_config( node_idx=node_offset, cmd=cmd, cmd_dir=None, nohup=False, popen=False ) + if not provider.is_remote(): + return + hostname = f"node{node_offset}.{provider.hostname_suffix}" # Sync entire directory (note the trailing slash to sync contents) cmd = 'rsync -azh -e "ssh {}" {}/ {}@{}:{}/'.format( @@ -88,6 +96,8 @@ def rsync_controller_client_configs( node_offset: int, ): """Sync controller client configurations to remote machine.""" + if not provider.is_remote(): + return hostname = f"node{node_offset}.{provider.hostname_suffix}" cmd = 'rsync -azh -e "ssh {}" {} {}@{}:{}/'.format( constants.SSH_OPTIONS, @@ -106,6 +116,8 @@ def rsync_controller_config_remote_to_local( node_offset: int, ): """Sync controller configuration from remote to local machine.""" + if not provider.is_remote(): + return hostname = f"node{node_offset}.{provider.hostname_suffix}" cmd = 'rsync -azh -e "ssh {}" {}@{}:{}/ {}/'.format( constants.SSH_OPTIONS, @@ -134,7 +146,6 @@ def rsync_dataset_file( Returns: The full remote path to the rsynced file. """ - hostname = f"node{node_offset}.{provider.hostname_suffix}" provider.execute_command( node_idx=node_offset, cmd=f"mkdir -p {remote_dir}", @@ -142,6 +153,11 @@ def rsync_dataset_file( nohup=False, popen=False, ) + if not provider.is_remote(): + # Same filesystem: no copy needed, the file is already at this path. + return local_data_file + + hostname = f"node{node_offset}.{provider.hostname_suffix}" cmd = 'rsync -azh --progress -e "ssh {}" {} {}@{}:{}/'.format( constants.SSH_OPTIONS, local_data_file, @@ -170,7 +186,6 @@ def rsync_streaming_configs( remote_dir: Absolute remote directory path (created if absent). node_offset: Index of the target node. """ - hostname = f"node{node_offset}.{provider.hostname_suffix}" provider.execute_command( node_idx=node_offset, cmd=f"mkdir -p {remote_dir}", @@ -178,6 +193,9 @@ def rsync_streaming_configs( nohup=False, popen=False, ) + if not provider.is_remote(): + return + hostname = f"node{node_offset}.{provider.hostname_suffix}" cmd = 'rsync -azh -e "ssh {}" {}/ {}@{}:{}/'.format( constants.SSH_OPTIONS, local_dir, diff --git a/asap-tools/experiments/experiment_utils/test_local_provider.py b/asap-tools/experiments/experiment_utils/test_local_provider.py new file mode 100644 index 00000000..90ce600e --- /dev/null +++ b/asap-tools/experiments/experiment_utils/test_local_provider.py @@ -0,0 +1,88 @@ +"""Self-check for local-vs-CloudLab provider dispatch (issue #424). + +Runnable standalone (`python3 test_local_provider.py`) or via pytest. +""" + +import os +import sys + +_EXPERIMENTS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, _EXPERIMENTS_DIR) + +from omegaconf import OmegaConf # noqa: E402 +import omegaconf.errors as oerr # noqa: E402 + +import config # noqa: E402 +from providers.factory import create_provider, detect_provider_type # noqa: E402 +from providers.local import LocalProvider # noqa: E402 +from providers.cloudlab import CloudLabProvider # noqa: E402 + + +def _base_cfg(**overrides): + base = OmegaConf.load(os.path.join(_EXPERIMENTS_DIR, "config", "config.yaml")) + del base["defaults"] # not relevant to provider dispatch + return OmegaConf.merge(base, {"experiment": {"name": "smoke"}, **overrides}) + + +def test_local_mode_never_touches_cloudlab_mandatory_fields(): + cfg = _base_cfg(local={"home_dir": "/tmp/sketchdb_home"}) + provider = create_provider(cfg) + assert isinstance(provider, LocalProvider) + assert provider.is_remote() is False + assert provider.get_node_ip(0) == "127.0.0.1" + assert provider.get_home_dir() == "/tmp/sketchdb_home" + assert detect_provider_type(cfg) == "local" + assert config.get_node_params(cfg) == (0, 0) + assert config.required_cloudlab_params(cfg) == [] + + try: + cfg.cloudlab.username + raise AssertionError("cfg.cloudlab.username should still be mandatory-missing") + except oerr.MissingMandatoryValue: + pass + + +def test_local_args_resolve_remote_write_ip_through_provider(): + cfg = _base_cfg(local={"home_dir": "/tmp/sketchdb_home"}) + provider = create_provider(cfg) + OmegaConf.register_new_resolver( + "remote_write_ip", provider.get_node_ip, replace=True + ) + args = config.Args(cfg) + assert args.num_nodes == 0 + assert args.node_offset == 0 + assert args.cloudlab_username is None + assert args.hostname_suffix is None + assert args.remote_write_ip == "127.0.0.1" + assert args.get_node_range(include_coordinator=True) == [0] + + +def test_cloudlab_mode_unchanged(): + cfg = _base_cfg( + cloudlab={ + "num_nodes": 2, + "username": "milindsr", + "hostname_suffix": "exp.cloudlab.us", + } + ) + provider = create_provider(cfg) + assert isinstance(provider, CloudLabProvider) + assert provider.is_remote() is True + assert provider.get_home_dir() == "/scratch/sketch_db_for_prometheus" + assert detect_provider_type(cfg) == "cloudlab" + assert config.get_node_params(cfg) == (2, 0) + assert len(config.required_cloudlab_params(cfg)) == 3 + + OmegaConf.register_new_resolver( + "remote_write_ip", provider.get_node_ip, replace=True + ) + args = config.Args(cfg) + assert args.cloudlab_username == "milindsr" + assert args.remote_write_ip == "10.10.1.1" + + +if __name__ == "__main__": + test_local_mode_never_touches_cloudlab_mandatory_fields() + test_local_args_resolve_remote_write_ip_through_provider() + test_cloudlab_mode_unchanged() + print("OK") From bc2b9dfca9c67be4bf6e93ae2eba759e87153d7b Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Thu, 25 Jun 2026 11:04:11 -0400 Subject: [PATCH 2/2] some fixes --- .../providers/cloudlab_local.py | 4 ++ .../experiment_utils/providers/local.py | 40 ++++++++++++++----- .../services/prometheus_client_service.py | 6 +-- 3 files changed, 37 insertions(+), 13 deletions(-) diff --git a/asap-tools/experiments/experiment_utils/providers/cloudlab_local.py b/asap-tools/experiments/experiment_utils/providers/cloudlab_local.py index ee799e27..2e7ed0e5 100644 --- a/asap-tools/experiments/experiment_utils/providers/cloudlab_local.py +++ b/asap-tools/experiments/experiment_utils/providers/cloudlab_local.py @@ -154,6 +154,10 @@ def execute_command_parallel( return processes + def is_remote(self) -> bool: + """Commands run locally on this node (no SSH), even though paths/usernames are CloudLab-shaped.""" + return False + def get_node_address(self, node_idx: int) -> str: """ Get the network address for the local node. diff --git a/asap-tools/experiments/experiment_utils/providers/local.py b/asap-tools/experiments/experiment_utils/providers/local.py index d2c320dc..15bfe2b9 100644 --- a/asap-tools/experiments/experiment_utils/providers/local.py +++ b/asap-tools/experiments/experiment_utils/providers/local.py @@ -49,15 +49,28 @@ def execute_command( if nohup: cmd = f"nohup {cmd}" + # subprocess treats cwd="" as an actual (invalid) path, unlike the + # CloudLab SSH path's `if cmd_dir: cd {cmd_dir}` which is falsy-safe. + cmd_dir = cmd_dir or None + if popen: - return subprocess.Popen( - cmd, - shell=True, - cwd=cmd_dir, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - ) + try: + return subprocess.Popen( + cmd, + shell=True, + cwd=cmd_dir, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + ) + except OSError as e: + # execute_command_parallel has no ignore_errors knob (see + # follow-up issue), so this can't be conditional yet. Mirrors + # CloudLab's SSH transport: launching `ssh` locally never + # fails even if the remote cmd_dir is missing — the failure + # is invisible unless something later checks output/health. + print(f"Warning: failed to launch '{cmd}' in {cmd_dir!r}: {e}") + return None try: return subprocess.run( @@ -68,7 +81,13 @@ def execute_command( text=True, check=not ignore_errors, ) - except subprocess.CalledProcessError as e: + except (subprocess.CalledProcessError, OSError) as e: + # Over SSH, a missing cmd_dir just makes the *remote* shell exit + # nonzero, which ignore_errors already swallows via check=False. + # Locally, a missing cmd_dir makes subprocess.run itself raise + # OSError (e.g. FileNotFoundError) before the command ever runs — + # same "best-effort cleanup, don't crash" intent, different + # exception type, so it needs the same ignore_errors treatment. if ignore_errors: return e raise @@ -97,7 +116,8 @@ def execute_command_parallel( processes = [process] if wait: for p in processes: - p.wait() + if p is not None: + p.wait() return processes def is_remote(self) -> bool: 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 518ed09e..9d763429 100644 --- a/asap-tools/experiments/experiment_utils/services/prometheus_client_service.py +++ b/asap-tools/experiments/experiment_utils/services/prometheus_client_service.py @@ -205,7 +205,7 @@ def _stop_containerized(self): try: if self.compose_file: cmd = f"docker compose -f {self.compose_file} down" - if self.provider.hostname_suffix == "localhost": + if not self.provider.is_remote(): utils.run_cmd(cmd, popen=False) self.compose_file = None else: @@ -220,7 +220,7 @@ def _stop_containerized(self): else: # Fallback: stop by container name on remote node cmd = f"docker stop {self.container_name}; docker rm {self.container_name}" - if self.provider.hostname_suffix == "localhost": + if not self.provider.is_remote(): utils.run_cmd(cmd, popen=False) else: self.provider.execute_command( @@ -238,7 +238,7 @@ def _stop_containerized(self): def _stop_bare_metal(self): """Kill Prometheus client processes.""" cmd = "pkill -f main_prometheus_client.py" - if self.provider.hostname_suffix == "localhost": + if not self.provider.is_remote(): # If running on localhost, use pkill to stop the process (e.g. from remote_monitor) utils.run_cmd(cmd, popen=False) else: