Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion asap-tools/experiments/config/config.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
9 changes: 3 additions & 6 deletions asap-tools/experiments/experiment_only_ingest_path.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand All @@ -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}"
Expand Down
9 changes: 3 additions & 6 deletions asap-tools/experiments/experiment_run_e2e.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand All @@ -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(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand All @@ -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}"
Expand Down
9 changes: 3 additions & 6 deletions asap-tools/experiments/experiment_run_grafana_demo.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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")
Expand All @@ -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}"
Expand Down
5 changes: 1 addition & 4 deletions asap-tools/experiments/experiment_teardown_everything.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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

Expand Down
29 changes: 25 additions & 4 deletions asap-tools/experiments/experiment_utils/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@
from omegaconf import DictConfig, ListConfig, OmegaConf

import constants
from experiment_utils.providers.factory import create_provider


def validate_basic_config(
Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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

Expand Down Expand Up @@ -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"):
"""
Expand Down
Loading