fix(tools): make Args the single source of truth for remote_write_ip

cfg.streaming.remote_write.ip has been a dead null since PR #469
replaced the OmegaConf resolver that used to populate it with a plain
per-script `args.remote_write_ip = provider.get_node_ip(...)` — a
value only arroyo.py ever read. generate_prometheus_config still reads
cfg.streaming.remote_write.ip directly, so it silently baked
http://None:<port>/... into prometheus.yml for every non-arroyo
streaming engine, causing remote_write to fail and precompute-engine
queries to hang forever waiting on data that never arrives (#546).

Args.__init__ now builds the provider itself and sets both
self.remote_write_ip and cfg.streaming.remote_write.ip from that one
place, so every script gets it automatically instead of each one
re-deriving (and potentially forgetting) it. Scripts that used to call
create_provider(cfg) right alongside config.Args(cfg) now just read
args.provider.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: milindsrivastava1997

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

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(

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

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

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
 

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