From 093e4a67eb754fb4b3878418815538b036d13828 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:27:19 +0000 Subject: [PATCH 1/4] feat(asap-tools): run sketch-bench workload tables end to end with the paced feeder - exporters.exporter_list.paced_feeder: experiment_run_e2e.py runs one paced_feeder process per (sink, metric) on the worker nodes, pushing over remote write to the query engine's ingest (sketchdb mode) and/or Prometheus (other modes, or query_prometheus_too). It starts once its sinks accept writes, shares one --start-ms and seed, and fails the run if a feeder dies. Prometheus does not forward fed metrics to the query engine. - Prometheus (bare-metal and containerized) starts with --web.enable-remote-write-receiver when the feeder is configured. - client_options.align_to_interval: repetition k evaluates at floor((start - query_time_offset) / T) * T + k * T and is sent query_time_offset after that time. - sketch_bench_e2e/sketch_bench_e2e_configs.py writes an experiment_type config from a table's e2e block: one query group per interval, the feeder block, metrics, and controller.prebuilt_output_dir; structured queries only with a SQL file. - setup_dependencies.sh builds the feeder. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .github/workflows/python.yml | 3 + .../CONFIG_PARAMETERS_REFERENCE.md | 43 ++ asap-tools/experiments/constants.py | 2 + asap-tools/experiments/experiment_run_e2e.py | 47 +++ .../experiments/experiment_utils/core.py | 31 +- .../experiment_utils/services/__init__.py | 22 +- .../services/docker_prometheus.py | 16 +- .../services/fake_exporters.py | 10 +- .../experiment_utils/services/paced_feeder.py | 223 +++++++++++ .../experiment_utils/services/prometheus.py | 19 +- .../sketch_bench_e2e_configs.py | 239 ++++++++++++ .../experiments/tests/test_paced_feeder.py | 367 ++++++++++++++++++ .../tests/test_sketch_bench_e2e_configs.py | 204 ++++++++++ .../installation/common/setup_dependencies.sh | 1 + .../prometheus-client/classes/query_time.py | 40 ++ .../main_prometheus_client.py | 74 ++-- .../tests/test_query_time.py | 51 +++ 17 files changed, 1356 insertions(+), 36 deletions(-) create mode 100644 asap-tools/experiments/experiment_utils/services/paced_feeder.py create mode 100644 asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py create mode 100644 asap-tools/experiments/tests/test_paced_feeder.py create mode 100644 asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py create mode 100644 asap-tools/queriers/prometheus-client/classes/query_time.py create mode 100644 asap-tools/queriers/prometheus-client/tests/test_query_time.py diff --git a/.github/workflows/python.yml b/.github/workflows/python.yml index 4a00b872..6b6cd229 100644 --- a/.github/workflows/python.yml +++ b/.github/workflows/python.yml @@ -92,6 +92,9 @@ jobs: - name: Type check with mypy working-directory: asap-tools/queriers/prometheus-client run: mypy . + - name: Run tests + working-directory: asap-tools/queriers/prometheus-client + run: python -m unittest discover -s tests -v test-utilities: needs: detect-changes diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index 35a8afed..a85db47e 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md @@ -457,6 +457,44 @@ These parameters come from the `experiment_type` config group and are prefixed w - **Example**: `"gauge"` - **Note**: Counter metrics have `_total` suffix, gauge metrics don't +#### Paced Feeder Parameters + +`paced_feeder` runs `asap-tools/data-sources/paced-feeder` (built by +`installation/common/setup_dependencies.sh`) bare-metal on the worker nodes, +one process per (sink, metric). It pushes over remote write instead of being +scraped: in `sketchdb` mode to the query engine's ingest +(`streaming.remote_write`, which must have `parallelism: 1`), plus Prometheus +when the mode sets `query_prometheus_too`; in every other mode to Prometheus, +which then starts with `--web.enable-remote-write-receiver`. All processes +share the seed and table, so every system gets the same data. Its metrics are +listed under `metrics` with `exporter: paced_feeder`; Prometheus does not +forward them to the query engine. Generate a whole config from a sketch-bench +table with `sketch_bench_e2e/sketch_bench_e2e_configs.py`. + +#### `experiment_params.exporters.exporter_list.paced_feeder.workload_table` (string, required) +- **Description**: Local path of the sketch-bench workload table; copied to each feeder node +- **Example**: `/path/to/synthetic-templatesclassic-shared1.json` + +#### `experiment_params.exporters.exporter_list.paced_feeder.metrics` (list, required) +- **Description**: Metrics of the table's `workloads[0].e2e.metrics` to send; gauge or counter only +- **Example**: `[data, data_total]` + +#### `experiment_params.exporters.exporter_list.paced_feeder.seed` (int, required) +- **Description**: `--seed` of every feeder process +- **Example**: `42` + +#### `experiment_params.exporters.exporter_list.paced_feeder.duration_secs` (int, required) +- **Description**: Seconds of data to send. Must be at least `flow.steady_state_wait` plus the query groups' running time +- **Example**: `4000` + +#### `experiment_params.exporters.exporter_list.paced_feeder.samples_per_sec` (int, optional) +- **Description**: Overrides the table's `samples_per_sec_per_series` +- **Example**: `20` + +#### `experiment_params.exporters.exporter_list.paced_feeder.extra_flags` (string, optional) +- **Description**: Appended to every feeder command line +- **Example**: `"--concurrency 32"` + ### Query Group Configuration #### `experiment_params.query_groups` (list, required) @@ -500,6 +538,11 @@ These parameters come from the `experiment_type` config group and are prefixed w - **Example**: `15` - **Usage**: Ensures data is available for lookback queries +#### `experiment_params.query_groups[].client_options.align_to_interval` (boolean, optional) +- **Description**: Evaluate repetition k at `t_k = floor((start - query_time_offset) / T) * T + k * T`, with `T = repetition_delay_ms` (whole seconds), sending it at wall-clock `t_k + query_time_offset` +- **Default**: `false` +- **Usage**: Every evaluation time is a multiple of `T`, so each system answers the same windows; `query_time_offset` is how long the data of the window ending at `t_k` gets to arrive. Overrides the query engine's per-aggregation alignment in sketchdb mode + #### `experiment_params.query_groups[].client_options.starting_delay` (int, required) - **Description**: Initial delay in seconds before starting query execution - **Example**: `610` diff --git a/asap-tools/experiments/constants.py b/asap-tools/experiments/constants.py index 487e7e2c..2faf6967 100644 --- a/asap-tools/experiments/constants.py +++ b/asap-tools/experiments/constants.py @@ -42,6 +42,8 @@ PROMETHEUS_CONFIG_DIR = "prometheus_config" PROMETHEUS_CONFIG_FILE = "prometheus.yml" PROMETHEUS_REMOTE_WRITE_ENDPOINT = "/api/v1/write" +# Lets Prometheus accept samples pushed to PROMETHEUS_REMOTE_WRITE_ENDPOINT. +PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG = "--web.enable-remote-write-receiver" HTTP_READINESS_CURL_TIMEOUT_SECONDS = 5 QUERY_ENGINE_RUNTIME_INFO_ENDPOINT = "/api/v1/status/runtimeinfo" diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index 9e653f8d..f1a6ef1a 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -25,6 +25,10 @@ DockerPrometheusService, DockerVictoriaMetricsService, SystemExportersService, + PACED_FEEDER_EXPORTER, + PacedFeederService, + check_paced_feeder_config, + paced_feeder_sinks, ) from experiment_utils.services.misc import DiscoveryBackend @@ -188,6 +192,22 @@ def main(cfg: DictConfig): cfg.experiment_params ) + paced_feeder_service = None + paced_feeder_config = exporter_config["exporter_list"].get(PACED_FEEDER_EXPORTER) + if paced_feeder_config is not None: + paced_feeder_config = OmegaConf.to_container(paced_feeder_config, resolve=True) + check_paced_feeder_config( + paced_feeder_config, + parallelism=args.parallelism, + min_feed_secs=(0 if skip_querying else args.steady_state_wait) + + minimum_experiment_running_time, + ) + paced_feeder_service = PacedFeederService(provider, args) + modes_by_name = { + m["mode"]: m + for m in OmegaConf.to_container(cfg.experiment_params.experiment, resolve=True) + } + for experiment_mode in experiment_modes: print(f"Running experiment mode: {experiment_mode}") experiment_output_dir = os.path.join( @@ -228,6 +248,8 @@ def main(cfg: DictConfig): exporter_service.stop() deathstar_service.stop() prometheus_service.reset() + if paced_feeder_service is not None: + paced_feeder_service.stop() # Also stop avalanche exporters if they were started if config.check_exporter_and_queries_exist("avalanche", cfg.experiment_params): @@ -469,6 +491,28 @@ def main(cfg: DictConfig): ), f"Expected PrometheusService but got {type(prometheus_service).__name__}" prometheus_service.start(experiment_output_dir) + # Pushes to its sinks, so it starts once they accept writes. + if paced_feeder_service is not None: + prometheus_url = ( + f"http://{args.remote_write_ip}:" + f"{prometheus_service.get_query_endpoint_port()}" + f"{constants.PROMETHEUS_REMOTE_WRITE_ENDPOINT}" + ) + sinks = paced_feeder_sinks( + modes_by_name[experiment_mode], + query_engine_url=f"http://{args.remote_write_ip}:" + f"{args.remote_write_base_port}{args.remote_write_path}", + prometheus_url=prometheus_url, + ) + if prometheus_url in sinks.values(): + prometheus_service.wait_until_ready() + paced_feeder_service.start( + config=paced_feeder_config, + sinks=sinks, + experiment_output_dir=experiment_output_dir, + local_experiment_dir=local_experiment_dir, + ) + # Start Prometheus throughput monitoring if enabled if args.throughput_prometheus: prometheus_throughput_monitor = PrometheusThroughputMonitor( @@ -546,6 +590,9 @@ def main(cfg: DictConfig): # Skip teardown if the no_teardown flag is set if not args.no_teardown: + # Before its sinks go away, so its log shows no spurious errors. + if paced_feeder_service is not None: + paced_feeder_service.stop() if experiment_mode == constants.SKETCHDB_EXPERIMENT_NAME: query_engine_service.stop() diff --git a/asap-tools/experiments/experiment_utils/core.py b/asap-tools/experiments/experiment_utils/core.py index 2bb24960..2a840130 100644 --- a/asap-tools/experiments/experiment_utils/core.py +++ b/asap-tools/experiments/experiment_utils/core.py @@ -134,6 +134,17 @@ def read_exporter_config(experiment_params: DictConfig) -> Tuple[Optional[Dict], "cluster_data_exporter with parts_mode='part-index' requires 'part_index'", ) + from experiment_utils.services.paced_feeder import ( + PACED_FEEDER_EXPORTER, + PACED_FEEDER_REQUIRED_KEYS, + ) + + if PACED_FEEDER_EXPORTER in exporters_config.exporter_list: + feeder_config = exporters_config.exporter_list[PACED_FEEDER_EXPORTER] + missing_keys = [k for k in PACED_FEEDER_REQUIRED_KEYS if k not in feeder_config] + if missing_keys: + return None, f"Missing keys in paced_feeder section: {missing_keys}" + return exporters_config, "" @@ -186,9 +197,25 @@ def call_generate_prometheus_config( import generate_prometheus_config import generate_victoriametrics_config + from experiment_utils.services.paced_feeder import paced_feeder_metric_names + + # The paced feeder writes its metrics to the query engine itself, so + # Prometheus forwarding them too would send them twice. + fed_metrics = paced_feeder_metric_names(cfg.experiment_params) + metrics_to_remote_write = [ + m + for m in get_metrics_to_remote_write(cfg.experiment_params) + if m not in fed_metrics + ] + # With only fed metrics there is nothing for Prometheus to forward. (An + # empty list without the feeder keeps meaning "forward everything".) + forward_to_query_engine = bool(metrics_to_remote_write) or not fed_metrics # Set up remote write if this is a SketchDB experiment - if experiment_mode == sketchdb_experiment_name and sketchdb_experiment_name: - metrics_to_remote_write = get_metrics_to_remote_write(cfg.experiment_params) + if ( + experiment_mode == sketchdb_experiment_name + and sketchdb_experiment_name + and forward_to_query_engine + ): args.remote_write_metric_names = metrics_to_remote_write # Build remote_write_url from experiment params diff --git a/asap-tools/experiments/experiment_utils/services/__init__.py b/asap-tools/experiments/experiment_utils/services/__init__.py index d560b97d..e4581dbb 100644 --- a/asap-tools/experiments/experiment_utils/services/__init__.py +++ b/asap-tools/experiments/experiment_utils/services/__init__.py @@ -34,6 +34,12 @@ ControllerService, ) from .grafana import GrafanaService +from .paced_feeder import ( + PACED_FEEDER_EXPORTER, + PacedFeederService, + check_paced_feeder_config, + paced_feeder_sinks, +) from .clickhouse_service import ClickHouseService, ClickHouseDataLoaderService @@ -71,6 +77,10 @@ def create_prometheus_service(cfg, provider, num_nodes: int, node_offset: int): ) monitoring = cfg.experiment_params.monitoring + # The paced feeder pushes over remote write instead of being scraped. + remote_write_receiver = ( + PACED_FEEDER_EXPORTER in cfg.experiment_params.exporters.exporter_list + ) # Validate required fields if not hasattr(monitoring, "tool"): @@ -108,7 +118,9 @@ def create_prometheus_service(cfg, provider, num_nodes: int, node_offset: int): # Create appropriate service based on deployment mode if monitoring.deployment_mode == "containerized": if monitoring.tool == "prometheus": - return DockerPrometheusService(provider, num_nodes, node_offset) + return DockerPrometheusService( + provider, num_nodes, node_offset, remote_write_receiver + ) elif monitoring.tool == "victoriametrics": return DockerVictoriaMetricsService(provider, num_nodes, node_offset) else: # bare_metal @@ -117,7 +129,9 @@ def create_prometheus_service(cfg, provider, num_nodes: int, node_offset: int): "ERROR: VictoriaMetrics only supports containerized deployment. " "Use tool: prometheus for bare_metal deployment." ) - return PrometheusService(provider, num_nodes, node_offset) + return PrometheusService( + provider, num_nodes, node_offset, remote_write_receiver + ) __all__ = [ @@ -147,4 +161,8 @@ def create_prometheus_service(cfg, provider, num_nodes: int, node_offset: int): "create_prometheus_service", "ClickHouseService", "ClickHouseDataLoaderService", + "PACED_FEEDER_EXPORTER", + "PacedFeederService", + "check_paced_feeder_config", + "paced_feeder_sinks", ] diff --git a/asap-tools/experiments/experiment_utils/services/docker_prometheus.py b/asap-tools/experiments/experiment_utils/services/docker_prometheus.py index 88c0e519..8475344c 100644 --- a/asap-tools/experiments/experiment_utils/services/docker_prometheus.py +++ b/asap-tools/experiments/experiment_utils/services/docker_prometheus.py @@ -6,14 +6,22 @@ from .base import DockerServiceBase from experiment_utils.providers.base import InfrastructureProvider -from constants import PROMETHEUS_CONFIG_DIR, PROMETHEUS_CONFIG_FILE +from constants import ( + PROMETHEUS_CONFIG_DIR, + PROMETHEUS_CONFIG_FILE, + PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG, +) class DockerPrometheusService(DockerServiceBase): """Docker-based Prometheus service with resource constraints.""" def __init__( - self, provider: InfrastructureProvider, num_nodes: int, node_offset: int + self, + provider: InfrastructureProvider, + num_nodes: int, + node_offset: int, + remote_write_receiver: bool = False, ): """ Initialize Docker Prometheus service. @@ -22,9 +30,11 @@ def __init__( provider: Infrastructure provider for node communication and management num_nodes: Number of nodes to manage node_offset: Starting node index offset + remote_write_receiver: Accept samples on /api/v1/write """ super().__init__(provider, num_nodes, node_offset) self.container_name = "prometheus-container" + self.remote_write_receiver = remote_write_receiver def get_container_name(self) -> str: """Get the Docker container name.""" @@ -100,6 +110,8 @@ def start( "--web.enable-lifecycle", ] ) + if self.remote_write_receiver: + docker_cmd_parts.append(PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG) docker_cmd = " ".join(docker_cmd_parts) diff --git a/asap-tools/experiments/experiment_utils/services/fake_exporters.py b/asap-tools/experiments/experiment_utils/services/fake_exporters.py index ade2eaef..e080084f 100644 --- a/asap-tools/experiments/experiment_utils/services/fake_exporters.py +++ b/asap-tools/experiments/experiment_utils/services/fake_exporters.py @@ -42,7 +42,7 @@ def get_fake_exporter_seed( return derived_seed -def get_fake_exporter_target_nodes(args) -> List[int]: +def get_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()] @@ -429,7 +429,7 @@ def _start_bare_metal( dataset = config["dataset"] commands_by_port: List[List[Tuple[int, str]]] = [] - target_nodes = get_fake_exporter_target_nodes(self.args) + target_nodes = get_exporter_target_nodes(self.args) for port_ordinal in range(num_ports): port_commands = [] for worker_ordinal, node_idx in enumerate(target_nodes): @@ -488,7 +488,7 @@ def _start_containerized( # Build docker run commands for each port docker_run_cmds: List[Tuple[int, str]] = [] container_names: List[str] = [] - target_nodes = get_fake_exporter_target_nodes(self.args) + target_nodes = get_exporter_target_nodes(self.args) for worker_ordinal, node_idx in enumerate(target_nodes): for port_ordinal in range(num_ports): @@ -579,7 +579,7 @@ 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) + target_nodes = get_exporter_target_nodes(self.args) self.provider.execute_command_parallel( node_idxs=target_nodes, cmd=cmd, @@ -591,7 +591,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) + target_nodes = get_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 diff --git a/asap-tools/experiments/experiment_utils/services/paced_feeder.py b/asap-tools/experiments/experiment_utils/services/paced_feeder.py new file mode 100644 index 00000000..c01abb3e --- /dev/null +++ b/asap-tools/experiments/experiment_utils/services/paced_feeder.py @@ -0,0 +1,223 @@ +""" +Paced feeder service: pushes a sketch-bench workload table's metrics in real +time over Prometheus remote write, one feeder process per (sink, metric). + +Unlike the fake exporters, nothing is scraped: the feeder writes straight to +ASAPQuery's ingest and/or Prometheus' remote-write receiver. Every process gets +the same seed and table, so each system receives identical data. +""" + +import json +import math +import os +import time +from typing import Any, Dict, List, Tuple + +import constants +from .base import BaseService +from .fake_exporters import get_exporter_target_nodes +from experiment_utils import sync +from experiment_utils.providers.base import InfrastructureProvider + +PACED_FEEDER_EXPORTER = "paced_feeder" +PACED_FEEDER_REQUIRED_KEYS = ("workload_table", "metrics", "seed", "duration_secs") +# Metric types a remote-write sink (ASAPQuery ingest, Prometheus) accepts. +REMOTE_WRITE_METRIC_TYPES = ("gauge", "counter") + +PACED_FEEDER_BINARY = "./target/release/paced_feeder" +# The brackets keep pgrep/pkill from matching the shell that runs them. +PACED_FEEDER_PROCESS_PATTERN = "[t]arget/release/paced_feeder" +# Feeders launched one after another over SSH all start at this much later +# event time, so the sinks of one mode get the same timestamps. +PACED_FEEDER_START_LEAD_SECS = 5 +# A feeder stops after its first tick if every request fails (e.g. a wrong +# URL), so check the processes one tick after the shared start. +PACED_FEEDER_STARTUP_CHECK_SECS = PACED_FEEDER_START_LEAD_SECS + 2 + +SINK_QUERY_ENGINE = "query_engine" +SINK_PROMETHEUS = "prometheus" + + +def paced_feeder_metric_names(experiment_params) -> List[str]: + """Metrics the paced feeder sends, per the experiment's `metrics` list.""" + return [ + m["metric"] + for m in experiment_params.get("metrics", []) + if m["exporter"] == PACED_FEEDER_EXPORTER + ] + + +def check_paced_feeder_config( + config: Dict[str, Any], parallelism: int, min_feed_secs: float +) -> None: + """Fail before the experiment starts on a config the feeder would reject + or one that stops sending before the queries end.""" + with open(config["workload_table"]) as f: + (workload,) = json.load(f)["workloads"] + table_metrics = workload["e2e"]["metrics"] + for metric in config["metrics"]: + if metric not in table_metrics: + raise ValueError( + f"paced_feeder metric {metric!r} is not in " + f"{config['workload_table']} (has {sorted(table_metrics)})" + ) + metric_type = table_metrics[metric]["type"] + if metric_type not in REMOTE_WRITE_METRIC_TYPES: + raise ValueError( + f"paced_feeder metric {metric!r} has type {metric_type!r}; " + f"remote write takes only {REMOTE_WRITE_METRIC_TYPES}" + ) + if parallelism != 1: + raise ValueError( + "paced_feeder sends to one ingest port; set streaming.parallelism=1" + ) + if config["duration_secs"] < min_feed_secs: + raise ValueError( + f"paced_feeder.duration_secs={config['duration_secs']} ends before the " + f"queries do: needs at least {min_feed_secs} " + "(flow.steady_state_wait + the query groups' running time)" + ) + + +def paced_feeder_sinks( + mode: Dict[str, Any], query_engine_url: str, prometheus_url: str +) -> Dict[str, str]: + """Sink name -> remote-write URL for one experiment mode. + + sketchdb feeds ASAPQuery's ingest, plus Prometheus when the mode also + queries it; every other mode queries Prometheus only. + """ + if mode["mode"] != constants.SKETCHDB_EXPERIMENT_NAME: + return {SINK_PROMETHEUS: prometheus_url} + sinks = {SINK_QUERY_ENGINE: query_engine_url} + if mode.get("query_prometheus_too", False): + sinks[SINK_PROMETHEUS] = prometheus_url + return sinks + + +def paced_feeder_command( + config: Dict[str, Any], + workload_table: str, + url: str, + metric: str, + start_ms: int, + stats_out: str, +) -> str: + cmd = ( + f"{PACED_FEEDER_BINARY} --sink remote-write --url {url}" + f" --workload-table {workload_table} --metric {metric}" + f" --seed {config['seed']} --duration-secs {config['duration_secs']}" + f" --start-ms {start_ms} --stats-out {stats_out}" + ) + if config.get("samples_per_sec") is not None: + cmd += f" --samples-per-sec {config['samples_per_sec']}" + if config.get("extra_flags"): + cmd += f" {config['extra_flags']}" + return cmd + + +class PacedFeederService(BaseService): + """Runs paced_feeder processes bare-metal, spread over the worker nodes + (the coordinator in local mode).""" + + def __init__(self, provider: InfrastructureProvider, args): + super().__init__(provider) + self.target_nodes = get_exporter_target_nodes(args) + self.cmd_dir = os.path.join( + provider.get_home_dir(), + "code", + "asap-tools", + "data-sources", + "paced-feeder", + ) + + def start( + self, + config: Dict[str, Any], + sinks: Dict[str, str], + experiment_output_dir: str, + local_experiment_dir: str, + **kwargs, + ) -> None: + output_dir = os.path.join(experiment_output_dir, "paced_feeder_output") + # One shared event-time origin for every process of this mode. + start_ms = (math.ceil(time.time()) + PACED_FEEDER_START_LEAD_SECS) * 1000 + + jobs = [(sink, metric) for sink in sinks for metric in config["metrics"]] + job_nodes = [ + self.target_nodes[i % len(self.target_nodes)] for i in range(len(jobs)) + ] + used_nodes = sorted(set(job_nodes)) + + for node_idx in used_nodes: + self.provider.execute_command( + node_idx=node_idx, + cmd=f"test -x {PACED_FEEDER_BINARY}", + cmd_dir=self.cmd_dir, + nohup=False, + popen=False, + ignore_errors=False, + ) + # Same path on every node; also creates output_dir. + workload_table = sync.rsync_dataset_file( + self.provider, config["workload_table"], output_dir, node_idx + ) + + node_cmds: List[Tuple[int, str]] = [] + for node_idx, (sink, metric) in zip(job_nodes, jobs): + name = f"{sink}_{metric}" + cmd = paced_feeder_command( + config, + workload_table, + sinks[sink], + metric, + start_ms, + os.path.join(output_dir, f"{name}_stats.json"), + ) + node_cmds.append( + (node_idx, f"{cmd} > {os.path.join(output_dir, name)}.out 2>&1 &") + ) + + local_config_dir = os.path.join(local_experiment_dir, "paced_feeder_config") + os.makedirs(local_config_dir, exist_ok=True) + with open(os.path.join(local_config_dir, "cmds.sh"), "w") as f: + f.write("\n".join(cmd for _, cmd in node_cmds) + "\n") + + for node_idx, cmd in node_cmds: + self.provider.execute_command( + node_idx=node_idx, + cmd=cmd, + cmd_dir=self.cmd_dir, + nohup=True, + popen=False, + ignore_errors=False, + ) + + time.sleep(PACED_FEEDER_STARTUP_CHECK_SECS) + for node_idx in used_nodes: + expected = job_nodes.count(node_idx) + # No $(...): the SSH command is double-quoted on this side. + result = self.provider.execute_command( + node_idx=node_idx, + cmd=f"pgrep -fc '{PACED_FEEDER_PROCESS_PATTERN}' | grep -qx {expected}", + cmd_dir=None, + nohup=False, + popen=False, + ignore_errors=True, + ) + if result.returncode != 0: + raise RuntimeError( + f"paced_feeder: expected {expected} processes running on " + f"node {node_idx}; see the .out files in {output_dir}" + ) + + def stop(self, **kwargs) -> None: + self.provider.execute_command_parallel( + node_idxs=self.target_nodes, + cmd=f"pkill -f '{PACED_FEEDER_PROCESS_PATTERN}'", + cmd_dir=None, + nohup=False, + popen=True, + wait=True, + ignore_errors=True, + ) diff --git a/asap-tools/experiments/experiment_utils/services/prometheus.py b/asap-tools/experiments/experiment_utils/services/prometheus.py index 07d54137..57697a7e 100644 --- a/asap-tools/experiments/experiment_utils/services/prometheus.py +++ b/asap-tools/experiments/experiment_utils/services/prometheus.py @@ -8,14 +8,22 @@ from .base import BaseService from experiment_utils.providers.base import InfrastructureProvider -from constants import PROMETHEUS_CONFIG_DIR, PROMETHEUS_CONFIG_FILE +from constants import ( + PROMETHEUS_CONFIG_DIR, + PROMETHEUS_CONFIG_FILE, + PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG, +) class PrometheusService(BaseService): """Service for managing Prometheus operations.""" def __init__( - self, provider: InfrastructureProvider, num_nodes: int, node_offset: int + self, + provider: InfrastructureProvider, + num_nodes: int, + node_offset: int, + remote_write_receiver: bool = False, ): """ Initialize Prometheus service. @@ -24,10 +32,12 @@ def __init__( provider: Infrastructure provider for node communication and management num_nodes: Number of nodes to manage node_offset: Starting node index offset + remote_write_receiver: Accept samples on /api/v1/write """ super().__init__(provider) self.num_nodes = num_nodes self.node_offset = node_offset + self.remote_write_receiver = remote_write_receiver def get_query_endpoint_port(self) -> int: """Get the query endpoint port for Prometheus.""" @@ -89,7 +99,10 @@ def _start_prometheus(self, experiment_output_dir: str) -> None: time.sleep(3) # Start prometheus - cmd = f"./prometheus --config.file={PROMETHEUS_CONFIG_FILE} > /dev/null 2>&1 < /dev/null &" + flags = f"--config.file={PROMETHEUS_CONFIG_FILE}" + if self.remote_write_receiver: + flags += f" {PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG}" + cmd = f"./prometheus {flags} > /dev/null 2>&1 < /dev/null &" self.provider.execute_command( node_idx=self.node_offset, cmd=cmd, diff --git a/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py b/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py new file mode 100644 index 00000000..27a5eb9d --- /dev/null +++ b/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py @@ -0,0 +1,239 @@ +"""Write an experiment_run_e2e.py config for a sketch-bench workload table. + +The table's `workloads[0].e2e` block lists the end-to-end queries and the +metrics the paced feeder generates. The config has: + +- one query group per repeat interval T (`interval_secs`), repeating every T + with evaluation times aligned to multiples of T (`align_to_interval`). PromQL + entries keep the table's order, as in `asap-planner --planner saved`. +- structured entries (no `promql`, the multi-grouping templates) as SQL query + groups, only when `--sql-queries` gives their SQL; otherwise they are + skipped with a note. The file holds one `;`-terminated statement per + structured entry, in table order. +- the table's gauge and counter metrics, sent by the paced feeder (records + metrics are skipped with a note: remote write cannot carry them). +- `controller.prebuilt_output_dir`, when given, so ASAPQuery runs those + planner configs (e.g. from `asap-planner --planner saved`). + +Usage (from asap-tools/experiments): + python sketch_bench_e2e/sketch_bench_e2e_configs.py \\ + --workload-table /.../synthetic-templatesclassic-shared1.json \\ + --seed 42 --starting-delay-secs 120 --query-duration-secs 3600 \\ + [--prebuilt-output-dir DIR] [--sql-queries FILE] +Then: + python experiment_run_e2e.py experiment_type=sketch_bench_e2e/ ... +""" + +import argparse +import json +import os +import sys +from typing import Dict, List, Optional, Tuple + +import yaml + +EXPERIMENTS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, EXPERIMENTS_DIR) + +from experiment_utils.config import ( # noqa: E402 + _load_sql_queries, + check_prebuilt_controller_output, +) +from experiment_utils.services.paced_feeder import ( # noqa: E402 + PACED_FEEDER_EXPORTER, + REMOTE_WRITE_METRIC_TYPES, +) + +DEFAULT_OUTPUT_DIR = os.path.join( + EXPERIMENTS_DIR, "config", "experiment_type", "sketch_bench_e2e" +) +SERVERS = [ + {"name": "prometheus", "url": "http://localhost:9090"}, + {"name": "sketchdb", "url": "http://localhost:8088"}, +] +MODES = [ + {"mode": "sketchdb", "server": "sketchdb"}, + {"mode": "baseline", "server": "prometheus"}, +] +# Seconds the window ending at an evaluation time gets for its data to arrive. +DEFAULT_QUERY_TIME_OFFSET_SECS = 10 +# Feeder time beyond the query phase: covers flow.steady_state_wait (60 s in +# config.yaml) and service startup. The runner checks the total. +FEEDER_SLACK_SECS = 300 + + +def load_e2e(table_path: str) -> dict: + with open(table_path) as f: + (workload,) = json.load(f)["workloads"] + return workload["e2e"] + + +def by_interval(entries: List[Tuple[int, str]]) -> Dict[int, List[str]]: + groups: Dict[int, List[str]] = {} + for interval_secs, query in entries: + groups.setdefault(interval_secs, []).append(query) + return dict(sorted(groups.items())) + + +def build_config( + table_path: str, + seed: int, + starting_delay_secs: int, + query_duration_secs: int, + query_time_offset_secs: int = DEFAULT_QUERY_TIME_OFFSET_SECS, + feeder_duration_secs: Optional[int] = None, + samples_per_sec: Optional[int] = None, + prebuilt_output_dir: Optional[str] = None, + sql_queries_file: Optional[str] = None, +) -> Tuple[dict, List[str]]: + """Return (config, notes on what was skipped).""" + e2e = load_e2e(table_path) + notes: List[str] = [] + + promql = [ + (q["interval_secs"], q["promql"]) for q in e2e["queries"] if "promql" in q + ] + structured = [q for q in e2e["queries"] if "promql" not in q] + sql: List[Tuple[int, str]] = [] + if structured and sql_queries_file is None: + notes.append( + f"skipped {len(structured)} structured (multi-grouping) queries: " + "no --sql-queries file" + ) + elif structured: + statements = _load_sql_queries(sql_queries_file) + if len(statements) != len(structured): + raise ValueError( + f"{sql_queries_file} has {len(statements)} statements; the table " + f"has {len(structured)} structured queries" + ) + sql = [(q["interval_secs"], s) for q, s in zip(structured, statements)] + + query_groups = [] + for queries_by_interval in (by_interval(promql), by_interval(sql)): + for interval_secs, queries in queries_by_interval.items(): + if query_duration_secs < interval_secs: + raise ValueError( + f"--query-duration-secs {query_duration_secs} is shorter than " + f"the {interval_secs} s repeat interval" + ) + query_groups.append( + { + "id": len(query_groups) + 1, + "queries": queries, + "repetition_delay_ms": interval_secs * 1000, + "client_options": { + "repetitions": query_duration_secs // interval_secs, + "starting_delay": starting_delay_secs, + "query_time_offset": query_time_offset_secs, + "align_to_interval": True, + }, + } + ) + if not query_groups: + raise ValueError(f"{table_path}: no queries to run") + + metrics = [] + for name, desc in e2e["metrics"].items(): + if desc["type"] not in REMOTE_WRITE_METRIC_TYPES: + notes.append(f"skipped {desc['type']} metric {name!r}: not remote-writable") + continue + metrics.append( + { + "metric": name, + "labels": [label["name"] for label in desc["labels"]], + "exporter": PACED_FEEDER_EXPORTER, + } + ) + + feeder = { + "workload_table": os.path.abspath(table_path), + "metrics": [m["metric"] for m in metrics], + "seed": seed, + "duration_secs": ( + feeder_duration_secs + if feeder_duration_secs is not None + else starting_delay_secs + query_duration_secs + FEEDER_SLACK_SECS + ), + } + if samples_per_sec is not None: + feeder["samples_per_sec"] = samples_per_sec + + config: dict = { + "experiment_params": { + "experiment": MODES, + "monitoring": {"tool": "prometheus", "deployment_mode": "bare_metal"}, + "servers": SERVERS, + "exporters": { + "only_start_if_queries_exist": False, + "exporter_list": {PACED_FEEDER_EXPORTER: feeder}, + }, + "query_groups": query_groups, + "metrics": metrics, + } + } + if prebuilt_output_dir is not None: + prebuilt_output_dir = os.path.abspath(prebuilt_output_dir) + check_prebuilt_controller_output(prebuilt_output_dir) + config["controller"] = {"prebuilt_output_dir": prebuilt_output_dir} + return config, notes + + +def write_config(config: dict, table_path: str, output_path: str) -> None: + os.makedirs(os.path.dirname(output_path), exist_ok=True) + with open(output_path, "w") as f: + f.write("# @package _global_\n") + f.write(f"# Generated by sketch_bench_e2e_configs.py from {table_path}\n") + yaml.safe_dump(config, f, sort_keys=False) + + +def main() -> None: + parser = argparse.ArgumentParser( + description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter + ) + parser.add_argument("--workload-table", required=True) + parser.add_argument("--seed", type=int, required=True) + parser.add_argument("--starting-delay-secs", type=int, required=True) + parser.add_argument( + "--query-duration-secs", + type=int, + required=True, + help="query phase length; each group repeats this // T times", + ) + parser.add_argument( + "--query-time-offset-secs", + type=int, + default=DEFAULT_QUERY_TIME_OFFSET_SECS, + ) + parser.add_argument( + "--feeder-duration-secs", + type=int, + help=f"default: starting delay + query duration + {FEEDER_SLACK_SECS}", + ) + parser.add_argument("--samples-per-sec", type=int) + parser.add_argument("--prebuilt-output-dir") + parser.add_argument("--sql-queries") + parser.add_argument("--output", help="default: /
.yaml") + args = parser.parse_args() + + config, notes = build_config( + args.workload_table, + seed=args.seed, + starting_delay_secs=args.starting_delay_secs, + query_duration_secs=args.query_duration_secs, + query_time_offset_secs=args.query_time_offset_secs, + feeder_duration_secs=args.feeder_duration_secs, + samples_per_sec=args.samples_per_sec, + prebuilt_output_dir=args.prebuilt_output_dir, + sql_queries_file=args.sql_queries, + ) + stem = os.path.splitext(os.path.basename(args.workload_table))[0] + output = args.output or os.path.join(DEFAULT_OUTPUT_DIR, f"{stem}.yaml") + write_config(config, args.workload_table, output) + for note in notes: + print(f"note: {note}") + print(output) + + +if __name__ == "__main__": + main() diff --git a/asap-tools/experiments/tests/test_paced_feeder.py b/asap-tools/experiments/tests/test_paced_feeder.py new file mode 100644 index 00000000..6e55fdc9 --- /dev/null +++ b/asap-tools/experiments/tests/test_paced_feeder.py @@ -0,0 +1,367 @@ +"""Tests for running the paced feeder from the e2e runner.""" + +import json +import os +import subprocess +import tempfile +import unittest +from unittest import mock + +from omegaconf import OmegaConf + +import constants +from experiment_utils.core import call_generate_prometheus_config, read_exporter_config +from experiment_utils.services import create_prometheus_service +from experiment_utils.services.docker_prometheus import DockerPrometheusService +from experiment_utils.services.paced_feeder import ( + PacedFeederService, + check_paced_feeder_config, + paced_feeder_command, + paced_feeder_sinks, +) +from experiment_utils.services.prometheus import PrometheusService + +QE_URL = "http://10.10.1.1:8080/api/v1/write" +PROM_URL = "http://10.10.1.1:9090/api/v1/write" + +# workloads[0].e2e.metrics of synthetic-templatesclassic-shared1.json, plus a +# records metric from the multi-grouping tables. +LABELS = [ + {"cardinality": 10, "name": "job"}, + {"cardinality": 1000, "child_of": "job", "name": "label_0"}, +] +E2E_METRICS = { + "data": { + "labels": LABELS, + "samples_per_sec_per_series": 200, + "type": "gauge", + "values": {"dist": {"pareto": 2.0}, "value": {"keys": 10000, "zipf": 1.1}}, + }, + "data_total": { + "labels": LABELS, + "samples_per_sec_per_series": 200, + "type": "counter", + "values": {"increment": {"keys": 10000, "zipf": 1.1}}, + }, + "flows": { + "labels": [{"cardinality": 3, "name": "proto", "skew": 0.0}], + "records_per_sec": 4000000, + "type": "records", + "values": {"src_ip": {"keys": 1000000, "zipf": 0.0}}, + }, +} + + +def write_table(directory, queries=()): + path = os.path.join(directory, "table.json") + with open(path, "w") as f: + json.dump( + {"workloads": [{"e2e": {"metrics": E2E_METRICS, "queries": queries}}]}, f + ) + return path + + +def feeder_config(table, **overrides): + return { + "workload_table": table, + "metrics": ["data", "data_total"], + "seed": 42, + "duration_secs": 600, + **overrides, + } + + +class PacedFeederCommandTest(unittest.TestCase): + def test_command_line(self): + cmd = paced_feeder_command( + feeder_config("/t/table.json"), + "/out/table.json", + QE_URL, + "data", + 1700000005000, + "/out/query_engine_data_stats.json", + ) + self.assertEqual( + cmd, + "./target/release/paced_feeder --sink remote-write" + f" --url {QE_URL} --workload-table /out/table.json --metric data" + " --seed 42 --duration-secs 600 --start-ms 1700000005000" + " --stats-out /out/query_engine_data_stats.json", + ) + + def test_optional_flags_are_appended(self): + cmd = paced_feeder_command( + feeder_config("/t", samples_per_sec=20, extra_flags="--concurrency 4"), + "/t", + QE_URL, + "data", + 0, + "/s", + ) + self.assertTrue(cmd.endswith(" --samples-per-sec 20 --concurrency 4")) + + +class PacedFeederSinksTest(unittest.TestCase): + def test_sketchdb_feeds_the_query_engine_only(self): + self.assertEqual( + paced_feeder_sinks({"mode": "sketchdb"}, QE_URL, PROM_URL), + {"query_engine": QE_URL}, + ) + + def test_sketchdb_querying_prometheus_too_feeds_both(self): + self.assertEqual( + paced_feeder_sinks( + {"mode": "sketchdb", "query_prometheus_too": True}, QE_URL, PROM_URL + ), + {"query_engine": QE_URL, "prometheus": PROM_URL}, + ) + + def test_baseline_feeds_prometheus(self): + self.assertEqual( + paced_feeder_sinks({"mode": "baseline"}, QE_URL, PROM_URL), + {"prometheus": PROM_URL}, + ) + + +class CheckPacedFeederConfigTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.table = write_table(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def test_valid_config(self): + check_paced_feeder_config(feeder_config(self.table), 1, 600) + + def test_unknown_metric(self): + with self.assertRaisesRegex(ValueError, "'cpu' is not in"): + check_paced_feeder_config( + feeder_config(self.table, metrics=["cpu"]), 1, 600 + ) + + def test_records_metric(self): + with self.assertRaisesRegex(ValueError, "type 'records'"): + check_paced_feeder_config( + feeder_config(self.table, metrics=["flows"]), 1, 600 + ) + + def test_sharded_ingest(self): + with self.assertRaisesRegex(ValueError, "parallelism=1"): + check_paced_feeder_config(feeder_config(self.table), 2, 600) + + def test_feeder_ending_before_the_queries(self): + with self.assertRaisesRegex(ValueError, "at least 601"): + check_paced_feeder_config(feeder_config(self.table), 1, 601) + + def test_missing_keys_rejected_with_the_other_exporters(self): + params = OmegaConf.create( + { + "exporters": { + "only_start_if_queries_exist": False, + "exporter_list": {"paced_feeder": {"workload_table": "t"}}, + } + } + ) + config, reason = read_exporter_config(params) + self.assertIsNone(config) + self.assertIn("duration_secs", reason) + + +class RecordingProvider: + def __init__(self, running_count): + self.commands = [] + self.running_count = running_count + + def get_home_dir(self): + return "/home/u" + + def is_remote(self): + return True + + def execute_command(self, node_idx, cmd, **kwargs): + self.commands.append((node_idx, cmd)) + returncode = 0 + if cmd.startswith("pgrep"): + returncode = 0 if cmd.endswith(f"-qx {self.running_count}") else 1 + return subprocess.CompletedProcess(cmd, returncode) + + +class Args: + def get_node_range(self, include_coordinator=True): + return [0, 1, 2] if include_coordinator else [1, 2] + + def get_coordinator_node(self): + return 0 + + +class PacedFeederServiceStartTest(unittest.TestCase): + def start(self, running_count): + provider = RecordingProvider(running_count) + service = PacedFeederService(provider, Args()) + with tempfile.TemporaryDirectory() as local_dir, mock.patch( + "experiment_utils.services.paced_feeder.time.sleep" + ), mock.patch( + "experiment_utils.services.paced_feeder.sync.rsync_dataset_file", + return_value="/exp/paced_feeder_output/table.json", + ): + service.start( + config=feeder_config("/local/table.json"), + sinks={"query_engine": QE_URL, "prometheus": PROM_URL}, + experiment_output_dir="/exp", + local_experiment_dir=local_dir, + ) + return provider.commands + + def test_one_process_per_sink_and_metric_spread_over_workers(self): + commands = self.start(running_count=2) + feeders = [(n, c) for n, c in commands if "--sink" in c] + self.assertEqual( + [ + ( + n, + c.split(" --url ")[1].split()[0], + c.split(" --metric ")[1].split()[0], + ) + for n, c in feeders + ], + [ + (1, QE_URL, "data"), + (2, QE_URL, "data_total"), + (1, PROM_URL, "data"), + (2, PROM_URL, "data_total"), + ], + ) + # One event-time origin, so the sinks get the same timestamps. + self.assertEqual( + len({c.split(" --start-ms ")[1].split()[0] for _, c in feeders}), 1 + ) + self.assertTrue( + all( + "--workload-table /exp/paced_feeder_output/table.json" in c + for _, c in feeders + ) + ) + + def test_dead_feeder_fails_the_experiment(self): + # A wrong URL stops a feeder after its first tick; without this check + # the run would carry on with no data. + with self.assertRaisesRegex(RuntimeError, "expected 2 processes"): + self.start(running_count=1) + + +class StartCommandProvider: + def __init__(self): + self.commands = [] + + def get_home_dir(self): + return "/home/u" + + def execute_command(self, node_idx, cmd, **kwargs): + self.commands.append(cmd) + # netstat finds nothing on the Prometheus port. + return subprocess.CompletedProcess(cmd, 1) + + +class PrometheusRemoteWriteReceiverTest(unittest.TestCase): + def bare_metal_command(self, receiver): + provider = StartCommandProvider() + PrometheusService(provider, 1, 0, remote_write_receiver=receiver).start("/exp") + return provider.commands[-1] + + def test_bare_metal_flag(self): + self.assertIn( + constants.PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG, + self.bare_metal_command(True), + ) + self.assertNotIn( + constants.PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG, + self.bare_metal_command(False), + ) + + def test_containerized_flag(self): + provider = StartCommandProvider() + service = DockerPrometheusService(provider, 1, 0, remote_write_receiver=True) + with mock.patch.object(service, "_force_cleanup_container"), mock.patch.object( + service, "_wait_for_service_ready" + ): + service.start("/exp") + self.assertIn( + constants.PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG, provider.commands[-1] + ) + + def test_enabled_when_the_paced_feeder_is_configured(self): + def service(exporter_list): + cfg = OmegaConf.create( + { + "experiment_params": { + "monitoring": { + "tool": "prometheus", + "deployment_mode": "bare_metal", + }, + "exporters": {"exporter_list": exporter_list}, + } + } + ) + return create_prometheus_service(cfg, StartCommandProvider(), 1, 0) + + self.assertTrue(service({"paced_feeder": {}}).remote_write_receiver) + self.assertFalse(service({"fake_exporter": {}}).remote_write_receiver) + + +class PrometheusForwardingTest(unittest.TestCase): + """In sketchdb mode Prometheus must not forward fed metrics to the query + engine, which the feeder already writes to.""" + + def remote_write_args(self, metrics): + cfg = OmegaConf.create( + { + "streaming": { + "parallelism": 1, + "remote_write": { + "ip": "10.10.1.1", + "base_port": 8080, + "path": "/api/v1/write", + }, + }, + "experiment_params": { + "exporters": { + "only_start_if_queries_exist": False, + "exporter_list": {}, + }, + "query_groups": [], + "metrics": metrics, + }, + } + ) + args = mock.Mock(remote_write_url=None, remote_write_metric_names=None) + with mock.patch("generate_prometheus_config.main"), mock.patch( + "generate_victoriametrics_config.main" + ): + call_generate_prometheus_config( + args, + cfg, + constants.SKETCHDB_EXPERIMENT_NAME, + constants.SKETCHDB_EXPERIMENT_NAME, + ) + return args + + def test_only_fed_metrics_means_no_remote_write(self): + args = self.remote_write_args( + [{"metric": "data", "labels": ["job"], "exporter": "paced_feeder"}] + ) + self.assertIsNone(args.remote_write_url) + + def test_scraped_metrics_are_still_forwarded(self): + args = self.remote_write_args( + [ + {"metric": "data", "labels": ["job"], "exporter": "paced_feeder"}, + {"metric": "node_cpu", "labels": ["job"], "exporter": "node_exporter"}, + ] + ) + self.assertEqual(args.remote_write_url, QE_URL) + self.assertEqual(args.remote_write_metric_names, ["node_cpu"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py b/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py new file mode 100644 index 00000000..f14dd2ba --- /dev/null +++ b/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py @@ -0,0 +1,204 @@ +"""Tests for sketch_bench_e2e_configs.py (e2e configs from a sketch-bench table).""" + +import json +import os +import shutil +import tempfile +import unittest + +from hydra import compose, initialize_config_dir +from omegaconf import OmegaConf + +from experiment_utils import config as experiment_config +from experiment_utils.core import read_exporter_config +from experiment_utils.services.paced_feeder import check_paced_feeder_config +from sketch_bench_e2e import sketch_bench_e2e_configs as sbe +from test_paced_feeder import E2E_METRICS + +CONFIG_DIR = os.path.join(sbe.EXPERIMENTS_DIR, "config") +PREFIX = "synthetic/templates=classic/shared=1" + +# A slice of synthetic-templatesclassic-shared1.json's e2e.queries, plus a +# structured entry from the multi-grouping tables. +QUERIES = [ + { + "interval_secs": 1, + "promql": "sum by (job) (data)", + "rqes": [f"{PREFIX}/sum_by_job/keys/1s"], + }, + { + "interval_secs": 60, + "promql": "sum_over_time(data[15m])", + "rqes": [f"{PREFIX}/sum_over_time/keys/15m"], + }, + { + "interval_secs": 1, + "promql": "quantile by (job) (0.5, data)", + "rqes": [f"{PREFIX}/quantile_by_job_p0.5/values/1s"], + }, + { + "interval_secs": 60, + "promql": "rate(data_total[15m])", + "rqes": [f"{PREFIX}/rate/keys/15m"], + }, + { + "capability": "cardinality", + "grouping": ["dst_subnet"], + "interval_secs": 60, + "rqes": ["synthetic/templates=all/shared=1/t11_distinct_src/keys/5m"], + "table": "flows", + "value": "src_ip", + "window_secs": 300, + }, +] + + +class BuildConfigTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.table = os.path.join(self.tmp.name, "table.json") + with open(self.table, "w") as f: + json.dump( + {"workloads": [{"e2e": {"metrics": E2E_METRICS, "queries": QUERIES}}]}, + f, + ) + + def tearDown(self): + self.tmp.cleanup() + + def build(self, **kwargs): + return sbe.build_config( + self.table, + seed=7, + starting_delay_secs=120, + query_duration_secs=600, + **kwargs, + ) + + def test_query_groups_by_interval_in_table_order(self): + config, _ = self.build() + groups = config["experiment_params"]["query_groups"] + self.assertEqual( + [(g["id"], g["repetition_delay_ms"], g["queries"]) for g in groups], + [ + (1, 1000, ["sum by (job) (data)", "quantile by (job) (0.5, data)"]), + (2, 60000, ["sum_over_time(data[15m])", "rate(data_total[15m])"]), + ], + ) + self.assertEqual( + [g["client_options"] for g in groups], + [ + { + "repetitions": repetitions, + "starting_delay": 120, + "query_time_offset": sbe.DEFAULT_QUERY_TIME_OFFSET_SECS, + "align_to_interval": True, + } + for repetitions in (600, 10) + ], + ) + + def test_feeder_sends_the_remote_writable_metrics(self): + config, notes = self.build() + params = config["experiment_params"] + self.assertEqual( + params["metrics"], + [ + { + "metric": "data", + "labels": ["job", "label_0"], + "exporter": "paced_feeder", + }, + { + "metric": "data_total", + "labels": ["job", "label_0"], + "exporter": "paced_feeder", + }, + ], + ) + self.assertEqual( + params["exporters"]["exporter_list"]["paced_feeder"], + { + "workload_table": self.table, + "metrics": ["data", "data_total"], + "seed": 7, + "duration_secs": 120 + 600 + sbe.FEEDER_SLACK_SECS, + }, + ) + self.assertIn("skipped records metric 'flows': not remote-writable", notes) + + def test_structured_queries_skipped_without_sql(self): + config, notes = self.build() + self.assertEqual(len(config["experiment_params"]["query_groups"]), 2) + self.assertIn( + "skipped 1 structured (multi-grouping) queries: no --sql-queries file", + notes, + ) + + def test_structured_queries_from_sql_file(self): + sql = os.path.join(self.tmp.name, "multigroup.sql") + with open(sql, "w") as f: + f.write("SELECT dst_subnet, uniq(src_ip) FROM flows GROUP BY dst_subnet;\n") + config, notes = self.build(sql_queries_file=sql) + group = config["experiment_params"]["query_groups"][2] + self.assertEqual(group["id"], 3) + self.assertEqual(group["repetition_delay_ms"], 60000) + self.assertEqual( + group["queries"], + ["SELECT dst_subnet, uniq(src_ip) FROM flows GROUP BY dst_subnet"], + ) + self.assertFalse(any("structured" in n for n in notes)) + + def test_sql_file_must_match_the_structured_queries(self): + sql = os.path.join(self.tmp.name, "multigroup.sql") + with open(sql, "w") as f: + f.write("SELECT 1;\nSELECT 2;\n") + with self.assertRaisesRegex(ValueError, "2 statements; the table has 1"): + self.build(sql_queries_file=sql) + + def test_query_phase_shorter_than_an_interval(self): + with self.assertRaisesRegex(ValueError, "shorter than the 60 s"): + sbe.build_config( + self.table, seed=7, starting_delay_secs=0, query_duration_secs=30 + ) + + def test_prebuilt_output_dir(self): + prebuilt = os.path.join(self.tmp.name, "prebuilt") + os.makedirs(prebuilt) + with self.assertRaisesRegex(ValueError, "streaming_config.yaml"): + self.build(prebuilt_output_dir=prebuilt) + for name in experiment_config.PLANNER_OUTPUT_FILES: + open(os.path.join(prebuilt, name), "w").close() + config, _ = self.build(prebuilt_output_dir=prebuilt) + self.assertEqual(config["controller"], {"prebuilt_output_dir": prebuilt}) + + def test_generated_config_composes_and_validates(self): + # A dry run of what experiment_run_e2e.py does with the file before it + # starts any service. + config_dir = os.path.join(self.tmp.name, "config") + shutil.copytree(CONFIG_DIR, config_dir) + sbe.write_config( + self.build()[0], + self.table, + os.path.join(config_dir, "experiment_type", "sketch_bench_e2e", "t.yaml"), + ) + with initialize_config_dir(version_base=None, config_dir=config_dir): + cfg = compose("config", overrides=["experiment_type=sketch_bench_e2e/t"]) + params = cfg.experiment_params + experiment_config.validate_experiment_config(params) + exporter_config, reason = read_exporter_config(params) + self.assertIsNotNone(exporter_config, reason) + check_paced_feeder_config( + OmegaConf.to_container(exporter_config.exporter_list.paced_feeder), + parallelism=cfg.streaming.parallelism, + min_feed_secs=cfg.flow.steady_state_wait + + experiment_config.get_minimum_experiment_running_time(params), + ) + modes, _ = experiment_config.generate_controller_client_configs( + params, self.tmp.name + ) + self.assertEqual(modes, ["sketchdb", "baseline"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/asap-tools/installation/common/setup_dependencies.sh b/asap-tools/installation/common/setup_dependencies.sh index 6120a78a..2e3930a4 100755 --- a/asap-tools/installation/common/setup_dependencies.sh +++ b/asap-tools/installation/common/setup_dependencies.sh @@ -24,6 +24,7 @@ else fi # (cd "$THIS_DIR/../../experiments/fake_exporter_rust/fake_exporter"; cargo build --release) (cd "$THIS_DIR/../../../asap-tools/data-sources/prometheus-exporters/fake_exporter/fake_exporter_rust/fake_exporter" && cargo build --release) +(cd "$THIS_DIR/../../../asap-tools/data-sources/paced-feeder" && cargo build --release) DOCKER_DATA_DIR=/scratch/var_lib_docker diff --git a/asap-tools/queriers/prometheus-client/classes/query_time.py b/asap-tools/queriers/prometheus-client/classes/query_time.py new file mode 100644 index 00000000..af208c19 --- /dev/null +++ b/asap-tools/queriers/prometheus-client/classes/query_time.py @@ -0,0 +1,40 @@ +"""Evaluation times aligned to a query group's repeat interval. + +With `client_options.align_to_interval`, repetition k of a group with repeat +interval T (`repetition_delay_ms`) and settle offset O +(`client_options.query_time_offset`, seconds, default 0) evaluates at + + t_k = floor((start - O) / T) * T + k * T + +and is sent at wall-clock time t_k + O. Every evaluation time is a multiple of +T, so all systems answer the same windows (e.g. a 60 s query at whole minutes), +and O gives the data of the window ending at t_k time to arrive first. For +example, T = 60 s and O = 10 s, started at 12:00:35: repetitions evaluate at +12:00:00, 12:01:00, ... and are sent at 12:00:35 (late, so at once), +12:01:10, 12:02:10, ... +""" + +MS_PER_SEC = 1000 + + +def first_aligned_eval_time( + now_secs: float, interval_ms: int, offset_secs: float +) -> int: + """The latest multiple of the interval at least `offset_secs` before now.""" + if interval_ms <= 0 or interval_ms % MS_PER_SEC != 0: + raise ValueError( + f"align_to_interval needs a whole-second repetition_delay_ms, got {interval_ms}" + ) + if offset_secs < 0: + raise ValueError(f"query_time_offset must be >= 0, got {offset_secs}") + interval_secs = interval_ms // MS_PER_SEC + return int((now_secs - offset_secs) // interval_secs) * interval_secs + + +def aligned_eval_time(first_eval_time: int, interval_ms: int, repetition: int) -> int: + return first_eval_time + repetition * (interval_ms // MS_PER_SEC) + + +def secs_until_send(eval_time: int, offset_secs: float, now_secs: float) -> float: + """How long to wait before evaluating at `eval_time`; 0 when already late.""" + return max(0.0, eval_time + offset_secs - now_secs) diff --git a/asap-tools/queriers/prometheus-client/main_prometheus_client.py b/asap-tools/queriers/prometheus-client/main_prometheus_client.py index 104e4d68..15c5e73d 100644 --- a/asap-tools/queriers/prometheus-client/main_prometheus_client.py +++ b/asap-tools/queriers/prometheus-client/main_prometheus_client.py @@ -30,6 +30,11 @@ from classes.query_client import QueryClient from classes.query_client_factory import QueryClientFactory from classes.query_template import QueryTemplate +from classes.query_time import ( + aligned_eval_time, + first_aligned_eval_time, + secs_until_send, +) from promql_utilities.query_results.classes import QueryResult, QueryResultAcrossTime from promql_utilities.query_results.serializers import SerializerFactory @@ -316,20 +321,36 @@ def handle_query_group( for server_name in servers } - for repetition_idx in range(query_group.repetitions): - current_time = datetime.datetime.now() - logger.debug("Current unix time: {}", int(current_time.timestamp())) + # See classes/query_time.py. Aligned groups ignore query_start_times. + align_to_interval = getattr(query_group, "align_to_interval", False) + if align_to_interval: + offset_secs = getattr(query_group, "query_time_offset", 0) + first_eval_time = first_aligned_eval_time( + time.time(), query_group.repetition_delay_ms, offset_secs + ) - if hasattr(query_group, "query_time_offset"): - current_time = current_time - datetime.timedelta( - seconds=query_group.query_time_offset - ) - logger.debug( - "Offsetting query time by {} seconds", query_group.query_time_offset + for repetition_idx in range(query_group.repetitions): + if align_to_interval: + query_unix_time = aligned_eval_time( + first_eval_time, query_group.repetition_delay_ms, repetition_idx ) + time.sleep(secs_until_send(query_unix_time, offset_secs, time.time())) + logger.debug("Aligned evaluation time: {}", query_unix_time) + else: + current_time = datetime.datetime.now() + logger.debug("Current unix time: {}", int(current_time.timestamp())) + + if hasattr(query_group, "query_time_offset"): + current_time = current_time - datetime.timedelta( + seconds=query_group.query_time_offset + ) + logger.debug( + "Offsetting query time by {} seconds", + query_group.query_time_offset, + ) - query_unix_time = int(current_time.timestamp()) - logger.debug("Unix time after query_time_offset: {}", query_unix_time) + query_unix_time = int(current_time.timestamp()) + logger.debug("Unix time after query_time_offset: {}", query_unix_time) if parallel: # Execute queries in parallel @@ -337,11 +358,15 @@ def handle_query_group( futures = [] for local_query_idx, query in enumerate(query_group.queries): global_query_idx = global_query_idx_start + local_query_idx - current_query_unix_time = get_query_unix_time( - query, - query_unix_time, - query_start_times, - query_group.repetition_delay_ms, + current_query_unix_time = ( + query_unix_time + if align_to_interval + else get_query_unix_time( + query, + query_unix_time, + query_start_times, + query_group.repetition_delay_ms, + ) ) for server_name, server_object in servers.items(): @@ -380,11 +405,15 @@ def handle_query_group( # Serial execution - use the same execute_single_query function for local_query_idx, query in enumerate(query_group.queries): global_query_idx = global_query_idx_start + local_query_idx - current_query_unix_time = get_query_unix_time( - query, - query_unix_time, - query_start_times, - query_group.repetition_delay_ms, + current_query_unix_time = ( + query_unix_time + if align_to_interval + else get_query_unix_time( + query, + query_unix_time, + query_start_times, + query_group.repetition_delay_ms, + ) ) logger.debug("Unix time for query: {}", current_query_unix_time) @@ -428,7 +457,8 @@ def handle_query_group( if latency_exporter is not None: latency_exporter.export_repetition(repetition_idx, result) - if repetition_idx < query_group.repetitions - 1: + # Aligned groups wait for their next evaluation time instead. + if not align_to_interval and repetition_idx < query_group.repetitions - 1: time.sleep(query_group.repetition_delay_ms / 1000) if latency_exporter is not None: diff --git a/asap-tools/queriers/prometheus-client/tests/test_query_time.py b/asap-tools/queriers/prometheus-client/tests/test_query_time.py new file mode 100644 index 00000000..3ae121ab --- /dev/null +++ b/asap-tools/queriers/prometheus-client/tests/test_query_time.py @@ -0,0 +1,51 @@ +"""Tests for evaluation times aligned to a query group's repeat interval.""" + +import unittest + +from classes.query_time import ( + aligned_eval_time, + first_aligned_eval_time, + secs_until_send, +) + +# 2026-10-13 12:00:00 UTC, a multiple of 60 s. +MINUTE = 1791892800 + + +class FirstAlignedEvalTimeTest(unittest.TestCase): + def test_floors_to_the_interval_after_the_offset(self) -> None: + # Started at 12:00:35 with a 10 s offset: 12:00:25 floors to 12:00:00. + self.assertEqual(first_aligned_eval_time(MINUTE + 35.4, 60000, 10), MINUTE) + # Started at 12:00:05: the 12:00:00 window has not settled yet. + self.assertEqual(first_aligned_eval_time(MINUTE + 5, 60000, 10), MINUTE - 60) + + def test_one_second_interval(self) -> None: + self.assertEqual(first_aligned_eval_time(MINUTE + 7.9, 1000, 2), MINUTE + 5) + + def test_sub_second_interval_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "whole-second"): + first_aligned_eval_time(MINUTE, 1500, 0) + + def test_negative_offset_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "query_time_offset"): + first_aligned_eval_time(MINUTE, 60000, -1) + + +class ScheduleTest(unittest.TestCase): + def test_repetitions_step_by_the_interval(self) -> None: + self.assertEqual( + [aligned_eval_time(MINUTE, 60000, k) for k in range(3)], + [MINUTE, MINUTE + 60, MINUTE + 120], + ) + + def test_waits_until_offset_after_the_eval_time(self) -> None: + self.assertEqual(secs_until_send(MINUTE + 60, 10, MINUTE + 65), 5) + + def test_late_repetition_runs_at_once(self) -> None: + # A slow repetition makes the next one late; it keeps its eval time + # rather than skipping it. + self.assertEqual(secs_until_send(MINUTE + 60, 10, MINUTE + 75), 0) + + +if __name__ == "__main__": + unittest.main() From f7e5a66c4b98c7f11b4a641474cf3e54b2e71073 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:41:26 +0000 Subject: [PATCH 2/4] feat(asap-tools): give Prometheus a 5 s out-of-order window for the paced feeder The feeder sends each tick's requests concurrently, so a series' samples can reach Prometheus' remote-write receiver out of order within a second, and Prometheus rejects them without an out-of-order window. When the paced feeder is configured, the generated prometheus.yml (used by bare-metal and containerized Prometheus) sets storage.tsdb.out_of_order_time_window: 5s. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../CONFIG_PARAMETERS_REFERENCE.md | 4 +- asap-tools/experiments/constants.py | 4 ++ .../experiments/generate_prometheus_config.py | 10 +++- .../experiments/tests/test_paced_feeder.py | 52 ++++++++++++++++++- 4 files changed, 67 insertions(+), 3 deletions(-) diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index a85db47e..5050abb1 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md @@ -465,7 +465,9 @@ one process per (sink, metric). It pushes over remote write instead of being scraped: in `sketchdb` mode to the query engine's ingest (`streaming.remote_write`, which must have `parallelism: 1`), plus Prometheus when the mode sets `query_prometheus_too`; in every other mode to Prometheus, -which then starts with `--web.enable-remote-write-receiver`. All processes +which then starts with `--web.enable-remote-write-receiver` and a 5 s +`storage.tsdb.out_of_order_time_window` (a tick's requests are sent +concurrently, so samples can arrive out of order within a second). All processes share the seed and table, so every system gets the same data. Its metrics are listed under `metrics` with `exporter: paced_feeder`; Prometheus does not forward them to the query engine. Generate a whole config from a sketch-bench diff --git a/asap-tools/experiments/constants.py b/asap-tools/experiments/constants.py index 2faf6967..151999af 100644 --- a/asap-tools/experiments/constants.py +++ b/asap-tools/experiments/constants.py @@ -44,6 +44,10 @@ PROMETHEUS_REMOTE_WRITE_ENDPOINT = "/api/v1/write" # Lets Prometheus accept samples pushed to PROMETHEUS_REMOTE_WRITE_ENDPOINT. PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG = "--web.enable-remote-write-receiver" +# The paced feeder sends a tick's requests concurrently, so a series' samples +# can arrive out of order within that second; Prometheus rejects them without +# an out-of-order window. +PROMETHEUS_OUT_OF_ORDER_TIME_WINDOW = "5s" HTTP_READINESS_CURL_TIMEOUT_SECONDS = 5 QUERY_ENGINE_RUNTIME_INFO_ENDPOINT = "/api/v1/status/runtimeinfo" diff --git a/asap-tools/experiments/generate_prometheus_config.py b/asap-tools/experiments/generate_prometheus_config.py index a988ee72..bcd3beee 100644 --- a/asap-tools/experiments/generate_prometheus_config.py +++ b/asap-tools/experiments/generate_prometheus_config.py @@ -4,7 +4,9 @@ import argparse from omegaconf import DictConfig +import constants import experiment_utils +from experiment_utils.services.paced_feeder import PACED_FEEDER_EXPORTER from utils import get_node_range @@ -158,6 +160,13 @@ def main(args, experiment_config=None): if exporter_config is None: raise ValueError("Invalid exporter config: {}".format(rejection_reason)) + if PACED_FEEDER_EXPORTER in exporter_config["exporter_list"]: + config["storage"] = { + "tsdb": { + "out_of_order_time_window": constants.PROMETHEUS_OUT_OF_ORDER_TIME_WINDOW + } + } + # add section for "scrape_configs" config["scrape_configs"] = [] # add target for self-monitoring prometheus @@ -437,7 +446,6 @@ def main(args, experiment_config=None): config["remote_write"].append(remote_write_config) # Write prometheus config to output_dir - import constants output_file = os.path.join(args.output_dir, constants.PROMETHEUS_CONFIG_FILE) os.makedirs(args.output_dir, exist_ok=True) diff --git a/asap-tools/experiments/tests/test_paced_feeder.py b/asap-tools/experiments/tests/test_paced_feeder.py index 6e55fdc9..2a61c5bb 100644 --- a/asap-tools/experiments/tests/test_paced_feeder.py +++ b/asap-tools/experiments/tests/test_paced_feeder.py @@ -7,10 +7,17 @@ import unittest from unittest import mock +import yaml + from omegaconf import OmegaConf import constants -from experiment_utils.core import call_generate_prometheus_config, read_exporter_config +import generate_prometheus_config +from experiment_utils.core import ( + GeneratePrometheusArgs, + call_generate_prometheus_config, + read_exporter_config, +) from experiment_utils.services import create_prometheus_service from experiment_utils.services.docker_prometheus import DockerPrometheusService from experiment_utils.services.paced_feeder import ( @@ -363,5 +370,48 @@ def test_scraped_metrics_are_still_forwarded(self): self.assertEqual(args.remote_write_metric_names, ["node_cpu"]) +class PrometheusOutOfOrderWindowTest(unittest.TestCase): + """The feeder sends a tick's requests concurrently, so Prometheus' receiver + sees a series' samples out of order within a second and, without an + out-of-order window, rejects them (400 "out of order sample"). Bare-metal + and containerized Prometheus both read this generated prometheus.yml.""" + + def storage(self, exporter_list): + with tempfile.TemporaryDirectory() as out: + prometheus_config = OmegaConf.create( + { + "scrape_interval": "1s", + "evaluation_interval": "10s", + "recording_rules": {"interval": "5s"}, + } + ) + args = GeneratePrometheusArgs( + 1, out, out, prometheus_config, "10.10.1.1", "10.10.1", 0 + ) + generate_prometheus_config.main( + args, + { + "exporters": { + "only_start_if_queries_exist": False, + "exporter_list": exporter_list, + }, + "query_groups": [], + "metrics": [], + }, + ) + with open(os.path.join(out, constants.PROMETHEUS_CONFIG_FILE)) as f: + return yaml.safe_load(f).get("storage") + + def test_window_set_when_the_feeder_targets_prometheus(self): + feeder = {"workload_table": "t", "metrics": [], "seed": 1, "duration_secs": 1} + self.assertEqual( + self.storage({"paced_feeder": feeder}), + {"tsdb": {"out_of_order_time_window": "5s"}}, + ) + + def test_no_window_without_the_feeder(self): + self.assertIsNone(self.storage({})) + + if __name__ == "__main__": unittest.main() From 0c6434a1d09990821cb9a49472ba43c48dfdbbda Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:52:13 +0000 Subject: [PATCH 3/4] feat(asap-tools): feed records metrics to ASAPQuery only Records metrics (http, flows) reach remote write as one sample per value field, many sharing a series and timestamp, which Prometheus' receiver rejects. check_paced_feeder_config now takes the experiment modes: it accepts records metrics when every mode feeds ASAPQuery only and refuses them when a mode feeds Prometheus (baseline or query_prometheus_too). The config generator feeds a table's records metrics only with --sql-queries (only the structured queries read them); the config then has no baseline mode, and records metrics stay out of `metrics`, which with prebuilt planner configs only keeps fed metrics out of Prometheus' forwarding. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../CONFIG_PARAMETERS_REFERENCE.md | 2 +- asap-tools/experiments/experiment_run_e2e.py | 9 ++-- .../experiment_utils/services/paced_feeder.py | 45 ++++++++++++++----- .../sketch_bench_e2e_configs.py | 42 ++++++++++++----- .../experiments/tests/test_paced_feeder.py | 36 ++++++++++++--- .../tests/test_sketch_bench_e2e_configs.py | 44 +++++++++++++++--- 6 files changed, 137 insertions(+), 41 deletions(-) diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index 5050abb1..55800114 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md @@ -478,7 +478,7 @@ table with `sketch_bench_e2e/sketch_bench_e2e_configs.py`. - **Example**: `/path/to/synthetic-templatesclassic-shared1.json` #### `experiment_params.exporters.exporter_list.paced_feeder.metrics` (list, required) -- **Description**: Metrics of the table's `workloads[0].e2e.metrics` to send; gauge or counter only +- **Description**: Metrics of the table's `workloads[0].e2e.metrics` to send. Gauge and counter metrics go to every sink; `records` metrics (one sample per value field, `.`) only to ASAPQuery, so they are refused when any mode feeds Prometheus (baseline, or `query_prometheus_too`) - **Example**: `[data, data_total]` #### `experiment_params.exporters.exporter_list.paced_feeder.seed` (int, required) diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index f1a6ef1a..805c65c9 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -192,21 +192,22 @@ def main(cfg: DictConfig): cfg.experiment_params ) + modes_by_name = { + m["mode"]: m + for m in OmegaConf.to_container(cfg.experiment_params.experiment, resolve=True) + } paced_feeder_service = None paced_feeder_config = exporter_config["exporter_list"].get(PACED_FEEDER_EXPORTER) if paced_feeder_config is not None: paced_feeder_config = OmegaConf.to_container(paced_feeder_config, resolve=True) check_paced_feeder_config( paced_feeder_config, + modes=list(modes_by_name.values()), parallelism=args.parallelism, min_feed_secs=(0 if skip_querying else args.steady_state_wait) + minimum_experiment_running_time, ) paced_feeder_service = PacedFeederService(provider, args) - modes_by_name = { - m["mode"]: m - for m in OmegaConf.to_container(cfg.experiment_params.experiment, resolve=True) - } for experiment_mode in experiment_modes: print(f"Running experiment mode: {experiment_mode}") diff --git a/asap-tools/experiments/experiment_utils/services/paced_feeder.py b/asap-tools/experiments/experiment_utils/services/paced_feeder.py index c01abb3e..40a31ece 100644 --- a/asap-tools/experiments/experiment_utils/services/paced_feeder.py +++ b/asap-tools/experiments/experiment_utils/services/paced_feeder.py @@ -21,8 +21,11 @@ PACED_FEEDER_EXPORTER = "paced_feeder" PACED_FEEDER_REQUIRED_KEYS = ("workload_table", "metrics", "seed", "duration_secs") -# Metric types a remote-write sink (ASAPQuery ingest, Prometheus) accepts. -REMOTE_WRITE_METRIC_TYPES = ("gauge", "counter") +# Metric types Prometheus' remote-write receiver accepts. +PROMETHEUS_METRIC_TYPES = ("gauge", "counter") +# Streams of records, sent as one sample per value field (`http.latency`). +# Many records share a series and timestamp, so only ASAPQuery takes them. +RECORDS_METRIC_TYPE = "records" PACED_FEEDER_BINARY = "./target/release/paced_feeder" # The brackets keep pgrep/pkill from matching the shell that runs them. @@ -47,11 +50,23 @@ def paced_feeder_metric_names(experiment_params) -> List[str]: ] +def feeds_prometheus(mode: Dict[str, Any]) -> bool: + """Whether the feeder sends to Prometheus in this experiment mode.""" + return mode["mode"] != constants.SKETCHDB_EXPERIMENT_NAME or mode.get( + "query_prometheus_too", False + ) + + def check_paced_feeder_config( - config: Dict[str, Any], parallelism: int, min_feed_secs: float + config: Dict[str, Any], + modes: List[Dict[str, Any]], + parallelism: int, + min_feed_secs: float, ) -> None: - """Fail before the experiment starts on a config the feeder would reject - or one that stops sending before the queries end.""" + """Fail before the experiment starts on a config the feeder would reject, + one sending records to Prometheus, or one that stops sending before the + queries end.""" + prometheus_modes = [m["mode"] for m in modes if feeds_prometheus(m)] with open(config["workload_table"]) as f: (workload,) = json.load(f)["workloads"] table_metrics = workload["e2e"]["metrics"] @@ -62,10 +77,16 @@ def check_paced_feeder_config( f"{config['workload_table']} (has {sorted(table_metrics)})" ) metric_type = table_metrics[metric]["type"] - if metric_type not in REMOTE_WRITE_METRIC_TYPES: + if metric_type == RECORDS_METRIC_TYPE: + if prometheus_modes: + raise ValueError( + f"paced_feeder metric {metric!r} is a records metric, which " + f"only ASAPQuery takes, but modes {prometheus_modes} feed " + "Prometheus (baseline or query_prometheus_too)" + ) + elif metric_type not in PROMETHEUS_METRIC_TYPES: raise ValueError( - f"paced_feeder metric {metric!r} has type {metric_type!r}; " - f"remote write takes only {REMOTE_WRITE_METRIC_TYPES}" + f"paced_feeder metric {metric!r} has unknown type {metric_type!r}" ) if parallelism != 1: raise ValueError( @@ -87,10 +108,10 @@ def paced_feeder_sinks( sketchdb feeds ASAPQuery's ingest, plus Prometheus when the mode also queries it; every other mode queries Prometheus only. """ - if mode["mode"] != constants.SKETCHDB_EXPERIMENT_NAME: - return {SINK_PROMETHEUS: prometheus_url} - sinks = {SINK_QUERY_ENGINE: query_engine_url} - if mode.get("query_prometheus_too", False): + sinks = {} + if mode["mode"] == constants.SKETCHDB_EXPERIMENT_NAME: + sinks[SINK_QUERY_ENGINE] = query_engine_url + if feeds_prometheus(mode): sinks[SINK_PROMETHEUS] = prometheus_url return sinks diff --git a/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py b/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py index 27a5eb9d..f1b2775c 100644 --- a/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py +++ b/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py @@ -10,8 +10,12 @@ groups, only when `--sql-queries` gives their SQL; otherwise they are skipped with a note. The file holds one `;`-terminated statement per structured entry, in table order. -- the table's gauge and counter metrics, sent by the paced feeder (records - metrics are skipped with a note: remote write cannot carry them). +- the table's gauge and counter metrics, sent by the paced feeder and listed + under `metrics`. Records metrics, which only the structured queries read, + are fed only with `--sql-queries`, and then only to ASAPQuery: the config + has no baseline (Prometheus) mode, and they are not listed under `metrics` + (with prebuilt planner configs that list only keeps fed metrics out of + Prometheus' forwarding, and records are never scraped). - `controller.prebuilt_output_dir`, when given, so ASAPQuery runs those planner configs (e.g. from `asap-planner --planner saved`). @@ -41,7 +45,8 @@ ) from experiment_utils.services.paced_feeder import ( # noqa: E402 PACED_FEEDER_EXPORTER, - REMOTE_WRITE_METRIC_TYPES, + PROMETHEUS_METRIC_TYPES, + RECORDS_METRIC_TYPE, ) DEFAULT_OUTPUT_DIR = os.path.join( @@ -51,10 +56,8 @@ {"name": "prometheus", "url": "http://localhost:9090"}, {"name": "sketchdb", "url": "http://localhost:8088"}, ] -MODES = [ - {"mode": "sketchdb", "server": "sketchdb"}, - {"mode": "baseline", "server": "prometheus"}, -] +SKETCHDB_MODE = {"mode": "sketchdb", "server": "sketchdb"} +BASELINE_MODE = {"mode": "baseline", "server": "prometheus"} # Seconds the window ending at an evaluation time gets for its data to arrive. DEFAULT_QUERY_TIME_OFFSET_SECS = 10 # Feeder time beyond the query phase: covers flow.steady_state_wait (60 s in @@ -134,10 +137,18 @@ def build_config( raise ValueError(f"{table_path}: no queries to run") metrics = [] + records_metrics = [] for name, desc in e2e["metrics"].items(): - if desc["type"] not in REMOTE_WRITE_METRIC_TYPES: - notes.append(f"skipped {desc['type']} metric {name!r}: not remote-writable") + if desc["type"] == RECORDS_METRIC_TYPE: + if sql: + records_metrics.append(name) + else: + notes.append( + f"skipped records metric {name!r}: only structured queries read it" + ) continue + if desc["type"] not in PROMETHEUS_METRIC_TYPES: + raise ValueError(f"metric {name!r} has unknown type {desc['type']!r}") metrics.append( { "metric": name, @@ -148,7 +159,7 @@ def build_config( feeder = { "workload_table": os.path.abspath(table_path), - "metrics": [m["metric"] for m in metrics], + "metrics": [m["metric"] for m in metrics] + records_metrics, "seed": seed, "duration_secs": ( feeder_duration_secs @@ -159,9 +170,18 @@ def build_config( if samples_per_sec is not None: feeder["samples_per_sec"] = samples_per_sec + modes = [SKETCHDB_MODE] + if records_metrics: + notes.append( + "no baseline mode: records metrics " + f"{records_metrics} go to ASAPQuery only" + ) + else: + modes.append(BASELINE_MODE) + config: dict = { "experiment_params": { - "experiment": MODES, + "experiment": modes, "monitoring": {"tool": "prometheus", "deployment_mode": "bare_metal"}, "servers": SERVERS, "exporters": { diff --git a/asap-tools/experiments/tests/test_paced_feeder.py b/asap-tools/experiments/tests/test_paced_feeder.py index 2a61c5bb..e2aa05e9 100644 --- a/asap-tools/experiments/tests/test_paced_feeder.py +++ b/asap-tools/experiments/tests/test_paced_feeder.py @@ -29,6 +29,9 @@ from experiment_utils.services.prometheus import PrometheusService QE_URL = "http://10.10.1.1:8080/api/v1/write" +SKETCHDB = {"mode": "sketchdb", "server": "sketchdb"} +BASELINE = {"mode": "baseline", "server": "prometheus"} +BOTH_MODES = [SKETCHDB, BASELINE] PROM_URL = "http://10.10.1.1:9090/api/v1/write" # workloads[0].e2e.metrics of synthetic-templatesclassic-shared1.json, plus a @@ -139,27 +142,46 @@ def tearDown(self): self.tmp.cleanup() def test_valid_config(self): - check_paced_feeder_config(feeder_config(self.table), 1, 600) + check_paced_feeder_config(feeder_config(self.table), BOTH_MODES, 1, 600) def test_unknown_metric(self): with self.assertRaisesRegex(ValueError, "'cpu' is not in"): check_paced_feeder_config( - feeder_config(self.table, metrics=["cpu"]), 1, 600 + feeder_config(self.table, metrics=["cpu"]), BOTH_MODES, 1, 600 ) - def test_records_metric(self): - with self.assertRaisesRegex(ValueError, "type 'records'"): + def test_records_metric_to_query_engine_only(self): + check_paced_feeder_config( + feeder_config(self.table, metrics=["data", "flows"]), [SKETCHDB], 1, 600 + ) + + def test_records_metric_refused_for_prometheus(self): + # Many records share a series and timestamp; Prometheus' receiver + # rejects those duplicates. + records = feeder_config(self.table, metrics=["flows"]) + with self.assertRaisesRegex(ValueError, r"records metric.*\['baseline'\]"): + check_paced_feeder_config(records, BOTH_MODES, 1, 600) + with self.assertRaisesRegex(ValueError, r"query_prometheus_too"): check_paced_feeder_config( - feeder_config(self.table, metrics=["flows"]), 1, 600 + records, [{**SKETCHDB, "query_prometheus_too": True}], 1, 600 ) + def test_unknown_metric_type(self): + with open(self.table) as f: + table = json.load(f) + table["workloads"][0]["e2e"]["metrics"]["data"]["type"] = "histogram" + with open(self.table, "w") as f: + json.dump(table, f) + with self.assertRaisesRegex(ValueError, "unknown type 'histogram'"): + check_paced_feeder_config(feeder_config(self.table), [SKETCHDB], 1, 600) + def test_sharded_ingest(self): with self.assertRaisesRegex(ValueError, "parallelism=1"): - check_paced_feeder_config(feeder_config(self.table), 2, 600) + check_paced_feeder_config(feeder_config(self.table), BOTH_MODES, 2, 600) def test_feeder_ending_before_the_queries(self): with self.assertRaisesRegex(ValueError, "at least 601"): - check_paced_feeder_config(feeder_config(self.table), 1, 601) + check_paced_feeder_config(feeder_config(self.table), BOTH_MODES, 1, 601) def test_missing_keys_rejected_with_the_other_exporters(self): params = OmegaConf.create( diff --git a/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py b/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py index f14dd2ba..1b31bc9d 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py +++ b/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py @@ -98,7 +98,7 @@ def test_query_groups_by_interval_in_table_order(self): ], ) - def test_feeder_sends_the_remote_writable_metrics(self): + def test_feeder_sends_the_classic_metrics_without_sql(self): config, notes = self.build() params = config["experiment_params"] self.assertEqual( @@ -125,7 +125,12 @@ def test_feeder_sends_the_remote_writable_metrics(self): "duration_secs": 120 + 600 + sbe.FEEDER_SLACK_SECS, }, ) - self.assertIn("skipped records metric 'flows': not remote-writable", notes) + self.assertIn( + "skipped records metric 'flows': only structured queries read it", notes + ) + self.assertEqual( + [m["mode"] for m in params["experiment"]], ["sketchdb", "baseline"] + ) def test_structured_queries_skipped_without_sql(self): config, notes = self.build() @@ -148,6 +153,20 @@ def test_structured_queries_from_sql_file(self): ["SELECT dst_subnet, uniq(src_ip) FROM flows GROUP BY dst_subnet"], ) self.assertFalse(any("structured" in n for n in notes)) + # The records metric goes to ASAPQuery only: fed, but no baseline + # mode and not under `metrics`. + params = config["experiment_params"] + self.assertEqual( + params["exporters"]["exporter_list"]["paced_feeder"]["metrics"], + ["data", "data_total", "flows"], + ) + self.assertEqual([m["mode"] for m in params["experiment"]], ["sketchdb"]) + self.assertEqual( + [m["metric"] for m in params["metrics"]], ["data", "data_total"] + ) + self.assertIn( + "no baseline mode: records metrics ['flows'] go to ASAPQuery only", notes + ) def test_sql_file_must_match_the_structured_queries(self): sql = os.path.join(self.tmp.name, "multigroup.sql") @@ -172,13 +191,13 @@ def test_prebuilt_output_dir(self): config, _ = self.build(prebuilt_output_dir=prebuilt) self.assertEqual(config["controller"], {"prebuilt_output_dir": prebuilt}) - def test_generated_config_composes_and_validates(self): + def compose_and_validate(self, config): # A dry run of what experiment_run_e2e.py does with the file before it - # starts any service. + # starts any service; returns the experiment modes. config_dir = os.path.join(self.tmp.name, "config") shutil.copytree(CONFIG_DIR, config_dir) sbe.write_config( - self.build()[0], + config, self.table, os.path.join(config_dir, "experiment_type", "sketch_bench_e2e", "t.yaml"), ) @@ -190,6 +209,7 @@ def test_generated_config_composes_and_validates(self): self.assertIsNotNone(exporter_config, reason) check_paced_feeder_config( OmegaConf.to_container(exporter_config.exporter_list.paced_feeder), + modes=OmegaConf.to_container(params.experiment), parallelism=cfg.streaming.parallelism, min_feed_secs=cfg.flow.steady_state_wait + experiment_config.get_minimum_experiment_running_time(params), @@ -197,7 +217,19 @@ def test_generated_config_composes_and_validates(self): modes, _ = experiment_config.generate_controller_client_configs( params, self.tmp.name ) - self.assertEqual(modes, ["sketchdb", "baseline"]) + return modes + + def test_generated_config_composes_and_validates(self): + self.assertEqual( + self.compose_and_validate(self.build()[0]), ["sketchdb", "baseline"] + ) + + def test_generated_config_with_records_validates(self): + sql = os.path.join(self.tmp.name, "multigroup.sql") + with open(sql, "w") as f: + f.write("SELECT 1;\n") + config, _ = self.build(sql_queries_file=sql) + self.assertEqual(self.compose_and_validate(config), ["sketchdb"]) if __name__ == "__main__": From 464edb6d5508d5e6e4b9bd9d28b70b7402ce6607 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 02:02:13 +0000 Subject: [PATCH 4/4] fix(asap-tools): address review on the paced feeder runner - Pass --samples-per-sec only to gauge and counter feeders and the new records_per_sec as --records-per-sec only to records feeders; the feeder exits on the other kind's flag. - SQL query groups get time_window_seconds: 0; their templates read only {{ end_time }} / {{ end_datetime }}, but the client refuses to render templates without a window. - Take the shared --start-ms after the per-node binary check and table copy, so the 5 s lead covers only the launches. - duration_secs must also cover the feeder's startup check before steady_state_wait and the queries. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../CONFIG_PARAMETERS_REFERENCE.md | 8 +- asap-tools/experiments/experiment_run_e2e.py | 7 +- .../experiment_utils/services/__init__.py | 2 + .../experiment_utils/services/paced_feeder.py | 41 +++++++-- .../sketch_bench_e2e_configs.py | 13 ++- .../experiments/tests/test_paced_feeder.py | 90 ++++++++++++++++++- .../tests/test_sketch_bench_e2e_configs.py | 17 ++++ 7 files changed, 160 insertions(+), 18 deletions(-) diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index 55800114..dbbef000 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md @@ -486,13 +486,17 @@ table with `sketch_bench_e2e/sketch_bench_e2e_configs.py`. - **Example**: `42` #### `experiment_params.exporters.exporter_list.paced_feeder.duration_secs` (int, required) -- **Description**: Seconds of data to send. Must be at least `flow.steady_state_wait` plus the query groups' running time +- **Description**: Seconds of data to send. Must be at least the feeder's startup check (7 s) plus `flow.steady_state_wait` plus the query groups' running time - **Example**: `4000` #### `experiment_params.exporters.exporter_list.paced_feeder.samples_per_sec` (int, optional) -- **Description**: Overrides the table's `samples_per_sec_per_series` +- **Description**: Overrides the table's `samples_per_sec_per_series` (gauge and counter metrics) - **Example**: `20` +#### `experiment_params.exporters.exporter_list.paced_feeder.records_per_sec` (int, optional) +- **Description**: Overrides the table's `records_per_sec` (records metrics) +- **Example**: `20000` + #### `experiment_params.exporters.exporter_list.paced_feeder.extra_flags` (string, optional) - **Description**: Appended to every feeder command line - **Example**: `"--concurrency 32"` diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index 805c65c9..bdf5268e 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -28,6 +28,7 @@ PACED_FEEDER_EXPORTER, PacedFeederService, check_paced_feeder_config, + paced_feeder_min_duration_secs, paced_feeder_sinks, ) from experiment_utils.services.misc import DiscoveryBackend @@ -204,8 +205,10 @@ def main(cfg: DictConfig): paced_feeder_config, modes=list(modes_by_name.values()), parallelism=args.parallelism, - min_feed_secs=(0 if skip_querying else args.steady_state_wait) - + minimum_experiment_running_time, + min_feed_secs=paced_feeder_min_duration_secs( + 0 if skip_querying else args.steady_state_wait, + minimum_experiment_running_time, + ), ) paced_feeder_service = PacedFeederService(provider, args) diff --git a/asap-tools/experiments/experiment_utils/services/__init__.py b/asap-tools/experiments/experiment_utils/services/__init__.py index e4581dbb..32cd0d53 100644 --- a/asap-tools/experiments/experiment_utils/services/__init__.py +++ b/asap-tools/experiments/experiment_utils/services/__init__.py @@ -38,6 +38,7 @@ PACED_FEEDER_EXPORTER, PacedFeederService, check_paced_feeder_config, + paced_feeder_min_duration_secs, paced_feeder_sinks, ) from .clickhouse_service import ClickHouseService, ClickHouseDataLoaderService @@ -164,5 +165,6 @@ def create_prometheus_service(cfg, provider, num_nodes: int, node_offset: int): "PACED_FEEDER_EXPORTER", "PacedFeederService", "check_paced_feeder_config", + "paced_feeder_min_duration_secs", "paced_feeder_sinks", ] diff --git a/asap-tools/experiments/experiment_utils/services/paced_feeder.py b/asap-tools/experiments/experiment_utils/services/paced_feeder.py index 40a31ece..94095324 100644 --- a/asap-tools/experiments/experiment_utils/services/paced_feeder.py +++ b/asap-tools/experiments/experiment_utils/services/paced_feeder.py @@ -57,6 +57,23 @@ def feeds_prometheus(mode: Dict[str, Any]) -> bool: ) +def paced_feeder_min_duration_secs( + steady_state_wait: float, query_running_secs: float +) -> float: + """Seconds of data that last until the final query: queries end at most + startup check + steady_state_wait + the query groups' running time after + launch, and the data starts PACED_FEEDER_START_LEAD_SECS after launch (not + subtracted, for slack).""" + return PACED_FEEDER_STARTUP_CHECK_SECS + steady_state_wait + query_running_secs + + +def table_metric_types(workload_table: str) -> Dict[str, str]: + """Metric name -> type in the table's `workloads[0].e2e.metrics`.""" + with open(workload_table) as f: + (workload,) = json.load(f)["workloads"] + return {name: m["type"] for name, m in workload["e2e"]["metrics"].items()} + + def check_paced_feeder_config( config: Dict[str, Any], modes: List[Dict[str, Any]], @@ -67,16 +84,14 @@ def check_paced_feeder_config( one sending records to Prometheus, or one that stops sending before the queries end.""" prometheus_modes = [m["mode"] for m in modes if feeds_prometheus(m)] - with open(config["workload_table"]) as f: - (workload,) = json.load(f)["workloads"] - table_metrics = workload["e2e"]["metrics"] + table_metrics = table_metric_types(config["workload_table"]) for metric in config["metrics"]: if metric not in table_metrics: raise ValueError( f"paced_feeder metric {metric!r} is not in " f"{config['workload_table']} (has {sorted(table_metrics)})" ) - metric_type = table_metrics[metric]["type"] + metric_type = table_metrics[metric] if metric_type == RECORDS_METRIC_TYPE: if prometheus_modes: raise ValueError( @@ -96,7 +111,8 @@ def check_paced_feeder_config( raise ValueError( f"paced_feeder.duration_secs={config['duration_secs']} ends before the " f"queries do: needs at least {min_feed_secs} " - "(flow.steady_state_wait + the query groups' running time)" + "(feeder startup + flow.steady_state_wait + the query groups' " + "running time)" ) @@ -121,6 +137,7 @@ def paced_feeder_command( workload_table: str, url: str, metric: str, + metric_type: str, start_ms: int, stats_out: str, ) -> str: @@ -130,7 +147,11 @@ def paced_feeder_command( f" --seed {config['seed']} --duration-secs {config['duration_secs']}" f" --start-ms {start_ms} --stats-out {stats_out}" ) - if config.get("samples_per_sec") is not None: + # The feeder refuses the other kind's rate flag. + if metric_type == RECORDS_METRIC_TYPE: + if config.get("records_per_sec") is not None: + cmd += f" --records-per-sec {config['records_per_sec']}" + elif config.get("samples_per_sec") is not None: cmd += f" --samples-per-sec {config['samples_per_sec']}" if config.get("extra_flags"): cmd += f" {config['extra_flags']}" @@ -161,9 +182,7 @@ def start( **kwargs, ) -> None: output_dir = os.path.join(experiment_output_dir, "paced_feeder_output") - # One shared event-time origin for every process of this mode. - start_ms = (math.ceil(time.time()) + PACED_FEEDER_START_LEAD_SECS) * 1000 - + metric_types = table_metric_types(config["workload_table"]) jobs = [(sink, metric) for sink in sinks for metric in config["metrics"]] job_nodes = [ self.target_nodes[i % len(self.target_nodes)] for i in range(len(jobs)) @@ -184,6 +203,9 @@ def start( self.provider, config["workload_table"], output_dir, node_idx ) + # One shared event-time origin for every process of this mode, taken + # after the SSH prep above so the lead covers only the launches. + start_ms = (math.ceil(time.time()) + PACED_FEEDER_START_LEAD_SECS) * 1000 node_cmds: List[Tuple[int, str]] = [] for node_idx, (sink, metric) in zip(job_nodes, jobs): name = f"{sink}_{metric}" @@ -192,6 +214,7 @@ def start( workload_table, sinks[sink], metric, + metric_types[metric], start_ms, os.path.join(output_dir, f"{name}_stats.json"), ) diff --git a/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py b/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py index f1b2775c..dd7bee6a 100644 --- a/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py +++ b/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py @@ -86,6 +86,7 @@ def build_config( query_time_offset_secs: int = DEFAULT_QUERY_TIME_OFFSET_SECS, feeder_duration_secs: Optional[int] = None, samples_per_sec: Optional[int] = None, + records_per_sec: Optional[int] = None, prebuilt_output_dir: Optional[str] = None, sql_queries_file: Optional[str] = None, ) -> Tuple[dict, List[str]]: @@ -113,7 +114,12 @@ def build_config( sql = [(q["interval_secs"], s) for q, s in zip(structured, statements)] query_groups = [] - for queries_by_interval in (by_interval(promql), by_interval(sql)): + for queries_by_interval, extra in ( + (by_interval(promql), {}), + # The SQL templates read only {{ end_time }} / {{ end_datetime }} (the + # window is in the SQL); the client requires a window to render them. + (by_interval(sql), {"time_window_seconds": 0}), + ): for interval_secs, queries in queries_by_interval.items(): if query_duration_secs < interval_secs: raise ValueError( @@ -131,6 +137,7 @@ def build_config( "query_time_offset": query_time_offset_secs, "align_to_interval": True, }, + **extra, } ) if not query_groups: @@ -169,6 +176,8 @@ def build_config( } if samples_per_sec is not None: feeder["samples_per_sec"] = samples_per_sec + if records_per_sec is not None and records_metrics: + feeder["records_per_sec"] = records_per_sec modes = [SKETCHDB_MODE] if records_metrics: @@ -231,6 +240,7 @@ def main() -> None: help=f"default: starting delay + query duration + {FEEDER_SLACK_SECS}", ) parser.add_argument("--samples-per-sec", type=int) + parser.add_argument("--records-per-sec", type=int) parser.add_argument("--prebuilt-output-dir") parser.add_argument("--sql-queries") parser.add_argument("--output", help="default: /
.yaml") @@ -244,6 +254,7 @@ def main() -> None: query_time_offset_secs=args.query_time_offset_secs, feeder_duration_secs=args.feeder_duration_secs, samples_per_sec=args.samples_per_sec, + records_per_sec=args.records_per_sec, prebuilt_output_dir=args.prebuilt_output_dir, sql_queries_file=args.sql_queries, ) diff --git a/asap-tools/experiments/tests/test_paced_feeder.py b/asap-tools/experiments/tests/test_paced_feeder.py index e2aa05e9..1484a80b 100644 --- a/asap-tools/experiments/tests/test_paced_feeder.py +++ b/asap-tools/experiments/tests/test_paced_feeder.py @@ -21,8 +21,11 @@ from experiment_utils.services import create_prometheus_service from experiment_utils.services.docker_prometheus import DockerPrometheusService from experiment_utils.services.paced_feeder import ( + PACED_FEEDER_START_LEAD_SECS, + PACED_FEEDER_STARTUP_CHECK_SECS, PacedFeederService, check_paced_feeder_config, + paced_feeder_min_duration_secs, paced_feeder_command, paced_feeder_sinks, ) @@ -88,6 +91,7 @@ def test_command_line(self): "/out/table.json", QE_URL, "data", + "gauge", 1700000005000, "/out/query_engine_data_stats.json", ) @@ -105,11 +109,28 @@ def test_optional_flags_are_appended(self): "/t", QE_URL, "data", + "gauge", 0, "/s", ) self.assertTrue(cmd.endswith(" --samples-per-sec 20 --concurrency 4")) + def test_each_metric_gets_its_own_kind_of_rate(self): + # The feeder exits on --samples-per-sec for a records metric (and on + # --records-per-sec for a gauge), which the startup check would only + # report as a dead process. + config = feeder_config("/t", samples_per_sec=20, records_per_sec=1000) + + def rate_flags(metric, metric_type): + cmd = paced_feeder_command( + config, "/t", QE_URL, metric, metric_type, 0, "/s" + ) + return [f for f in cmd.split() if f.endswith("-per-sec")] + + self.assertEqual(rate_flags("data", "gauge"), ["--samples-per-sec"]) + self.assertEqual(rate_flags("data_total", "counter"), ["--samples-per-sec"]) + self.assertEqual(rate_flags("flows", "records"), ["--records-per-sec"]) + class PacedFeederSinksTest(unittest.TestCase): def test_sketchdb_feeds_the_query_engine_only(self): @@ -183,6 +204,18 @@ def test_feeder_ending_before_the_queries(self): with self.assertRaisesRegex(ValueError, "at least 601"): check_paced_feeder_config(feeder_config(self.table), BOTH_MODES, 1, 601) + def test_duration_covers_startup_wait_and_queries(self): + # 60 s steady_state_wait and 533 s of queries: the feeder's startup + # check comes on top, so 600 s is exactly enough and 599 s is not. + min_secs = paced_feeder_min_duration_secs(60, 533) + self.assertEqual(min_secs, PACED_FEEDER_STARTUP_CHECK_SECS + 60 + 533) + self.assertEqual(min_secs, 600) + check_paced_feeder_config(feeder_config(self.table), BOTH_MODES, 1, min_secs) + with self.assertRaisesRegex(ValueError, "at least 600"): + check_paced_feeder_config( + feeder_config(self.table, duration_secs=599), BOTH_MODES, 1, min_secs + ) + def test_missing_keys_rejected_with_the_other_exporters(self): params = OmegaConf.create( { @@ -197,10 +230,15 @@ def test_missing_keys_rejected_with_the_other_exporters(self): self.assertIn("duration_secs", reason) +# Fake seconds each SSH prep step (binary check, table copy) takes. +PREP_SECS = 10 + + class RecordingProvider: def __init__(self, running_count): self.commands = [] self.running_count = running_count + self.clock = 1700000000.0 def get_home_dir(self): return "/home/u" @@ -210,6 +248,8 @@ def is_remote(self): def execute_command(self, node_idx, cmd, **kwargs): self.commands.append((node_idx, cmd)) + if cmd.startswith("test -x"): + self.clock += PREP_SECS returncode = 0 if cmd.startswith("pgrep"): returncode = 0 if cmd.endswith(f"-qx {self.running_count}") else 1 @@ -225,23 +265,65 @@ def get_coordinator_node(self): class PacedFeederServiceStartTest(unittest.TestCase): - def start(self, running_count): + def start(self, running_count, sinks=None, **config_overrides): provider = RecordingProvider(running_count) + self.provider = provider service = PacedFeederService(provider, Args()) + + def rsync(*args): + provider.clock += PREP_SECS + return "/exp/paced_feeder_output/table.json" + with tempfile.TemporaryDirectory() as local_dir, mock.patch( "experiment_utils.services.paced_feeder.time.sleep" + ), mock.patch( + "experiment_utils.services.paced_feeder.time.time", + lambda: provider.clock, ), mock.patch( "experiment_utils.services.paced_feeder.sync.rsync_dataset_file", - return_value="/exp/paced_feeder_output/table.json", + side_effect=rsync, ): service.start( - config=feeder_config("/local/table.json"), - sinks={"query_engine": QE_URL, "prometheus": PROM_URL}, + config=feeder_config(write_table(local_dir), **config_overrides), + sinks=sinks or {"query_engine": QE_URL, "prometheus": PROM_URL}, experiment_output_dir="/exp", local_experiment_dir=local_dir, ) return provider.commands + def test_start_time_leads_the_launches_not_the_prep(self): + # The SSH prep (2 nodes x binary check + table copy) takes 40 fake + # seconds; the shared start must still be the lead after it. + commands = self.start(running_count=2) + start_secs = { + int(c.split(" --start-ms ")[1].split()[0]) / 1000 + for _, c in commands + if "--sink" in c + } + self.assertEqual( + start_secs, {1700000000 + 4 * PREP_SECS + PACED_FEEDER_START_LEAD_SECS} + ) + + def test_records_and_gauge_processes_get_their_rate_flags(self): + commands = self.start( + running_count=1, + sinks={"query_engine": QE_URL}, + metrics=["data", "flows"], + samples_per_sec=20, + records_per_sec=1000, + ) + rate_flags = { + c.split(" --metric ")[1].split()[0]: [ + f for f in c.split() if f.endswith("-per-sec") + ] + for _, c in commands + if "--sink" in c + } + self.assertEqual( + rate_flags, + {"data": ["--samples-per-sec"], "flows": ["--records-per-sec"]}, + ) + def test_one_process_per_sink_and_metric_spread_over_workers(self): commands = self.start(running_count=2) feeders = [(n, c) for n, c in commands if "--sink" in c] diff --git a/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py b/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py index 1b31bc9d..72f795ad 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py +++ b/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py @@ -152,6 +152,12 @@ def test_structured_queries_from_sql_file(self): group["queries"], ["SELECT dst_subnet, uniq(src_ip) FROM flows GROUP BY dst_subnet"], ) + # The client refuses to render {{ end_time }} templates without a + # window; the SQL carries its own, so 0. PromQL groups have none. + self.assertEqual(group["time_window_seconds"], 0) + self.assertNotIn( + "time_window_seconds", config["experiment_params"]["query_groups"][0] + ) self.assertFalse(any("structured" in n for n in notes)) # The records metric goes to ASAPQuery only: fed, but no baseline # mode and not under `metrics`. @@ -168,6 +174,17 @@ def test_structured_queries_from_sql_file(self): "no baseline mode: records metrics ['flows'] go to ASAPQuery only", notes ) + def test_records_rate_only_with_records_metrics(self): + sql = os.path.join(self.tmp.name, "multigroup.sql") + with open(sql, "w") as f: + f.write("SELECT 1;\n") + config, _ = self.build(sql_queries_file=sql, records_per_sec=1000) + feeder = config["experiment_params"]["exporters"]["exporter_list"] + self.assertEqual(feeder["paced_feeder"]["records_per_sec"], 1000) + config, _ = self.build(records_per_sec=1000) + feeder = config["experiment_params"]["exporters"]["exporter_list"] + self.assertNotIn("records_per_sec", feeder["paced_feeder"]) + def test_sql_file_must_match_the_structured_queries(self): sql = os.path.join(self.tmp.name, "multigroup.sql") with open(sql, "w") as f: