diff --git a/asap-tools/experiments/config/config.yaml b/asap-tools/experiments/config/config.yaml index 7b697450..0c6d5ef6 100644 --- a/asap-tools/experiments/config/config.yaml +++ b/asap-tools/experiments/config/config.yaml @@ -70,7 +70,7 @@ streaming: # Prometheus configuration prometheus: # Global Prometheus settings - scrape_interval: "10s" # How frequently to scrape targets + scrape_interval: "1s" # How frequently to scrape targets evaluation_interval: "10s" # How frequently to evaluate rules # query_log_file: "/scratch/sketch_db_for_prometheus/prometheus/queries.log" # Disabled to avoid permission issues in Docker # Recording rules settings diff --git a/asap-tools/experiments/experiment_only_ingest_path.py b/asap-tools/experiments/experiment_only_ingest_path.py index 942e72fc..a7c16dc7 100644 --- a/asap-tools/experiments/experiment_only_ingest_path.py +++ b/asap-tools/experiments/experiment_only_ingest_path.py @@ -8,7 +8,6 @@ import constants import experiment_utils from experiment_utils import sync, config -from experiment_utils.providers.factory import create_provider from experiment_utils.services import ( KafkaService, ExporterServiceFactory, @@ -74,11 +73,9 @@ def main(cfg: DictConfig): is_v2 = experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME # Convert config to args-like object for backward compatibility + # (also constructs the infrastructure provider, exposed as args.provider) args = config.Args(cfg) - - # Create infrastructure provider - provider = create_provider(cfg) - args.remote_write_ip = provider.get_node_ip(args.node_offset) + provider = args.provider local_experiment_root_dir = os.path.join( constants.LOCAL_EXPERIMENT_DIR, args.experiment_name @@ -91,7 +88,7 @@ def main(cfg: DictConfig): # Also dump args to a file for backward compatibility with open(os.path.join(local_experiment_root_dir, "cmdline_args.txt"), "w") as f: - json.dump(vars(args), f) + json.dump(args.to_dict(), f) experiment_root_output_dir = ( f"{constants.CLOUDLAB_HOME_DIR}/experiment_outputs/{args.experiment_name}" diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index 6edbf074..ce9ccba3 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -8,7 +8,6 @@ import constants import experiment_utils from experiment_utils import sync, config -from experiment_utils.providers.factory import create_provider from experiment_utils.services import ( KafkaService, FlinkService, @@ -55,12 +54,10 @@ def main(cfg: DictConfig): # Validate experiment configuration config.validate_experiment_config(cfg.experiment_params) - # Create infrastructure provider - provider = create_provider(cfg) - # Convert config to args-like object for backward compatibility + # (also constructs the infrastructure provider, exposed as args.provider) args = config.Args(cfg) - args.remote_write_ip = provider.get_node_ip(args.node_offset) + provider = args.provider if provider.is_remote(): local_experiment_root_dir = os.path.join( @@ -85,7 +82,7 @@ def main(cfg: DictConfig): # Also dump args to a file for backward compatibility with open(os.path.join(local_experiment_root_dir, "cmdline_args.txt"), "w") as f: - json.dump(vars(args), f) + json.dump(args.to_dict(), f) global CONTROLLER_REMOTE_OUTPUT_DIR, CONTROLLER_LOCAL_OUTPUT_DIR CONTROLLER_LOCAL_OUTPUT_DIR = os.path.join( diff --git a/asap-tools/experiments/experiment_run_exporters_and_prometheus.py b/asap-tools/experiments/experiment_run_exporters_and_prometheus.py index ba11bab3..414d7da7 100644 --- a/asap-tools/experiments/experiment_run_exporters_and_prometheus.py +++ b/asap-tools/experiments/experiment_run_exporters_and_prometheus.py @@ -7,7 +7,6 @@ import constants import experiment_utils from experiment_utils import sync, config -from experiment_utils.providers.factory import create_provider from experiment_utils.services import ( ExporterServiceFactory, SystemExportersService, @@ -30,10 +29,9 @@ def main(cfg: DictConfig): # Validate experiment configuration config.validate_experiment_config(cfg.experiment_params) # Convert config to args-like object for backward compatibility + # (also constructs the infrastructure provider, exposed as args.provider) args = config.Args(cfg) - - # Create infrastructure provider - provider = create_provider(cfg) + provider = args.provider local_experiment_root_dir = os.path.join( constants.LOCAL_EXPERIMENT_DIR, args.experiment_name @@ -46,7 +44,7 @@ def main(cfg: DictConfig): # Also dump args to a file for backward compatibility with open(os.path.join(local_experiment_root_dir, "cmdline_args.txt"), "w") as f: - json.dump(vars(args), f) + json.dump(args.to_dict(), f) experiment_root_output_dir = ( f"{constants.CLOUDLAB_HOME_DIR}/experiment_outputs/{args.experiment_name}" diff --git a/asap-tools/experiments/experiment_run_grafana_demo.py b/asap-tools/experiments/experiment_run_grafana_demo.py index b7383bad..d7e98b48 100644 --- a/asap-tools/experiments/experiment_run_grafana_demo.py +++ b/asap-tools/experiments/experiment_run_grafana_demo.py @@ -8,7 +8,6 @@ import constants import experiment_utils from experiment_utils import sync, config -from experiment_utils.providers.factory import create_provider from experiment_utils.services import ( KafkaService, QueryEngineRustService, @@ -51,11 +50,9 @@ def main(cfg: DictConfig): # Validate experiment configuration config.validate_experiment_config(cfg.experiment_params) # Convert config to args-like object for backward compatibility + # (also constructs the infrastructure provider, exposed as args.provider) args = config.Args(cfg) - - # Create infrastructure provider - provider = create_provider(cfg) - args.remote_write_ip = provider.get_node_ip(args.node_offset) + provider = args.provider args.forward_unsupported_queries = True print("Forcing forward_unsupported_queries to True for Grafana demo") @@ -71,7 +68,7 @@ def main(cfg: DictConfig): # Also dump args to a file for backward compatibility with open(os.path.join(local_experiment_root_dir, "cmdline_args.txt"), "w") as f: - json.dump(vars(args), f) + json.dump(args.to_dict(), f) experiment_root_output_dir = ( f"{constants.CLOUDLAB_HOME_DIR}/experiment_outputs/{args.experiment_name}" diff --git a/asap-tools/experiments/experiment_teardown_everything.py b/asap-tools/experiments/experiment_teardown_everything.py index 534fbd39..c14276f1 100644 --- a/asap-tools/experiments/experiment_teardown_everything.py +++ b/asap-tools/experiments/experiment_teardown_everything.py @@ -14,7 +14,6 @@ import constants from experiment_utils import config -from experiment_utils.providers.factory import create_provider from experiment_utils.services import ( KafkaService, FlinkService, @@ -55,9 +54,7 @@ def main(cfg: DictConfig): # Validate configuration (minimal validation for provider setup) config.validate_config(cfg) args = config.Args(cfg) - - # Create infrastructure provider - provider = create_provider(cfg) + provider = args.provider num_nodes_in_experiment = args.num_nodes diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py index 64bcba49..891f1596 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/asap-tools/experiments/experiment_utils/config.py @@ -11,6 +11,7 @@ from omegaconf import DictConfig, ListConfig, OmegaConf import constants +from experiment_utils.providers.factory import create_provider def validate_basic_config( @@ -532,6 +533,19 @@ def __init__(self, cfg: DictConfig): self.cloudlab_username = cfg.providers.cloudlab.username self.hostname_suffix = cfg.providers.cloudlab.hostname_suffix + # Single source of truth for the infrastructure provider: build it here + # (from cfg alone) so every node-dependent value derived from it — + # remote_write_ip included — is computed in exactly one place instead + # of being re-derived (and potentially forgotten) in each script. + self.provider = create_provider(cfg) + + # Remote write IP (127.0.0.1 for local mode, CloudLab's 10.10.1.x + # scheme otherwise). Written back onto cfg.streaming.remote_write.ip + # too, since generate_prometheus_config reads it from cfg directly + # rather than through this Args object. + self.remote_write_ip = self.provider.get_node_ip(self.node_offset) + cfg.streaming.remote_write.ip = self.remote_write_ip + # Logging and debugging self.log_level = cfg.logging.level @@ -565,10 +579,8 @@ def __init__(self, cfg: DictConfig): self.do_local_flink = cfg.streaming.do_local_flink self.forward_unsupported_queries = cfg.streaming.forward_unsupported_queries self.use_kafka_ingest = cfg.streaming.use_kafka_ingest - # Remote write configuration - self.remote_write_ip = ( - None # set by caller: provider.get_node_ip(self.node_offset) - ) + # Remote write configuration (self.remote_write_ip set above, near + # provider construction) self.remote_write_base_port = cfg.streaming.remote_write.base_port self.remote_write_path = cfg.streaming.remote_write.path @@ -624,6 +636,15 @@ def get_coordinator_node(self) -> int: """Get the coordinator node index (first node in the range).""" return self.node_offset + def to_dict(self) -> Dict[str, Any]: + """JSON-serializable snapshot of this Args instance. + + `provider` holds a live InfrastructureProvider object, so it's + swapped for its repr() (e.g. "CloudLabProvider(username='...', ...)") + rather than dropped. + """ + return {k: (repr(v) if k == "provider" else v) for k, v in vars(self).items()} + def validate_config(cfg: DictConfig, script_name: str = "experiment_run_e2e"): """