feat(asap-tools): run sketch-bench workload tables end to end with the paced feeder

.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

asap-tools/experiments/CONFIG_PARAMETERS_REFERENCE.md

@@ -457,6 +457,50 @@ 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` 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
+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, `<metric>.<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 +544,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`

asap-tools/experiments/constants.py

@@ -42,6 +42,12 @@
 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"
+# 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"
 

asap-tools/experiments/experiment_run_e2e.py

@@ -25,6 +25,11 @@
     DockerPrometheusService,
     DockerVictoriaMetricsService,
     SystemExportersService,
+    PACED_FEEDER_EXPORTER,
+    PacedFeederService,
+    check_paced_feeder_config,
+    paced_feeder_min_duration_secs,
+    paced_feeder_sinks,
 )
 from experiment_utils.services.misc import DiscoveryBackend
 
@@ -188,6 +193,25 @@ 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=paced_feeder_min_duration_secs(
+                0 if skip_querying else args.steady_state_wait,
+                minimum_experiment_running_time,
+            ),
+        )
+        paced_feeder_service = PacedFeederService(provider, args)
+
     for experiment_mode in experiment_modes:
         print(f"Running experiment mode: {experiment_mode}")
         experiment_output_dir = os.path.join(
@@ -228,6 +252,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 +495,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 +594,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()
 

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

asap-tools/experiments/experiment_utils/services/__init__.py

@@ -34,6 +34,13 @@
     ControllerService,
 )
 from .grafana import GrafanaService
+from .paced_feeder import (
+    PACED_FEEDER_EXPORTER,
+    PacedFeederService,
+    check_paced_feeder_config,
+    paced_feeder_min_duration_secs,
+    paced_feeder_sinks,
+)
 from .clickhouse_service import ClickHouseService, ClickHouseDataLoaderService
 
 
@@ -71,6 +78,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 +119,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 +130,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 +162,9 @@ 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_min_duration_secs",
+    "paced_feeder_sinks",
 ]

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)
 

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