Skip to content

Commit 8a6e92e

Browse files
feat(planner): emit streaming and inference configs from the MILP plan
plan_to_planner_output maps each planned deployment's sketch-bench variant to an ASAPQuery aggregation and writes the planner's streaming and inference YAML: exact-sum (sum/count) -> MultipleSum, exact-min/max -> MultipleMinMax, exact-increase -> MultipleIncrease, kll-percall -> DatasketchesKLL, hll -> HLL, cms-heap -> CountMinSketchWithHeap with heapsize = the largest k it serves. Any other variant (e.g. hydra-kll), avg queries, and a query split across deployments are errors. Cleanup is NoCleanup. asap-optimizer-cli --milp gains --output-dir and --allow-undeployable-families (prints the plan, writes no configs). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 parent b1ac24b commit 8a6e92e

6 files changed

Lines changed: 490 additions & 20 deletions

File tree

‎asap-planner-rs/src/bin/optimizer_cli.rs‎

Lines changed: 38 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,8 @@ use std::path::PathBuf;
88

99
use asap_planner::optimizer::{
1010
build_milp_workload, load_flat_atomic_cost_table, load_optional_selected_atomic_cost_table,
11-
load_workload_facts, run_greedy_pipeline, solve_milp, AtomicCostTable, LabelSetFacts,
12-
LabelSetFactsError,
11+
load_workload_facts, plan_to_planner_output, run_greedy_pipeline, solve_milp, AtomicCostTable,
12+
LabelSetFacts, LabelSetFactsError,
1313
};
1414
use asap_planner::ControllerConfig;
1515
use clap::Parser;
@@ -59,11 +59,24 @@ struct Args {
5959
)]
6060
atomic_cost_workload: Option<PathBuf>,
6161

62-
/// Plan with sketch-bench's rqe-optimizer MILP and print the plan; writes
63-
/// no configs yet.
62+
/// Plan with sketch-bench's rqe-optimizer MILP and print the plan.
6463
#[arg(long)]
6564
milp: bool,
6665

66+
/// MILP only. Write `streaming_config.yaml` and `inference_config.yaml`
67+
/// for the plan here.
68+
#[arg(long = "output-dir", requires = "milp")]
69+
output_dir: Option<PathBuf>,
70+
71+
/// MILP only. Also plan with families the engine can't deploy; prints the
72+
/// plan and writes no configs.
73+
#[arg(
74+
long = "allow-undeployable-families",
75+
requires = "milp",
76+
conflicts_with = "output_dir"
77+
)]
78+
allow_undeployable_families: bool,
79+
6780
/// MILP only. YAML workload facts: per metric, `cardinality` per label
6881
/// set, including the set of all its labels (the series count).
6982
#[arg(
@@ -188,7 +201,13 @@ fn run_milp(args: &Args, config: &ControllerConfig) -> anyhow::Result<()> {
188201
tracing::debug!(?objective, cost_rows = costs.len(), "milp: inputs loaded");
189202

190203
let workload = build_milp_workload(config, &facts, args.data_ingestion_interval_ms)?;
191-
let solution = solve_milp(&workload, &facts, &costs, objective)?;
204+
let solution = solve_milp(
205+
&workload,
206+
&facts,
207+
&costs,
208+
objective,
209+
args.allow_undeployable_families,
210+
)?;
192211

193212
println!("=== Deployments: {} ===", solution.deployments.len());
194213
for (d, planned) in solution.deployments.iter().enumerate() {
@@ -237,6 +256,20 @@ fn run_milp(args: &Args, config: &ControllerConfig) -> anyhow::Result<()> {
237256
c.memory_bytes / 1e6
238257
);
239258
}
259+
260+
if let Some(dir) = &args.output_dir {
261+
let output = plan_to_planner_output(config, &workload, &solution)?;
262+
std::fs::create_dir_all(dir)?;
263+
std::fs::write(
264+
dir.join("streaming_config.yaml"),
265+
output.to_streaming_yaml_string()?,
266+
)?;
267+
std::fs::write(
268+
dir.join("inference_config.yaml"),
269+
output.to_inference_yaml_string()?,
270+
)?;
271+
println!("\nwrote configs to {}", dir.display());
272+
}
240273
Ok(())
241274
}
242275

‎asap-planner-rs/src/optimizer/aqe_extractor.rs‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -157,6 +157,16 @@ fn decompose_to_leaves(query: &str) -> Vec<String> {
157157
leaves
158158
}
159159

160+
/// Whether any leaf of `query` is an `avg` that `decompose_to_leaves` rewrites.
161+
pub(crate) fn contains_avg(query: &str) -> bool {
162+
match parse_binary_arms(query) {
163+
Some((lhs, rhs)) => [lhs, rhs]
164+
.into_iter()
165+
.any(|arm| matches!(arm, BinaryArm::Query(q) if contains_avg(&q))),
166+
None => rewrite_avg(query).len() > 1,
167+
}
168+
}
169+
160170
/// Rewrite an `avg` leaf into the sum and count leaves it is computed from:
161171
/// `avg by (l) (x)` → `sum by (l) (x)`, `count by (l) (x)`, and
162172
/// `avg_over_time(x[5m])` → `sum_over_time(x[5m])`, `count_over_time(x[5m])`.

‎asap-planner-rs/src/optimizer/milp.rs‎

Lines changed: 16 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -46,14 +46,17 @@ pub enum MilpError {
4646
}
4747

4848
/// The cheapest plan. Only families in sketch-bench's `DEPLOYABLE_FAMILIES`
49-
/// are candidates.
49+
/// are candidates unless `allow_undeployable_families`, whose plan can be
50+
/// studied but not deployed.
5051
pub fn solve_milp(
5152
workload: &MilpWorkload,
5253
facts: &WorkloadFacts,
5354
costs: &[AtomicCostEntry],
5455
objective: Objective,
56+
allow_undeployable_families: bool,
5557
) -> Result<MilpSolution, MilpError> {
56-
let deployments = build_all_candidates(&workload.raqes, costs, facts, false);
58+
let deployments =
59+
build_all_candidates(&workload.raqes, costs, facts, allow_undeployable_families);
5760
tracing::debug!(
5861
candidates = deployments.len(),
5962
cost_rows = costs.len(),
@@ -251,13 +254,13 @@ fn accuracy_target(
251254
}
252255

253256
#[cfg(test)]
254-
mod tests {
257+
pub(super) mod tests {
255258
use std::collections::BTreeMap;
256259

257260
use super::*;
258261
use crate::optimizer::workload_facts::parse_workload_facts;
259262

260-
const SCRAPE_MS: u64 = 15_000;
263+
pub(crate) const SCRAPE_MS: u64 = 15_000;
261264

262265
const FACTS: &str = r#"
263266
metrics:
@@ -271,20 +274,20 @@ metrics:
271274
cardinality: 1
272275
"#;
273276

274-
fn config(groups: &str) -> ControllerConfig {
277+
pub(crate) fn config(groups: &str) -> ControllerConfig {
275278
let yaml = format!(
276279
"query_groups:\n{groups}\nmetrics:\n - metric: http_requests_total\n labels: [instance, job]\n"
277280
);
278281
serde_yaml::from_str(&yaml).unwrap()
279282
}
280283

281-
fn group(query: &str, accuracy_sla: f64) -> String {
284+
pub(crate) fn group(query: &str, accuracy_sla: f64) -> String {
282285
format!(
283286
" - queries: [\"{query}\"]\n repetition_delay_ms: 60000\n controller_options: {{accuracy_sla: {accuracy_sla}}}\n"
284287
)
285288
}
286289

287-
fn facts(config: &ControllerConfig) -> WorkloadFacts {
290+
pub(crate) fn facts(config: &ControllerConfig) -> WorkloadFacts {
288291
parse_workload_facts(FACTS, config.metrics.as_deref().unwrap(), SCRAPE_MS).unwrap()
289292
}
290293

@@ -311,7 +314,7 @@ metrics:
311314
}
312315
}
313316

314-
fn costs() -> Vec<AtomicCostEntry> {
317+
pub(crate) fn costs() -> Vec<AtomicCostEntry> {
315318
vec![
316319
cost("exact-sum", &[(RELATIVE_ERROR, 0.0)]),
317320
cost("kll-percall", &[(MAX_RANK_ERROR, 0.005)]),
@@ -469,7 +472,7 @@ metrics:
469472
let config = config(&(group(query, 0.99) + &group(query, 0.99)));
470473
let facts = facts(&config);
471474
let w = build_milp_workload(&config, &facts, SCRAPE_MS).unwrap();
472-
let solution = solve_milp(&w, &facts, &costs(), Objective::default()).unwrap();
475+
let solution = solve_milp(&w, &facts, &costs(), Objective::default(), false).unwrap();
473476
assert_eq!(solution.deployments.len(), 1);
474477
assert_eq!(solution.raqes[0].deployment, solution.raqes[1].deployment);
475478
assert_eq!(
@@ -487,7 +490,7 @@ metrics:
487490
));
488491
let facts = facts(&config);
489492
let w = build_milp_workload(&config, &facts, SCRAPE_MS).unwrap();
490-
let err = solve_milp(&w, &facts, &costs(), Objective::default()).unwrap_err();
493+
let err = solve_milp(&w, &facts, &costs(), Objective::default(), false).unwrap_err();
491494
assert!(matches!(err, MilpError::Unservable(ids) if ids.len() == 1));
492495
}
493496

@@ -498,7 +501,7 @@ metrics:
498501
let facts = facts(&config);
499502
let w = build_milp_workload(&config, &facts, SCRAPE_MS).unwrap();
500503
assert_eq!(w.raqes.len(), 2);
501-
let err = solve_milp(&w, &facts, &costs(), Objective::default()).unwrap_err();
504+
let err = solve_milp(&w, &facts, &costs(), Objective::default(), false).unwrap_err();
502505
assert!(matches!(err, MilpError::Unservable(ids) if ids == [w.raqes[0].id.clone()]));
503506
}
504507

@@ -510,7 +513,7 @@ metrics:
510513
let config = config(&groups);
511514
let facts = facts(&config);
512515
let w = build_milp_workload(&config, &facts, SCRAPE_MS).unwrap();
513-
let solution = solve_milp(&w, &facts, &costs(), Objective::default()).unwrap();
516+
let solution = solve_milp(&w, &facts, &costs(), Objective::default(), false).unwrap();
514517
let chosen: BTreeMap<Capability, &str> = w
515518
.raqes
516519
.iter()
@@ -538,7 +541,7 @@ metrics:
538541
let caps: Vec<_> = w.raqes.iter().map(|r| r.capability).collect();
539542
assert_eq!(caps.len(), 2);
540543
assert!(caps.contains(&Capability::Sum) && caps.contains(&Capability::Count));
541-
let solution = solve_milp(&w, &facts, &costs(), Objective::default()).unwrap();
544+
let solution = solve_milp(&w, &facts, &costs(), Objective::default(), false).unwrap();
542545
assert_eq!(solution.deployments.len(), 2);
543546
}
544547
}

0 commit comments

Comments
 (0)