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/data-sources/paced-feeder/README.md b/asap-tools/data-sources/paced-feeder/README.md index f5c1acbc..ab41bd70 100644 --- a/asap-tools/data-sources/paced-feeder/README.md +++ b/asap-tools/data-sources/paced-feeder/README.md @@ -160,8 +160,7 @@ http: 2000000 records/s ``` Event time starts at the next whole wall-clock second (or `--start-ms`), so -`NOW()`-relative queries see the data as it arrives. `--send-at-ms` sends the -first tick at a later wall-clock time, backdating the data by the difference. +`NOW()`-relative queries see the data as it arrives. ## Usage diff --git a/asap-tools/data-sources/paced-feeder/src/main.rs b/asap-tools/data-sources/paced-feeder/src/main.rs index 9c7b3eb2..1a106ba2 100644 --- a/asap-tools/data-sources/paced-feeder/src/main.rs +++ b/asap-tools/data-sources/paced-feeder/src/main.rs @@ -93,11 +93,6 @@ struct Args { /// next whole wall-clock second, so `NOW()`-relative queries see live data. #[arg(long)] start_ms: Option, - /// Wall-clock time to send the first tick, in ms since the epoch. Defaults - /// to `--start-ms`; a later time backdates the data by the difference, so - /// feeders launched at different moments share one event-time origin. - #[arg(long, requires = "start_ms")] - send_at_ms: Option, /// Rows per HTTP request, for gauges and counters rounded down to whole /// series (at least one). Defaults to 50,000 for remote write, which keeps @@ -139,7 +134,6 @@ struct RunStats { samples_per_sec: Option, seed: u64, start_ms: i64, - send_at_ms: i64, duration_secs: u64, target_rows_per_sec: usize, rows_sent: usize, @@ -279,14 +273,6 @@ fn next_whole_second_ms() -> i64 { (now_ms / 1_000 + 1) * 1_000 } -/// Data sent before its own timestamp would be in the future. -fn check_send_at(start_ms: i64, send_at_ms: i64) -> Result<()> { - if send_at_ms < start_ms { - bail!("--send-at-ms ({send_at_ms}) must not be before --start-ms ({start_ms})"); - } - Ok(()) -} - /// Sleep until wall-clock `epoch_ms`, then return the monotonic instant it maps to. async fn wait_for_wall_clock(epoch_ms: i64) -> Instant { let now_ms = SystemTime::now() @@ -452,8 +438,6 @@ async fn main() -> Result<()> { let sink = Arc::new(Sink::new(&args, sink_kind, url, &workload, batch_rows)?); let permits = Arc::new(Semaphore::new(args.concurrency)); let start_ms = args.start_ms.unwrap_or_else(next_whole_second_ms); - let send_at_ms = args.send_at_ms.unwrap_or(start_ms); - check_send_at(start_ms, send_at_ms)?; let metric_names = workload.names(); let series = if workload.metrics.is_empty() { @@ -466,12 +450,12 @@ async fn main() -> Result<()> { ) }; eprintln!( - "paced_feeder: {sink_kind:?} -> {url}, {}{series} = {} rows/s for {duration_secs}s, start_ms={start_ms}, send_at_ms={send_at_ms}", + "paced_feeder: {sink_kind:?} -> {url}, {}{series} = {} rows/s for {duration_secs}s, start_ms={start_ms}", metric_names.join(", "), workload.rows_per_tick(), ); - let base = wait_for_wall_clock(send_at_ms).await; + let base = wait_for_wall_clock(start_ms).await; let mut ticks = Vec::with_capacity(duration_secs as usize); let mut errors: Vec = Vec::new(); let mut failed_requests = 0; @@ -564,7 +548,6 @@ async fn main() -> Result<()> { samples_per_sec: (!workload.metrics.is_empty()).then_some(workload.samples_per_sec), seed: args.seed, start_ms, - send_at_ms, duration_secs, target_rows_per_sec: workload.rows_per_tick(), rows_sent, @@ -609,11 +592,4 @@ mod tests { assert!(validate_start_ms(1_700_000_000_000).is_ok()); assert!(validate_start_ms(1_700_000_000_500).is_err()); } - - #[test] - fn send_at_may_backdate_but_not_postdate_the_data() { - assert!(check_send_at(1_000, 1_000).is_ok()); - assert!(check_send_at(1_000, 301_000).is_ok()); - assert!(check_send_at(1_000, 999).is_err()); - } } diff --git a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md index 35a8afed..53bc78a8 100644 --- a/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md +++ b/asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md @@ -457,6 +457,48 @@ 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 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) +- **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 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` (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"` + ### Query Group Configuration #### `experiment_params.query_groups` (list, required) @@ -500,6 +542,15 @@ 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 = data_start_time + (max(1, ceil(starting_delay / T)) + k) * T`, with `T = repetition_delay_ms` (whole seconds), sending it at `t_k + query_time_offset`. `starting_delay` counts from `data_start_time`, not as a sleep from client start +- **Default**: `false` +- **Usage**: Every evaluation time is a multiple of `T`, and repetition k covers the same samples in every mode however late the mode started; `query_time_offset` (at least 5 s, ASAPQuery's allowed lateness) 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. Requires `data_start_time`; the LCM of the aligned intervals must be at most 600 s + +#### `experiment_params.query_groups[].client_options.data_start_time` (int) +- **Description**: Time (epoch seconds) the paced feeder's data starts at +- **Usage**: Not set by hand: `experiment_run_e2e.py` writes it into each mode's client config after starting the paced feeder, which waits for the next multiple of the aligned groups' intervals' LCM + #### `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 fc9cdd40..f8ab3f33 100644 --- a/asap-tools/experiments/constants.py +++ b/asap-tools/experiments/constants.py @@ -48,6 +48,8 @@ # Peak query memory is roughly 16 B per sample (~8 GB at this cap). PROMETHEUS_QUERY_MAX_SAMPLES = 500_000_000 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 13273a12..6653aad6 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -1,8 +1,10 @@ import os +import atexit import json import time import hydra +import yaml from omegaconf import DictConfig, OmegaConf import constants @@ -25,6 +27,13 @@ DockerPrometheusService, DockerVictoriaMetricsService, SystemExportersService, + PACED_FEEDER_EXPORTER, + PacedFeederService, + aligned_interval_lcm, + check_paced_feeder_config, + paced_feeder_min_duration_secs, + paced_feeder_sinks, + set_data_start, ) from experiment_utils.services.misc import DiscoveryBackend @@ -188,6 +197,39 @@ 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) + # The feeder runs from its own directory. + paced_feeder_config["workload_table"] = os.path.abspath( + paced_feeder_config["workload_table"] + ) + check_paced_feeder_config( + paced_feeder_config, + modes=list(modes_by_name.values()), + parallelism=args.parallelism, + min_feed_secs=paced_feeder_min_duration_secs( + 0 if skip_querying else args.steady_state_wait, + minimum_experiment_running_time, + ), + forward_unsupported_queries=args.forward_unsupported_queries, + auto_discover_labels=args.controller_auto_discover_labels, + ) + # Also checks the aligned query groups before any service starts. + paced_feeder_align_secs = aligned_interval_lcm( + OmegaConf.to_container(cfg.experiment_params, resolve=True) + ) + paced_feeder_service = PacedFeederService(provider, args) + if not args.no_teardown: + # Otherwise a failed run leaves feeders pushing into the next + # run's sinks. + atexit.register(paced_feeder_service.stop) + for experiment_mode in experiment_modes: print(f"Running experiment mode: {experiment_mode}") experiment_output_dir = os.path.join( @@ -228,6 +270,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): @@ -476,6 +520,45 @@ 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() + local_client_config = os.path.join( + local_experiment_root_dir, + "controller_client_configs", + f"{experiment_mode}.yaml", + ) + with open(local_client_config) as f: + client_config = yaml.safe_load(f) + start_ms = paced_feeder_service.start( + config=paced_feeder_config, + sinks=sinks, + experiment_output_dir=experiment_output_dir, + local_experiment_dir=local_experiment_dir, + align_secs=paced_feeder_align_secs, + ) + set_data_start(client_config, start_ms) + with open(local_client_config, "w") as f: + yaml.safe_dump(client_config, f, sort_keys=False) + sync.rsync_controller_client_configs( + provider, + experiment_root_output_dir, + local_experiment_root_dir, + node_offset=args.node_offset, + ) + # Start Prometheus throughput monitoring if enabled if args.throughput_prometheus: prometheus_throughput_monitor = PrometheusThroughputMonitor( @@ -553,6 +636,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..8882b831 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,36 @@ def call_generate_prometheus_config( import generate_prometheus_config import generate_victoriametrics_config + from experiment_utils.services.paced_feeder import ( + PACED_FEEDER_EXPORTER, + 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 + exporter_list = cfg.experiment_params.get("exporters", {}).get("exporter_list", {}) + other_exporters = sorted(set(exporter_list) - {PACED_FEEDER_EXPORTER}) + if not forward_to_query_engine and other_exporters: + raise ValueError( + f"exporters {other_exporters} run next to the paced feeder but none " + "of their metrics are listed under experiment_params.metrics, so " + "Prometheus would forward nothing of theirs to the query engine" + ) # 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..fbb16d94 100644 --- a/asap-tools/experiments/experiment_utils/services/__init__.py +++ b/asap-tools/experiments/experiment_utils/services/__init__.py @@ -34,6 +34,15 @@ ControllerService, ) from .grafana import GrafanaService +from .paced_feeder import ( + PACED_FEEDER_EXPORTER, + PacedFeederService, + aligned_interval_lcm, + check_paced_feeder_config, + paced_feeder_min_duration_secs, + paced_feeder_sinks, + set_data_start, +) from .clickhouse_service import ClickHouseService, ClickHouseDataLoaderService @@ -71,6 +80,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 +121,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 +132,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 +164,11 @@ def create_prometheus_service(cfg, provider, num_nodes: int, node_offset: int): "create_prometheus_service", "ClickHouseService", "ClickHouseDataLoaderService", + "PACED_FEEDER_EXPORTER", + "PacedFeederService", + "aligned_interval_lcm", + "check_paced_feeder_config", + "paced_feeder_min_duration_secs", + "paced_feeder_sinks", + "set_data_start", ] diff --git a/asap-tools/experiments/experiment_utils/services/docker_prometheus.py b/asap-tools/experiments/experiment_utils/services/docker_prometheus.py index 90ac419b..7bfdccc7 100644 --- a/asap-tools/experiments/experiment_utils/services/docker_prometheus.py +++ b/asap-tools/experiments/experiment_utils/services/docker_prometheus.py @@ -11,6 +11,7 @@ PROMETHEUS_CONFIG_FILE, PROMETHEUS_QUERY_MAX_SAMPLES, PROMETHEUS_QUERY_TIMEOUT, + PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG, ) @@ -18,7 +19,11 @@ 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. @@ -27,9 +32,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.""" @@ -107,6 +114,8 @@ def start( f"--query.max-samples={PROMETHEUS_QUERY_MAX_SAMPLES}", ] ) + 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..cc9da43c --- /dev/null +++ b/asap-tools/experiments/experiment_utils/services/paced_feeder.py @@ -0,0 +1,363 @@ +""" +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 functools +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 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. +PACED_FEEDER_PROCESS_PATTERN = "[t]arget/release/paced_feeder" +# Feeders are launched over SSH, one node after another, and all start at the +# first aligned time at least this long after the launches begin: a base plus, +# per node, enough for one SSH retry (run_cmd_with_retry waits 5 s). +PACED_FEEDER_START_LEAD_SECS = 5 +PACED_FEEDER_LAUNCH_SECS_PER_NODE = 10 +# 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 = 2 +# Feeders wait for the next multiple of the aligned intervals' LCM before +# sending; refuse intervals that would make that wait longer than this. +PACED_FEEDER_MAX_ALIGN_SECS = 600 +# ASAPQuery's default allowed lateness: a window closes this long after its +# end, so an aligned query must not be sent sooner. +QUERY_ENGINE_ALLOWED_LATENESS_SECS = 5 + +SINK_QUERY_ENGINE = "query_engine" +SINK_PROMETHEUS = "prometheus" + + +def paced_feeder_metric_names(experiment_params) -> List[str]: + """Metrics the paced feeder sends, per its own config block.""" + exporter_list = experiment_params.get("exporters", {}).get("exporter_list", {}) + feeder = exporter_list.get(PACED_FEEDER_EXPORTER) + return list(feeder["metrics"]) if feeder is not None else [] + + +def aligned_start_ms(earliest_ms: int, align_secs: int) -> int: + """The first multiple of `align_secs` at or after `earliest_ms`: every + mode's data starts on the same boundary of each aligned query group's + interval.""" + align_ms = align_secs * 1000 + return -(-earliest_ms // align_ms) * align_ms + + +def aligned_interval_lcm(client_config: Dict[str, Any]) -> int: + """LCM, in seconds, of the repeat intervals of the client config's + `align_to_interval` groups (1 if there are none), after checking that + those groups can be aligned.""" + intervals = [] + for group in client_config["query_groups"]: + options = group.get("client_options", {}) + if not options.get("align_to_interval"): + continue + interval_ms = group["repetition_delay_ms"] + if interval_ms <= 0 or interval_ms % 1000 != 0: + raise ValueError( + f"query group {group['id']}: align_to_interval needs a " + f"whole-second repetition_delay_ms, got {interval_ms}" + ) + offset = options.get("query_time_offset", 0) + if offset < QUERY_ENGINE_ALLOWED_LATENESS_SECS: + raise ValueError( + f"query group {group['id']}: align_to_interval needs " + f"query_time_offset >= {QUERY_ENGINE_ALLOWED_LATENESS_SECS} " + f"(ASAPQuery's allowed lateness), got {offset}" + ) + intervals.append(interval_ms // 1000) + # math.lcm needs Python 3.9. + lcm = functools.reduce(lambda a, b: a * b // math.gcd(a, b), intervals, 1) + if lcm > PACED_FEEDER_MAX_ALIGN_SECS: + raise ValueError( + f"aligned repeat intervals {sorted(set(intervals))} s have an LCM of " + f"{lcm} s; the feeder would wait up to that long to start " + f"(limit {PACED_FEEDER_MAX_ALIGN_SECS} s)" + ) + return lcm + + +def set_data_start(client_config: Dict[str, Any], start_ms: int) -> None: + """Tell the client's aligned groups when the data starts (see the + client's query_time.py).""" + for group in client_config["query_groups"]: + options = group.get("client_options", {}) + if options.get("align_to_interval"): + options["data_start_time"] = start_ms // 1000 + + +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 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 + the data starts. Aligned groups need less (their last evaluation is at + most the starting delay + running time into the data).""" + 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]], + parallelism: int, + min_feed_secs: float, + forward_unsupported_queries: bool, + auto_discover_labels: bool, +) -> None: + """Fail before the experiment starts on a config the feeder would reject, + one sending records to Prometheus, one that stops sending before the + queries end, or one whose queries would read a Prometheus without the + data.""" + prometheus_modes = [m["mode"] for m in modes if feeds_prometheus(m)] + sketchdb_modes = [ + m for m in modes if m["mode"] == constants.SKETCHDB_EXPERIMENT_NAME + ] + if forward_unsupported_queries and not all(map(feeds_prometheus, sketchdb_modes)): + raise ValueError( + "sketchdb mode without query_prometheus_too gives Prometheus no " + "paced_feeder data, so queries ASAPQuery forwards there would come " + "back empty; set streaming.forward_unsupported_queries=false" + ) + if auto_discover_labels: + raise ValueError( + "controller.auto_discover_labels reads labels from Prometheus before " + "the paced_feeder starts; list the labels under experiment_params.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] + 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 unknown type {metric_type!r}" + ) + if sketchdb_modes and 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} " + "(feeder startup + 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. + """ + 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 + + +def paced_feeder_command( + config: Dict[str, Any], + workload_table: str, + url: str, + metric: str, + metric_type: 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}" + ) + # 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']}" + 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, + align_secs: int = 1, + **kwargs, + ) -> int: + """Launch the feeders; returns when their data starts, in epoch ms, + once they are running.""" + output_dir = os.path.join(experiment_output_dir, "paced_feeder_output") + 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)) + ] + 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 + ) + + # One shared start for every process of this mode, taken after the SSH + # prep above so the lead covers only the launches. + lead_secs = ( + PACED_FEEDER_START_LEAD_SECS + + PACED_FEEDER_LAUNCH_SECS_PER_NODE * len(used_nodes) + ) + start_ms = aligned_start_ms( + (math.ceil(time.time()) + lead_secs) * 1000, align_secs + ) + 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, + metric_types[metric], + start_ms, + os.path.join(output_dir, f"{name}_stats.json"), + ) + node_cmds.append( + ( + node_idx, + f"nohup {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 in used_nodes: + self.provider.execute_command( + node_idx=node_idx, + cmd=" ".join(cmd for n, cmd in node_cmds if n == node_idx), + cmd_dir=self.cmd_dir, + nohup=False, + popen=False, + ignore_errors=False, + ) + if time.time() * 1000 > start_ms: + # A feeder launched after the start begins late, so its data + # would trail the other sinks'. + self.stop() + raise RuntimeError( + f"paced_feeder: launching on {len(used_nodes)} nodes took longer " + f"than the {lead_secs} s lead" + ) + + time.sleep(start_ms / 1000 - time.time() + 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}" + ) + return start_ms + + 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 60d210ea..49b3fe6d 100644 --- a/asap-tools/experiments/experiment_utils/services/prometheus.py +++ b/asap-tools/experiments/experiment_utils/services/prometheus.py @@ -13,6 +13,7 @@ PROMETHEUS_CONFIG_FILE, PROMETHEUS_QUERY_MAX_SAMPLES, PROMETHEUS_QUERY_TIMEOUT, + PROMETHEUS_REMOTE_WRITE_RECEIVER_FLAG, ) @@ -20,7 +21,11 @@ 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. @@ -29,10 +34,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.""" @@ -94,12 +101,14 @@ def _start_prometheus(self, experiment_output_dir: str) -> None: time.sleep(3) # Start prometheus - cmd = ( - f"./prometheus --config.file={PROMETHEUS_CONFIG_FILE}" + flags = ( + f"--config.file={PROMETHEUS_CONFIG_FILE}" f" --query.timeout={PROMETHEUS_QUERY_TIMEOUT}" f" --query.max-samples={PROMETHEUS_QUERY_MAX_SAMPLES}" - " > /dev/null 2>&1 < /dev/null &" ) + 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..9e6b487d --- /dev/null +++ b/asap-tools/experiments/sketch_bench_e2e/sketch_bench_e2e_configs.py @@ -0,0 +1,277 @@ +"""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 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`). +- `streaming.forward_unsupported_queries=false`: Prometheus gets no data in + sketchdb mode, so a forwarded query would come back empty. + +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, + PROMETHEUS_METRIC_TYPES, + RECORDS_METRIC_TYPE, +) + +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"}, +] +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 +# 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, + records_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).""" + if sql_queries_file is not None and prebuilt_output_dir is None: + # The planner rejects the SQL groups' `time_window_seconds` and can't + # plan them; they need configs from `asap-planner --planner saved`. + raise ValueError("--sql-queries needs --prebuilt-output-dir") + 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, 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( + 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, + }, + **extra, + } + ) + if not query_groups: + raise ValueError(f"{table_path}: no queries to run") + + metrics = [] + records_metrics = [] + for name, desc in e2e["metrics"].items(): + 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, + "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] + records_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 + if records_per_sec is not None and records_metrics: + feeder["records_per_sec"] = records_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, + "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, + }, + "streaming": {"forward_unsupported_queries": False}, + } + 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("--records-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, + records_per_sec=args.records_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..c297519a --- /dev/null +++ b/asap-tools/experiments/tests/test_paced_feeder.py @@ -0,0 +1,659 @@ +"""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 import paced_feeder +from experiment_utils.services.paced_feeder import ( + PACED_FEEDER_LAUNCH_SECS_PER_NODE, + PACED_FEEDER_START_LEAD_SECS, + PACED_FEEDER_STARTUP_CHECK_SECS, + PacedFeederService, + aligned_interval_lcm, + aligned_start_ms, + paced_feeder_min_duration_secs, + paced_feeder_command, + paced_feeder_sinks, + set_data_start, +) +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 +# 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 check_paced_feeder_config( + config, + modes, + parallelism, + min_feed_secs, + forward_unsupported_queries=False, + auto_discover_labels=False, +): + paced_feeder.check_paced_feeder_config( + config, + modes, + parallelism, + min_feed_secs, + forward_unsupported_queries=forward_unsupported_queries, + auto_discover_labels=auto_discover_labels, + ) + + +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", + "gauge", + 1700000000000, + "/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 1700000000000" + " --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", + "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): + 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), 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"]), BOTH_MODES, 1, 600 + ) + + 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( + 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), BOTH_MODES, 2, 600) + + def test_sharded_ingest_is_fine_without_sketchdb(self): + check_paced_feeder_config(feeder_config(self.table), [BASELINE], 2, 600) + + def test_forwarding_to_a_prometheus_without_data_refused(self): + # In sketchdb mode Prometheus gets no fed data, so a query ASAPQuery + # can't answer would be forwarded there and come back empty. + with self.assertRaisesRegex(ValueError, "forward_unsupported_queries"): + check_paced_feeder_config( + feeder_config(self.table), + BOTH_MODES, + 1, + 600, + forward_unsupported_queries=True, + ) + check_paced_feeder_config( + feeder_config(self.table), + [{**SKETCHDB, "query_prometheus_too": True}, BASELINE], + 1, + 600, + forward_unsupported_queries=True, + ) + + def test_label_discovery_refused(self): + # It reads Prometheus before the feeder has sent anything. + with self.assertRaisesRegex(ValueError, "auto_discover_labels"): + check_paced_feeder_config( + feeder_config(self.table), + BOTH_MODES, + 1, + 600, + auto_discover_labels=True, + ) + + 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 538 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, 538) + self.assertEqual(min_secs, PACED_FEEDER_STARTUP_CHECK_SECS + 60 + 538) + 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( + { + "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) + + +def client_config(*intervals_ms, aligned=True, offset=10): + return { + "query_groups": [ + { + "id": i, + "repetition_delay_ms": t, + "client_options": { + "repetitions": 2, + "align_to_interval": aligned, + "query_time_offset": offset, + }, + } + for i, t in enumerate(intervals_ms) + ] + } + + +class DataStartTest(unittest.TestCase): + def test_waits_for_the_next_boundary(self): + # Ready at 10:02:23, starts at 10:05:00. + self.assertEqual(aligned_start_ms(1791885743000, 300), 1791885900000) + self.assertEqual(aligned_start_ms(1791885600000, 300), 1791885600000) + self.assertEqual(aligned_start_ms(1791885743000, 1), 1791885743000) + + def test_start_fits_every_aligned_interval(self): + lcm = aligned_interval_lcm + self.assertEqual(lcm(client_config(1000, 60000, 300000)), 300) + self.assertEqual(lcm(client_config(60000, 90000)), 180) + self.assertEqual(lcm(client_config(60000, aligned=False)), 1) + + def test_long_wait_refused(self): + # 420 s and 3600 s intervals would wait up to 7 h for an aligned start. + with self.assertRaisesRegex(ValueError, "LCM of 25200 s"): + aligned_interval_lcm(client_config(420000, 3600000)) + + def test_sub_second_interval_refused(self): + # Truncating to whole seconds would give an LCM of 0 (or misalign 1.5 s). + for interval_ms in (500, 1500): + with self.assertRaisesRegex(ValueError, "whole-second"): + aligned_interval_lcm(client_config(interval_ms)) + + def test_offset_below_the_allowed_lateness_refused(self): + # Sent sooner, a query reads an ASAPQuery window that is still open. + with self.assertRaisesRegex(ValueError, "query_time_offset >= 5"): + aligned_interval_lcm(client_config(60000, offset=0)) + aligned_interval_lcm(client_config(60000, offset=5)) + + def test_aligned_groups_learn_the_data_start(self): + config = client_config(60000) + config["query_groups"].append( + {"id": 9, "repetition_delay_ms": 1000, "client_options": {}} + ) + set_data_start(config, 1791885600000) + self.assertEqual( + config["query_groups"][0]["client_options"]["data_start_time"], + 1791885600, + ) + self.assertEqual(config["query_groups"][1]["client_options"], {}) + + +# Fake seconds each SSH prep step (binary check, table copy) takes. +PREP_SECS = 10 + + +class RecordingProvider: + def __init__(self, running_count, launch_secs=0): + self.commands = [] + self.launches = 0 + self.running_count = running_count + self.launch_secs = launch_secs + self.clock = 1700000000.0 + + def get_home_dir(self): + return "/home/u" + + def is_remote(self): + return True + + def execute_command_parallel(self, node_idxs, cmd, **kwargs): + self.commands += [(n, cmd) for n in node_idxs] + + def execute_command(self, node_idx, cmd, **kwargs): + if "--sink" in cmd: + # One SSH launches every feeder of a node. + self.launches += 1 + self.clock += self.launch_secs + self.commands += [ + (node_idx, f"nohup {c.strip()}") for c in cmd.split("nohup ") if c + ] + else: + 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 + 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, + sinks=None, + align_secs=1, + launch_secs=0, + **config_overrides, + ): + provider = RecordingProvider(running_count, launch_secs) + 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", + side_effect=rsync, + ): + self.started = service.start( + 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, + align_secs=align_secs, + ) + return provider.commands + + @staticmethod + def flag_values(commands, flag): + return { + int(c.split(f" {flag} ")[1].split()[0]) + for _, c in commands + if "--sink" in c + } + + # Lead for the 2 worker nodes the tests launch on. + LEAD_SECS = PACED_FEEDER_START_LEAD_SECS + 2 * PACED_FEEDER_LAUNCH_SECS_PER_NODE + + def test_start_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_ms = (1700000000 + 4 * PREP_SECS + self.LEAD_SECS) * 1000 + self.assertEqual(self.flag_values(commands, "--start-ms"), {start_ms}) + self.assertEqual(self.started, start_ms) + + def test_start_waits_for_the_aligned_boundary(self): + # Ready at 1700000065 s; the next multiple of 300 s is 1700000100 s. + commands = self.start(running_count=2, align_secs=300) + self.assertEqual(self.flag_values(commands, "--start-ms"), {1700000100000}) + self.assertEqual(self.started, 1700000100000) + + def test_one_launch_per_node(self): + self.start(running_count=2) + self.assertEqual(self.provider.launches, 2) + + def test_launches_get_time_per_node(self): + # A retried SSH (5 s) on each node still fits. + self.start(running_count=2, launch_secs=PACED_FEEDER_LAUNCH_SECS_PER_NODE) + + def test_launch_slower_than_the_lead_fails(self): + # A feeder launched after the start would trail the other sinks. + per_node = self.LEAD_SECS // 2 + 1 + with self.assertRaisesRegex( + RuntimeError, f"longer than the {self.LEAD_SECS} s lead" + ): + self.start(running_count=2, launch_secs=per_node) + self.assertTrue(any("pkill" in c for _, c in self.provider.commands)) + + 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] + self.assertEqual( + [ + ( + n, + c.split(" --url ")[1].split()[0], + c.split(" --metric ")[1].split()[0], + ) + for n, c in feeders + ], + [ + (1, QE_URL, "data"), + (1, PROM_URL, "data"), + (2, QE_URL, "data_total"), + (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, fed=("data",), other_exporters=()): + 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": { + "paced_feeder": {"metrics": list(fed)}, + **{name: {} for name in other_exporters}, + }, + }, + "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_unlisted_exporter_next_to_the_feeder_refused(self): + # With nothing left to forward, remote write is off, so the other + # exporter's metrics would never reach ASAPQuery. + with self.assertRaisesRegex(ValueError, r"\['node_exporter'\]"): + self.remote_write_args( + [{"metric": "data", "labels": ["job"], "exporter": "paced_feeder"}], + other_exporters=["node_exporter"], + ) + + 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"]) + + def test_fed_metrics_come_from_the_feeder_block(self): + # The feeder sends what its own block lists; a metric not tagged + # paced_feeder in `metrics` would otherwise reach ASAPQuery twice. + args = self.remote_write_args( + [ + {"metric": "data", "labels": ["job"], "exporter": "node_exporter"}, + {"metric": "node_cpu", "labels": ["job"], "exporter": "node_exporter"}, + ] + ) + 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..fc38e390 --- /dev/null +++ b/asap-tools/experiments/tests/test_sketch_bench_e2e_configs.py @@ -0,0 +1,295 @@ +"""Tests for sketch_bench_e2e_configs.py (e2e configs from a sketch-bench table).""" + +import json +import os +import shutil +import tempfile +import unittest + +import yaml +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 ( + aligned_interval_lcm, + check_paced_feeder_config, + set_data_start, +) +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): + if "sql_queries_file" in kwargs and "prebuilt_output_dir" not in kwargs: + prebuilt = os.path.join(self.tmp.name, "sql_prebuilt") + os.makedirs(prebuilt, exist_ok=True) + for name in experiment_config.PLANNER_OUTPUT_FILES: + open(os.path.join(prebuilt, name), "w").close() + kwargs["prebuilt_output_dir"] = prebuilt + 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_classic_metrics_without_sql(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': 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() + 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"], + ) + # 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`. + 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_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: + 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_sql_queries_need_prebuilt_planner_configs(self): + # The planner rejects the SQL groups' time_window_seconds. + with self.assertRaisesRegex(ValueError, "--prebuilt-output-dir"): + sbe.build_config( + self.table, + seed=7, + starting_delay_secs=120, + query_duration_secs=600, + sql_queries_file=os.path.join(self.tmp.name, "q.sql"), + ) + + 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 compose_and_validate(self, config): + # A dry run of what experiment_run_e2e.py does with the file before it + # 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( + config, + 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), + 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), + # The generated config overrides config.yaml's `true`. + forward_unsupported_queries=cfg.streaming.forward_unsupported_queries, + auto_discover_labels=cfg.controller.auto_discover_labels, + ) + modes, _ = experiment_config.generate_controller_client_configs( + params, self.tmp.name + ) + return modes + + def test_generated_config_composes_and_validates(self): + self.assertEqual( + self.compose_and_validate(self.build()[0]), ["sketchdb", "baseline"] + ) + + def test_runner_can_anchor_the_generated_client_config(self): + # What experiment_run_e2e.py does to the mode's client config once the + # feeder starts: every group must learn where the data starts. + self.compose_and_validate(self.build()[0]) + path = os.path.join(self.tmp.name, "controller_client_configs", "sketchdb.yaml") + with open(path) as f: + client_config = yaml.safe_load(f) + self.assertEqual(aligned_interval_lcm(client_config), 60) + set_data_start(client_config, 1791885600000) + self.assertEqual( + { + g["client_options"]["data_start_time"] + for g in client_config["query_groups"] + }, + {1791885600}, + ) + + 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__": + 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..b35c5d78 --- /dev/null +++ b/asap-tools/queriers/prometheus-client/classes/query_time.py @@ -0,0 +1,50 @@ +"""Evaluation times aligned to a query group's repeat interval. + +With `client_options.align_to_interval`, the data starts at time S +(`client_options.data_start_time`, epoch seconds, a multiple of every aligned +group's interval). Repetition k of a group with repeat interval T +(`repetition_delay_ms`), starting delay D (`starting_delay`) and settle offset +O (`query_time_offset`, default 0) evaluates at + + t_k = S + (max(1, ceil(D / T)) + k) * T + +and is sent at t_k + O, once its data has arrived. Every evaluation time is a +multiple of T, matching ASAPQuery's epoch-aligned windows, and repetition k +covers the same samples in every run regardless of when the run started. For +example, T = 60 s, D = 120 s, O = 10 s and data from 10:00:00: repetitions +evaluate at 10:02:00, 10:03:00, ... and are sent at 10:02:10, 10:03:10, ... +""" + +import math + +MS_PER_SEC = 1000 + + +def first_aligned_eval_time( + data_start_time: int, interval_ms: int, starting_delay_secs: float +) -> int: + """The first multiple of the interval at least `starting_delay_secs` (and + one full interval) after the data starts.""" + 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}" + ) + interval_secs = interval_ms // MS_PER_SEC + if data_start_time % interval_secs != 0: + raise ValueError( + f"data_start_time {data_start_time} is not a multiple of the " + f"{interval_secs} s interval" + ) + intervals = max(1, math.ceil(starting_delay_secs / interval_secs)) + return data_start_time + intervals * 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.""" + if offset_secs < 0: + raise ValueError(f"query_time_offset must be >= 0, got {offset_secs}") + 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 84be87f4..1382c7c3 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 @@ -288,7 +293,10 @@ def handle_query_group( streaming_serializer: Optional[Any] = None, ) -> ResultDict: logger.debug(f"Starting query group {query_group.id}") - if query_group.starting_delay: + # See classes/query_time.py. Aligned groups apply the starting delay in + # event time and ignore query_start_times. + align_to_interval = getattr(query_group, "align_to_interval", False) + if query_group.starting_delay and not align_to_interval: logger.debug( f"Waiting for {query_group.starting_delay} seconds before starting" ) @@ -317,30 +325,54 @@ def handle_query_group( } simulated_start_time = getattr(query_group, "simulated_start_time", None) + if align_to_interval and simulated_start_time is not None: + raise ValueError( + "align_to_interval follows data_start_time; it can't be combined " + "with simulated_start_time" + ) + if align_to_interval: + data_start_time = getattr(query_group, "data_start_time", None) + if data_start_time is None: + raise ValueError( + "align_to_interval needs client_options.data_start_time " + "(set by the paced feeder runner)" + ) + offset_secs = getattr(query_group, "query_time_offset", 0) + first_eval_time = first_aligned_eval_time( + data_start_time, query_group.repetition_delay_ms, query_group.starting_delay + ) for repetition_idx in range(query_group.repetitions): - if simulated_start_time is None: - current_time = datetime.datetime.now() - else: - # Pre-loaded data has fixed timestamps, so query time follows a - # clock anchored at the data start instead of the wall clock. - current_time = datetime.datetime.fromtimestamp( - simulated_start_time - + query_group.starting_delay - + repetition_idx * query_group.repetition_delay_ms / 1000 + if align_to_interval: + query_unix_time = aligned_eval_time( + first_eval_time, query_group.repetition_delay_ms, repetition_idx ) - logger.debug("Current unix time: {}", int(current_time.timestamp())) + time.sleep(secs_until_send(query_unix_time, offset_secs, time.time())) + logger.debug("Aligned evaluation time: {}", query_unix_time) + else: + if simulated_start_time is None: + current_time = datetime.datetime.now() + else: + # Pre-loaded data has fixed timestamps, so query time follows a + # clock anchored at the data start instead of the wall clock. + current_time = datetime.datetime.fromtimestamp( + simulated_start_time + + query_group.starting_delay + + repetition_idx * query_group.repetition_delay_ms / 1000 + ) + 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 - ) + 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 @@ -348,11 +380,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(): @@ -391,11 +427,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) @@ -439,7 +479,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..acac7c1c --- /dev/null +++ b/asap-tools/queriers/prometheus-client/tests/test_query_time.py @@ -0,0 +1,58 @@ +"""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 10:00:00 UTC, a multiple of 300 s. +START = 1791885600 + + +class FirstAlignedEvalTimeTest(unittest.TestCase): + def test_first_multiple_after_the_starting_delay(self) -> None: + self.assertEqual(first_aligned_eval_time(START, 60000, 120), START + 120) + self.assertEqual(first_aligned_eval_time(START, 60000, 121), START + 180) + self.assertEqual(first_aligned_eval_time(START, 1000, 120), START + 120) + + def test_first_window_is_always_full(self) -> None: + # A starting delay shorter than the interval would evaluate a window + # that began before the data did. + self.assertEqual(first_aligned_eval_time(START, 300000, 120), START + 300) + self.assertEqual(first_aligned_eval_time(START, 60000, 0), START + 60) + + def test_start_off_the_interval_rejected(self) -> None: + # Eval times off the interval would miss ASAPQuery's aligned windows. + with self.assertRaisesRegex(ValueError, "not a multiple"): + first_aligned_eval_time(START + 60, 300000, 0) + + def test_sub_second_interval_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "whole-second"): + first_aligned_eval_time(START, 1500, 0) + + +class ScheduleTest(unittest.TestCase): + def test_repetitions_step_by_the_interval(self) -> None: + self.assertEqual( + [aligned_eval_time(START, 60000, k) for k in range(3)], + [START, START + 60, START + 120], + ) + + def test_waits_for_the_offset_after_the_eval_time(self) -> None: + self.assertEqual(secs_until_send(START + 60, 10, START + 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(START + 60, 10, START + 75), 0) + + def test_negative_offset_rejected(self) -> None: + with self.assertRaisesRegex(ValueError, "query_time_offset"): + secs_until_send(START, -1, START) + + +if __name__ == "__main__": + unittest.main()