From bb2bef5486a159006a9ce855a8cad2a633bf3b31 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 2 Sep 2026 10:53:13 -0400 Subject: [PATCH 1/4] feat(asap-tools): seed fake exporters per instance --- .../fake_exporter/docker-compose.yml.j2 | 3 +- .../fake_exporter/src/main.rs | 9 +- .../CONFIG_PARAMETERS_REFERENCE.md | 6 + .../services/fake_exporters.py | 130 +++++++++++------- .../generate_fake_exporter_compose.py | 7 + asap-tools/experiments/generate_workload.py | 1 + 6 files changed, 104 insertions(+), 52 deletions(-) diff --git a/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter/docker-compose.yml.j2 b/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter/docker-compose.yml.j2 index a8b67b4..7f0878b 100644 --- a/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter/docker-compose.yml.j2 +++ b/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter/docker-compose.yml.j2 @@ -13,6 +13,7 @@ services: "--dataset", "{{ dataset }}", "--num-labels", "{{ num_labels }}", "--num-values-per-label", "{{ num_values_per_label | string }}", - "--metric-type", "{{ metric_type }}" + "--metric-type", "{{ metric_type }}"{% if seed is defined %}, + "--seed", "{{ seed }}"{% endif %} ] restart: unless-stopped diff --git a/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter/src/main.rs b/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter/src/main.rs index 99e1bbd..9be2e9f 100644 --- a/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter/src/main.rs +++ b/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter/src/main.rs @@ -73,7 +73,7 @@ const CONST_1M: u64 = 1_000_000; const CONST_2M: u64 = 2_000_000; const CONST_3M: u64 = 3_000_000; -const RNG_SEED: u64 = 0; // seed for rng used by all distributions +const DEFAULT_RNG_SEED: u64 = 0; // default seed for rng used by all distributions const ZIPF_ALPHA: f64 = 1.01; // zipf parameter @@ -217,6 +217,7 @@ impl FakeCollector { label_names: Option, label_value_prefixes: Option, add_pattern_label: bool, + seed: u64, ) -> Self { let num_values_per_label = get_num_vals_per_label(num_values_per_label, num_labels); let prefixes: Option> = label_value_prefixes @@ -308,7 +309,7 @@ impl FakeCollector { metric_name, label_names, add_pattern_label, - rng: Mutex::new(SmallRng::seed_from_u64(RNG_SEED)), + rng: Mutex::new(SmallRng::seed_from_u64(seed)), zipf_dist, normal_dist, uniform_dist, @@ -626,6 +627,9 @@ struct Args { #[arg(long, default_value = "false", help = "Add 'pattern' label to metrics with dataset name")] add_pattern_label: bool, + + #[arg(long, default_value_t = DEFAULT_RNG_SEED, help = "Seed for random value generation")] + seed: u64, } #[tokio::main] @@ -642,6 +646,7 @@ async fn main() -> Result<(), BoxedErr> { args.label_names, args.label_value_prefixes, args.add_pattern_label, + args.seed, )); // Register collector and start serving diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index 469321b..07591c0 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md @@ -482,6 +482,11 @@ These parameters come from the `experiment_type` config group and are prefixed w - **Example**: `5` - **SCHEMA ISSUE**: Values vary widely (1-10) across configs without clear pattern +#### `experiment_params.exporters.exporter_list.fake_exporter.seed` (int, optional) +- **Description**: Base seed for Rust fake exporter data generation. Each exporter receives a distinct seed derived from this base seed, worker ordinal, and port ordinal. +- **Default**: `0` +- **Example**: `42` + #### `experiment_params.exporters.exporter_list.fake_exporter.start_port` (int, optional) - **Description**: Starting port number for fake exporters - **Default**: `50000` @@ -727,6 +732,7 @@ python experiment_run_e2e.py \ python experiment_run_e2e.py \ experiment_type=cloud_demo \ experiment_params.exporters.exporter_list.fake_exporter.num_ports_per_server=5 \ + experiment_params.exporters.exporter_list.fake_exporter.seed=42 \ experiment_params.exporters.exporter_list.fake_exporter.metric_type=gauge \ [required params...] diff --git a/asap-tools/experiments/experiment_utils/services/fake_exporters.py b/asap-tools/experiments/experiment_utils/services/fake_exporters.py index 1ae533a..22d8f1b 100644 --- a/asap-tools/experiments/experiment_utils/services/fake_exporters.py +++ b/asap-tools/experiments/experiment_utils/services/fake_exporters.py @@ -9,6 +9,24 @@ from .base import BaseService from experiment_utils.providers.base import InfrastructureProvider +DEFAULT_FAKE_EXPORTER_SEED = 0 + + +def get_fake_exporter_seed( + config: Dict[str, Any], worker_ordinal: int, port_ordinal: int +) -> int: + """Derive a unique, reproducible seed for one exporter instance.""" + base_seed = config.get("seed", DEFAULT_FAKE_EXPORTER_SEED) + num_ports_per_server = config["num_ports_per_server"] + exporter_ordinal = worker_ordinal * num_ports_per_server + port_ordinal + return base_seed + exporter_ordinal + + +def get_fake_exporter_target_nodes(args) -> List[int]: + """Return exporter nodes, preserving the single-node local fallback.""" + worker_nodes = args.get_node_range(include_coordinator=False) + return worker_nodes or [args.get_coordinator_node()] + class BaseExporterService(BaseService): """Base class for exporter services.""" @@ -363,16 +381,20 @@ def _start_bare_metal( dataset = config["dataset"] cmds = [] - for port in range(num_ports): - cmd = "./target/release/fake_exporter --port {} --valuescale {} --dataset {} --num-labels {} --num-values-per-label {} --metric-type {}".format( - port + config["start_port"], - config["synthetic_data_value_scale"], - dataset, - config["num_labels"], - config["num_values_per_label"], - config["metric_type"], - ) - cmds.append(cmd) + target_nodes = get_fake_exporter_target_nodes(self.args) + for worker_ordinal, node_idx in enumerate(target_nodes): + for port_ordinal in range(num_ports): + seed = get_fake_exporter_seed(config, worker_ordinal, port_ordinal) + cmd = "./target/release/fake_exporter --port {} --valuescale {} --dataset {} --num-labels {} --num-values-per-label {} --metric-type {} --seed {}".format( + port_ordinal + config["start_port"], + config["synthetic_data_value_scale"], + dataset, + config["num_labels"], + config["num_values_per_label"], + config["metric_type"], + seed, + ) + cmds.append((node_idx, cmd)) cmd_dir = f"{self.provider.get_home_dir()}/code/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter" @@ -383,12 +405,12 @@ def _start_bare_metal( with open( os.path.join(local_experiment_dir, "fake_exporter_config", "cmds.sh"), "w" ) as f: - f.write("\n".join(cmds)) + f.write("\n".join(cmd for _, cmd in cmds)) # Run commands in parallel across nodes - for cmd in cmds: + for node_idx, cmd in cmds: self.provider.execute_command_parallel( - node_idxs=self.args.get_node_range(include_coordinator=False), + node_idxs=[node_idx], cmd=cmd, cmd_dir=cmd_dir, nohup=False, @@ -410,30 +432,34 @@ def _start_containerized( dataset = config["dataset"] # Build docker run commands for each port - docker_run_cmds: List[str] = [] + docker_run_cmds: List[Tuple[int, str]] = [] container_names: List[str] = [] + target_nodes = get_fake_exporter_target_nodes(self.args) + + for worker_ordinal, node_idx in enumerate(target_nodes): + for port_ordinal in range(num_ports): + actual_port = port_ordinal + config["start_port"] + seed = get_fake_exporter_seed(config, worker_ordinal, port_ordinal) + container_name = f"{BaseExporterService.FAKE_EXPORTER_BASE_CONTAINER_NAME}-{actual_port}-rust" + + # Build docker run command + docker_cmd = ( + f"docker run -d " + f"--name {container_name} " + f"-p {actual_port}:{actual_port} " + f"--restart unless-stopped " + f"sketchdb-fake-exporter-rust:latest " + f"--port {actual_port} " + f"--valuescale {config['synthetic_data_value_scale']} " + f"--dataset {dataset} " + f"--num-labels {config['num_labels']} " + f"--num-values-per-label {config['num_values_per_label']} " + f"--metric-type {config['metric_type']} " + f"--seed {seed}" + ) - for port in range(num_ports): - actual_port = port + config["start_port"] - container_name = f"{BaseExporterService.FAKE_EXPORTER_BASE_CONTAINER_NAME}-{actual_port}-rust" - - # Build docker run command - docker_cmd = ( - f"docker run -d " - f"--name {container_name} " - f"-p {actual_port}:{actual_port} " - f"--restart unless-stopped " - f"sketchdb-fake-exporter-rust:latest " - f"--port {actual_port} " - f"--valuescale {config['synthetic_data_value_scale']} " - f"--dataset {dataset} " - f"--num-labels {config['num_labels']} " - f"--num-values-per-label {config['num_values_per_label']} " - f"--metric-type {config['metric_type']}" - ) - - container_names.append(container_name) - docker_run_cmds.append(docker_cmd) + container_names.append(container_name) + docker_run_cmds.append((node_idx, docker_cmd)) self.container_names = container_names @@ -447,24 +473,30 @@ def _start_containerized( ), "w", ) as f: - f.write("\n".join(docker_run_cmds)) + f.write("\n".join(cmd for _, cmd in docker_run_cmds)) # Start containers in batches to avoid overwhelming Docker daemon BATCH_SIZE = 5 - for i in range(0, len(docker_run_cmds), BATCH_SIZE): - batch = docker_run_cmds[i : i + BATCH_SIZE] - # Combine docker run commands in batch into single SSH command - batch_cmd = "; ".join(batch) + for node_idx in target_nodes: + node_commands = [ + cmd + for command_node_idx, cmd in docker_run_cmds + if command_node_idx == node_idx + ] + for i in range(0, len(node_commands), BATCH_SIZE): + batch = node_commands[i : i + BATCH_SIZE] + # Combine docker run commands in batch for one node. + batch_cmd = "; ".join(batch) - self.provider.execute_command_parallel( - node_idxs=self.args.get_node_range(include_coordinator=False), - cmd=batch_cmd, - cmd_dir="", - nohup=False, - popen=True, - redirect=True, - wait=True, # Wait for batch to complete - ) + self.provider.execute_command_parallel( + node_idxs=[node_idx], + cmd=batch_cmd, + cmd_dir="", + nohup=False, + popen=True, + redirect=True, + wait=True, # Wait for batch to complete + ) return diff --git a/asap-tools/experiments/generate_fake_exporter_compose.py b/asap-tools/experiments/generate_fake_exporter_compose.py index 7fcb91d..10254ba 100644 --- a/asap-tools/experiments/generate_fake_exporter_compose.py +++ b/asap-tools/experiments/generate_fake_exporter_compose.py @@ -88,6 +88,11 @@ def main(): required=True, help="Type of metric to generate (e.g., gauge, counter)", ) + parser.add_argument( + "--seed", + type=int, + help="Base seed for random value generation", + ) parser.add_argument( "--template-path", @@ -132,6 +137,8 @@ def main(): # Only include container_name if provided, so Jinja2 default filter can work if args.container_name: template_vars["container_name"] = args.container_name + if args.seed is not None: + template_vars["seed"] = args.seed # Set up file paths script_dir = Path(__file__).parent diff --git a/asap-tools/experiments/generate_workload.py b/asap-tools/experiments/generate_workload.py index 1259f7b..9fc7ccc 100644 --- a/asap-tools/experiments/generate_workload.py +++ b/asap-tools/experiments/generate_workload.py @@ -366,6 +366,7 @@ def get_base_config() -> Dict: "fake_exporter": { "num_ports_per_server": 1, "start_port": 50000, + "seed": 0, "dataset": "zipf", "synthetic_data_value_scale": 10000, "num_labels": 3, From de264bc8071b40a7746371991a068fd40b76de1e Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 2 Sep 2026 11:01:43 -0400 Subject: [PATCH 2/4] fix(asap-tools): validate fake exporter seed range --- .../services/fake_exporters.py | 21 ++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/asap-tools/experiments/experiment_utils/services/fake_exporters.py b/asap-tools/experiments/experiment_utils/services/fake_exporters.py index 22d8f1b..977a49b 100644 --- a/asap-tools/experiments/experiment_utils/services/fake_exporters.py +++ b/asap-tools/experiments/experiment_utils/services/fake_exporters.py @@ -10,6 +10,7 @@ from experiment_utils.providers.base import InfrastructureProvider DEFAULT_FAKE_EXPORTER_SEED = 0 +MAX_FAKE_EXPORTER_SEED = (1 << 64) - 1 def get_fake_exporter_seed( @@ -17,9 +18,27 @@ def get_fake_exporter_seed( ) -> int: """Derive a unique, reproducible seed for one exporter instance.""" base_seed = config.get("seed", DEFAULT_FAKE_EXPORTER_SEED) + if ( + not isinstance(base_seed, int) + or isinstance(base_seed, bool) + or not 0 <= base_seed <= MAX_FAKE_EXPORTER_SEED + ): + raise ValueError( + "fake_exporter.seed must be an integer in the range " + f"0..{MAX_FAKE_EXPORTER_SEED}, got {base_seed!r}" + ) + num_ports_per_server = config["num_ports_per_server"] exporter_ordinal = worker_ordinal * num_ports_per_server + port_ordinal - return base_seed + exporter_ordinal + derived_seed = base_seed + exporter_ordinal + if derived_seed > MAX_FAKE_EXPORTER_SEED: + raise ValueError( + "derived fake exporter seed exceeds the Rust u64 range: " + f"base seed {base_seed} with exporter ordinal {exporter_ordinal} " + f"produces {derived_seed}, maximum {MAX_FAKE_EXPORTER_SEED}" + ) + + return derived_seed def get_fake_exporter_target_nodes(args) -> List[int]: From 00732efc89ea5fd7c96f9e2c6b9fe252436e5adc Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 2 Sep 2026 12:29:57 -0400 Subject: [PATCH 3/4] fix(asap-tools): preserve fake exporter launch fan-out --- .../services/fake_exporters.py | 81 +++++++++++++------ 1 file changed, 58 insertions(+), 23 deletions(-) diff --git a/asap-tools/experiments/experiment_utils/services/fake_exporters.py b/asap-tools/experiments/experiment_utils/services/fake_exporters.py index 977a49b..e2cd26c 100644 --- a/asap-tools/experiments/experiment_utils/services/fake_exporters.py +++ b/asap-tools/experiments/experiment_utils/services/fake_exporters.py @@ -4,6 +4,7 @@ import os from abc import abstractmethod +from concurrent.futures import ThreadPoolExecutor from typing import Tuple, List, Dict, Any from .base import BaseService @@ -47,6 +48,29 @@ def get_fake_exporter_target_nodes(args) -> List[int]: return worker_nodes or [args.get_coordinator_node()] +def execute_fake_exporter_commands_in_parallel( + provider: InfrastructureProvider, + node_commands: List[Tuple[int, str]], + **kwargs, +) -> None: + """Run node-specific exporter commands concurrently.""" + if not node_commands: + return + + with ThreadPoolExecutor(max_workers=len(node_commands)) as executor: + futures = [ + executor.submit( + provider.execute_command_parallel, + node_idxs=[node_idx], + cmd=cmd, + **kwargs, + ) + for node_idx, cmd in node_commands + ] + for future in futures: + future.result() + + class BaseExporterService(BaseService): """Base class for exporter services.""" @@ -399,10 +423,11 @@ def _start_bare_metal( num_ports = config["num_ports_per_server"] dataset = config["dataset"] - cmds = [] + commands_by_port: List[List[Tuple[int, str]]] = [] target_nodes = get_fake_exporter_target_nodes(self.args) - for worker_ordinal, node_idx in enumerate(target_nodes): - for port_ordinal in range(num_ports): + for port_ordinal in range(num_ports): + port_commands = [] + for worker_ordinal, node_idx in enumerate(target_nodes): seed = get_fake_exporter_seed(config, worker_ordinal, port_ordinal) cmd = "./target/release/fake_exporter --port {} --valuescale {} --dataset {} --num-labels {} --num-values-per-label {} --metric-type {} --seed {}".format( port_ordinal + config["start_port"], @@ -413,7 +438,12 @@ def _start_bare_metal( config["metric_type"], seed, ) - cmds.append((node_idx, cmd)) + port_commands.append((node_idx, cmd)) + commands_by_port.append(port_commands) + + cmds = [ + command for port_commands in commands_by_port for command in port_commands + ] cmd_dir = f"{self.provider.get_home_dir()}/code/asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter" @@ -426,11 +456,11 @@ def _start_bare_metal( ) as f: f.write("\n".join(cmd for _, cmd in cmds)) - # Run commands in parallel across nodes - for node_idx, cmd in cmds: - self.provider.execute_command_parallel( - node_idxs=[node_idx], - cmd=cmd, + # Preserve cross-node fan-out while assigning each node its own seed. + for port_commands in commands_by_port: + execute_fake_exporter_commands_in_parallel( + self.provider, + port_commands, cmd_dir=cmd_dir, nohup=False, popen=True, @@ -459,7 +489,7 @@ def _start_containerized( for port_ordinal in range(num_ports): actual_port = port_ordinal + config["start_port"] seed = get_fake_exporter_seed(config, worker_ordinal, port_ordinal) - container_name = f"{BaseExporterService.FAKE_EXPORTER_BASE_CONTAINER_NAME}-{actual_port}-rust" + container_name = f"{BaseExporterService.FAKE_EXPORTER_BASE_CONTAINER_NAME}-{node_idx}-{actual_port}-rust" # Build docker run command docker_cmd = ( @@ -494,8 +524,9 @@ def _start_containerized( ) as f: f.write("\n".join(cmd for _, cmd in docker_run_cmds)) - # Start containers in batches to avoid overwhelming Docker daemon + # Start containers in batches to avoid overwhelming Docker daemon. BATCH_SIZE = 5 + node_batch_commands: List[Tuple[int, str]] = [] for node_idx in target_nodes: node_commands = [ cmd @@ -506,16 +537,18 @@ def _start_containerized( batch = node_commands[i : i + BATCH_SIZE] # Combine docker run commands in batch for one node. batch_cmd = "; ".join(batch) + node_batch_commands.append((node_idx, batch_cmd)) - self.provider.execute_command_parallel( - node_idxs=[node_idx], - cmd=batch_cmd, - cmd_dir="", - nohup=False, - popen=True, - redirect=True, - wait=True, # Wait for batch to complete - ) + # Start each worker's batches concurrently. + execute_fake_exporter_commands_in_parallel( + self.provider, + node_batch_commands, + cmd_dir="", + nohup=False, + popen=True, + redirect=True, + wait=True, # Wait for batch to complete + ) return @@ -539,8 +572,9 @@ def _stop_bare_metal(self, **kwargs) -> None: **kwargs: Additional configuration (currently unused) """ cmd = "pkill -f fake_exporter" + target_nodes = get_fake_exporter_target_nodes(self.args) self.provider.execute_command_parallel( - node_idxs=self.args.get_node_range(include_coordinator=False), + node_idxs=target_nodes, cmd=cmd, cmd_dir=None, nohup=False, @@ -550,6 +584,7 @@ def _stop_bare_metal(self, **kwargs) -> None: def _stop_containerized(self, **kwargs) -> None: """Stop fake exporters using containerized deployment.""" + target_nodes = get_fake_exporter_target_nodes(self.args) try: if self.container_names is not None and len(self.container_names) > 0: # Stop and remove containers by name @@ -561,7 +596,7 @@ def _stop_containerized(self, **kwargs) -> None: cmd = f"docker stop {container_list} 2>/dev/null || true; docker rm {container_list} 2>/dev/null || true" self.provider.execute_command_parallel( - node_idxs=self.args.get_node_range(include_coordinator=False), + node_idxs=target_nodes, cmd=cmd, cmd_dir=None, nohup=False, @@ -572,7 +607,7 @@ def _stop_containerized(self, **kwargs) -> None: # Fallback: stop all containers matching the base name pattern cmd = f"docker ps -a --filter name={BaseExporterService.FAKE_EXPORTER_BASE_CONTAINER_NAME} --format '{{{{.Names}}}}' | xargs -r docker stop; docker ps -a --filter name={BaseExporterService.FAKE_EXPORTER_BASE_CONTAINER_NAME} --format '{{{{.Names}}}}' | xargs -r docker rm" self.provider.execute_command_parallel( - node_idxs=self.args.get_node_range(include_coordinator=False), + node_idxs=target_nodes, cmd=cmd, cmd_dir=None, nohup=False, From b39d8d1871afc5bfc51ecff496efa313f51e6083 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 2 Sep 2026 12:43:55 -0400 Subject: [PATCH 4/4] docs(asap-tools): explain exporter command fan-out --- .../experiment_utils/services/fake_exporters.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/asap-tools/experiments/experiment_utils/services/fake_exporters.py b/asap-tools/experiments/experiment_utils/services/fake_exporters.py index e2cd26c..bbd3f94 100644 --- a/asap-tools/experiments/experiment_utils/services/fake_exporters.py +++ b/asap-tools/experiments/experiment_utils/services/fake_exporters.py @@ -53,7 +53,12 @@ def execute_fake_exporter_commands_in_parallel( node_commands: List[Tuple[int, str]], **kwargs, ) -> None: - """Run node-specific exporter commands concurrently.""" + """Run node-specific exporter commands concurrently. + + The provider method applies one command to every node in ``node_idxs``. + Fake exporters need different commands per node because their seeds are + node-specific, so invoke it once per node while preserving concurrency. + """ if not node_commands: return