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..36ce8772 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,21 +141,38 @@ 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(()); + 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" + ); } - 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::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" + ); + } + 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(|| { @@ -208,44 +250,46 @@ 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(()) } -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)?; @@ -274,3 +318,27 @@ fn run_milp(args: &Args) -> anyhow::Result<()> { .write_to_dir(&args.output_dir)?; Ok(()) } + +fn run_saved(args: &Args) -> anyhow::Result<()> { + let scrape_interval_ms = check_optimizer_flags(args, "saved")?; + anyhow::ensure!( + args.input_config.is_none(), + "--planner saved reads its queries from --workload-table, not --input_config" + ); + 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..929f271f --- /dev/null +++ b/asap-planner-rs/src/optimizer/saved_plan.rs @@ -0,0 +1,494 @@ +//! 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::{heap_capacity, AtomicCostEntry, Deployment, LabelSet, Raqe}; +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, reject_unwritable_queries}; + +#[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 + /// (a roll-up), which the converter refuses. + 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 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(); + 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 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:?}"))?; + let deployment = Deployment { + capability: raqe.capability, + metric: raqe.metric.clone(), + spatial_filter: raqe.spatial_filter.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, + 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!( + 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, + )?) +} + +/// 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 +/// 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 { + run_on(TABLE, plan.to_string().as_bytes(), sla_ms) + } + + 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).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 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()).to_string().as_bytes(), None) + .err() + .expect("refused") + .to_string(); + assert!(err.contains("avg"), "{err}"); + 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()); + 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); + 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..13273a12 100644 --- a/asap-tools/experiments/experiment_run_e2e.py +++ b/asap-tools/experiments/experiment_run_e2e.py @@ -316,20 +316,42 @@ 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}" + ) + config.copy_prebuilt_controller_output( + 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, + 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..5f8ce768 100644 --- a/asap-tools/experiments/experiment_utils/config.py +++ b/asap-tools/experiments/experiment_utils/config.py @@ -5,10 +5,13 @@ import math import os +import re +import shutil import copy import yaml from typing import Any, Dict, List, Tuple +import hydra.utils from omegaconf import DictConfig, ListConfig, OmegaConf import constants @@ -535,6 +538,65 @@ 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 _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) + 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 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) + 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.""" @@ -605,6 +667,12 @@ 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 + ) + 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..b140933c --- /dev/null +++ b/asap-tools/experiments/tests/test_prebuilt_controller_output.py @@ -0,0 +1,87 @@ +"""Tests for running e2e experiments on prebuilt planner configs.""" + +import os +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) + 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") + self.copy() + 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"): + self.copy() + + 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()