From 45734b3bd6decb637f1d64afd261d88000555219 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 7 Oct 2026 15:00:08 -0400 Subject: [PATCH 1/5] refactor(planner): move MILP load-and-solve into the library plan_milp loads the facts and costs, builds the objective and solves, so asap-planner can share it with asap-optimizer-cli. PlannerOutput gains write_to_dir. Refs #790 Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/bin/optimizer_cli.rs | 72 +++++---------------- asap-planner-rs/src/optimizer/milp.rs | 80 ++++++++++++++++++++++++ asap-planner-rs/src/optimizer/mod.rs | 5 +- asap-planner-rs/src/planner_output.rs | 12 ++++ 4 files changed, 111 insertions(+), 58 deletions(-) diff --git a/asap-planner-rs/src/bin/optimizer_cli.rs b/asap-planner-rs/src/bin/optimizer_cli.rs index d9db9193..6bf54dbb 100644 --- a/asap-planner-rs/src/bin/optimizer_cli.rs +++ b/asap-planner-rs/src/bin/optimizer_cli.rs @@ -4,12 +4,10 @@ use std::path::PathBuf; use asap_planner::optimizer::{ - build_milp_workload, load_flat_atomic_cost_table, load_workload_facts, plan_to_planner_output, - reject_avg_queries, solve_milp, MilpError, + parse_weight, plan_milp, plan_to_planner_output, reject_avg_queries, MilpInputs, MilpPlan, }; use asap_planner::ControllerConfig; use clap::Parser; -use rqe_optimizer::milp::Objective; #[derive(Parser, Debug)] #[command( @@ -74,44 +72,25 @@ fn main() -> anyhow::Result<()> { run_milp(&args, &config) } -/// Objective weights must be finite and non-negative: a negative weight -/// rewards cost, and NaN poisons every coefficient. -fn parse_weight(s: &str) -> Result { - match s.parse::() { - Ok(w) if w.is_finite() && w >= 0.0 => Ok(w), - Ok(w) => Err(format!("must be finite and >= 0, got {w}")), - Err(e) => Err(e.to_string()), - } -} - fn run_milp(args: &Args, config: &ControllerConfig) -> anyhow::Result<()> { - config.warn_default_slas(); - let Some(hints) = config.metrics.as_deref() else { - return Err(MilpError::MissingMetricHints.into()); - }; - let facts = load_workload_facts(&args.workload_facts, hints, args.data_ingestion_interval_ms)?; - let costs = load_flat_atomic_cost_table(&args.atomic_costs)?; - let Objective::AUCCost { w_cpu, w_mem } = Objective::default(); - let (w_cpu, w_mem) = (args.w_cpu.unwrap_or(w_cpu), args.w_mem.unwrap_or(w_mem)); - // All-zero weights make every plan cost 0, so the solver's pick is arbitrary. - anyhow::ensure!( - w_cpu > 0.0 || w_mem > 0.0, - "--w-cpu and --w-mem are both 0; at least one must be positive" - ); - let objective = Objective::AUCCost { w_cpu, w_mem }; - tracing::debug!(?objective, cost_rows = costs.len(), "milp: inputs loaded"); - // Fail before solving when the plan would be written but can't be. if args.output_dir.is_some() { reject_avg_queries(config)?; } - let workload = build_milp_workload(config, &facts, args.data_ingestion_interval_ms)?; - let solution = solve_milp( - &workload, - &facts, - &costs, + let MilpPlan { + workload, + solution, objective, - args.allow_undeployable_families, + } = plan_milp( + config, + &MilpInputs { + workload_facts: &args.workload_facts, + atomic_costs: &args.atomic_costs, + scrape_interval_ms: args.data_ingestion_interval_ms, + w_cpu: args.w_cpu, + w_mem: args.w_mem, + allow_undeployable_families: args.allow_undeployable_families, + }, )?; println!("=== Deployments: {} ===", solution.deployments.len()); @@ -163,29 +142,8 @@ fn run_milp(args: &Args, config: &ControllerConfig) -> anyhow::Result<()> { } if let Some(dir) = &args.output_dir { - let output = plan_to_planner_output(config, &workload, &solution)?; - // Serialize both before writing either, so a failure can't leave a - // new streaming config next to a stale inference config. - let streaming = output.to_streaming_yaml_string()?; - let inference = output.to_inference_yaml_string()?; - std::fs::create_dir_all(dir)?; - std::fs::write(dir.join("streaming_config.yaml"), streaming)?; - std::fs::write(dir.join("inference_config.yaml"), inference)?; + plan_to_planner_output(config, &workload, &solution)?.write_to_dir(dir)?; println!("\nwrote configs to {}", dir.display()); } Ok(()) } - -#[cfg(test)] -mod tests { - use super::parse_weight; - - #[test] - fn weights_must_be_finite_and_non_negative() { - assert_eq!(parse_weight("0.5"), Ok(0.5)); - assert_eq!(parse_weight("0"), Ok(0.0)); - for bad in ["-1", "NaN", "inf", "x"] { - assert!(parse_weight(bad).is_err(), "{bad}"); - } - } -} diff --git a/asap-planner-rs/src/optimizer/milp.rs b/asap-planner-rs/src/optimizer/milp.rs index 9f2d4a9f..79906a9f 100644 --- a/asap-planner-rs/src/optimizer/milp.rs +++ b/asap-planner-rs/src/optimizer/milp.rs @@ -12,9 +12,13 @@ use thiserror::Error; use crate::config::input::ControllerConfig; +use std::path::Path; + use super::aqe_extractor::{extract_aqes, RQE}; +use super::atomic_costs::load_flat_atomic_cost_table; use super::error::OptimizerError; use super::solution::OptimizerItem; +use super::workload_facts::load_workload_facts; /// Slack on accuracy tolerances so `1 - sla` rounding (`1 - 0.9 = /// 0.0999...98`) doesn't reject a row measured exactly at the boundary. @@ -100,6 +104,73 @@ pub fn solve_milp( Ok(solution) } +/// What a MILP planning run reads besides the workload config. +pub struct MilpInputs<'a> { + pub workload_facts: &'a Path, + pub atomic_costs: &'a Path, + pub scrape_interval_ms: u64, + /// Objective weights; `None` takes rqe-optimizer's default. + pub w_cpu: Option, + pub w_mem: Option, + pub allow_undeployable_families: bool, +} + +pub struct MilpPlan { + pub workload: MilpWorkload, + pub solution: MilpSolution, + pub objective: Objective, +} + +/// Loads the facts and costs and solves for the cheapest plan. +pub fn plan_milp(config: &ControllerConfig, inputs: &MilpInputs) -> anyhow::Result { + config.warn_default_slas(); + let Some(hints) = config.metrics.as_deref() else { + return Err(MilpError::MissingMetricHints.into()); + }; + let facts = load_workload_facts(inputs.workload_facts, hints, inputs.scrape_interval_ms)?; + let costs = load_flat_atomic_cost_table(inputs.atomic_costs)?; + let objective = objective(inputs.w_cpu, inputs.w_mem)?; + tracing::debug!(?objective, cost_rows = costs.len(), "milp: inputs loaded"); + + let workload = build_milp_workload(config, &facts, inputs.scrape_interval_ms)?; + let solution = solve_milp( + &workload, + &facts, + &costs, + objective, + inputs.allow_undeployable_families, + )?; + Ok(MilpPlan { + workload, + solution, + objective, + }) +} + +fn objective(w_cpu: Option, w_mem: Option) -> anyhow::Result { + let Objective::AUCCost { + w_cpu: default_cpu, + w_mem: default_mem, + } = Objective::default(); + let (w_cpu, w_mem) = (w_cpu.unwrap_or(default_cpu), w_mem.unwrap_or(default_mem)); + // All-zero weights make every plan cost 0, so the solver's pick is arbitrary. + anyhow::ensure!( + w_cpu > 0.0 || w_mem > 0.0, + "--w-cpu and --w-mem are both 0; at least one must be positive" + ); + Ok(Objective::AUCCost { w_cpu, w_mem }) +} + +/// Parses an objective weight flag. Weights must be finite and non-negative: +/// a negative weight rewards cost, and NaN poisons every coefficient. +pub fn parse_weight(s: &str) -> Result { + match s.parse::() { + Ok(w) if w.is_finite() && w >= 0.0 => Ok(w), + Ok(w) => Err(format!("must be finite and >= 0, got {w}")), + Err(e) => Err(e.to_string()), + } +} + /// The MILP's view of a workload. #[derive(Debug)] pub struct MilpWorkload { @@ -590,6 +661,15 @@ metrics: assert_eq!(solution.deployments.len(), 2); } + #[test] + fn weights_must_be_finite_and_non_negative() { + assert_eq!(parse_weight("0.5"), Ok(0.5)); + assert_eq!(parse_weight("0"), Ok(0.0)); + for bad in ["-1", "NaN", "inf", "x"] { + assert!(parse_weight(bad).is_err(), "{bad}"); + } + } + #[test] fn workload_requires_metric_hints() { let config: ControllerConfig = serde_yaml::from_str(&format!( diff --git a/asap-planner-rs/src/optimizer/mod.rs b/asap-planner-rs/src/optimizer/mod.rs index f7e1cea5..ec86dc1c 100644 --- a/asap-planner-rs/src/optimizer/mod.rs +++ b/asap-planner-rs/src/optimizer/mod.rs @@ -9,7 +9,10 @@ pub mod workload_facts; 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, solve_milp, MilpError, MilpWorkload}; +pub use milp::{ + build_milp_workload, parse_weight, plan_milp, solve_milp, MilpError, MilpInputs, MilpPlan, + MilpWorkload, +}; pub use milp_output::{plan_to_planner_output, reject_avg_queries, MilpOutputError}; pub use solution::OptimizerItem; pub use workload_facts::{load_workload_facts, parse_workload_facts, WorkloadFactsError}; diff --git a/asap-planner-rs/src/planner_output.rs b/asap-planner-rs/src/planner_output.rs index 036a8c79..d34202f9 100644 --- a/asap-planner-rs/src/planner_output.rs +++ b/asap-planner-rs/src/planner_output.rs @@ -171,6 +171,18 @@ impl PlannerOutput { Ok(serde_yaml::to_string(&self.inference_yaml)?) } + /// Writes `streaming_config.yaml` and `inference_config.yaml`. Both are + /// serialized first, so a failure can't leave a new streaming config next + /// to a stale inference config. + pub fn write_to_dir(&self, dir: &std::path::Path) -> Result<(), anyhow::Error> { + let streaming = self.to_streaming_yaml_string()?; + let inference = self.to_inference_yaml_string()?; + std::fs::create_dir_all(dir)?; + std::fs::write(dir.join("streaming_config.yaml"), streaming)?; + std::fs::write(dir.join("inference_config.yaml"), inference)?; + Ok(()) + } + pub fn to_streaming_config( &self, query_language: QueryLanguage, From 2639950ec0cbc7124bea2328e27b8038cb70f7f1 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 7 Oct 2026 15:02:08 -0400 Subject: [PATCH 2/5] feat(planner): add asap-planner --planner legacy|milp --planner milp plans with the rqe-optimizer MILP and writes its configs; legacy stays the default. milp takes --workload-facts and --atomic-costs (required) and --w-cpu/--w-mem, and supports only PromQL with --input_config. It rejects --query-log, --prometheus-url, --enable-punting, --range-duration-ms, --step-ms, and per-group step_ms/range_duration_ms, whose retention MILP configs don't model yet (#800). Refs #790 Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/main.rs | 96 ++++++++++++++++++++++++++- asap-planner-rs/src/optimizer/milp.rs | 31 +++++++++ 2 files changed, 125 insertions(+), 2 deletions(-) diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index cb345361..557d3bf8 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -1,6 +1,9 @@ +use asap_planner::optimizer::{ + parse_weight, plan_milp, plan_to_planner_output, reject_avg_queries, MilpInputs, +}; use asap_planner::{ - Controller, ElasticController, ElasticRuntimeOptions, RuntimeOptions, SQLController, - SQLRuntimeOptions, StreamingEngine, + Controller, ControllerConfig, ElasticController, ElasticRuntimeOptions, RuntimeOptions, + SQLController, SQLRuntimeOptions, StreamingEngine, }; use asap_types::enums::QueryLanguage; use clap::Parser; @@ -53,6 +56,29 @@ struct Args { #[arg(long = "clickhouse-database", required = false)] clickhouse_database: Option, + /// `milp` plans with sketch-bench's rqe-optimizer: PromQL with + /// --input_config only, labels from its `metrics:` hints. + #[arg(long, value_enum, default_value = "legacy")] + planner: PlannerArg, + + /// MILP only. YAML workload facts: per metric, `cardinality` per label + /// set, including the set of all its labels (the series count). + #[arg(long = "workload-facts", required_if_eq("planner", "milp"))] + workload_facts: Option, + + /// MILP only. The flat cost table `export_rqe_optimizer_costs.sh` writes + /// (`rqe_atomic_costs.json`). + #[arg(long = "atomic-costs", required_if_eq("planner", "milp"))] + atomic_costs: Option, + + /// MILP only. Objective weight on CPU-sec/sec. Default: rqe-optimizer's. + #[arg(long = "w-cpu", value_parser = parse_weight)] + w_cpu: Option, + + /// MILP only. Objective weight on memory GiB. Default: rqe-optimizer's. + #[arg(long = "w-mem", value_parser = parse_weight)] + w_mem: Option, + #[arg(short, long, action = clap::ArgAction::Count)] verbose: u8, } @@ -62,6 +88,12 @@ enum EngineArg { Precompute, } +#[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq, Eq)] +enum PlannerArg { + Legacy, + Milp, +} + fn main() -> anyhow::Result<()> { let args = Args::parse(); @@ -77,6 +109,19 @@ fn main() -> anyhow::Result<()> { EngineArg::Precompute => StreamingEngine::Precompute, }; + if args.planner == PlannerArg::Milp { + run_milp(&args)?; + println!("Generated configs in {}", args.output_dir.display()); + return Ok(()); + } + anyhow::ensure!( + args.workload_facts.is_none() + && args.atomic_costs.is_none() + && args.w_cpu.is_none() + && args.w_mem.is_none(), + "--workload-facts, --atomic-costs, --w-cpu and --w-mem require --planner milp" + ); + match args.query_language { QueryLanguage::promql => { let scrape_interval_ms = args.data_ingestion_interval_ms.ok_or_else(|| { @@ -158,3 +203,50 @@ fn main() -> anyhow::Result<()> { println!("Generated configs in {}", args.output_dir.display()); Ok(()) } + +fn run_milp(args: &Args) -> anyhow::Result<()> { + anyhow::ensure!( + matches!(args.query_language, QueryLanguage::promql), + "--planner milp supports only --query-language promql" + ); + anyhow::ensure!( + args.query_log.is_none(), + "--planner milp needs --input_config: 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" + ); + 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" + ); + 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!("--data-ingestion-interval-ms is required for PromQL mode") + })?; + let config: ControllerConfig = serde_yaml::from_str(&std::fs::read_to_string(config_path)?)?; + // Fail before solving: an avg query can't be written as a config. + reject_avg_queries(&config)?; + let plan = plan_milp( + &config, + &MilpInputs { + workload_facts: args + .workload_facts + .as_deref() + .expect("clap requires --workload-facts with --planner milp"), + atomic_costs: args + .atomic_costs + .as_deref() + .expect("clap requires --atomic-costs with --planner milp"), + scrape_interval_ms, + w_cpu: args.w_cpu, + w_mem: args.w_mem, + allow_undeployable_families: false, + }, + )?; + plan_to_planner_output(&config, &plan.workload, &plan.solution)?.write_to_dir(&args.output_dir) +} diff --git a/asap-planner-rs/src/optimizer/milp.rs b/asap-planner-rs/src/optimizer/milp.rs index 79906a9f..c2b02e70 100644 --- a/asap-planner-rs/src/optimizer/milp.rs +++ b/asap-planner-rs/src/optimizer/milp.rs @@ -26,6 +26,11 @@ const SLA_EPSILON: f64 = 1e-9; #[derive(Debug, Error)] pub enum MilpError { + #[error( + "query groups set step_ms or range_duration_ms, which the MILP planner doesn't \ + support yet (#800): {0:?}" + )] + RangeQueryOverrides(Vec), #[error( "workload config has no `metrics:` hints; they are required to resolve grouping labels" )] @@ -188,6 +193,16 @@ pub fn build_milp_workload( facts: &WorkloadFacts, scrape_interval_ms: u64, ) -> Result { + // They only size retention, which MILP configs don't set yet. + let range_queries: Vec = config + .query_groups + .iter() + .filter(|qg| qg.step_ms.is_some() || qg.range_duration_ms.is_some()) + .flat_map(|qg| qg.queries.iter().cloned()) + .collect(); + if !range_queries.is_empty() { + return Err(MilpError::RangeQueryOverrides(range_queries)); + } let mut items = extract_hinted_items(config, scrape_interval_ms)?; // Stable Raqe order and ids across runs. items.sort_by(|a, b| { @@ -582,6 +597,22 @@ metrics: assert!(matches!(err, MilpError::InvalidInputs(_))); } + /// Their retention isn't modeled, so planning them would ignore them silently. + #[test] + fn range_query_overrides_are_rejected() { + for field in ["step_ms: 60000", "range_duration_ms: 3600000"] { + let config = config(&format!( + "{} {field}\n", + group("sum(http_requests_total)", 0.99) + )); + let err = build_milp_workload(&config, &facts(&config), SCRAPE_MS).unwrap_err(); + assert!( + matches!(&err, MilpError::RangeQueryOverrides(q) if q == &["sum(http_requests_total)"]), + "{field}: {err}" + ); + } + } + #[test] fn solve_shares_one_deployment_across_repeated_queries() { let query = "sum by (job) (http_requests_total)"; From ac73396d951351d9f09c61985109fc6afd1811bb Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 7 Oct 2026 15:07:56 -0400 Subject: [PATCH 3/5] docs(planner): mention value_range in asap-planner --workload-facts help Refs #790 Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/main.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index 557d3bf8..aa2db711 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -61,8 +61,8 @@ struct Args { #[arg(long, value_enum, default_value = "legacy")] planner: PlannerArg, - /// MILP only. YAML workload facts: per metric, `cardinality` per label - /// set, including the set of all its labels (the series count). + /// MILP only. YAML workload facts: per metric, positive `value_range` and + /// `cardinality` per label set, including all labels (the series count). #[arg(long = "workload-facts", required_if_eq("planner", "milp"))] workload_facts: Option, From ce4483f799e529d2c615cfc7d48019e05de8d037 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 7 Oct 2026 15:55:56 -0400 Subject: [PATCH 4/5] refactor(planner): write controller outputs with PlannerOutput::write_to_dir The PromQL, SQL and Elastic controllers each repeated the two-file write; the output file names now live in one place. Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/elastic_dsl/controller.rs | 6 +----- asap-planner-rs/src/main.rs | 4 +++- asap-planner-rs/src/planner_output.rs | 15 ++++----------- asap-planner-rs/src/promql/controller.rs | 6 +----- asap-planner-rs/src/sql/controller.rs | 6 +----- 5 files changed, 10 insertions(+), 27 deletions(-) diff --git a/asap-planner-rs/src/elastic_dsl/controller.rs b/asap-planner-rs/src/elastic_dsl/controller.rs index 22df6872..313fdb63 100644 --- a/asap-planner-rs/src/elastic_dsl/controller.rs +++ b/asap-planner-rs/src/elastic_dsl/controller.rs @@ -35,11 +35,7 @@ impl ElasticController { pub fn generate_to_dir(&self, dir: &Path) -> Result { let output = self.generate()?; - std::fs::create_dir_all(dir)?; - let streaming_str = serde_yaml::to_string(output.streaming_yaml())?; - let inference_str = serde_yaml::to_string(output.inference_yaml())?; - std::fs::write(dir.join("streaming_config.yaml"), streaming_str)?; - std::fs::write(dir.join("inference_config.yaml"), inference_str)?; + output.write_to_dir(dir)?; Ok(output) } } diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index aa2db711..fef09a8d 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -248,5 +248,7 @@ fn run_milp(args: &Args) -> anyhow::Result<()> { allow_undeployable_families: false, }, )?; - plan_to_planner_output(&config, &plan.workload, &plan.solution)?.write_to_dir(&args.output_dir) + plan_to_planner_output(&config, &plan.workload, &plan.solution)? + .write_to_dir(&args.output_dir)?; + Ok(()) } diff --git a/asap-planner-rs/src/planner_output.rs b/asap-planner-rs/src/planner_output.rs index d34202f9..3d72be25 100644 --- a/asap-planner-rs/src/planner_output.rs +++ b/asap-planner-rs/src/planner_output.rs @@ -4,6 +4,7 @@ use asap_types::enums::QueryLanguage; use asap_types::inference_config::InferenceConfig; use asap_types::streaming_config::StreamingConfig; +use crate::error::ControllerError; use crate::generator::{ GeneratorOutput, PuntedQuery, KEY_AGGREGATIONS, KEY_AGG_SUB_TYPE, KEY_AGG_TYPE, KEY_LABELS, KEY_NUM_AGG_TO_RETAIN, KEY_PARAMETERS, KEY_QUERIES, KEY_QUERY, KEY_READ_COUNT_THRESHOLD, @@ -30,14 +31,6 @@ impl PlannerOutput { } } - pub(crate) fn streaming_yaml(&self) -> &YamlValue { - &self.streaming_yaml - } - - pub(crate) fn inference_yaml(&self) -> &YamlValue { - &self.inference_yaml - } - pub fn streaming_aggregation_count(&self) -> usize { self.aggregation_count } @@ -174,9 +167,9 @@ impl PlannerOutput { /// Writes `streaming_config.yaml` and `inference_config.yaml`. Both are /// serialized first, so a failure can't leave a new streaming config next /// to a stale inference config. - pub fn write_to_dir(&self, dir: &std::path::Path) -> Result<(), anyhow::Error> { - let streaming = self.to_streaming_yaml_string()?; - let inference = self.to_inference_yaml_string()?; + pub fn write_to_dir(&self, dir: &std::path::Path) -> Result<(), ControllerError> { + let streaming = serde_yaml::to_string(&self.streaming_yaml)?; + let inference = serde_yaml::to_string(&self.inference_yaml)?; std::fs::create_dir_all(dir)?; std::fs::write(dir.join("streaming_config.yaml"), streaming)?; std::fs::write(dir.join("inference_config.yaml"), inference)?; diff --git a/asap-planner-rs/src/promql/controller.rs b/asap-planner-rs/src/promql/controller.rs index a24d6edc..354a7e84 100644 --- a/asap-planner-rs/src/promql/controller.rs +++ b/asap-planner-rs/src/promql/controller.rs @@ -162,11 +162,7 @@ impl Controller { pub fn generate_to_dir(&self, dir: &Path) -> Result { let output = self.generate()?; - std::fs::create_dir_all(dir)?; - let streaming_str = serde_yaml::to_string(output.streaming_yaml())?; - let inference_str = serde_yaml::to_string(output.inference_yaml())?; - std::fs::write(dir.join("streaming_config.yaml"), streaming_str)?; - std::fs::write(dir.join("inference_config.yaml"), inference_str)?; + output.write_to_dir(dir)?; Ok(output) } } diff --git a/asap-planner-rs/src/sql/controller.rs b/asap-planner-rs/src/sql/controller.rs index d9da13b9..5587315e 100644 --- a/asap-planner-rs/src/sql/controller.rs +++ b/asap-planner-rs/src/sql/controller.rs @@ -90,11 +90,7 @@ impl SQLController { pub fn generate_to_dir(&self, dir: &Path) -> Result { let output = self.generate()?; - std::fs::create_dir_all(dir)?; - let streaming_str = serde_yaml::to_string(output.streaming_yaml())?; - let inference_str = serde_yaml::to_string(output.inference_yaml())?; - std::fs::write(dir.join("streaming_config.yaml"), streaming_str)?; - std::fs::write(dir.join("inference_config.yaml"), inference_str)?; + output.write_to_dir(dir)?; Ok(output) } } From 0c4bcf5f848bdf6b626c659fb650a0d4e4356459 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 7 Oct 2026 15:57:42 -0400 Subject: [PATCH 5/5] fix(planner): reject inputs the MILP planner would silently ignore - windowing, sketch_parameters and aggregate_cleanup are rejected by build_milp_workload, since the MILP picks windows and parameters itself. - Per-group step_ms/range_duration_ms move to reject_unwritable_queries (formerly reject_avg_queries), so print-only asap-optimizer-cli runs still plan them. - asap-planner --planner milp rejects a zero scrape interval and the ClickHouse flags. Refs #790 Co-Authored-By: Claude Opus 5.5 --- asap-planner-rs/src/bin/optimizer_cli.rs | 5 +- asap-planner-rs/src/main.rs | 21 +++++-- asap-planner-rs/src/optimizer/milp.rs | 60 +++++++++++--------- asap-planner-rs/src/optimizer/milp_output.rs | 45 +++++++++++++-- asap-planner-rs/src/optimizer/mod.rs | 2 +- 5 files changed, 91 insertions(+), 42 deletions(-) diff --git a/asap-planner-rs/src/bin/optimizer_cli.rs b/asap-planner-rs/src/bin/optimizer_cli.rs index 6bf54dbb..7353c5a8 100644 --- a/asap-planner-rs/src/bin/optimizer_cli.rs +++ b/asap-planner-rs/src/bin/optimizer_cli.rs @@ -4,7 +4,8 @@ use std::path::PathBuf; use asap_planner::optimizer::{ - parse_weight, plan_milp, plan_to_planner_output, reject_avg_queries, MilpInputs, MilpPlan, + parse_weight, plan_milp, plan_to_planner_output, reject_unwritable_queries, MilpInputs, + MilpPlan, }; use asap_planner::ControllerConfig; use clap::Parser; @@ -75,7 +76,7 @@ fn main() -> anyhow::Result<()> { fn run_milp(args: &Args, config: &ControllerConfig) -> anyhow::Result<()> { // Fail before solving when the plan would be written but can't be. if args.output_dir.is_some() { - reject_avg_queries(config)?; + reject_unwritable_queries(config)?; } let MilpPlan { workload, diff --git a/asap-planner-rs/src/main.rs b/asap-planner-rs/src/main.rs index fef09a8d..6717b0de 100644 --- a/asap-planner-rs/src/main.rs +++ b/asap-planner-rs/src/main.rs @@ -1,5 +1,5 @@ use asap_planner::optimizer::{ - parse_weight, plan_milp, plan_to_planner_output, reject_avg_queries, MilpInputs, + parse_weight, plan_milp, plan_to_planner_output, reject_unwritable_queries, MilpInputs, }; use asap_planner::{ Controller, ControllerConfig, ElasticController, ElasticRuntimeOptions, RuntimeOptions, @@ -221,16 +221,25 @@ fn run_milp(args: &Args) -> anyhow::Result<()> { !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" ); + anyhow::ensure!( + args.clickhouse_url.is_none() && args.clickhouse_database.is_none(), + "--clickhouse-url and --clickhouse-database don't apply to --planner 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!("--data-ingestion-interval-ms is required for PromQL mode") - })?; + 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: an avg query can't be written as a config. - reject_avg_queries(&config)?; + // Fail before solving: the plan is always written. + reject_unwritable_queries(&config)?; let plan = plan_milp( &config, &MilpInputs { diff --git a/asap-planner-rs/src/optimizer/milp.rs b/asap-planner-rs/src/optimizer/milp.rs index c2b02e70..d4849e2a 100644 --- a/asap-planner-rs/src/optimizer/milp.rs +++ b/asap-planner-rs/src/optimizer/milp.rs @@ -26,11 +26,8 @@ const SLA_EPSILON: f64 = 1e-9; #[derive(Debug, Error)] pub enum MilpError { - #[error( - "query groups set step_ms or range_duration_ms, which the MILP planner doesn't \ - support yet (#800): {0:?}" - )] - RangeQueryOverrides(Vec), + #[error("the MILP planner doesn't support these workload config fields: {0:?}")] + UnsupportedFields(Vec<&'static str>), #[error( "workload config has no `metrics:` hints; they are required to resolve grouping labels" )] @@ -193,15 +190,17 @@ pub fn build_milp_workload( facts: &WorkloadFacts, scrape_interval_ms: u64, ) -> Result { - // They only size retention, which MILP configs don't set yet. - let range_queries: Vec = config - .query_groups - .iter() - .filter(|qg| qg.step_ms.is_some() || qg.range_duration_ms.is_some()) - .flat_map(|qg| qg.queries.iter().cloned()) - .collect(); - if !range_queries.is_empty() { - return Err(MilpError::RangeQueryOverrides(range_queries)); + // The MILP chooses windows and sketch parameters itself. + let unsupported: Vec<&'static str> = [ + ("windowing", config.windowing.is_some()), + ("sketch_parameters", config.sketch_parameters.is_some()), + ("aggregate_cleanup", config.aggregate_cleanup.is_some()), + ] + .into_iter() + .filter_map(|(field, set)| set.then_some(field)) + .collect(); + if !unsupported.is_empty() { + return Err(MilpError::UnsupportedFields(unsupported)); } let mut items = extract_hinted_items(config, scrape_interval_ms)?; // Stable Raqe order and ids across runs. @@ -597,20 +596,27 @@ metrics: assert!(matches!(err, MilpError::InvalidInputs(_))); } - /// Their retention isn't modeled, so planning them would ignore them silently. + /// The MILP picks windows and sketch parameters itself, so these would be + /// dropped silently. #[test] - fn range_query_overrides_are_rejected() { - for field in ["step_ms: 60000", "range_duration_ms: 3600000"] { - let config = config(&format!( - "{} {field}\n", - group("sum(http_requests_total)", 0.99) - )); - let err = build_milp_workload(&config, &facts(&config), SCRAPE_MS).unwrap_err(); - assert!( - matches!(&err, MilpError::RangeQueryOverrides(q) if q == &["sum(http_requests_total)"]), - "{field}: {err}" - ); - } + fn unsupported_config_fields_are_rejected() { + let mut config = config(&group("sum(http_requests_total)", 0.99)); + config.windowing = + Some(serde_yaml::from_str("{type: tumbling, window_size_ms: 60000}").unwrap()); + config.sketch_parameters = Some(Default::default()); + let err = build_milp_workload(&config, &facts(&config), SCRAPE_MS).unwrap_err(); + assert!( + matches!(&err, MilpError::UnsupportedFields(f) if f == &["windowing", "sketch_parameters"]), + "{err}" + ); + config.windowing = None; + config.sketch_parameters = None; + config.aggregate_cleanup = Some(serde_yaml::from_str("{}").unwrap()); + let err = build_milp_workload(&config, &facts(&config), SCRAPE_MS).unwrap_err(); + assert!( + matches!(&err, MilpError::UnsupportedFields(f) if f == &["aggregate_cleanup"]), + "{err}" + ); } #[test] diff --git a/asap-planner-rs/src/optimizer/milp_output.rs b/asap-planner-rs/src/optimizer/milp_output.rs index fbc517e5..a98f9136 100644 --- a/asap-planner-rs/src/optimizer/milp_output.rs +++ b/asap-planner-rs/src/optimizer/milp_output.rs @@ -29,6 +29,11 @@ use super::solution::OptimizerItem; pub enum MilpOutputError { #[error("query {0:?} uses avg, which the engine can't answer from sum and count yet")] AvgQuery(String), + #[error( + "query groups set step_ms or range_duration_ms, whose retention MILP configs \ + don't set yet (#800): {0:?}" + )] + RangeQueryOverrides(Vec), #[error("sketch-bench variant {variant} ({capability:?}) has no ASAPQuery aggregation type")] UndeployableVariant { variant: String, @@ -49,17 +54,27 @@ pub enum MilpOutputError { Generator(#[from] ControllerError), } -/// Errors on the first avg query: its plan can be costed but not deployed. -pub fn reject_avg_queries(config: &ControllerConfig) -> Result<(), MilpOutputError> { - match config +/// Errors on queries whose plan can be costed but not written as configs: +/// avg queries, and range-query overrides, which only size retention. +pub fn reject_unwritable_queries(config: &ControllerConfig) -> Result<(), MilpOutputError> { + if let Some(query) = config .query_groups .iter() .flat_map(|group| &group.queries) .find(|query| contains_avg(query)) { - Some(query) => Err(MilpOutputError::AvgQuery(query.clone())), - None => Ok(()), + return Err(MilpOutputError::AvgQuery(query.clone())); } + let range_queries: Vec = config + .query_groups + .iter() + .filter(|qg| qg.step_ms.is_some() || qg.range_duration_ms.is_some()) + .flat_map(|qg| qg.queries.iter().cloned()) + .collect(); + if !range_queries.is_empty() { + return Err(MilpOutputError::RangeQueryOverrides(range_queries)); + } + Ok(()) } /// Streaming and inference YAML for `solution`. Aggregation ids follow the @@ -70,7 +85,7 @@ pub fn plan_to_planner_output( workload: &MilpWorkload, solution: &MilpSolution, ) -> Result { - reject_avg_queries(config)?; + reject_unwritable_queries(config)?; let item_of = |raqe: usize| &workload.items[workload.raqe_items[raqe]]; // Each item once per deployment, however many occurrences it has. @@ -471,6 +486,24 @@ mod tests { assert!(matches!(err, MilpOutputError::AvgQuery(q) if q == query)); } + /// Range overrides only size retention, so they block writing configs + /// but not planning. + #[test] + fn range_query_overrides_block_only_the_written_configs() { + let query = "sum(http_requests_total)"; + for field in ["step_ms: 60000", "range_duration_ms: 3600000"] { + let (config, workload, solution) = + plan(&format!("{} {field}\n", group(query, 0.99))); + let err = plan_to_planner_output(&config, &workload, &solution) + .err() + .unwrap(); + assert!( + matches!(&err, MilpOutputError::RangeQueryOverrides(q) if q == &[query]), + "{field}: {err}" + ); + } + } + #[test] fn hydra_kll_is_not_deployed() { // Its DeltaSet key tracker has no ASAPQuery pairing yet. diff --git a/asap-planner-rs/src/optimizer/mod.rs b/asap-planner-rs/src/optimizer/mod.rs index ec86dc1c..7731c14c 100644 --- a/asap-planner-rs/src/optimizer/mod.rs +++ b/asap-planner-rs/src/optimizer/mod.rs @@ -13,6 +13,6 @@ pub use milp::{ build_milp_workload, parse_weight, plan_milp, solve_milp, MilpError, MilpInputs, MilpPlan, MilpWorkload, }; -pub use milp_output::{plan_to_planner_output, reject_avg_queries, MilpOutputError}; +pub use milp_output::{plan_to_planner_output, reject_unwritable_queries, MilpOutputError}; pub use solution::OptimizerItem; pub use workload_facts::{load_workload_facts, parse_workload_facts, WorkloadFactsError};