From be66ac5211e32d71c6161fdc3c3bb0f4704cc380 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Sun, 11 Oct 2026 20:11:33 -0400 Subject: [PATCH 1/3] feat(planner): asap-planner --planner saved writes a sketch-bench eval plan without re-solving Reads a plan from sketch-bench's autosketch_vs_asap plan file and the eval table's `e2e` block (PromQL queries, the plan RQEs each reads, metric labels), and writes streaming and inference YAML through the same converter as --planner milp. experiment_run_e2e.py gains controller.prebuilt_output_dir to deploy those configs instead of running the planner. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + asap-planner-rs/Cargo.toml | 1 + asap-planner-rs/src/main.rs | 96 ++++- .../src/optimizer/aqe_extractor.rs | 2 +- asap-planner-rs/src/optimizer/milp.rs | 13 +- asap-planner-rs/src/optimizer/milp_output.rs | 42 +- asap-planner-rs/src/optimizer/mod.rs | 11 +- asap-planner-rs/src/optimizer/saved_plan.rs | 407 ++++++++++++++++++ asap-tools/experiments/config/config.yaml | 3 + asap-tools/experiments/experiment_run_e2e.py | 46 +- .../experiments/experiment_utils/config.py | 12 + 11 files changed, 600 insertions(+), 34 deletions(-) create mode 100644 asap-planner-rs/src/optimizer/saved_plan.rs diff --git a/Cargo.lock b/Cargo.lock index e1d15191..50c54223 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -166,6 +166,7 @@ dependencies = [ "chrono", "clap 4.6.1", "elastic_dsl_utilities", + "flate2", "indexmap 2.14.0", "pretty_assertions", "promql-parser", diff --git a/asap-planner-rs/Cargo.toml b/asap-planner-rs/Cargo.toml index 5e9bf33d..0c89320d 100644 --- a/asap-planner-rs/Cargo.toml +++ b/asap-planner-rs/Cargo.toml @@ -37,6 +37,7 @@ indexmap.workspace = true chrono.workspace = true promql-parser = "0.5.0" reqwest = { version = "0.11", features = ["blocking", "json"] } +flate2 = "1.1" rqe-optimizer = { git = "https://github.com/ProjectASAP/sketch-bench", branch = "main" } [dev-dependencies] diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index c4b2e20b..e4948501 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -1,5 +1,6 @@ use asap_planner::optimizer::{ - parse_weight, plan_milp, plan_to_planner_output, reject_unwritable_queries, MilpInputs, + parse_weight, plan_milp, plan_to_planner_output, reject_unwritable_queries, saved_plan_output, + MilpInputs, PlanSelector, }; use asap_planner::{ Controller, ControllerConfig, ElasticController, ElasticRuntimeOptions, RuntimeOptions, @@ -57,10 +58,33 @@ struct Args { clickhouse_database: Option, /// `milp` plans with sketch-bench's rqe-optimizer: PromQL with - /// --input_config only, labels from its `metrics:` hints. + /// --input_config only, labels from its `metrics:` hints. `saved` writes + /// a plan sketch-bench's `autosketch_vs_asap` already solved. #[arg(long, value_enum, default_value = "legacy")] planner: PlannerArg, + /// Saved only. sketch-bench `autosketch_vs_asap` plan file + /// (`.plans.json.gz`). + #[arg(long, required_if_eq("planner", "saved"))] + plan: Option, + + /// Saved only. The eval workload table the plan was solved for; its + /// `e2e` block gives the PromQL queries and metric labels. + #[arg(long = "workload-table", required_if_eq("planner", "saved"))] + workload_table: Option, + + /// Saved only. The plan's method, e.g. `asap`. + #[arg(long, required_if_eq("planner", "saved"))] + method: Option, + + /// Saved only. The plan's objective weights, e.g. `cpu` or `fargate`. + #[arg(long, required_if_eq("planner", "saved"))] + weights: Option, + + /// Saved only. The plan at this latency SLA; without it, the cheapest. + #[arg(long = "sla-ms")] + sla_ms: Option, + /// MILP only. YAML workload facts: per metric, positive `value_range` and /// `cardinality` per label set, including all labels (the series count), /// plus the `shape` of each grouping sketches may serve. @@ -99,6 +123,7 @@ enum EngineArg { enum PlannerArg { Legacy, Milp, + Saved, } fn main() -> anyhow::Result<()> { @@ -116,8 +141,22 @@ fn main() -> anyhow::Result<()> { EngineArg::Precompute => StreamingEngine::Precompute, }; - if args.planner == PlannerArg::Milp { - run_milp(&args)?; + if args.planner != PlannerArg::Saved { + anyhow::ensure!( + args.plan.is_none() + && args.workload_table.is_none() + && args.method.is_none() + && args.weights.is_none() + && args.sla_ms.is_none(), + "--plan, --workload-table, --method, --weights and --sla-ms require --planner saved" + ); + } + if args.planner != PlannerArg::Legacy { + match args.planner { + PlannerArg::Milp => run_milp(&args)?, + PlannerArg::Saved => run_saved(&args)?, + PlannerArg::Legacy => unreachable!(), + } println!("Generated configs in {}", args.output_dir.display()); return Ok(()); } @@ -274,3 +313,52 @@ fn run_milp(args: &Args) -> anyhow::Result<()> { .write_to_dir(&args.output_dir)?; Ok(()) } + +fn run_saved(args: &Args) -> anyhow::Result<()> { + anyhow::ensure!( + args.input_config.is_none() && args.query_log.is_none(), + "--planner saved reads its queries from --workload-table, not --input_config or --query-log" + ); + anyhow::ensure!( + matches!(args.query_language, QueryLanguage::promql) + && args.prometheus_url.is_none() + && !args.enable_punting + && args.range_duration_ms == 0 + && args.step_ms == 0 + && args.clickhouse_url.is_none() + && args.clickhouse_database.is_none(), + "--planner saved takes only PromQL and none of --prometheus-url, --enable-punting, \ + --range-duration-ms, --step-ms, --clickhouse-url, --clickhouse-database" + ); + anyhow::ensure!( + args.workload_facts.is_none() + && args.atomic_costs.is_none() + && args.saturation_dir.is_none() + && args.w_cpu.is_none() + && args.w_mem.is_none(), + "--workload-facts, --atomic-costs, --saturation-dir, --w-cpu and --w-mem require \ + --planner milp" + ); + let scrape_interval_ms = args + .data_ingestion_interval_ms + .filter(|&ms| ms > 0) + .ok_or_else(|| { + anyhow::anyhow!("--planner saved requires a positive --data-ingestion-interval-ms") + })?; + saved_plan_output( + args.workload_table + .as_deref() + .expect("clap requires --workload-table with --planner saved"), + args.plan + .as_deref() + .expect("clap requires --plan with --planner saved"), + &PlanSelector { + method: args.method.as_deref().expect("clap requires --method"), + weights: args.weights.as_deref().expect("clap requires --weights"), + sla_ms: args.sla_ms, + }, + scrape_interval_ms, + )? + .write_to_dir(&args.output_dir)?; + Ok(()) +} diff --git a/asap-planner-rs/src/optimizer/aqe_extractor.rs b/asap-planner-rs/src/optimizer/aqe_extractor.rs index 1a7b3262..00ff2415 100644 --- a/asap-planner-rs/src/optimizer/aqe_extractor.rs +++ b/asap-planner-rs/src/optimizer/aqe_extractor.rs @@ -132,7 +132,7 @@ fn normalized_f64_bits(value: f64) -> u64 { /// dropped — they contribute no AQE. Only arithmetic operators are split; /// comparison and set operators are left as-is (treated as opaque leaves). /// An `avg` leaf becomes its sum and count leaves (see `rewrite_avg`). -fn decompose_to_leaves(query: &str) -> Vec { +pub(crate) fn decompose_to_leaves(query: &str) -> Vec { let (lhs, rhs) = match parse_binary_arms(query) { Some(arms) => arms, None => return rewrite_avg(query), diff --git a/asap-planner-rs/src/optimizer/milp.rs b/asap-planner-rs/src/optimizer/milp.rs index 71d8d83c..2f53fca9 100644 --- a/asap-planner-rs/src/optimizer/milp.rs +++ b/asap-planner-rs/src/optimizer/milp.rs @@ -243,6 +243,17 @@ pub fn build_milp_workload( config: &ControllerConfig, facts: &WorkloadFacts, scrape_interval_ms: u64, +) -> Result { + let workload = workload_from_config(config, scrape_interval_ms)?; + validate_facts(&workload.raqes, facts).map_err(MilpError::InvalidInputs)?; + Ok(workload) +} + +/// [`build_milp_workload`] without the facts check, for a plan solved +/// elsewhere. +pub fn workload_from_config( + config: &ControllerConfig, + scrape_interval_ms: u64, ) -> Result { // The MILP chooses windows and sketch parameters itself. let unsupported: Vec<&'static str> = [ @@ -298,10 +309,8 @@ pub fn build_milp_workload( tracing::debug!( items = items.len(), raqes = raqes.len(), - metrics_with_facts = facts.len(), "milp inputs: built raqes" ); - validate_facts(&raqes, facts).map_err(MilpError::InvalidInputs)?; Ok(MilpWorkload { items, raqes, diff --git a/asap-planner-rs/src/optimizer/milp_output.rs b/asap-planner-rs/src/optimizer/milp_output.rs index 8f934931..f822edb0 100644 --- a/asap-planner-rs/src/optimizer/milp_output.rs +++ b/asap-planner-rs/src/optimizer/milp_output.rs @@ -92,14 +92,36 @@ pub fn plan_to_planner_output( workload: &MilpWorkload, solution: &MilpSolution, ) -> Result { + let deployments: Vec = solution + .deployments + .iter() + .map(|planned| planned.deployment.clone()) + .collect(); + let served_by: Vec = solution.raqes.iter().map(|r| r.deployment).collect(); + deployments_to_planner_output(config, workload, &deployments, &served_by) +} + +/// [`plan_to_planner_output`] for a plan given as its deployments and, per +/// Raqe of `workload`, the index of the deployment serving it. +pub fn deployments_to_planner_output( + config: &ControllerConfig, + workload: &MilpWorkload, + deployments: &[Deployment], + served_by: &[usize], +) -> Result { + assert_eq!( + served_by.len(), + workload.raqes.len(), + "one deployment per Raqe" + ); reject_unwritable_queries(config)?; // sketch-bench lets a deployment serve a RAQE grouped by a subset of its // grouping (a roll-up, sketch-bench#190). The engine reads a deployment // only at its own grouping, and `aggregation_config` takes the labels // from the first RAQE served, so refuse a roll-up rather than write // configs at the wrong grouping. - for (raqe, planned) in workload.raqes.iter().zip(&solution.raqes) { - let deployment = &solution.deployments[planned.deployment].deployment; + for (raqe, &index) in workload.raqes.iter().zip(served_by) { + let deployment = &deployments[index]; if raqe.grouping_labels != deployment.grouping_labels { return Err(MilpOutputError::RollupUnsupported { query: raqe.id.clone(), @@ -111,26 +133,26 @@ pub fn plan_to_planner_output( let item_of = |raqe: usize| &workload.items[workload.raqe_items[raqe]]; // Each item once per deployment, however many occurrences it has. - let mut served: Vec> = vec![Vec::new(); solution.deployments.len()]; + let mut served: Vec> = vec![Vec::new(); deployments.len()]; let mut seen = HashSet::new(); - for (raqe, planned) in solution.raqes.iter().enumerate() { - if seen.insert((planned.deployment, workload.raqe_items[raqe])) { - served[planned.deployment].push(item_of(raqe)); + for (raqe, &index) in served_by.iter().enumerate() { + if seen.insert((index, workload.raqe_items[raqe])) { + served[index].push(item_of(raqe)); } } let mut aggregations: IndexMap = IndexMap::new(); let mut deployment_keys = Vec::with_capacity(served.len()); - for (planned, items) in solution.deployments.iter().zip(&served) { - let aggregation = aggregation_config(&planned.deployment, items)?; + for (deployment, items) in deployments.iter().zip(&served) { + let aggregation = aggregation_config(deployment, items)?; let key = aggregation.identifying_key(); aggregations.entry(key.clone()).or_insert(aggregation); deployment_keys.push(key); } let mut queries: IndexMap = IndexMap::new(); - for (raqe, planned) in solution.raqes.iter().enumerate() { - let key = &deployment_keys[planned.deployment]; + for (raqe, &index) in served_by.iter().enumerate() { + let key = &deployment_keys[index]; for query in &item_of(raqe).query_strings { match queries.entry(query.clone()) { Entry::Vacant(entry) => { diff --git a/asap-planner-rs/src/optimizer/mod.rs b/asap-planner-rs/src/optimizer/mod.rs index 7731c14c..2b0b46cc 100644 --- a/asap-planner-rs/src/optimizer/mod.rs +++ b/asap-planner-rs/src/optimizer/mod.rs @@ -3,6 +3,7 @@ pub mod atomic_costs; pub mod error; pub mod milp; pub mod milp_output; +pub mod saved_plan; pub mod solution; pub mod workload_facts; @@ -10,9 +11,13 @@ pub use aqe_extractor::{extract_aqes, RQE}; pub use atomic_costs::{load_flat_atomic_cost_table, AtomicCostEntry, AtomicCostTable}; pub use error::OptimizerError; pub use milp::{ - build_milp_workload, parse_weight, plan_milp, solve_milp, MilpError, MilpInputs, MilpPlan, - MilpWorkload, + build_milp_workload, parse_weight, plan_milp, solve_milp, workload_from_config, MilpError, + MilpInputs, MilpPlan, MilpWorkload, }; -pub use milp_output::{plan_to_planner_output, reject_unwritable_queries, MilpOutputError}; +pub use milp_output::{ + deployments_to_planner_output, plan_to_planner_output, reject_unwritable_queries, + MilpOutputError, +}; +pub use saved_plan::{saved_plan_output, PlanSelector}; pub use solution::OptimizerItem; pub use workload_facts::{load_workload_facts, parse_workload_facts, WorkloadFactsError}; diff --git a/asap-planner-rs/src/optimizer/saved_plan.rs b/asap-planner-rs/src/optimizer/saved_plan.rs new file mode 100644 index 00000000..02ceecab --- /dev/null +++ b/asap-planner-rs/src/optimizer/saved_plan.rs @@ -0,0 +1,407 @@ +//! Writes a plan solved elsewhere, sketch-bench `autosketch_vs_asap`'s plan +//! file, as the planner's streaming and inference YAML, without re-solving. +//! The workload comes from the eval table's `e2e` block: its PromQL queries, +//! the plan RQEs each reads, and the metrics' labels. + +use std::collections::{BTreeMap, HashMap}; +use std::io::Read; +use std::path::Path; + +use anyhow::{bail, ensure, Context}; +use rqe_optimizer::{AtomicCostEntry, Deployment}; +use serde::Deserialize; +use serde_json::{json, Value}; + +use crate::config::input::ControllerConfig; +use crate::planner_output::PlannerOutput; + +use super::aqe_extractor::decompose_to_leaves; +use super::milp::workload_from_config; +use super::milp_output::deployments_to_planner_output; + +#[derive(Deserialize)] +struct Table { + workloads: Vec, +} + +#[derive(Deserialize)] +struct TableWorkload { + e2e: Option, +} + +#[derive(Deserialize)] +struct E2e { + queries: Vec, + metrics: BTreeMap, +} + +#[derive(Deserialize)] +struct E2eQuery { + promql: String, + interval_secs: u64, + /// Plan RQE ids, one per leaf of `promql` in order. + rqes: Vec, +} + +#[derive(Deserialize)] +struct E2eMetric { + labels: Vec, +} + +#[derive(Deserialize)] +struct E2eLabel { + name: String, +} + +#[derive(Deserialize)] +struct PlanFile { + plans: Vec, +} + +#[derive(Deserialize)] +struct Plan { + method: String, + weights: String, + kind: String, + sla_ms: Option, + chosen: Vec, +} + +#[derive(Deserialize)] +struct Chosen { + rqe: String, + sketch: String, + params: Value, + window_secs: u64, + slide_secs: u64, + /// Index of the plan's deployment. + deployment: usize, + /// Present only when the deployment's grouping differs from the RQE's. + deployment_grouping: Option, +} + +/// Which plan of the file to write. +pub struct PlanSelector<'a> { + pub method: &'a str, + pub weights: &'a str, + /// `None` for the cheapest plan, else the plan at this latency SLA. + pub sla_ms: Option, +} + +/// The controller config the table's `e2e` block describes: one query group +/// per repeat interval. +fn e2e_config(e2e: &E2e) -> anyhow::Result { + let mut by_interval: BTreeMap> = BTreeMap::new(); + for q in &e2e.queries { + by_interval + .entry(q.interval_secs) + .or_default() + .push(&q.promql); + } + let config = json!({ + "query_groups": by_interval.iter().map(|(secs, queries)| json!({ + "queries": queries, + "repetition_delay_ms": secs * 1000, + })).collect::>(), + "metrics": e2e.metrics.iter().map(|(metric, m)| json!({ + "metric": metric, + "labels": m.labels.iter().map(|l| &l.name).collect::>(), + })).collect::>(), + }); + Ok(serde_json::from_value(config)?) +} + +fn read_json(path: &Path) -> anyhow::Result { + let bytes = std::fs::read(path).with_context(|| format!("reading {}", path.display()))?; + let text = if bytes.starts_with(&[0x1f, 0x8b]) { + let mut text = String::new(); + flate2::read::GzDecoder::new(&bytes[..]).read_to_string(&mut text)?; + text + } else { + String::from_utf8(bytes)? + }; + serde_json::from_str(&text).with_context(|| format!("parsing {}", path.display())) +} + +/// Streaming and inference YAML for the selected plan of `plan_path` (JSON, +/// gzipped or not) over the workload of `table_path`. +pub fn saved_plan_output( + table_path: &Path, + plan_path: &Path, + selector: &PlanSelector, + scrape_interval_ms: u64, +) -> anyhow::Result { + let table: Table = read_json(table_path)?; + let [TableWorkload { e2e: Some(e2e) }] = table.workloads.as_slice() else { + bail!( + "{}: expected one workload with an `e2e` block", + table_path.display() + ); + }; + let plans: PlanFile = read_json(plan_path)?; + let kind = if selector.sla_ms.is_some() { + "sla" + } else { + "cheapest" + }; + let matching: Vec<&Plan> = plans + .plans + .iter() + .filter(|p| { + p.method == selector.method + && p.weights == selector.weights + && p.kind == kind + && p.sla_ms == selector.sla_ms + }) + .collect(); + let [plan] = matching.as_slice() else { + bail!( + "{}: {} plans match method {:?}, weights {:?}, {kind} {:?}; expected one", + plan_path.display(), + matching.len(), + selector.method, + selector.weights, + selector.sla_ms, + ); + }; + let chosen: HashMap<&str, &Chosen> = plan.chosen.iter().map(|c| (c.rqe.as_str(), c)).collect(); + + // Leaf query string and repeat interval -> the plan RQE it reads. + let mut rqe_of: HashMap<(String, u64), &str> = HashMap::new(); + for q in &e2e.queries { + let leaves = decompose_to_leaves(&q.promql); + ensure!( + leaves.len() == q.rqes.len(), + "query {:?} has {} leaves but names {} RQEs", + q.promql, + leaves.len(), + q.rqes.len() + ); + for (leaf, rqe) in leaves.into_iter().zip(&q.rqes) { + let key = (leaf, q.interval_secs * 1000); + if let Some(other) = rqe_of.insert(key.clone(), rqe) { + ensure!( + other == rqe, + "leaf {:?} every {} ms reads both {other:?} and {rqe:?}", + key.0, + key.1 + ); + } + } + } + + let config = e2e_config(e2e)?; + let workload = workload_from_config(&config, scrape_interval_ms)?; + + // Plan deployment index -> its position in `deployments`, in order of + // the first Raqe each serves. + let mut position: HashMap = HashMap::new(); + let mut deployments: Vec = Vec::new(); + let mut served_by = Vec::with_capacity(workload.raqes.len()); + for (raqe, &item) in workload.raqes.iter().zip(&workload.raqe_items) { + let item = &workload.items[item]; + // An item's leaves (say p90 and p50 over one window) read one + // deployment, so their picks must name the same sketch. + let mut index = None; + for leaf in &item.query_strings { + let rqe = rqe_of + .get(&(leaf.clone(), item.t_repeat_ms)) + .with_context(|| format!("leaf {leaf:?} has no RQE in the table's e2e block"))?; + let pick = chosen + .get(rqe) + .with_context(|| format!("the plan doesn't serve RQE {rqe:?}"))?; + ensure!( + pick.deployment_grouping.is_none(), + "RQE {rqe:?} is served by a roll-up, which the engine can't read yet" + ); + let deployment = Deployment { + capability: raqe.capability, + metric: raqe.metric.clone(), + spatial_filter: raqe.spatial_filter.clone(), + grouping_labels: raqe.grouping_labels.clone(), + config: cost_entry(pick)?, + window_ms: pick.window_secs * 1000, + slide_ms: pick.slide_secs * 1000, + key_tracker: None, + }; + let at = *position.entry(pick.deployment).or_insert_with(|| { + deployments.push(deployment.clone()); + deployments.len() - 1 + }); + // One plan deployment is one sketch: every Raqe it serves must agree. + ensure!( + same_sketch(&deployments[at], &deployment), + "plan deployment {} serves Raqes that need different sketches: {:?} and {deployment:?}", + pick.deployment, + deployments[at] + ); + ensure!( + index.is_none_or(|index| index == at), + "query {:?} is served by two deployments", + item.query_strings + ); + index = Some(at); + } + let index = index.expect("an item has a query string"); + served_by.push(index); + } + Ok(deployments_to_planner_output( + &config, + &workload, + &deployments, + &served_by, + )?) +} + +fn same_sketch(a: &Deployment, b: &Deployment) -> bool { + ( + a.capability, + &a.metric, + &a.spatial_filter, + &a.grouping_labels, + &a.config.sketch, + &a.config.sketch_config, + a.window_ms, + a.slide_ms, + ) == ( + b.capability, + &b.metric, + &b.spatial_filter, + &b.grouping_labels, + &b.config.sketch, + &b.config.sketch_config, + b.window_ms, + b.slide_ms, + ) +} + +/// The sketch config the YAML is written from. Costs and accuracy don't +/// reach the YAML, so they are zero. +fn cost_entry(pick: &Chosen) -> anyhow::Result { + Ok(serde_json::from_value(json!({ + "sketch": pick.sketch, + "sketch_config": {"algorithm": pick.sketch, "params": pick.params}, + "mem_bytes_per_instance": 0.0, + "insert_cpu_secs": 0.0, + "merge_cpu_secs": 0.0, + "query_cpu_secs": 0.0, + "query_accuracy": {}, + "accuracy_metric": "", + "measured_at": { + "items_per_instance": 0, + "keys_per_instance": null, + "value_range": null, + "merge_operand_items": null, + "distribution": null, + }, + }))?) +} + +#[cfg(test)] +mod tests { + use super::*; + + const TABLE: &str = r#"{"workloads": [{"e2e": { + "queries": [ + {"promql": "sum by (job) (data)", "interval_secs": 1, "rqes": ["sum"]}, + {"promql": "quantile_over_time(0.9, data[5m]) / quantile_over_time(0.5, data[5m])", + "interval_secs": 60, "rqes": ["p90", "p50"]}, + {"promql": "quantile_over_time(0.5, data[5m])", "interval_secs": 60, "rqes": ["p50"]} + ], + "metrics": {"data": {"type": "gauge", "labels": [ + {"name": "job", "cardinality": 10}, + {"name": "label_0", "cardinality": 1000, "child_of": "job"} + ]}} + }}]}"#; + + fn pick(rqe: &str, deployment: usize, sketch: &str, params: Value, window: u64) -> Value { + json!({"rqe": rqe, "deployment": deployment, "sketch": sketch, "params": params, + "window_secs": window, "slide_secs": 60.min(window)}) + } + + /// The cheapest `asap`/`cpu` plan: the ratio's legs share one sliding + /// DDSketch. The SLA plan differs only in alpha. + fn plans(p50: Value) -> Value { + let dd = |alpha: f64| json!({"alpha": alpha}); + json!({"plans": [ + {"method": "asap", "weights": "cpu", "kind": "cheapest", "chosen": [ + pick("sum", 3, "exact-sum", json!({}), 1), + pick("p90", 0, "dd", dd(0.02), 300), + p50, + ]}, + {"method": "asap", "weights": "cpu", "kind": "sla", "sla_ms": 100.0, "chosen": [ + pick("sum", 3, "exact-sum", json!({}), 1), + pick("p90", 0, "dd", dd(0.01), 300), + pick("p50", 0, "dd", dd(0.01), 300), + ]}, + ]}) + } + + fn run(plan: Value, sla_ms: Option) -> anyhow::Result { + let dir = tempfile::tempdir().unwrap(); + let (table_path, plan_path) = (dir.path().join("table.json"), dir.path().join("plan.json")); + std::fs::write(&table_path, TABLE).unwrap(); + std::fs::write(&plan_path, plan.to_string()).unwrap(); + let selector = PlanSelector { + method: "asap", + weights: "cpu", + sla_ms, + }; + saved_plan_output(&table_path, &plan_path, &selector, 1000) + } + + fn refusal(plan: Value, sla_ms: Option) -> String { + run(plan, sla_ms).err().expect("refused").to_string() + } + + fn shared_p50() -> Value { + pick("p50", 0, "dd", json!({"alpha": 0.02}), 300) + } + + #[test] + fn writes_one_aggregation_per_plan_deployment() { + let output = run(plans(shared_p50()), None).unwrap(); + assert_eq!(output.streaming_aggregation_count(), 2); + // The ratio's two leaves and the sum; the standalone p50 is a leaf already. + assert_eq!(output.inference_query_count(), 3); + let yaml = output.to_streaming_yaml_string().unwrap(); + assert!(yaml.contains("windowSizeMs: 300000"), "{yaml}"); + assert!(yaml.contains("slideIntervalMs: 60000"), "{yaml}"); + assert!(yaml.contains("windowType: sliding"), "{yaml}"); + assert!(yaml.contains("alpha: 0.02"), "{yaml}"); + } + + #[test] + fn sla_ms_selects_the_plan_at_that_sla() { + let yaml = run(plans(shared_p50()), Some(100.0)) + .unwrap() + .to_streaming_yaml_string() + .unwrap(); + assert!(yaml.contains("alpha: 0.01"), "{yaml}"); + let err = refusal(plans(shared_p50()), Some(300.0)); + assert!(err.contains("0 plans match"), "{err}"); + } + + #[test] + fn refuses_a_roll_up() { + let mut p50 = shared_p50(); + p50["deployment_grouping"] = json!(["job"]); + let err = refusal(plans(p50), None); + assert!(err.contains("roll-up"), "{err}"); + } + + #[test] + fn refuses_one_deployment_with_two_sketches() { + let p50 = pick("p50", 0, "dd", json!({"alpha": 0.05}), 300); + let err = refusal(plans(p50), None); + assert!(err.contains("different sketches"), "{err}"); + } + + #[test] + fn refuses_a_plan_missing_an_rqe() { + let mut plan = plans(shared_p50()); + plan["plans"][0]["chosen"].as_array_mut().unwrap().pop(); + let err = refusal(plan, None); + assert!(err.contains(r#"doesn't serve RQE "p50""#), "{err}"); + } +} diff --git a/asap-tools/experiments/config/config.yaml b/asap-tools/experiments/config/config.yaml index dfd4abd9..8539b4ff 100644 --- a/asap-tools/experiments/config/config.yaml +++ b/asap-tools/experiments/config/config.yaml @@ -90,6 +90,9 @@ query_engine: controller: punting: ??? # REQUIRED: Enable query punting based on performance heuristics (should_be_performant check) auto_discover_labels: false # Query Prometheus for metric labels instead of using config hints + # Skip the planner and deploy the streaming_config.yaml and inference_config.yaml + # already in this directory, e.g. from `asap-planner --planner saved`. + prebuilt_output_dir: null # Optional manual planner windowing override. Omit this block to preserve the # default tumbling-window behavior. For tumbling windows, omit diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index 9b288ad8..e7188f5b 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -1,5 +1,6 @@ import os import json +import shutil import time import hydra @@ -316,20 +317,37 @@ def main(cfg: DictConfig): "Prometheus label discovery is disabled." ) - controller_service.start( - controller_input_file=controller_input_config, - streaming_engine=args.streaming_engine, - controller_remote_output_dir=CONTROLLER_REMOTE_OUTPUT_DIR, - punting=args.controller_punting, - discovery_backend=discovery_backend, - data_ingestion_interval_ms=data_ingestion_interval_ms, - ) - sync.rsync_controller_config_remote_to_local( - provider, - CONTROLLER_REMOTE_OUTPUT_DIR, - CONTROLLER_LOCAL_OUTPUT_DIR, - node_offset=args.node_offset, - ) + if args.controller_prebuilt_output_dir: + print( + f"Skipping the planner: using configs from " + f"{args.controller_prebuilt_output_dir}" + ) + shutil.copytree( + args.controller_prebuilt_output_dir, + CONTROLLER_LOCAL_OUTPUT_DIR, + dirs_exist_ok=True, + ) + sync.rsync_streaming_configs( + provider, + CONTROLLER_LOCAL_OUTPUT_DIR, + CONTROLLER_REMOTE_OUTPUT_DIR, + node_offset=args.node_offset, + ) + else: + controller_service.start( + controller_input_file=controller_input_config, + streaming_engine=args.streaming_engine, + controller_remote_output_dir=CONTROLLER_REMOTE_OUTPUT_DIR, + punting=args.controller_punting, + discovery_backend=discovery_backend, + data_ingestion_interval_ms=data_ingestion_interval_ms, + ) + sync.rsync_controller_config_remote_to_local( + provider, + CONTROLLER_REMOTE_OUTPUT_DIR, + CONTROLLER_LOCAL_OUTPUT_DIR, + node_offset=args.node_offset, + ) if ( config.check_exporter_and_queries_exist( diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py index a731c005..a1a363aa 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/asap-tools/experiments/experiment_utils/config.py @@ -9,6 +9,7 @@ import yaml from typing import Any, Dict, List, Tuple +import hydra.utils from omegaconf import DictConfig, ListConfig, OmegaConf import constants @@ -605,6 +606,17 @@ def __init__(self, cfg: DictConfig): # Controller configuration self.controller_punting = cfg.controller.punting self.controller_auto_discover_labels = cfg.controller.auto_discover_labels + self.controller_prebuilt_output_dir = cfg.controller.get("prebuilt_output_dir") + if self.controller_prebuilt_output_dir: + self.controller_prebuilt_output_dir = hydra.utils.to_absolute_path( + self.controller_prebuilt_output_dir + ) + for name in ("streaming_config.yaml", "inference_config.yaml"): + path = os.path.join(self.controller_prebuilt_output_dir, name) + if not os.path.isfile(path): + raise ValueError( + f"controller.prebuilt_output_dir has no {name}: {path}" + ) # Aggregate cleanup configuration # Valid policies: "circular_buffer", "read_based", "no_cleanup" From e7a49eb8f5b965d0c98186f413f8c6852e9d108e Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 12 Oct 2026 18:05:37 -0400 Subject: [PATCH 2/3] fix(planner): address review on --planner saved Check saved picks against the RAQE's window rules, refuse duplicate picks, reject avg queries before the leaf split, and take a roll-up's grouping from the plan so the shared guard catches it. Share the flag checks of --planner milp and saved. Copy only the two prebuilt configs into a cleared output dir, refusing the dir itself. Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/main.rs | 83 +++++++---------- asap-planner-rs/src/optimizer/saved_plan.rs | 91 ++++++++++++++++--- asap-tools/experiments/experiment_run_e2e.py | 7 +- .../experiments/experiment_utils/config.py | 31 +++++-- .../tests/test_prebuilt_controller_output.py | 45 +++++++++ 5 files changed, 180 insertions(+), 77 deletions(-) create mode 100644 asap-tools/experiments/tests/test_prebuilt_controller_output.py diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index e4948501..01f556da 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -151,6 +151,17 @@ fn main() -> anyhow::Result<()> { "--plan, --workload-table, --method, --weights and --sla-ms require --planner saved" ); } + if args.planner != PlannerArg::Milp { + anyhow::ensure!( + args.workload_facts.is_none() + && args.atomic_costs.is_none() + && args.saturation_dir.is_none() + && args.w_cpu.is_none() + && args.w_mem.is_none(), + "--workload-facts, --atomic-costs, --saturation-dir, --w-cpu and --w-mem require \ + --planner milp" + ); + } if args.planner != PlannerArg::Legacy { match args.planner { PlannerArg::Milp => run_milp(&args)?, @@ -160,15 +171,6 @@ fn main() -> anyhow::Result<()> { println!("Generated configs in {}", args.output_dir.display()); return Ok(()); } - anyhow::ensure!( - args.workload_facts.is_none() - && args.atomic_costs.is_none() - && args.saturation_dir.is_none() - && args.w_cpu.is_none() - && args.w_mem.is_none(), - "--workload-facts, --atomic-costs, --saturation-dir, --w-cpu and --w-mem require \ - --planner milp" - ); match args.query_language { QueryLanguage::promql => { @@ -252,39 +254,43 @@ fn main() -> anyhow::Result<()> { Ok(()) } -fn run_milp(args: &Args) -> anyhow::Result<()> { +/// Checks the flags `--planner milp` and `--planner saved` share: PromQL +/// with labels from `metrics:` hints, no legacy-only flags. Returns the +/// positive scrape interval, which per-series sample rates divide by. +fn check_optimizer_flags(args: &Args, planner: &str) -> anyhow::Result { anyhow::ensure!( matches!(args.query_language, QueryLanguage::promql), - "--planner milp supports only --query-language promql" + "--planner {planner} supports only --query-language promql" ); anyhow::ensure!( args.query_log.is_none(), - "--planner milp needs --input_config: query logs have no `metrics:` hints" + "--planner {planner} doesn't take --query-log: query logs have no `metrics:` hints" ); anyhow::ensure!( args.prometheus_url.is_none(), - "--planner milp takes labels from the config's `metrics:` hints, not --prometheus-url" + "--planner {planner} takes labels from `metrics:` hints, not --prometheus-url" ); anyhow::ensure!( !args.enable_punting && args.range_duration_ms == 0 && args.step_ms == 0, - "--enable-punting, --range-duration-ms and --step-ms don't apply to --planner milp" + "--enable-punting, --range-duration-ms and --step-ms don't apply to --planner {planner}" ); anyhow::ensure!( args.clickhouse_url.is_none() && args.clickhouse_database.is_none(), - "--clickhouse-url and --clickhouse-database don't apply to --planner milp" + "--clickhouse-url and --clickhouse-database don't apply to --planner {planner}" ); + args.data_ingestion_interval_ms + .filter(|&ms| ms > 0) + .ok_or_else(|| { + anyhow::anyhow!("--planner {planner} requires a positive --data-ingestion-interval-ms") + }) +} + +fn run_milp(args: &Args) -> anyhow::Result<()> { + let scrape_interval_ms = check_optimizer_flags(args, "milp")?; let config_path = args .input_config .as_deref() .ok_or_else(|| anyhow::anyhow!("--planner milp requires --input_config"))?; - let scrape_interval_ms = args - .data_ingestion_interval_ms - .ok_or_else(|| anyhow::anyhow!("--planner milp requires --data-ingestion-interval-ms"))?; - // Per-series sample rates divide by it. - anyhow::ensure!( - scrape_interval_ms > 0, - "--data-ingestion-interval-ms must be positive" - ); let config: ControllerConfig = serde_yaml::from_str(&std::fs::read_to_string(config_path)?)?; // Fail before solving: the plan is always written. reject_unwritable_queries(&config)?; @@ -315,36 +321,11 @@ fn run_milp(args: &Args) -> anyhow::Result<()> { } fn run_saved(args: &Args) -> anyhow::Result<()> { + let scrape_interval_ms = check_optimizer_flags(args, "saved")?; anyhow::ensure!( - args.input_config.is_none() && args.query_log.is_none(), - "--planner saved reads its queries from --workload-table, not --input_config or --query-log" - ); - anyhow::ensure!( - matches!(args.query_language, QueryLanguage::promql) - && args.prometheus_url.is_none() - && !args.enable_punting - && args.range_duration_ms == 0 - && args.step_ms == 0 - && args.clickhouse_url.is_none() - && args.clickhouse_database.is_none(), - "--planner saved takes only PromQL and none of --prometheus-url, --enable-punting, \ - --range-duration-ms, --step-ms, --clickhouse-url, --clickhouse-database" - ); - anyhow::ensure!( - args.workload_facts.is_none() - && args.atomic_costs.is_none() - && args.saturation_dir.is_none() - && args.w_cpu.is_none() - && args.w_mem.is_none(), - "--workload-facts, --atomic-costs, --saturation-dir, --w-cpu and --w-mem require \ - --planner milp" + args.input_config.is_none(), + "--planner saved reads its queries from --workload-table, not --input_config" ); - let scrape_interval_ms = args - .data_ingestion_interval_ms - .filter(|&ms| ms > 0) - .ok_or_else(|| { - anyhow::anyhow!("--planner saved requires a positive --data-ingestion-interval-ms") - })?; saved_plan_output( args.workload_table .as_deref() diff --git a/asap-planner-rs/src/optimizer/saved_plan.rs b/asap-planner-rs/src/optimizer/saved_plan.rs index 02ceecab..c383e6cd 100644 --- a/asap-planner-rs/src/optimizer/saved_plan.rs +++ b/asap-planner-rs/src/optimizer/saved_plan.rs @@ -8,7 +8,7 @@ use std::io::Read; use std::path::Path; use anyhow::{bail, ensure, Context}; -use rqe_optimizer::{AtomicCostEntry, Deployment}; +use rqe_optimizer::{AtomicCostEntry, Deployment, LabelSet}; use serde::Deserialize; use serde_json::{json, Value}; @@ -17,7 +17,7 @@ use crate::planner_output::PlannerOutput; use super::aqe_extractor::decompose_to_leaves; use super::milp::workload_from_config; -use super::milp_output::deployments_to_planner_output; +use super::milp_output::{deployments_to_planner_output, reject_unwritable_queries}; #[derive(Deserialize)] struct Table { @@ -76,8 +76,9 @@ struct Chosen { slide_secs: u64, /// Index of the plan's deployment. deployment: usize, - /// Present only when the deployment's grouping differs from the RQE's. - deployment_grouping: Option, + /// Present only when the deployment's grouping differs from the RQE's + /// (a roll-up), which the converter refuses. + deployment_grouping: Option, } /// Which plan of the file to write. @@ -164,7 +165,17 @@ pub fn saved_plan_output( selector.sla_ms, ); }; - let chosen: HashMap<&str, &Chosen> = plan.chosen.iter().map(|c| (c.rqe.as_str(), c)).collect(); + let mut chosen: HashMap<&str, &Chosen> = HashMap::new(); + for c in &plan.chosen { + ensure!( + chosen.insert(&c.rqe, c).is_none(), + "the plan picks RQE {:?} twice", + c.rqe + ); + } + let config = e2e_config(e2e)?; + // Before the leaf split, which turns an avg into two leaves. + reject_unwritable_queries(&config)?; // Leaf query string and repeat interval -> the plan RQE it reads. let mut rqe_of: HashMap<(String, u64), &str> = HashMap::new(); @@ -190,7 +201,6 @@ pub fn saved_plan_output( } } - let config = e2e_config(e2e)?; let workload = workload_from_config(&config, scrape_interval_ms)?; // Plan deployment index -> its position in `deployments`, in order of @@ -210,18 +220,32 @@ pub fn saved_plan_output( let pick = chosen .get(rqe) .with_context(|| format!("the plan doesn't serve RQE {rqe:?}"))?; + let (window_ms, slide_ms) = (pick.window_secs * 1000, pick.slide_secs * 1000); + // sketch-bench's `is_eligible` window rules: a pick that breaks + // them belongs to another workload table. ensure!( - pick.deployment_grouping.is_none(), - "RQE {rqe:?} is served by a roll-up, which the engine can't read yet" + slide_ms != 0 + && window_ms.is_multiple_of(slide_ms) + && raqe.lookback_ms.is_multiple_of(window_ms) + && raqe.interval_ms.is_multiple_of(slide_ms), + "RQE {rqe:?}'s {}s window sliding every {}s can't serve {leaf:?} \ + ({} ms lookback every {} ms); does --plan match --workload-table?", + pick.window_secs, + pick.slide_secs, + raqe.lookback_ms, + raqe.interval_ms ); let deployment = Deployment { capability: raqe.capability, metric: raqe.metric.clone(), spatial_filter: raqe.spatial_filter.clone(), - grouping_labels: raqe.grouping_labels.clone(), + grouping_labels: pick + .deployment_grouping + .clone() + .unwrap_or_else(|| raqe.grouping_labels.clone()), config: cost_entry(pick)?, - window_ms: pick.window_secs * 1000, - slide_ms: pick.slide_secs * 1000, + window_ms, + slide_ms, key_tracker: None, }; let at = *position.entry(pick.deployment).or_insert_with(|| { @@ -338,9 +362,13 @@ mod tests { } fn run(plan: Value, sla_ms: Option) -> anyhow::Result { + run_on(TABLE, plan, sla_ms) + } + + fn run_on(table: &str, plan: Value, sla_ms: Option) -> anyhow::Result { let dir = tempfile::tempdir().unwrap(); let (table_path, plan_path) = (dir.path().join("table.json"), dir.path().join("plan.json")); - std::fs::write(&table_path, TABLE).unwrap(); + std::fs::write(&table_path, table).unwrap(); std::fs::write(&plan_path, plan.to_string()).unwrap(); let selector = PlanSelector { method: "asap", @@ -384,12 +412,45 @@ mod tests { #[test] fn refuses_a_roll_up() { - let mut p50 = shared_p50(); - p50["deployment_grouping"] = json!(["job"]); - let err = refusal(plans(p50), None); + let mut plan = plans(shared_p50()); + plan["plans"][0]["chosen"][0]["deployment_grouping"] = json!(["job", "label_0"]); + let err = refusal(plan, None); assert!(err.contains("roll-up"), "{err}"); } + #[test] + fn refuses_a_pick_from_another_table() { + // A 7s window doesn't divide the 5m lookback. + let p50 = pick("p50", 0, "dd", json!({"alpha": 0.02}), 7); + let err = refusal(plans(p50), None); + assert!(err.contains("does --plan match --workload-table"), "{err}"); + } + + #[test] + fn refuses_an_avg_before_splitting_it() { + let table = TABLE.replace( + r#""sum by (job) (data)", "interval_secs": 1"#, + r#""avg by (job) (data)", "interval_secs": 1"#, + ); + let err = run_on(&table, plans(shared_p50()), None) + .err() + .expect("refused") + .to_string(); + assert!(err.contains("avg"), "{err}"); + assert!(!err.contains("leaves"), "{err}"); + } + + #[test] + fn refuses_an_rqe_picked_twice() { + let mut plan = plans(shared_p50()); + plan["plans"][0]["chosen"] + .as_array_mut() + .unwrap() + .push(shared_p50()); + let err = refusal(plan, None); + assert!(err.contains(r#"picks RQE "p50" twice"#), "{err}"); + } + #[test] fn refuses_one_deployment_with_two_sketches() { let p50 = pick("p50", 0, "dd", json!({"alpha": 0.05}), 300); diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index e7188f5b..9e653f8d 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -1,6 +1,5 @@ import os import json -import shutil import time import hydra @@ -322,10 +321,8 @@ def main(cfg: DictConfig): f"Skipping the planner: using configs from " f"{args.controller_prebuilt_output_dir}" ) - shutil.copytree( - args.controller_prebuilt_output_dir, - CONTROLLER_LOCAL_OUTPUT_DIR, - dirs_exist_ok=True, + config.copy_prebuilt_controller_output( + args.controller_prebuilt_output_dir, CONTROLLER_LOCAL_OUTPUT_DIR ) sync.rsync_streaming_configs( provider, diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py index a1a363aa..b988cf98 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/asap-tools/experiments/experiment_utils/config.py @@ -5,6 +5,7 @@ import math import os +import shutil import copy import yaml from typing import Any, Dict, List, Tuple @@ -536,6 +537,29 @@ def get_prometheus_data_ingestion_interval_ms(prometheus_config): raise ValueError(f"Invalid scrape interval string: {s}") +PLANNER_OUTPUT_FILES = ("streaming_config.yaml", "inference_config.yaml") + + +def check_prebuilt_controller_output(src): + for name in PLANNER_OUTPUT_FILES: + path = os.path.join(src, name) + if not os.path.isfile(path): + raise ValueError(f"controller.prebuilt_output_dir has no {name}: {path}") + + +def copy_prebuilt_controller_output(src, dest): + """Replaces `dest` with the planner configs in `src`.""" + check_prebuilt_controller_output(src) + if os.path.realpath(src) == os.path.realpath(dest): + raise ValueError( + f"controller.prebuilt_output_dir is the experiment's own output dir: {dest}" + ) + shutil.rmtree(dest, ignore_errors=True) + os.makedirs(dest) + for name in PLANNER_OUTPUT_FILES: + shutil.copy2(os.path.join(src, name), dest) + + class Args: """Helper class to convert Hydra config to argparse-like namespace for backward compatibility.""" @@ -611,12 +635,7 @@ def __init__(self, cfg: DictConfig): self.controller_prebuilt_output_dir = hydra.utils.to_absolute_path( self.controller_prebuilt_output_dir ) - for name in ("streaming_config.yaml", "inference_config.yaml"): - path = os.path.join(self.controller_prebuilt_output_dir, name) - if not os.path.isfile(path): - raise ValueError( - f"controller.prebuilt_output_dir has no {name}: {path}" - ) + check_prebuilt_controller_output(self.controller_prebuilt_output_dir) # Aggregate cleanup configuration # Valid policies: "circular_buffer", "read_based", "no_cleanup" diff --git a/asap-tools/experiments/tests/test_prebuilt_controller_output.py b/asap-tools/experiments/tests/test_prebuilt_controller_output.py new file mode 100644 index 00000000..4dd227fc --- /dev/null +++ b/asap-tools/experiments/tests/test_prebuilt_controller_output.py @@ -0,0 +1,45 @@ +"""Tests for running e2e experiments on prebuilt planner configs.""" + +import os +import tempfile +import unittest + +from experiment_utils.config import copy_prebuilt_controller_output + + +class CopyPrebuiltControllerOutputTest(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.src = os.path.join(self.tmp.name, "src") + self.dest = os.path.join(self.tmp.name, "dest") + os.makedirs(self.src) + for name in ("streaming_config.yaml", "inference_config.yaml"): + with open(os.path.join(self.src, name), "w") as f: + f.write(name) + + def tearDown(self): + self.tmp.cleanup() + + def test_replaces_stale_output(self): + os.makedirs(self.dest) + with open(os.path.join(self.dest, "controller.log"), "w") as f: + f.write("stale") + copy_prebuilt_controller_output(self.src, self.dest) + self.assertEqual( + sorted(os.listdir(self.dest)), + ["inference_config.yaml", "streaming_config.yaml"], + ) + + def test_refuses_missing_config(self): + os.remove(os.path.join(self.src, "inference_config.yaml")) + with self.assertRaisesRegex(ValueError, "inference_config.yaml"): + copy_prebuilt_controller_output(self.src, self.dest) + + def test_refuses_own_output_dir(self): + with self.assertRaisesRegex(ValueError, "own output dir"): + copy_prebuilt_controller_output(self.src, self.src) + self.assertTrue(os.path.isfile(os.path.join(self.src, "streaming_config.yaml"))) + + +if __name__ == "__main__": + unittest.main() From c9c48e394bb2cbdc6d95a1164c308e41276d694f Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 12 Oct 2026 22:41:45 -0400 Subject: [PATCH 3/3] fix(planner): address second review on --planner saved Check saved picks against every eligibility rule that needs no workload facts (mergeability, heap size, key tracker), compare deployments with PartialEq, and dispatch planners in one match. Before copying prebuilt configs, refuse nested source and output dirs, queries the configs don't plan, and slides off the run's scrape interval. Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/main.rs | 19 ++- asap-planner-rs/src/optimizer/saved_plan.rs | 112 +++++++++++------- asap-tools/experiments/experiment_run_e2e.py | 9 +- .../experiments/experiment_utils/config.py | 45 ++++++- .../tests/test_prebuilt_controller_output.py | 58 +++++++-- 5 files changed, 177 insertions(+), 66 deletions(-) diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index 01f556da..36ce8772 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -162,16 +162,17 @@ fn main() -> anyhow::Result<()> { --planner milp" ); } - if args.planner != PlannerArg::Legacy { - match args.planner { - PlannerArg::Milp => run_milp(&args)?, - PlannerArg::Saved => run_saved(&args)?, - PlannerArg::Legacy => unreachable!(), - } - println!("Generated configs in {}", args.output_dir.display()); - return Ok(()); + let output_dir = args.output_dir.clone(); + match args.planner { + PlannerArg::Milp => run_milp(&args)?, + PlannerArg::Saved => run_saved(&args)?, + PlannerArg::Legacy => run_legacy(args, engine)?, } + println!("Generated configs in {}", output_dir.display()); + Ok(()) +} +fn run_legacy(args: Args, engine: StreamingEngine) -> anyhow::Result<()> { match args.query_language { QueryLanguage::promql => { let scrape_interval_ms = args.data_ingestion_interval_ms.ok_or_else(|| { @@ -249,8 +250,6 @@ fn main() -> anyhow::Result<()> { ElasticController::from_file(&config_path, opts)?.generate_to_dir(&args.output_dir)?; } } - - println!("Generated configs in {}", args.output_dir.display()); Ok(()) } diff --git a/asap-planner-rs/src/optimizer/saved_plan.rs b/asap-planner-rs/src/optimizer/saved_plan.rs index c383e6cd..929f271f 100644 --- a/asap-planner-rs/src/optimizer/saved_plan.rs +++ b/asap-planner-rs/src/optimizer/saved_plan.rs @@ -8,7 +8,7 @@ use std::io::Read; use std::path::Path; use anyhow::{bail, ensure, Context}; -use rqe_optimizer::{AtomicCostEntry, Deployment, LabelSet}; +use rqe_optimizer::{heap_capacity, AtomicCostEntry, Deployment, LabelSet, Raqe}; use serde::Deserialize; use serde_json::{json, Value}; @@ -220,21 +220,6 @@ pub fn saved_plan_output( let pick = chosen .get(rqe) .with_context(|| format!("the plan doesn't serve RQE {rqe:?}"))?; - let (window_ms, slide_ms) = (pick.window_secs * 1000, pick.slide_secs * 1000); - // sketch-bench's `is_eligible` window rules: a pick that breaks - // them belongs to another workload table. - ensure!( - slide_ms != 0 - && window_ms.is_multiple_of(slide_ms) - && raqe.lookback_ms.is_multiple_of(window_ms) - && raqe.interval_ms.is_multiple_of(slide_ms), - "RQE {rqe:?}'s {}s window sliding every {}s can't serve {leaf:?} \ - ({} ms lookback every {} ms); does --plan match --workload-table?", - pick.window_secs, - pick.slide_secs, - raqe.lookback_ms, - raqe.interval_ms - ); let deployment = Deployment { capability: raqe.capability, metric: raqe.metric.clone(), @@ -244,17 +229,27 @@ pub fn saved_plan_output( .clone() .unwrap_or_else(|| raqe.grouping_labels.clone()), config: cost_entry(pick)?, - window_ms, - slide_ms, + window_ms: pick.window_secs * 1000, + slide_ms: pick.slide_secs * 1000, key_tracker: None, }; + ensure!( + can_serve(raqe, &deployment), + "RQE {rqe:?}'s {} {}s window sliding every {}s can't serve {leaf:?} \ + ({} ms lookback every {} ms); does --plan match --workload-table?", + pick.sketch, + pick.window_secs, + pick.slide_secs, + raqe.lookback_ms, + raqe.interval_ms + ); let at = *position.entry(pick.deployment).or_insert_with(|| { deployments.push(deployment.clone()); deployments.len() - 1 }); // One plan deployment is one sketch: every Raqe it serves must agree. ensure!( - same_sketch(&deployments[at], &deployment), + deployments[at] == deployment, "plan deployment {} serves Raqes that need different sketches: {:?} and {deployment:?}", pick.deployment, deployments[at] @@ -277,26 +272,22 @@ pub fn saved_plan_output( )?) } -fn same_sketch(a: &Deployment, b: &Deployment) -> bool { - ( - a.capability, - &a.metric, - &a.spatial_filter, - &a.grouping_labels, - &a.config.sketch, - &a.config.sketch_config, - a.window_ms, - a.slide_ms, - ) == ( - b.capability, - &b.metric, - &b.spatial_filter, - &b.grouping_labels, - &b.config.sketch, - &b.config.sketch_config, - b.window_ms, - b.slide_ms, - ) +/// The rules of sketch-bench's `is_eligible` that need no workload facts or +/// accuracy: a pick that breaks them belongs to another workload table. +/// Grouping is left to the converter's roll-up check. +fn can_serve(r: &Raqe, d: &Deployment) -> bool { + let properties = d.properties(); + d.window_ms != 0 + && d.slide_ms != 0 + && d.window_ms.is_multiple_of(d.slide_ms) + && r.lookback_ms.is_multiple_of(d.window_ms) + && r.interval_ms.is_multiple_of(d.slide_ms) + && (properties.mergeable_across_windows || r.lookback_ms == d.window_ms) + && !properties.needs_delta_set_key_tracker + && heap_capacity(&d.config).is_none_or(|heap| { + d.heap_needed(r.lookback_ms, r.topk_k()) + .is_some_and(|need| heap >= need) + }) } /// The sketch config the YAML is written from. Costs and accuracy don't @@ -362,14 +353,14 @@ mod tests { } fn run(plan: Value, sla_ms: Option) -> anyhow::Result { - run_on(TABLE, plan, sla_ms) + run_on(TABLE, plan.to_string().as_bytes(), sla_ms) } - fn run_on(table: &str, plan: Value, sla_ms: Option) -> anyhow::Result { + fn run_on(table: &str, plan: &[u8], sla_ms: Option) -> anyhow::Result { let dir = tempfile::tempdir().unwrap(); let (table_path, plan_path) = (dir.path().join("table.json"), dir.path().join("plan.json")); std::fs::write(&table_path, table).unwrap(); - std::fs::write(&plan_path, plan.to_string()).unwrap(); + std::fs::write(&plan_path, plan).unwrap(); let selector = PlanSelector { method: "asap", weights: "cpu", @@ -432,7 +423,7 @@ mod tests { r#""sum by (job) (data)", "interval_secs": 1"#, r#""avg by (job) (data)", "interval_secs": 1"#, ); - let err = run_on(&table, plans(shared_p50()), None) + let err = run_on(&table, plans(shared_p50()).to_string().as_bytes(), None) .err() .expect("refused") .to_string(); @@ -440,6 +431,41 @@ mod tests { assert!(!err.contains("leaves"), "{err}"); } + #[test] + fn reads_a_gzipped_plan() { + use std::io::Write; + let mut gz = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + gz.write_all(plans(shared_p50()).to_string().as_bytes()) + .unwrap(); + let output = run_on(TABLE, &gz.finish().unwrap(), None).unwrap(); + assert_eq!(output.streaming_aggregation_count(), 2); + } + + #[test] + fn refuses_a_query_naming_too_few_rqes() { + let table = TABLE.replace(r#""rqes": ["p90", "p50"]"#, r#""rqes": ["p90"]"#); + let err = run_on(&table, plans(shared_p50()).to_string().as_bytes(), None) + .err() + .expect("refused") + .to_string(); + assert!(err.contains("has 2 leaves but names 1 RQEs"), "{err}"); + } + + #[test] + fn refuses_a_query_on_two_deployments() { + // The ratio's legs share one item, so they must share a deployment. + let p50 = pick("p50", 1, "dd", json!({"alpha": 0.02}), 300); + let err = refusal(plans(p50), None); + assert!(err.contains("served by two deployments"), "{err}"); + } + + #[test] + fn refuses_a_pick_needing_a_key_tracker() { + let p50 = pick("p50", 0, "hydra-kll", json!({}), 300); + let err = refusal(plans(p50), None); + assert!(err.contains("does --plan match --workload-table"), "{err}"); + } + #[test] fn refuses_an_rqe_picked_twice() { let mut plan = plans(shared_p50()); diff --git a/asap-tools/experiments/experiment_run_e2e.py b/asap-tools/experiments/experiment_run_e2e.py index 9e653f8d..13273a12 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -322,7 +322,14 @@ def main(cfg: DictConfig): f"{args.controller_prebuilt_output_dir}" ) config.copy_prebuilt_controller_output( - args.controller_prebuilt_output_dir, CONTROLLER_LOCAL_OUTPUT_DIR + args.controller_prebuilt_output_dir, + CONTROLLER_LOCAL_OUTPUT_DIR, + os.path.join( + local_experiment_root_dir, + "controller_client_configs", + f"{experiment_mode}_controller_input.yaml", + ), + data_ingestion_interval_ms, ) sync.rsync_streaming_configs( provider, diff --git a/asap-tools/experiments/experiment_utils/config.py b/asap-tools/experiments/experiment_utils/config.py index b988cf98..5f8ce768 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/asap-tools/experiments/experiment_utils/config.py @@ -5,6 +5,7 @@ import math import os +import re import shutil import copy import yaml @@ -547,12 +548,48 @@ def check_prebuilt_controller_output(src): raise ValueError(f"controller.prebuilt_output_dir has no {name}: {path}") -def copy_prebuilt_controller_output(src, dest): - """Replaces `dest` with the planner configs in `src`.""" +def _is_planned(query, planned): + """Whether `query` is planned whole, or is arithmetic over planned leaves + (the planner splits binary queries into leaves).""" + # ponytail: textual check, not a PromQL parse; a leaf that is a substring + # of another is removed longest first. + for leaf in sorted(planned, key=len, reverse=True): + query = query.replace(leaf, "") + return re.fullmatch(r"[\s()+\-*/0-9.]*", query) is not None + + +def copy_prebuilt_controller_output( + src, dest, controller_input_config, scrape_interval_ms +): + """Replaces `dest` with the planner configs in `src`, after checking they + plan every query of `controller_input_config` at this scrape interval.""" check_prebuilt_controller_output(src) - if os.path.realpath(src) == os.path.realpath(dest): + src_real, dest_real = os.path.realpath(src), os.path.realpath(dest) + if os.path.commonpath([src_real, dest_real]) in (src_real, dest_real): + raise ValueError( + f"controller.prebuilt_output_dir {src} and the experiment's output " + f"dir {dest} must not contain one another" + ) + with open(os.path.join(src, "inference_config.yaml")) as f: + planned = {q["query"] for q in yaml.safe_load(f)["queries"]} + with open(controller_input_config) as f: + groups = yaml.safe_load(f)["query_groups"] + unplanned = [q for g in groups for q in g["queries"] if not _is_planned(q, planned)] + if unplanned: + raise ValueError( + f"controller.prebuilt_output_dir doesn't plan these queries: {unplanned}" + ) + with open(os.path.join(src, "streaming_config.yaml")) as f: + aggregations = yaml.safe_load(f)["aggregations"] + bad = [ + a["aggregationId"] + for a in aggregations + if a["slideIntervalMs"] % scrape_interval_ms + ] + if bad: raise ValueError( - f"controller.prebuilt_output_dir is the experiment's own output dir: {dest}" + f"controller.prebuilt_output_dir aggregations {bad} slide off the " + f"{scrape_interval_ms} ms scrape interval: planned for another one?" ) shutil.rmtree(dest, ignore_errors=True) os.makedirs(dest) diff --git a/asap-tools/experiments/tests/test_prebuilt_controller_output.py b/asap-tools/experiments/tests/test_prebuilt_controller_output.py index 4dd227fc..b140933c 100644 --- a/asap-tools/experiments/tests/test_prebuilt_controller_output.py +++ b/asap-tools/experiments/tests/test_prebuilt_controller_output.py @@ -4,27 +4,48 @@ import tempfile import unittest +import yaml + from experiment_utils.config import copy_prebuilt_controller_output +QUERY = "sum by (job) (data)" + class CopyPrebuiltControllerOutputTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.src = os.path.join(self.tmp.name, "src") self.dest = os.path.join(self.tmp.name, "dest") + self.client = os.path.join(self.tmp.name, "client.yaml") os.makedirs(self.src) - for name in ("streaming_config.yaml", "inference_config.yaml"): - with open(os.path.join(self.src, name), "w") as f: - f.write(name) + self.write( + os.path.join(self.src, "inference_config.yaml"), + {"queries": [{"query": QUERY}]}, + ) + self.write( + os.path.join(self.src, "streaming_config.yaml"), + {"aggregations": [{"aggregationId": 1, "slideIntervalMs": 1000}]}, + ) + self.write(self.client, {"query_groups": [{"queries": [QUERY]}]}) def tearDown(self): self.tmp.cleanup() + @staticmethod + def write(path, content): + with open(path, "w") as f: + yaml.safe_dump(content, f) + + def copy(self, src=None, dest=None, scrape_interval_ms=1000): + copy_prebuilt_controller_output( + src or self.src, dest or self.dest, self.client, scrape_interval_ms + ) + def test_replaces_stale_output(self): os.makedirs(self.dest) with open(os.path.join(self.dest, "controller.log"), "w") as f: f.write("stale") - copy_prebuilt_controller_output(self.src, self.dest) + self.copy() self.assertEqual( sorted(os.listdir(self.dest)), ["inference_config.yaml", "streaming_config.yaml"], @@ -33,13 +54,34 @@ def test_replaces_stale_output(self): def test_refuses_missing_config(self): os.remove(os.path.join(self.src, "inference_config.yaml")) with self.assertRaisesRegex(ValueError, "inference_config.yaml"): - copy_prebuilt_controller_output(self.src, self.dest) + self.copy() - def test_refuses_own_output_dir(self): - with self.assertRaisesRegex(ValueError, "own output dir"): - copy_prebuilt_controller_output(self.src, self.src) + def test_refuses_nested_dirs(self): + for src, dest in [ + (self.src, self.src), + (self.src, os.path.dirname(self.src)), + (self.src, os.path.join(self.src, "out")), + ]: + with self.assertRaisesRegex(ValueError, "contain one another"): + self.copy(src, dest) self.assertTrue(os.path.isfile(os.path.join(self.src, "streaming_config.yaml"))) + def test_accepts_arithmetic_over_planned_leaves(self): + self.write( + self.client, + {"query_groups": [{"queries": [f"({QUERY}) / {QUERY} * 100"]}]}, + ) + self.copy() + + def test_refuses_an_unplanned_query(self): + self.write(self.client, {"query_groups": [{"queries": [QUERY, "other"]}]}) + with self.assertRaisesRegex(ValueError, "other"): + self.copy() + + def test_refuses_another_scrape_interval(self): + with self.assertRaisesRegex(ValueError, "15000 ms scrape interval"): + self.copy(scrape_interval_ms=15000) + if __name__ == "__main__": unittest.main()