From 889b0b466bf269bd0dff10ec8113574b4a943528 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 11:19:08 -0600 Subject: [PATCH 1/5] refactor: remove the unreferenced legacy planning endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Six endpoints on the legacy planning API have no caller anywhere in the workspace — not the data plane, not the process e2e tests, not the docs: `/plan/auto`, `/plan/pareto`, `/plan/:metric/rollback`, `/plan/:metric/diff`, `/agents`, and `/config/:metric`. Removing them retires the autonomous allocation chain that existed only to serve `/plan/auto`: `epsilon_alloc::build_auto_plan` and its knob allocation, which pulled in `query_planning`, and through it `sketch_selection` and `asap_tier_implement`. `epsilon_alloc` itself stays. `derive_sample_p` and `split_budget` are live on the physical path (`physical/compiler.rs` and `types.rs`), so the module keeps those two and loses the rest. `POST /api/v1/plan`, `GET /api/v1/plan/:metric` and `/api/v1/cost-model` are untouched: the data plane's capability-miss notifier and config fetcher call the first two, and `control_plane/tests/component_process_e2e.rs` uses the third as its readiness probe. Co-Authored-By: Claude Opus 5 (1M context) --- control_plane/src/asap_tier_implement.rs | 333 -------------- control_plane/src/epsilon_alloc.rs | 328 -------------- control_plane/src/lib.rs | 3 - control_plane/src/main.rs | 409 ----------------- .../src/physical/deployment_cost/mod.rs | 1 - .../src/physical/deployment_cost/pareto.rs | 418 ------------------ control_plane/src/query_planning.rs | 328 -------------- control_plane/src/sketch_selection.rs | 169 ------- 8 files changed, 1989 deletions(-) delete mode 100644 control_plane/src/asap_tier_implement.rs delete mode 100644 control_plane/src/physical/deployment_cost/pareto.rs delete mode 100644 control_plane/src/query_planning.rs delete mode 100644 control_plane/src/sketch_selection.rs diff --git a/control_plane/src/asap_tier_implement.rs b/control_plane/src/asap_tier_implement.rs deleted file mode 100644 index 858c8e3c9..000000000 --- a/control_plane/src/asap_tier_implement.rs +++ /dev/null @@ -1,333 +0,0 @@ -//! PromQL → compositional L4 plan (`asap_aware_mapping::bind::implement_tree` -//! adoption, Step A of the plan-shaped-serving scoping). -//! -//! ## Terminology: "implementation", not "bind" -//! -//! This module (and the rest of `asap_tier_analysis`) has historically -//! used "bind"/"binding" loosely for several different things. `asap-plan` -//! itself is explicit about *not* doing that — its crate-level doc -//! (`crates/plan/src/lib.rs` in `ASAPController`) lays out a table of -//! four already-distinct senses of "bind" in this stack (L2 name -//! resolution `ColumnRef → ColumnId`; the L3→L4 per-node physical -//! choice; a deployment's own L4→L5 *placement* decision, e.g. -//! `control_plane::physical::post_asap::rules::bind_*`; and this module's own -//! prior usage) and deliberately names its L3→L4 step **"implementation"** -//! instead (Cascades/Volcano terminology: an *implementation rule* is -//! logical → physical, distinct from a *transformation rule*, logical → -//! logical) so it doesn't become a fifth colliding sense. Following that -//! lead here: [`collect_aggregate_roots`] *finds* realizable `Aggregate` -//! subtrees, [`implement_promql_for_asap_tier`] *implements* (realizes) -//! them via `asap_aware_mapping::bind::implement_tree_in_with` — "bind" is used -//! below only where citing `asap-plan`'s own module/function names -//! (`asap_aware_mapping::bind`, `implement_tree_in_with`) or the specific -//! deployment-side "Bind #2" placement decision that table names. -//! -//! ## The gap this closes -//! -//! `asap_tier_analysis::analyze_promql_for_asap_tier` collects every -//! `AggIntent` anywhere in the query tree into one **flat** list, then -//! checks each independently against `capability_for` — "does *some* -//! pre-registered accumulator answer *this one* intent." It never -//! composes multiple accumulators into a multi-stage plan (`Avg = Sum / -//! Count` stays archive-only for exactly this reason). -//! -//! `asap_aware_mapping::bind::implement_tree_in_with` already builds a genuinely -//! compositional plan (`Rc` / `SummaryExpr` — `SummaryAgg`, -//! `SummaryEstimate`, `SummaryMerge`, ...) from an L3 `QueryExpr` — but -//! it is deliberately conservative about *where* it looks for a -//! realizable `Aggregate`: hitting any non-`Aggregate` node (`Filter`, -//! `Window`, `Project`, ...) immediately wraps the **whole** subtree as -//! `SummaryExpr::Logical`, with no attempt to recurse past it looking -//! for a realizable `Aggregate` further down. Per `asap-plan`'s own -//! module doc: "rewriting through logical parents is the L4 rule -//! engine's job" — the deployment-specific "Bind #2" placement decision -//! that crate explicitly does not model. Finding realizable `Aggregate` -//! subtrees wherever they occur in the tree is squarely this crate's -//! job. -//! -//! [`collect_aggregate_roots`] does that: it mirrors -//! `asap_tier_analysis::collect_agg_intents`'s full recursion through -//! every `QueryExpr` variant, but instead of flattening `AggIntent`s it -//! collects the `Aggregate` subtree **roots** `implement_tree_in_with` -//! can realize. [`implement_promql_for_asap_tier`] parses PromQL, finds -//! those roots, and implements each independently. -//! -//! ## Known gap: `Extension`/`Frequency` under-realizes -//! -//! `asap-plan` has no opinion on `AggIntent::Extension` (control_plane's -//! deployment-specific `Frequency` intent rides in one) — it always -//! returns `Implementation::PassThrough`, by design (the crate -//! genuinely cannot see into an opaque extension payload). So today, an -//! `Aggregate` root whose one intent is a `Frequency` extension -//! implements to `SummaryExpr::Logical` here — i.e. this walker -//! currently *under-realizes* exactly the query shape `capability_for`'s -//! `as_frequency` special case already handles correctly on the flat -//! path. `implement_frequency_as_agg_test` below pins this as a known, -//! tracked gap (not a silent trap) — closing it needs either an -//! `asap-plan` extension hook or a hand-rolled `SummaryAgg`/ -//! `SummaryEstimate` construction here (the SummaryNode-building helpers in -//! `asap_aware_mapping::bind` are private, so today there's no way to do the -//! latter without reimplementing them). Not yet wired into the live -//! serving path for this reason — see module doc for the broader -//! plan-shaped-serving scope this is Step A of. - -use std::rc::Rc; - -use asap_aware_mapping::DefaultCostModel; -use planner_types::post_asap::SummaryNode; - -use crate::query_parser::parse_query_expr_canonical; -use crate::types::AccuracyTarget; -use planner_types::pre_asap::QueryExpr; - -/// Fixed accuracy target for this L1 call site (L1 adoption, -/// design-target-architecture.md Part B) -- matches -/// `asap_tier_analysis::WARM_TIER_ANALYSIS_ACCURACY`; this module has no -/// per-query accuracy bound available either (see its own module doc: -/// "not yet a drop-in replacement for `analyze_promql_for_asap_tier`"). -const IMPLEMENT_PROMQL_ACCURACY: AccuracyTarget = AccuracyTarget::Epsilon(0.01); - -/// Find every independently-realizable `Aggregate` subtree in `expr`. -/// -/// Mirrors `asap_tier_analysis::collect_agg_intents`'s exact recursion -/// through every `QueryExpr` variant, but collects `Aggregate` node -/// references instead of flattening their `AggIntent`s. -/// -/// A **realizable-shaped** `Aggregate` (exactly one intent, no `HAVING` -/// — what `implement_tree_in_with`'s own internal comment calls its -/// "bindable shape") is pushed as a root and NOT recursed into further: -/// `implement_tree_in_with` already recurses into its own `child` via -/// `bind_summary_agg`, so walking past it here would find the same -/// nested aggregates twice. -/// -/// A **non-realizable** `Aggregate` (multiple intents, or a `HAVING` -/// predicate) is different: `implement_tree_in_with` gives up on it -/// entirely (falls straight to `Logical`, without recursing into -/// `child` at all — see its own shape check). So a realizable -/// `Aggregate` nested inside a non-realizable one's `child` would -/// otherwise never be found. Recurse into `child` by hand for exactly -/// this case. -fn collect_aggregate_roots<'a>(expr: &'a QueryExpr, out: &mut Vec<&'a QueryExpr>) { - match expr { - QueryExpr::Aggregate { - measures: aggs, - having, - child, - .. - } => { - if let ([_], None) = (aggs.as_slice(), having) { - out.push(expr); - } else { - collect_aggregate_roots(child, out); - } - } - QueryExpr::TimeRange { child, .. } => collect_aggregate_roots(child, out), - QueryExpr::Scan { .. } => {} - // A-variants lifted in Batch 2 of the relational migration. They - // carry no AggIntent themselves — recurse into their children to - // find Aggregates further down the tree. `Partition` no longer - // exists in the canonical IR — its keys fold into `Aggregate.by` - // at construction time (`intent_algebra::lower`). - QueryExpr::Filter { child, .. } - | QueryExpr::Project { child, .. } - | QueryExpr::Dedup { child, .. } - | QueryExpr::Sort { child, .. } - | QueryExpr::Limit { child, .. } - | QueryExpr::PromqlSubquery { child, .. } => collect_aggregate_roots(child, out), - QueryExpr::Concat { children, .. } => { - for c in children { - collect_aggregate_roots(c, out); - } - } - QueryExpr::Join { left, right, .. } - | QueryExpr::SetOp { left, right, .. } - | QueryExpr::BinaryOp { - lhs: left, - rhs: right, - .. - } => { - collect_aggregate_roots(left, out); - collect_aggregate_roots(right, out); - } - // The PromQL-surface superset (Scalar/EvalTime/VectorFromScalar/ - // ScalarFromVector/Relabel/InfoJoin/Sample/TimeRange/TimeShift/ - // WindowFunc) isn't constructed by this parser today; the - // single-child wrappers among them carry no `AggIntent` either - // way, so a no-op default is safe (mirrors `collect_agg_intents`). - _ => {} - } -} - -/// Errors from [`implement_promql_for_asap_tier`]. -#[derive(Debug)] -pub enum ImplementPromqlError { - /// PromQL failed to parse / lower to canonical `QueryExpr`. - UnparseableMetricsql(String), - /// L3→L4 implementation failed for a found `Aggregate` root (schema - /// derivation error — see `asap_aware_mapping::bind::ImplementError`). - Implement(crate::planner_selection::SelectionError), -} - -/// Parse `metricsql`, find every independently-realizable `Aggregate` -/// subtree ([`collect_aggregate_roots`]), and implement each into a -/// compositional L4 plan via `asap_aware_mapping::bind::implement_tree_in_with`. -/// -/// Returns one `Rc` per root found, in tree order. Empty (not an -/// error) when the query has no realizable `Aggregate` at all (a bare -/// selector, or a window-bound exact-aggregation the lowerer didn't -/// emit an `Aggregate` for) — same "no candidates" contract -/// `analyze_promql_for_asap_tier` uses. -/// -/// See the module doc for the current `Extension`/`Frequency` -/// under-realization gap — not yet a drop-in replacement for -/// `analyze_promql_for_asap_tier`, and not wired into the live serving -/// path. -pub fn implement_promql_for_asap_tier( - metricsql: &str, -) -> Result>, ImplementPromqlError> { - let expr = parse_query_expr_canonical(metricsql, IMPLEMENT_PROMQL_ACCURACY) - .map_err(|e| ImplementPromqlError::UnparseableMetricsql(e.to_string()))?; - - let mut roots: Vec<&QueryExpr> = Vec::new(); - collect_aggregate_roots(&expr, &mut roots); - - roots - .into_iter() - .map(|root| { - crate::planner_selection::select_summary(root, &DefaultCostModel) - .map_err(ImplementPromqlError::Implement) - }) - .collect() -} - -#[cfg(test)] -mod tests { - use super::*; - use planner_types::post_asap::SummaryExpr; - - fn root_count(metricsql: &str) -> usize { - implement_promql_for_asap_tier(metricsql) - .expect("parses and implements") - .len() - } - - #[test] - fn bare_selector_has_no_aggregate_root_to_implement() { - // A bare selector has no `Aggregate` node, so there is no ASAP-tier root. - let roots = - implement_promql_for_asap_tier("http_requests_total").expect("parses and implements"); - assert!(roots.is_empty(), "{roots:?}"); - } - - #[test] - fn single_sum_finds_one_root() { - assert_eq!(root_count("sum(http_requests_total)"), 1); - } - - #[test] - fn quantile_implements_to_a_sketch_agg_estimate() { - let roots = implement_promql_for_asap_tier( - "quantile_over_time(0.99, http_requests_total_latency_ms[5m])", - ) - .expect("parses and implements"); - assert_eq!(roots.len(), 1); - assert!( - matches!(roots[0].expr, SummaryExpr::SummaryEstimate { .. }), - "quantile is sketch-realized (SummaryAgg wrapped in a SummaryEstimate \ - readout), not archive-only Logical: {:?}", - roots[0].expr, - ); - } - - #[test] - fn sum_implements_to_an_exact_accumulator_agg_with_no_estimate() { - let roots = implement_promql_for_asap_tier("sum(http_requests_total)") - .expect("parses and implements"); - assert_eq!(roots.len(), 1); - assert!( - matches!(roots[0].expr, SummaryExpr::SummaryAgg { .. }), - "Sum is an exact mergeable accumulator -- SummaryAgg with no \ - wrapping SummaryEstimate (the partial state IS the value): {:?}", - roots[0].expr, - ); - } - - #[test] - fn avg_over_time_is_not_yet_realizable_matching_capability_for_today() { - // Avg = Sum / Count needs a cross-policy join implement_tree_in_with - // doesn't build (matches capability_for(&AggIntent::Avg) => None - // on the flat path -- see asap_tier_analysis.rs and lower.rs's - // AggFunc::Avg comment). Use avg_over_time (a range-vector - // function), not bare instant avg(...) -- only the former is - // guaranteed to lower through AggFunc::Avg in this frontend. - let roots = implement_promql_for_asap_tier("avg_over_time(http_requests_total[5m])") - .expect("parses and implements"); - assert_eq!(roots.len(), 1); - assert!( - matches!(roots[0].expr, SummaryExpr::KeepPreAsap(_)), - "Avg has no ASAP-tier realization yet on either path: {:?}", - roots[0].expr, - ); - } - - #[test] - fn nested_aggregate_under_aggregate_finds_both_roots_independently() { - // topk(5, quantile_over_time(0.9, m[5m])) -- outer TopK Aggregate - // wraps an inner Quantile Aggregate. implement_tree_in_with's own - // recursion into `child` handles this once we hand it the OUTER - // root; collect_aggregate_roots must not ALSO independently - // re-find the inner one (that would double-implement it). - let roots = implement_promql_for_asap_tier( - "topk(5, quantile_over_time(0.9, http_requests_total_latency_ms[5m]))", - ) - .expect("parses and implements"); - assert_eq!( - roots.len(), - 1, - "the outer TopK root's own recursion already covers the nested \ - Quantile -- collect_aggregate_roots must find exactly one \ - independent root here, not two" - ); - } - - #[test] - fn filter_wrapping_a_realizable_aggregate_is_still_found() { - // Filter { child: Aggregate{ Sum } } -- implement_tree_in_with - // alone would give up at the Filter and wrap the whole subtree - // as Logical (archive-only). collect_aggregate_roots must recurse - // past the Filter to find the Sum root underneath -- this is the - // exact gap Step A closes. - let roots = implement_promql_for_asap_tier("sum(http_requests_total{zone=\"us-east\"})") - .expect("parses and implements"); - assert_eq!(roots.len(), 1); - assert!( - matches!(roots[0].expr, SummaryExpr::SummaryAgg { .. }), - "a label filter above the aggregate must not prevent implementation: {:?}", - roots[0].expr, - ); - } - - /// KNOWN GAP (see module doc): a bare frequency point-query - /// (`count_over_time(m[5m])` -- a *windowed* Count is one of - /// control_plane's two triggers for the `Extension`/`Frequency` - /// intent, per `lower.rs`'s `frequency_trigger = !by.is_empty() || - /// windowed`, pinned by its own `windowed_count_is_frequency` unit - /// test) binds to `Logical` here today, even though - /// `capability_for`'s `as_frequency` special case on the flat path - /// correctly realizes it as `FrequencyEstimate`. Pinned as a test so - /// closing this gap is a deliberate, visible change, not a silent - /// behavior shift. - #[test] - fn implement_frequency_as_agg_test() { - // Approximate `count_over_time` lowers to a first-class `Count` intent, - // which Planner realizes as CMS-backed `SummaryAgg` + `SummaryEstimate`. - let roots = implement_promql_for_asap_tier("count_over_time(http_requests_total[5m])") - .expect("parses and implements"); - assert_eq!(roots.len(), 1); - assert!( - matches!(roots[0].expr, SummaryExpr::SummaryEstimate { .. }), - "expected a realized SummaryEstimate, got: {:?}", - roots[0].expr, - ); - } -} diff --git a/control_plane/src/epsilon_alloc.rs b/control_plane/src/epsilon_alloc.rs index 7a7fc7a79..308db7c95 100644 --- a/control_plane/src/epsilon_alloc.rs +++ b/control_plane/src/epsilon_alloc.rs @@ -19,13 +19,8 @@ //! control_plane has no dependency on data_plane, so the canonical one-liner is //! re-stated here — keep the two in sync.) -use std::collections::HashMap; -use serde::Serialize; -use crate::query_planning::{plan_sketches_for_queries, QuerySetPlan}; -use crate::runtime_samples::RuntimeSamplesStore; -use crate::types::SketchType; /// Admission sampling probability under the unified ε-floor law, /// `p = 1/(1 + ε²·rate)`. @@ -40,17 +35,6 @@ pub fn derive_sample_p(epsilon: f64, rate: f64) -> f64 { 1.0 / (1.0 + epsilon * epsilon * rate) } -/// Per-site CDM delta-transmission slack, `δ = ε·τ / k`. -/// -/// The global threshold `τ` is the monitored value of interest (a user/query -/// input). The even-split CDM construction gives each of `k` sites a local -/// slack whose deviations sum to at most the global `ε·τ` budget. `k` is -/// clamped to ≥ 1. (A rate-weighted split is a possible refinement; the even -/// split is the baseline CDM allocation.) -pub fn derive_delta_threshold(epsilon: f64, tau: f64, n_sites: u32) -> f64 { - let k = n_sites.max(1) as f64; - epsilon * tau / k -} /// GOS budget split (design §7, Layer B) — **exact linear peel**. /// @@ -90,194 +74,18 @@ pub fn split_budget(eps_total: f64, eps_sketch: f64, w_edge: f64, w_comm: f64) - (eps_sa, eps_st) } -/// The runtime knobs derived for one metric. -#[derive(Debug, Clone, PartialEq)] -pub struct MetricKnobs { - /// Admission sampling probability (ε-floor). `1.0` = no sampling. - pub sample_p: f64, - /// Per-site CDM delta-transmission slack, present only for metrics that - /// carry a monitored threshold `τ`. - pub delta_threshold: Option, -} - -/// A fully-allocated metric: the sketches to stand up plus the runtime knobs. -#[derive(Debug, Clone, PartialEq)] -pub struct AllocatedMetric { - pub metric_name: String, - pub sketches: Vec, - pub knobs: MetricKnobs, -} -/// Attach ε-derived knobs to every metric in a [`QuerySetPlan`]. -/// -/// * `rate_for(metric)` → the metric's estimated per-window update rate (drives -/// `p`). At call sites this comes from runtime telemetry or a workload -/// estimate; kept as a closure so the deriver stays pure and testable. -/// * `monitor_for(metric)` → `Some((τ, k))` if the metric carries a CDM -/// threshold (drives `δ`), else `None` (no delta-transmission). -pub fn allocate_knobs( - epsilon: f64, - plan: &QuerySetPlan, - mut rate_for: impl FnMut(&str) -> f64, - mut monitor_for: impl FnMut(&str) -> Option<(f64, u32)>, -) -> Vec { - plan.per_metric - .iter() - .map(|m| { - let sample_p = derive_sample_p(epsilon, rate_for(&m.metric_name)); - let delta_threshold = - monitor_for(&m.metric_name).map(|(tau, k)| derive_delta_threshold(epsilon, tau, k)); - AllocatedMetric { - metric_name: m.metric_name.clone(), - sketches: m.sketches.clone(), - knobs: MetricKnobs { - sample_p, - delta_threshold, - }, - } - }) - .collect() -} -/// Wire-serializable view of one planned metric (`POST /api/v1/plan/auto`). -#[derive(Debug, Clone, PartialEq, Serialize)] -pub struct PlannedMetric { - pub metric: String, - pub sketches: Vec, - pub sample_p: f64, - pub delta_threshold: Option, -} -/// A query routed to the cold tier, with the human-readable reason. -#[derive(Debug, Clone, PartialEq, Serialize)] -pub struct ColdEntry { - pub query: String, - pub reason: Option, -} -/// The full autonomous-allocation response: per-metric warm plan + cold-tier -/// fallthrough, for a given `(ε, queries)` (and optional rate/monitor inputs). -#[derive(Debug, Clone, PartialEq, Serialize)] -pub struct AutoPlanResponse { - pub epsilon: f64, - pub metrics: Vec, - pub cold_only: Vec, - /// Number of metrics actually applied (monitor injected → repost emits it → - /// coordinator derives live `p`) when the request set `apply: true`. `None` - /// for a dry-run plan. - #[serde(skip_serializing_if = "Option::is_none")] - pub applied: Option, -} -/// Best-effort per-metric update rate from runtime telemetry. -/// -/// The runtime-samples store is keyed by `(source, sketch, impl)` and carries -/// no metric name, so a metric's rate can only be approximated by the -/// throughput of the sketch *families* it was allocated: this sums the freshest -/// `throughput_items_per_sec.mean` across every telemetry key whose sketch tag -/// matches one of `sketches`. Returns `None` when no matching telemetry exists -/// (caller falls back to a default). The throughput is per-second; it feeds the -/// plan-time ε-floor `p` as a display estimate — the authoritative live `p` is -/// the data-plane coordinator's (computed from the monitor ε). -pub fn metric_rate_from_telemetry( - store: &RuntimeSamplesStore, - sketches: &[SketchType], -) -> Option { - let mut total = 0.0; - let mut matched = false; - for key in store.keys() { - if !sketches.iter().any(|s| sketch_tag_matches(&key.sketch, s)) { - continue; - } - if let Some(rec) = store.latest(&key) { - if let Some(tp) = rec.payload["bench"]["throughput_items_per_sec"]["mean"].as_f64() { - total += tp; - matched = true; - } - } - } - matched.then_some(total) -} -/// Loose match of a telemetry `sketch` tag to a [`SketchType`] family. -fn sketch_tag_matches(tag: &str, want: &SketchType) -> bool { - let t = tag.to_lowercase(); - match want { - SketchType::DDSketch => t.contains("ddsketch"), - SketchType::KLL => t.contains("kll"), - SketchType::HLL => t.contains("hll"), - SketchType::CountSketch => t.contains("countsketch"), - SketchType::CountMinSketch => t.contains("cms") || t.contains("countmin"), - } -} -/// End-to-end autonomous allocation: `(ε, queries) → AutoPlanResponse`. -/// -/// Runs slice 2 ([`plan_sketches_for_queries`]) then slice 3 -/// ([`allocate_knobs`]), flattening to a serializable response. Per metric the -/// ε-floor `p` rate is resolved by `rate_for(metric, &sketches)` (e.g. -/// [`metric_rate_from_telemetry`]); when it returns `None`, `default_rate` is -/// used. `monitors[metric] = (τ, sites)` adds the CDM `δ` for monitored metrics. -/// Kept in the lib so it is unit-testable without standing up the HTTP server. -pub fn build_auto_plan( - epsilon: f64, - queries: &[S], - default_rate: f64, - mut rate_for: R, - monitors: &HashMap, -) -> AutoPlanResponse -where - S: AsRef, - R: FnMut(&str, &[SketchType]) -> Option, -{ - let plan = plan_sketches_for_queries(queries.iter().map(|s| s.as_ref())); - // Resolve a rate per metric up front (telemetry → default fallback) so the - // ε-floor closure below is a simple lookup. - let rate_map: HashMap = plan - .per_metric - .iter() - .map(|m| { - let r = rate_for(&m.metric_name, &m.sketches).unwrap_or(default_rate); - (m.metric_name.clone(), r) - }) - .collect(); - let allocated = allocate_knobs( - epsilon, - &plan, - |m| rate_map.get(m).copied().unwrap_or(default_rate), - |m| monitors.get(m).copied(), - ); - let metrics = allocated - .into_iter() - .map(|a| PlannedMetric { - metric: a.metric_name, - sketches: a.sketches, - sample_p: a.knobs.sample_p, - delta_threshold: a.knobs.delta_threshold, - }) - .collect(); - let cold_only = plan - .cold_only - .into_iter() - .map(|c| ColdEntry { - query: c.query, - reason: c.reason.map(|r| format!("{r:?}")), - }) - .collect(); - - AutoPlanResponse { - epsilon, - metrics, - cold_only, - applied: None, - } -} #[cfg(test)] mod tests { use super::*; - use crate::query_planning::plan_sketches_for_queries; #[test] fn sample_p_follows_epsilon_floor() { @@ -320,145 +128,9 @@ mod tests { assert!(derive_sample_p(0.01, 1000.0) > derive_sample_p(0.10, 1000.0)); } - #[test] - fn delta_threshold_is_even_split_of_epsilon_tau() { - // ε·τ/k = 0.05 * 7000 / 5 = 70 - assert!((derive_delta_threshold(0.05, 7000.0, 5) - 70.0).abs() < 1e-9); - // k clamps to ≥ 1 - assert!((derive_delta_threshold(0.05, 7000.0, 0) - 350.0).abs() < 1e-9); - } - #[test] - fn allocate_knobs_attaches_p_and_optional_delta() { - let plan = plan_sketches_for_queries(["quantile_over_time(0.99, latency_ms[5m])"]); - // latency_ms has a monitored threshold τ=7000 over k=4 sites; rate 2000/win - let allocated = allocate_knobs( - 0.05, - &plan, - |_m| 2000.0, - |m| { - if m == "latency_ms" { - Some((7000.0, 4)) - } else { - None - } - }, - ); - let a = allocated - .iter() - .find(|a| a.metric_name == "latency_ms") - .expect("metric present"); - // p = 1/(1+0.0025*2000) = 1/6 = 0.1667… - assert!( - (a.knobs.sample_p - 1.0 / 6.0).abs() < 1e-9, - "p={}", - a.knobs.sample_p - ); - // δ = 0.05*7000/4 = 87.5 - assert_eq!(a.knobs.delta_threshold, Some(87.5)); - // the sketch set survived from slice 2 - assert!( - a.sketches.contains(&SketchType::DDSketch) || a.sketches.contains(&SketchType::KLL) - ); - } - #[test] - fn build_auto_plan_end_to_end_over_query_set() { - // (ε, queries) → full response. A quantile read + a monitored threshold. - let mut monitors = HashMap::new(); - monitors.insert("latency_ms".to_string(), (7000.0, 4)); - let resp = build_auto_plan( - 0.05, - &[ - "quantile_over_time(0.99, latency_ms[5m])".to_string(), - "not valid promql @@@".to_string(), - ], - 2000.0, - |_m, _s| None, // no telemetry → default_rate - &monitors, - ); - assert_eq!(resp.epsilon, 0.05); - // the quantile metric is planned with a quantile sketch + ε-floor p + δ - let m = resp - .metrics - .iter() - .find(|m| m.metric == "latency_ms") - .expect("latency_ms planned"); - assert!( - m.sketches.contains(&SketchType::DDSketch) || m.sketches.contains(&SketchType::KLL) - ); - assert!((m.sample_p - 1.0 / 6.0).abs() < 1e-9); // 1/(1+0.0025*2000) - assert_eq!(m.delta_threshold, Some(87.5)); // 0.05*7000/4 - // the unparseable query falls through to cold - assert_eq!(resp.cold_only.len(), 1); - // and the response serializes to JSON cleanly (the handler's job) - let v = serde_json::to_value(&resp).expect("serializes"); - assert!((v["metrics"][0]["sample_p"].as_f64().unwrap() - 1.0 / 6.0).abs() < 1e-9); - } - #[test] - fn metric_without_monitor_gets_no_delta() { - let plan = plan_sketches_for_queries(["quantile_over_time(0.99, latency_ms[5m])"]); - let allocated = allocate_knobs(0.05, &plan, |_m| 500.0, |_m| None); - let a = &allocated[0]; - assert_eq!(a.knobs.delta_threshold, None); - assert!(a.knobs.sample_p < 1.0); // sampling still applies - } - #[test] - fn telemetry_rate_read_by_sketch_family() { - use crate::runtime_samples::{RuntimeRecord, RuntimeSamplesStore, SampleKey}; - let store = RuntimeSamplesStore::new(8); - // seed a ddsketch throughput of 4000 items/sec from one source - store.append_for_test(RuntimeRecord { - source: "dc-a".into(), - sketch: "ddsketch".into(), - impl_name: "oxide".into(), - schema_version: 1, - payload: serde_json::json!({ - "bench": { "throughput_items_per_sec": { "mean": 4000.0, "stddev": 0.0 } } - }), - }); - // a quantile metric is allocated DDSketch → telemetry rate found - let r = metric_rate_from_telemetry(&store, &[SketchType::DDSketch]); - assert_eq!(r, Some(4000.0)); - // a metric allocated only HLL has no matching telemetry → None (→ default) - assert_eq!(metric_rate_from_telemetry(&store, &[SketchType::HLL]), None); - let _ = SampleKey { - source: "dc-a".into(), - sketch: "ddsketch".into(), - impl_name: "oxide".into(), - }; - } - #[test] - fn build_auto_plan_uses_telemetry_rate_when_available() { - use crate::runtime_samples::{RuntimeRecord, RuntimeSamplesStore}; - let store = RuntimeSamplesStore::new(8); - store.append_for_test(RuntimeRecord { - source: "dc-a".into(), - sketch: "ddsketch".into(), - impl_name: "oxide".into(), - schema_version: 1, - payload: serde_json::json!({ - "bench": { "throughput_items_per_sec": { "mean": 2000.0, "stddev": 0.0 } } - }), - }); - let monitors = HashMap::new(); - // default_rate is a tiny 1.0, but telemetry says 2000 → p must reflect 2000 - let resp = build_auto_plan( - 0.05, - &["quantile_over_time(0.99, latency_ms[5m])".to_string()], - 1.0, - |_m, _sks| metric_rate_from_telemetry(&store, &[SketchType::DDSketch]), - &monitors, - ); - let m = resp - .metrics - .iter() - .find(|m| m.metric == "latency_ms") - .unwrap(); - // p = 1/(1+0.0025*2000) = 1/6 (telemetry), NOT 1/(1+0.0025*1) (default) - assert!((m.sample_p - 1.0 / 6.0).abs() < 1e-9, "p={}", m.sample_p); - } } diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs index b4f413029..da3261d7e 100644 --- a/control_plane/src/lib.rs +++ b/control_plane/src/lib.rs @@ -29,17 +29,14 @@ pub mod pipeline; pub mod planner_selection; pub mod query_parser; pub mod query_plan; -pub mod query_planning; pub mod replan; pub mod runtime_samples; -pub mod sketch_selection; pub mod store; pub mod types; pub mod workload; /// PromQL → ASAPPlanner's canonical post-ASAP plan via /// `asap_aware_mapping::bind::implement_tree`. -pub mod asap_tier_implement; /// Crate-wide test-only utilities for serialising access to process-global /// state (environment variables). diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index a255ddcb2..20b864a67 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -3,7 +3,6 @@ use control_plane::backend_client; use control_plane::clickhouse; use control_plane::emit; -use control_plane::epsilon_alloc; use control_plane::metrics_exposer; use control_plane::monitor; use control_plane::opamp; @@ -38,7 +37,6 @@ use opamp::{AgentRole, OpampServer, RemoteConfig}; use physical::colored_dag::emitter::BackendStageConfig; use physical::deployment_cost::online as online_cost_model; use physical::deployment_cost::online::{init_store as init_online_store, OnlineMetricsStore}; -use physical::deployment_cost::pareto::{pareto_frontier, select_best, ObjectiveWeights}; use physical::deployment_cost::tco; use physical::deployment_cost::DeploymentCostPlanner; use physical::plan_cache::CachedDeploymentPlanner; @@ -489,13 +487,7 @@ async fn main() { "/api/v1/clickhouse-plan/automatic/compile-and-publish", post(handle_compile_and_publish_automatic_clickhouse_plan), ) - .route("/api/v1/plan/auto", post(handle_plan_auto)) - .route("/api/v1/plan/pareto", post(handle_pareto)) .route("/api/v1/plan/:metric", get(handle_get_plan)) - .route("/api/v1/plan/:metric/rollback", post(handle_rollback)) - .route("/api/v1/plan/:metric/diff", get(handle_plan_diff)) - .route("/api/v1/agents", get(handle_agents)) - .route("/api/v1/config/:metric", get(handle_get_config)) .route( "/api/v1/collector-config/agent", get(handle_bootstrap_agent_config), @@ -1356,228 +1348,11 @@ async fn handle_plan(State(st): State, Json(spec): Json) -> .into_response() } -/// Request body for `POST /api/v1/plan/auto` — autonomous capability-match -/// allocation. Given a target accuracy `epsilon` and a set of `queries`, the -/// control plane derives, per metric: which sketch families to allocate (from -/// the queries' required capabilities), the admission sampling probability -/// `p = 1/(1+ε²·rate)`, and the CDM delta-transmission slack `δ = ε·τ/sites` -/// for any metric carrying a monitored threshold. Queries the warm sketch tier -/// can't answer are returned under `cold_only`. -#[derive(serde::Deserialize)] -struct AutoPlanRequest { - epsilon: f64, - queries: Vec, - /// Estimated per-window update rate feeding the ε-floor `p`; applied to - /// every metric (a per-metric / runtime-telemetry source is a future - /// refinement). Omitted → 1.0 (≈ no sampling). - #[serde(default)] - default_rate: Option, - /// Optional per-metric CDM monitor thresholds `{metric: {tau, sites}}`. - /// Metrics absent here get no delta-transmission knob. - #[serde(default)] - monitors: std::collections::HashMap, - /// When true, APPLY the plan: inject a runtime monitor entry per allocated - /// metric into the workload registry and trigger a replan, so the backend - /// `StreamingConfig` carries the monitor (ε, τ) and the data-plane - /// coordinator derives the live ε-floor `p`. Default false = dry-run plan. - #[serde(default)] - apply: bool, -} - -#[derive(serde::Deserialize)] -struct AutoMonitor { - tau: f64, - #[serde(default = "default_sites")] - sites: u32, - /// CDM epoch (seconds) the monitor is bound to. MUST equal the edge's - /// reporting epoch (its SDK window) or the coordinator's alignment guard - /// silently refuses to grant. Omit to AUTO-DERIVE from the deployment's edge - /// epoch (`CONTROLLER_MONITOR_WINDOW_SECS`, default 1s) — see - /// `autonomous_monitor_window_secs`. - #[serde(default)] - window_secs: Option, - /// CDM functional the coordinator monitors: "sum"/"f2" (whole-sketch, - /// empty key) or "cms_point" (per-key). Must match what the edge reports. - #[serde(default = "default_functional")] - functional: String, - /// The monitored key (cms_point series id). Empty for sum/f2. Must match - /// the key the edge registers under, or the coordinator rejects it. - #[serde(default)] - key: String, -} -fn default_functional() -> String { - "sum".to_string() -} -fn default_sites() -> u32 { - 1 -} - -/// The CDM epoch (seconds) autonomous monitors bind to, when the request does -/// not pin one. This MUST equal the edge's reporting epoch (its SDK window): -/// the coordinator's `on_register` alignment guard refuses to grant when -/// `monitor.window_ms != edge.epoch_window_ms`, so a mismatch silently no-ops -/// the whole sampling loop. Defaults to 1s (the standard edge SDK window); -/// override per-deployment with `CONTROLLER_MONITOR_WINDOW_SECS`. -fn autonomous_monitor_window_secs() -> u64 { - std::env::var("CONTROLLER_MONITOR_WINDOW_SECS") - .ok() - .and_then(|s| s.parse::().ok()) - .filter(|&w| w > 0) - .unwrap_or(1) -} - -/// Run the autonomous allocation pipeline (`query_planning` → `epsilon_alloc`) -/// and return the derived per-metric sketch+knob plan plus the cold-tier -/// fallthrough set. The per-metric ε-floor `p` rate is sourced from runtime -/// telemetry (by allocated sketch family) with the request `default_rate` as -/// fallback. -async fn handle_plan_auto( - State(st): State, - Json(req): Json, -) -> impl IntoResponse { - let monitors: std::collections::HashMap = req - .monitors - .iter() - .map(|(m, am)| (m.clone(), (am.tau, am.sites))) - .collect(); - let mut resp = epsilon_alloc::build_auto_plan( - req.epsilon, - &req.queries, - req.default_rate.unwrap_or(1.0), - |_m, sketches| epsilon_alloc::metric_rate_from_telemetry(&st.runtime_samples, sketches), - &monitors, - ); - - if req.apply { - use control_plane::workload::{derive_agg_role, MonitorDecl, WorkloadEntry}; - let mut applied = 0usize; - for m in &resp.metrics { - let metric = m.metric.clone(); - let sketch = m.sketches.first().cloned(); - let monitor = req.monitors.get(&metric).map(|am| MonitorDecl { - tau: am.tau, - functional: am.functional.clone(), - key: am.key.clone(), - epsilon: req.epsilon, - // Auto-derive the CDM window to the edge epoch unless pinned — - // a mismatch silently no-ops the grant (alignment guard). - window_secs: am - .window_secs - .unwrap_or_else(autonomous_monitor_window_secs), - }); - let entry = WorkloadEntry { - metric_name: metric.clone(), - query_string: None, - accuracy_sla: (1.0 - req.epsilon).clamp(0.0, 1.0), - assign_to_role: "agent".to_string(), - sketch_family_override: sketch.clone(), - target_path: None, - grouping_labels: Vec::new(), - // Coordinator derives the live p from the monitor ε; keep static 1.0. - sample_p: 1.0, - distinct_keys_per_window: None, - item_label: None, - monitor, - repeat_every: None, - }; - let role = derive_agg_role(&entry); - // (1) Inject the monitor unconditionally — only needs the metric name; - // monitor_intents() overlays it so the repost emits monitor(ε,τ) and - // the data-plane coordinator derives the live ε-floor p. - st.workload_registry.insert_runtime(entry); - // (2) Best-effort SKETCH registration: analyze a query for this metric - // and register the QueryWorkload (with the chosen sketch) so replan - // emits a fresh sketch plan. The pipeline analyzer accepts a narrower - // grammar than the planner's, so this is non-fatal on rejection. - for q in &req.queries { - let Ok(spec) = - serde_json::from_value::(json!({ "query_string": q })) - else { - continue; - }; - if let Ok(mut wl) = st.analyzer.analyze(spec) { - if wl.metric_name == metric { - wl.sketch_type_override = sketch.clone(); - st.workload_store.set( - &metric, - role, - wl, - control_plane::types::WorkloadCharacteristics::default(), - ); - break; - } - } - } - applied += 1; - } - // Push now: replan_all re-posts the cumulative backend config, which - // carries the overlaid monitor intents (the 60s repost ticker would - // otherwise pick them up). - st.replanner.replan_all().await; - resp.applied = Some(applied); - } - - (StatusCode::OK, Json(resp)).into_response() -} -/// Request body for `POST /api/v1/plan/pareto`. -#[derive(serde::Deserialize)] -struct ParetoRequest { - #[serde(flatten)] - spec: QuerySpec, - #[serde(default)] - weights: ObjectiveWeights, -} -/// Returns the Pareto frontier of collection plans for the given workload. -/// Each point is annotated with bandwidth, CPU, memory and accuracy objectives. -/// The caller can specify `weights` to get the frontier sorted by their -/// preferred trade-off. -async fn handle_pareto( - State(st): State, - Json(req): Json, -) -> impl IntoResponse { - let wc = req.spec.workload.clone(); - let workload = match st.analyzer.analyze(req.spec) { - Ok(w) => w, - Err(e) => return (StatusCode::UNPROCESSABLE_ENTITY, e.to_string()).into_response(), - }; - let frontier = pareto_frontier(&workload, &wc, req.weights, Some(&st.online_store)); - if frontier.is_empty() { - return ( - StatusCode::UNPROCESSABLE_ENTITY, - "no sketch meets the accuracy SLA for the given workload", - ) - .into_response(); - } - let best = select_best(&frontier, req.weights).map(|p| p.sketch_type.to_string()); - - let points: Vec = frontier - .iter() - .map(|p| { - json!({ - "sketch_type": p.sketch_type.to_string(), - "bandwidth_bytes_per_sec": p.bandwidth_bytes_per_sec, - "cpu_micros_per_sample": p.cpu_micros_per_sample, - "memory_bytes": p.memory_bytes, - "estimated_error": p.estimated_error, - }) - }) - .collect(); - - ( - StatusCode::OK, - Json(json!({ - "metric": workload.metric_name, - "frontier": points, - "best": best, - })), - ) - .into_response() -} async fn handle_get_plan( State(st): State, @@ -1620,108 +1395,8 @@ async fn handle_get_plan( .into_response() } -async fn handle_rollback( - State(st): State, - Path(metric): Path, -) -> impl IntoResponse { - // Reset the baseline so the next POST /api/v1/plan re-runs the cost - // model and establishes a fresh baseline plan for this metric. - // - // B2 (metric, role): rollback ALL roles for this metric. The - // response surfaces the per-role outcome so clients can see which - // roles had a previous plan and which were no-ops. Pre-B2 callers - // who fired a rollback on a metric got back `{rolled_back: true}` - // unconditionally for a single role; the new shape stays additive - // (carries `rolled_back: true` when ≥1 role rolled back). - st.planner.reset(&metric); - let plans = st.store.get_all_for_metric(&metric); - if plans.is_empty() { - return ( - StatusCode::BAD_REQUEST, - format!("plan not found for metric {metric:?}"), - ) - .into_response(); - } - let mut per_role = Vec::with_capacity(plans.len()); - let mut any_rolled_back = false; - for (role, _) in plans { - match st.store.rollback(&metric, role) { - Ok(plan) => { - if let Ok(yaml) = - generate_agent_collector_config(&plan.agent_config, &st.opamp_endpoint) - { - st.opamp - .push_to_role( - AgentRole::Agent, - RemoteConfig { - config_hash: short_hash(&yaml), - yaml, - }, - ) - .await; - } - per_role.push(json!({ "role": role.as_str(), "rolled_back": true })); - any_rolled_back = true; - } - Err(e) => { - per_role.push(json!({ - "role": role.as_str(), - "rolled_back": false, - "reason": e.to_string(), - })); - } - } - } - // Pre-B2 contract: return BAD_REQUEST when no role could roll - // back (e.g. every role's plan has no `previous` slot). The - // multi-role variants are surfaced in the `roles` array so - // callers can distinguish "rolled back N of K" cases. - let status = if any_rolled_back { - StatusCode::OK - } else { - StatusCode::BAD_REQUEST - }; - ( - status, - Json(json!({ - "metric": metric, - "rolled_back": any_rolled_back, - "roles": per_role, - })), - ) - .into_response() -} -async fn handle_agents(State(st): State) -> impl IntoResponse { - Json(st.opamp.connected_agents_with_roles().await) -} -/// Returns a complete OTel collector YAML for the named metric's current plan. -/// Collectors can use this with the HTTP config provider: -/// --config=http://control_plane:8080/api/v1/config/ -async fn handle_get_config( - State(st): State, - Path(metric): Path, -) -> impl IntoResponse { - // B2 (metric, role): the legacy `generate_agent_collector_config` - // emits a SINGLE-pipeline YAML — when a metric has multiple roles, - // pick the FIRST registered role's plan. The 5-sketch routing- - // connector emit path (the `USE_TYPED_STAGE_SPLIT` typed pipeline) - // is the supported multi-role wire shape; this endpoint stays - // legacy-compat by picking one role's plan. - let plan = st.store.get_all_for_metric(&metric).into_iter().next(); - let Some((_role, plan)) = plan else { - return ( - StatusCode::NOT_FOUND, - format!("plan not found for metric {metric:?}"), - ) - .into_response(); - }; - match generate_agent_collector_config(&plan.agent_config, &st.opamp_endpoint) { - Ok(yaml) => (StatusCode::OK, [("content-type", "application/yaml")], yaml).into_response(), - Err(e) => (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()).into_response(), - } -} /// Bootstrap YAML config for agent collectors. /// @@ -2041,70 +1716,6 @@ async fn emit_bootstrap_typed( .with_context(|| format!("emit_for_runtime failed for `{metric}`")) } -/// Returns the diff between the current and previous plan for `metric`. -/// 404 if the metric has no plan, 200 with `null` data if no previous plan exists. -async fn handle_plan_diff( - State(st): State, - Path(metric): Path, -) -> impl IntoResponse { - // B2 (metric, role): a metric may carry multiple roles; surface a - // per-role `roles` array. The top-level `has_diff` is true iff at - // least one role has a diff. Pre-B2 single-role clients see - // `has_diff` and the `diff` field of the first role with one; - // the new shape stays additive (no URL change, response keys - // preserved). - let plans = st.store.get_all_for_metric(&metric); - if plans.is_empty() { - return ( - StatusCode::NOT_FOUND, - format!("plan not found for metric {metric:?}"), - ) - .into_response(); - } - let mut roles: Vec = Vec::with_capacity(plans.len()); - let mut first_diff: Option = None; - let mut any_has_diff = false; - for (role, _) in plans { - match st.store.diff(&metric, role) { - Ok(Some(diff)) => { - let diff_json = serde_json::to_value(&diff).unwrap_or(serde_json::Value::Null); - if first_diff.is_none() { - first_diff = Some(diff_json.clone()); - } - any_has_diff = true; - roles.push(json!({ - "role": role.as_str(), - "has_diff": true, - "diff": diff_json, - })); - } - Ok(None) => { - roles.push(json!({ "role": role.as_str(), "has_diff": false })); - } - Err(e) => { - roles.push(json!({ - "role": role.as_str(), - "error": e.to_string(), - })); - } - } - } - let body = if any_has_diff { - json!({ - "metric": metric, - "has_diff": true, - "diff": first_diff, - "roles": roles, - }) - } else { - json!({ - "metric": metric, - "has_diff": false, - "roles": roles, - }) - }; - (StatusCode::OK, Json(body)).into_response() -} /// Returns the current EMA cost model state — blended benchmark + observed costs /// per sketch type. Useful for diagnosing whether the online cost model has @@ -2211,17 +1822,7 @@ fn test_app_with_backend(backend_url: Option) -> (AppState, axum::Router }; let router = axum::Router::new() .route("/api/v1/plan", axum::routing::post(handle_plan)) - .route("/api/v1/plan/pareto", axum::routing::post(handle_pareto)) .route("/api/v1/plan/:metric", axum::routing::get(handle_get_plan)) - .route( - "/api/v1/plan/:metric/rollback", - axum::routing::post(handle_rollback), - ) - .route( - "/api/v1/plan/:metric/diff", - axum::routing::get(handle_plan_diff), - ) - .route("/api/v1/agents", axum::routing::get(handle_agents)) .route("/api/v1/cost-model", axum::routing::get(handle_cost_model)) .route("/api/v1/tco", axum::routing::post(handle_tco)) .route( @@ -3243,17 +2844,7 @@ mod api_tests { // 5. Rebuild the router with the updated state. let router = axum::Router::new() .route("/api/v1/plan", axum::routing::post(handle_plan)) - .route("/api/v1/plan/pareto", axum::routing::post(handle_pareto)) .route("/api/v1/plan/:metric", axum::routing::get(handle_get_plan)) - .route( - "/api/v1/plan/:metric/rollback", - axum::routing::post(handle_rollback), - ) - .route( - "/api/v1/plan/:metric/diff", - axum::routing::get(handle_plan_diff), - ) - .route("/api/v1/agents", axum::routing::get(handle_agents)) .route("/api/v1/cost-model", axum::routing::get(handle_cost_model)) .route("/api/v1/tco", axum::routing::post(handle_tco)) .route( diff --git a/control_plane/src/physical/deployment_cost/mod.rs b/control_plane/src/physical/deployment_cost/mod.rs index 557ff4cf4..1f294fea9 100644 --- a/control_plane/src/physical/deployment_cost/mod.rs +++ b/control_plane/src/physical/deployment_cost/mod.rs @@ -4,7 +4,6 @@ use std::collections::HashMap; pub mod delta; pub mod online; -pub mod pareto; pub mod sketch_capability; pub mod tco; pub mod wire; diff --git a/control_plane/src/physical/deployment_cost/pareto.rs b/control_plane/src/physical/deployment_cost/pareto.rs deleted file mode 100644 index 29c2a0a7c..000000000 --- a/control_plane/src/physical/deployment_cost/pareto.rs +++ /dev/null @@ -1,418 +0,0 @@ -//! SP-6 Pareto frontier reporting for physical plans. -//! -//! The cost-model planner minimises bandwidth while meeting the accuracy SLA. -//! This module extends it by enumerating **all** Pareto-optimal plans across -//! three objectives — bandwidth, CPU and memory — so operators can choose the -//! trade-off that best fits their infrastructure. -//! -//! # Algorithm -//! -//! For each sketch candidate the planner generates one `ParetoPoint` (a plan -//! annotated with its scored objectives). A point is Pareto-optimal if no -//! other point is better on *every* objective simultaneously. The frontier is -//! returned sorted by weighted-sum score so that `select_best` returns the -//! single plan that best matches caller-supplied objective weights. -//! -//! # Usage -//! -//! ```ignore -//! let weights = ObjectiveWeights { bandwidth: 0.6, cpu: 0.3, memory: 0.1 }; -//! let frontier = pareto_frontier(&planner, &workload, &wc, weights); -//! let best = select_best(&frontier, weights); -//! ``` - -use std::collections::HashMap; - -use crate::physical::deployment_cost::delta::decide_delta; -use crate::physical::deployment_cost::online; -use crate::physical::deployment_cost::{benchmark_table_pub, score_with, SketchCosts}; -use crate::physical::workload_planner::{ - default_sketch_params, select_window_strategy, DeploymentPlanCompiler, -}; -use crate::types::*; - -// ── Types ───────────────────────────────────────────────────────────────────── - -/// Caller-supplied relative weights for the three objectives. -/// Values are normalised internally so they need not sum to 1. -#[derive(Debug, Clone, Copy, serde::Deserialize, serde::Serialize)] -pub struct ObjectiveWeights { - /// Weight given to minimising bandwidth (bytes/sec). - pub bandwidth: f64, - /// Weight given to minimising CPU (µs/sample). - pub cpu: f64, - /// Weight given to minimising memory (bytes). - pub memory: f64, -} - -impl Default for ObjectiveWeights { - fn default() -> Self { - Self { - bandwidth: 1.0, - cpu: 0.0, - memory: 0.0, - } - } -} - -impl ObjectiveWeights { - fn normalised(self) -> Self { - let total = self.bandwidth + self.cpu + self.memory; - if total <= 0.0 { - return Self { - bandwidth: 1.0, - cpu: 0.0, - memory: 0.0, - }; - } - Self { - bandwidth: self.bandwidth / total, - cpu: self.cpu / total, - memory: self.memory / total, - } - } -} - -/// A single candidate on the Pareto frontier. -#[derive(Debug, Clone)] -pub struct ParetoPoint { - pub plan: CollectionPlan, - pub sketch_type: SketchType, - pub bandwidth_bytes_per_sec: f64, - pub cpu_micros_per_sample: f64, - pub memory_bytes: f64, - pub estimated_error: f64, - pub meets_sla: bool, -} - -// ── Frontier computation ────────────────────────────────────────────────────── - -/// Enumerate all Pareto-optimal plans for the given workload. -/// -/// Only plans that meet the accuracy SLA are included. The returned slice is -/// sorted by weighted-sum score (lowest = best) according to `weights`. -/// -/// If `online_store` is `Some` the scoring uses EMA-blended costs; otherwise -/// it falls back to the static benchmark table. -pub fn pareto_frontier( - workload: &QueryWorkload, - wc: &WorkloadCharacteristics, - weights: ObjectiveWeights, - online_store: Option<&online::OnlineMetricsStore>, -) -> Vec { - if !matches!(workload.accuracy, crate::types::AccuracyTarget::Epsilon(epsilon) if epsilon > 0.0) - { - return Vec::new(); - } - let table: HashMap = match online_store { - Some(s) => online::effective_table(s), - None => benchmark_table_pub(), - }; - - let candidates = all_sketch_types(); - let rules = DeploymentPlanCompiler::new(); - let mut points: Vec = Vec::new(); - - for st in candidates { - let params = default_sketch_params(&st, workload.error_bound()); - let (mode, window_duration) = select_window_strategy(workload); - - let mut plan = rules.plan(workload); - plan.agent_config.sketch_type = st.clone(); - plan.agent_config.sketch_params = params; - plan.agent_config.mode = mode; - plan.agent_config.window_duration = window_duration; - - // Apply delta decision using the cost table. - apply_delta(st.clone(), &mut plan, workload, wc, &table); - - let s = score_with(&plan, workload, &table); - if !s.meets_sla { - continue; - } - - points.push(ParetoPoint { - sketch_type: st, - bandwidth_bytes_per_sec: s.bandwidth_bytes_per_sec, - cpu_micros_per_sample: s.cpu_micros_per_sample, - memory_bytes: s.memory_bytes, - estimated_error: s.estimated_error, - meets_sla: s.meets_sla, - plan, - }); - } - - // Filter to Pareto-optimal subset. - let optimal = pareto_filter(&points); - - // Sort by weighted score. - let w = weights.normalised(); - let mut sorted = optimal; - sorted.sort_by(|a, b| { - weighted_score(a, w) - .partial_cmp(&weighted_score(b, w)) - .unwrap_or(std::cmp::Ordering::Equal) - }); - sorted -} - -/// Returns the single plan from `frontier` with the lowest weighted-sum score. -/// Returns `None` if the frontier is empty. -pub fn select_best(frontier: &[ParetoPoint], weights: ObjectiveWeights) -> Option<&ParetoPoint> { - let w = weights.normalised(); - frontier.iter().min_by(|a, b| { - weighted_score(a, w) - .partial_cmp(&weighted_score(b, w)) - .unwrap_or(std::cmp::Ordering::Equal) - }) -} - -// ── Private helpers ─────────────────────────────────────────────────────────── - -fn all_sketch_types() -> Vec { - vec![ - SketchType::DDSketch, - SketchType::KLL, - SketchType::HLL, - SketchType::CountSketch, - SketchType::CountMinSketch, - ] -} - -fn apply_delta( - st: SketchType, - plan: &mut CollectionPlan, - w: &QueryWorkload, - wc: &WorkloadCharacteristics, - table: &HashMap, -) { - let bytes_per_series_per_sec = table - .get(&st) - .map(|c| c.bytes_per_series_per_sec) - .unwrap_or(200.0); - - let (decision, summary) = decide_delta(plan, w, wc, bytes_per_series_per_sec); - - match &decision { - DeltaDecision::UseDelta { threshold, .. } => { - plan.agent_config.delta_transmission = true; - plan.agent_config.delta_threshold = *threshold; - plan.agent_config.gos = (plan.agent_config.sketch_type == SketchType::CountSketch) - .then(|| GosKnobs::derive(w.error_bound(), 1, 0.0, 1.0, false)); - } - _ => { - plan.agent_config.delta_transmission = false; - plan.agent_config.delta_threshold = 0.0; - plan.agent_config.gos = None; - } - } - plan.delta_decision = decision; - plan.transmission_cost_summary = summary; -} - -fn pareto_filter(points: &[ParetoPoint]) -> Vec { - points - .iter() - .filter(|p| { - !points.iter().any(|q| { - q.bandwidth_bytes_per_sec <= p.bandwidth_bytes_per_sec - && q.cpu_micros_per_sample <= p.cpu_micros_per_sample - && q.memory_bytes <= p.memory_bytes - && (q.bandwidth_bytes_per_sec < p.bandwidth_bytes_per_sec - || q.cpu_micros_per_sample < p.cpu_micros_per_sample - || q.memory_bytes < p.memory_bytes) - }) - }) - .cloned() - .collect() -} - -fn weighted_score(p: &ParetoPoint, w: ObjectiveWeights) -> f64 { - w.bandwidth * p.bandwidth_bytes_per_sec - + w.cpu * p.cpu_micros_per_sample - + w.memory * p.memory_bytes -} - -// ── Tests ───────────────────────────────────────────────────────────────────── - -#[cfg(test)] -mod tests { - use super::*; - use std::collections::HashMap; - use std::time::Duration; - - fn quantile_workload() -> QueryWorkload { - QueryWorkload { - metric_name: "latency".into(), - label_filters: HashMap::new(), - group_by_labels: vec![], - aggregations: vec![AggType::Quantile], - time_window: Duration::from_secs(300), - repeat_every: None, - accuracy_sla: 0.02, - accuracy: crate::types::AccuracyTarget::Epsilon(0.02), - latency_sla: None, - sketch_type_override: None, - exact_required: false, - quantiles: vec![], - } - } - - fn default_wc() -> WorkloadCharacteristics { - WorkloadCharacteristics::default() - } - - #[test] - fn quantile_frontier_non_empty_and_kll_present() { - // KLL has lower BW, CPU, and memory than DDSketch, so DDSketch is - // dominated and excluded. KLL and HLL form the frontier (HLL has - // lower BW/CPU but higher memory than KLL). - let f = pareto_frontier( - &quantile_workload(), - &default_wc(), - ObjectiveWeights::default(), - None, - ); - assert!( - !f.is_empty(), - "frontier should not be empty for quantile workload" - ); - let types: Vec<_> = f.iter().map(|p| &p.sketch_type).collect(); - assert!( - types.contains(&&SketchType::KLL), - "KLL expected in frontier" - ); - } - - #[test] - fn all_frontier_points_meet_sla() { - let f = pareto_frontier( - &quantile_workload(), - &default_wc(), - ObjectiveWeights::default(), - None, - ); - for p in &f { - assert!( - p.meets_sla, - "{} does not meet SLA (error={})", - p.sketch_type, p.estimated_error - ); - } - } - - #[test] - fn bandwidth_weight_selects_lowest_bw() { - let f = pareto_frontier( - &quantile_workload(), - &default_wc(), - ObjectiveWeights { - bandwidth: 1.0, - cpu: 0.0, - memory: 0.0, - }, - None, - ); - let best = select_best( - &f, - ObjectiveWeights { - bandwidth: 1.0, - cpu: 0.0, - memory: 0.0, - }, - ) - .unwrap(); - for p in &f { - assert!( - p.bandwidth_bytes_per_sec >= best.bandwidth_bytes_per_sec, - "best ({}) should have minimum bandwidth", - best.sketch_type - ); - } - } - - #[test] - fn memory_weight_selects_lowest_memory() { - let f = pareto_frontier( - &quantile_workload(), - &default_wc(), - ObjectiveWeights { - bandwidth: 0.0, - cpu: 0.0, - memory: 1.0, - }, - None, - ); - let best = select_best( - &f, - ObjectiveWeights { - bandwidth: 0.0, - cpu: 0.0, - memory: 1.0, - }, - ) - .unwrap(); - for p in &f { - assert!( - p.memory_bytes >= best.memory_bytes, - "best ({}) should have minimum memory", - best.sketch_type - ); - } - } - - #[test] - fn select_best_empty_returns_none() { - let result = select_best(&[], ObjectiveWeights::default()); - assert!(result.is_none()); - } - - #[test] - fn tight_sla_excludes_inaccurate_sketches() { - let w = QueryWorkload { - accuracy_sla: 0.001, // very tight — only DDSketch at 0.1% accuracy can meet this - accuracy: crate::types::AccuracyTarget::Epsilon(0.001), - ..quantile_workload() - }; - let f = pareto_frontier(&w, &default_wc(), ObjectiveWeights::default(), None); - for p in &f { - assert!( - p.estimated_error <= 0.001, - "{} error {} exceeds 0.001 SLA", - p.sketch_type, - p.estimated_error - ); - } - } - - #[test] - fn pareto_no_dominated_points() { - let f = pareto_frontier( - &quantile_workload(), - &default_wc(), - ObjectiveWeights::default(), - None, - ); - // Verify no point in f is dominated by another. - for i in 0..f.len() { - for j in 0..f.len() { - if i == j { - continue; - } - let a = &f[i]; - let b = &f[j]; - let b_dominates_a = b.bandwidth_bytes_per_sec <= a.bandwidth_bytes_per_sec - && b.cpu_micros_per_sample <= a.cpu_micros_per_sample - && b.memory_bytes <= a.memory_bytes - && (b.bandwidth_bytes_per_sec < a.bandwidth_bytes_per_sec - || b.cpu_micros_per_sample < a.cpu_micros_per_sample - || b.memory_bytes < a.memory_bytes); - assert!( - !b_dominates_a, - "{} dominates {} but both are in frontier", - b.sketch_type, a.sketch_type - ); - } - } - } -} diff --git a/control_plane/src/query_planning.rs b/control_plane/src/query_planning.rs deleted file mode 100644 index e99512012..000000000 --- a/control_plane/src/query_planning.rs +++ /dev/null @@ -1,328 +0,0 @@ -//! Autonomous allocation, slice 2: a **set of query strings → per-metric sketch -//! plan**. -//! -//! This composes two existing pieces with slice 1's inverse map: -//! 1. ASAPPlanner lowers each query to its canonical post-ASAP summary DAG. -//! 2. [`required_sketches_for_capabilities`](crate::sketch_selection::required_sketches_for_capabilities) -//! (slice 1) maps a capability set to the `SketchType` families that satisfy it. -//! -//! The result, [`QuerySetPlan`], is the warm-tier half of autonomous -//! allocation: "given these queries, allocate these sketches per metric", plus -//! the list of queries that fall through to the cold tier. Slice 3 layers the -//! `ε → (p, τ)` knobs onto each metric; slice 4 emits a `StreamingConfig`. - -use std::collections::BTreeMap; - -use crate::physical::runtime_capability::Capability; -use crate::sketch_selection::required_sketches_for_capabilities; -use crate::types::SketchType; - -/// The sketch allocation derived for one metric across the whole query set. -#[derive(Debug, Clone, PartialEq)] -pub struct MetricSketchPlan { - pub metric_name: String, - /// The distinct warm-tier capabilities the query set demands of this - /// metric (order-stable, first-seen wins). - pub capabilities: Vec, - /// The sketch families to allocate so every demanded capability is - /// satisfied (de-duplicated union over `capabilities`). - pub sketches: Vec, -} - -/// A query (or query fragment) that the warm sketch tier cannot answer and that -/// must be routed to the cold tier. -#[derive(Debug, Clone, PartialEq)] -pub struct ColdQuery { - pub query: String, - /// Why it isn't ASAP-tier-answerable (`None` only in the degenerate case - /// where the analyzer returned no candidates and no explicit reason). - pub reason: Option, -} - -/// The full allocation plan for a query set: what to allocate warm, and what -/// falls through to cold. -#[derive(Debug, Clone, PartialEq, Default)] -pub struct QuerySetPlan { - /// Per-metric warm sketch allocation, keyed/ordered by metric name. - pub per_metric: Vec, - /// Queries (or partially-supported queries) needing the cold tier. - pub cold_only: Vec, -} - -impl QuerySetPlan { - #[cfg(test)] - /// All sketch families this plan allocates, across every metric - /// (de-duplicated, order-stable). Convenience for capacity sizing. - pub fn all_sketches(&self) -> Vec { - let mut out: Vec = Vec::new(); - for m in &self.per_metric { - for s in &m.sketches { - if !out.contains(s) { - out.push(s.clone()); - } - } - } - out - } -} - -/// Plan the warm sketch allocation for a set of queries. -/// -/// Each query is lowered via the ASAP-tier analyzer; every answerable candidate -/// contributes its `required_capability` to its metric's bucket, and any -/// unsupported reason (or a fully-unsupported query) is recorded under -/// `cold_only`. Per metric, the accumulated capabilities are unioned into the -/// sketch families to allocate. -/// -/// A single query can land in BOTH halves — e.g. `sum(quantile_over_time(...))` -/// yields a warm quantile candidate (→ DDSketch/KLL) while an exact-`sum` -/// fragment routes cold. That partial-support split is intentional. -pub fn plan_sketches_for_queries(queries: I) -> QuerySetPlan -where - I: IntoIterator, - S: AsRef, -{ - // BTreeMap → deterministic metric ordering regardless of query order. - let mut by_metric: BTreeMap> = BTreeMap::new(); - let mut cold_only: Vec = Vec::new(); - - for q in queries { - let q = q.as_ref(); - let planned = crate::asap_tier_implement::implement_promql_for_asap_tier(q); - let mut found = false; - if let Ok(nodes) = &planned { - for node in nodes { - if let Some((metric, capability)) = planned_capability(node) { - found = true; - let caps = by_metric.entry(metric).or_default(); - if !caps.contains(&capability) { - caps.push(capability); - } - } - } - } - if !found { - cold_only.push(ColdQuery { - query: q.to_string(), - reason: planned.err().map(|error| format!("{error:?}")), - }); - } - } - - let per_metric = by_metric - .into_iter() - .map(|(metric_name, capabilities)| { - let sketches = required_sketches_for_capabilities(&capabilities); - MetricSketchPlan { - metric_name, - capabilities, - sketches, - } - }) - .collect(); - - QuerySetPlan { - per_metric, - cold_only, - } -} - -fn planned_capability( - node: &planner_types::post_asap::SummaryNode, -) -> Option<(String, Capability)> { - use crate::physical::runtime_capability::SketchAlgorithm; - use planner_types::post_asap::{SketchQuery, SummaryExpr, SummaryFamilyType}; - - fn metric(node: &planner_types::post_asap::SummaryNode) -> Option { - fn from_query(expr: &planner_types::pre_asap::QueryExpr) -> Option { - use planner_types::pre_asap::{QueryExpr, Source}; - match expr { - QueryExpr::Scan { - source: Source::TimeSeries { metric }, - .. - } => Some(metric.clone()), - QueryExpr::Filter { child, .. } - | QueryExpr::Project { child, .. } - | QueryExpr::Aggregate { child, .. } - | QueryExpr::Dedup { child, .. } - | QueryExpr::Sort { child, .. } - | QueryExpr::Limit { child, .. } - | QueryExpr::PromqlSubquery { child, .. } - | QueryExpr::TimeRange { child, .. } - | QueryExpr::TimeShift { child, .. } - | QueryExpr::SQLWindowFunc { child, .. } => from_query(child), - _ => None, - } - } - match &node.expr { - SummaryExpr::KeepPreAsap(expr) => from_query(expr), - SummaryExpr::SummaryAgg { child, .. } => metric(child), - SummaryExpr::SummaryEstimate { summary_input, .. } => metric(summary_input), - SummaryExpr::SummaryMerge { children } => { - children.iter().find_map(|child| metric(child)) - } - _ => None, - } - } - - fn handle(family: &SummaryFamilyType) -> Option { - let SummaryFamilyType::Sketch(kind, _) = family else { - return None; - }; - Some(match kind.algorithm() { - planner_types::post_asap::SketchAlgorithm::UnivMon => return None, - planner_types::post_asap::SketchAlgorithm::DDSketch => SketchAlgorithm::DDSketch, - planner_types::post_asap::SketchAlgorithm::Kll => SketchAlgorithm::Kll, - planner_types::post_asap::SketchAlgorithm::Hll => SketchAlgorithm::Hll, - planner_types::post_asap::SketchAlgorithm::Cms => SketchAlgorithm::Cms, - planner_types::post_asap::SketchAlgorithm::CmsWithHeap => SketchAlgorithm::CmsWithHeap, - planner_types::post_asap::SketchAlgorithm::CountSketch => SketchAlgorithm::CountSketch, - planner_types::post_asap::SketchAlgorithm::CountSketchWithHeap => { - SketchAlgorithm::CountSketchWithHeap - } - planner_types::post_asap::SketchAlgorithm::Kmv - | planner_types::post_asap::SketchAlgorithm::Theta => return None, - }) - } - - let metric = metric(node)?; - let capability = match &node.expr { - SummaryExpr::SummaryEstimate { - summary_input, - query, - } => { - let SummaryExpr::SummaryAgg { family, .. } = &summary_input.expr else { - return None; - }; - match query { - SketchQuery::FrequencyL2 | SketchQuery::FrequencyEntropy => return None, - SketchQuery::Quantile { .. } => Capability::QuantileApprox(Some(handle(family)?)), - SketchQuery::Cardinality => Capability::CardinalityApprox, - SketchQuery::PointCount { .. } => { - Capability::FrequencyEstimate(Some(handle(family)?)) - } - SketchQuery::TopK { .. } => Capability::FrequencyTopk(Some(handle(family)?)), - } - } - SummaryExpr::SummaryAgg { - family: SummaryFamilyType::ExactAggregate(kind, _), - .. - } => { - let agg = match kind { - planner_types::post_asap::ExactKind::Sum - | planner_types::post_asap::ExactKind::Count => asap_types::AggregationType::Sum, - planner_types::post_asap::ExactKind::Increase - | planner_types::post_asap::ExactKind::Rate - | planner_types::post_asap::ExactKind::IRate => { - asap_types::AggregationType::Increase - } - planner_types::post_asap::ExactKind::MinMax => asap_types::AggregationType::MinMax, - }; - Capability::ExactAgg(agg) - } - _ => return None, - }; - Some((metric, capability)) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn quantile_query_allocates_a_quantile_sketch() { - let plan = plan_sketches_for_queries([ - "quantile_over_time(0.99, http_requests_total_latency_ms[5m])", - ]); - let m = plan - .per_metric - .iter() - .find(|m| m.metric_name == "http_requests_total_latency_ms") - .expect("metric present"); - // a quantile demands a quantile-family sketch - assert!( - m.sketches.contains(&SketchType::DDSketch) || m.sketches.contains(&SketchType::KLL) - ); - } - - #[test] - fn distinct_metrics_get_separate_plans() { - let plan = plan_sketches_for_queries([ - "quantile_over_time(0.99, latency_ms[5m])", - "count(http_requests_total)", - ]); - // at least the quantile metric must appear with a quantile sketch - let latency = plan - .per_metric - .iter() - .find(|m| m.metric_name == "latency_ms"); - assert!(latency.is_some(), "latency_ms should be planned: {plan:?}"); - } - - #[test] - fn unparseable_query_falls_through_to_cold() { - let plan = plan_sketches_for_queries(["this is not valid promql @@@"]); - assert!(plan.per_metric.is_empty()); - assert_eq!(plan.cold_only.len(), 1); - } - - #[test] - fn same_metric_two_queries_unions_capabilities_without_dup() { - // two quantile queries on the same metric → one metric plan, the - // quantile capability not duplicated. - let plan = plan_sketches_for_queries([ - "quantile_over_time(0.99, m_latency[5m])", - "quantile_over_time(0.50, m_latency[5m])", - ]); - let m = plan - .per_metric - .iter() - .find(|m| m.metric_name == "m_latency") - .expect("metric present"); - // capabilities are de-duplicated (both queries need the same quantile cap) - assert_eq!( - m.capabilities.len(), - 1, - "caps deduped: {:?}", - m.capabilities - ); - } - - #[test] - fn canonical_e2e_query_suite_produces_a_plan() { - // The real queries-e2e.json shapes: nested aggregations, topk, count. - // We don't pin each classification (that's the analyzer's contract) — - // we assert the planner survives them and produces a sensible plan: - // the quantile query must yield a quantile-family sketch somewhere. - let plan = plan_sketches_for_queries([ - "quantile_over_time(0.99, http_requests_total_latency_ms[5m])", - "max by (zone) (quantile_over_time(0.99, http_requests_total_latency_ms[5m]))", - "sum by (zone) (http_requests_total)", - "sum by (zone) (rate(http_requests_total[5m]))", - "topk(5, sum by (zone) (rate(http_requests_total[5m])))", - "count(http_requests_total{zone=\"z0\"})", - ]); - // every allocated sketch must trace back to a demanded capability - for m in &plan.per_metric { - assert!( - !m.capabilities.is_empty(), - "metric {} has sketches but no capabilities", - m.metric_name - ); - } - // the quantile query guarantees a quantile-family sketch is planned - let all = plan.all_sketches(); - assert!( - all.contains(&SketchType::DDSketch) || all.contains(&SketchType::KLL), - "quantile query should allocate a quantile sketch; got {all:?} / {plan:?}" - ); - } - - #[test] - fn empty_query_set_yields_empty_plan() { - let plan = plan_sketches_for_queries(Vec::<&str>::new()); - assert!(plan.per_metric.is_empty()); - assert!(plan.cold_only.is_empty()); - assert!(plan.all_sketches().is_empty()); - } -} diff --git a/control_plane/src/sketch_selection.rs b/control_plane/src/sketch_selection.rs deleted file mode 100644 index 75ea2353d..000000000 --- a/control_plane/src/sketch_selection.rs +++ /dev/null @@ -1,169 +0,0 @@ -//! Inverse of [`capability_for`](crate::physical::runtime_capability::capability_for): -//! given a required [`Capability`], enumerate the concrete [`SketchType`] -//! families that can satisfy it. -//! -//! This is the keystone of autonomous allocation. `capability_for` lowers an -//! `AggIntent` (parsed from a query) to the *capability* it needs; this module -//! closes the loop by naming the *sketches* that provide that capability, so a -//! planner handed a set of queries can decide which sketches to allocate -//! without a hand-written workload YAML. -//! -//! Policy encoded here (matches `Capability::is_satisfied_by` on the matching -//! side, `physical::runtime_capability`): -//! * `QuantileApprox` → DDSketch | KLL -//! * `CardinalityApprox`→ HLL -//! * `FrequencyEstimate`→ CountSketch | CountMinSketch (heap-less point query) -//! * `FrequencyTopk` → CountSketch | CountMinSketch (heap layered on the same matrix) -//! * `ExactAgg` → ∅ (served by the exact-aggregation / archive path, not a sketch) -//! -//! When a capability binds a *concrete* `SketchAlgorithm` (not `Any`), the -//! result is exactly that one family; `Any` expands to the full candidate set. -//! -//! Moved out of `physical::post_asap` (Stage 4 of the `physical::post_asap` -//! re-layering) — this is a query-planning concern (its one caller is -//! [`crate::query_planning`]), not L4 IR. - -use crate::physical::runtime_capability::{Capability, SketchAlgorithm}; -use crate::types::SketchType; - -/// Map a sketch handle to the allocatable control-plane [`SketchType`]. -/// -/// Heap-bearing dispatch hints collapse to their underlying matrix family -/// (`CmsWithHeap` → CountMinSketch, `CountSketchWithHeap` → CountSketch) — the -/// heap is an allocation detail layered on the same sketch, not a distinct -/// allocatable family. `Any` is an analysis-time wildcard with no single -/// concrete family, so it returns `None`. -pub fn sketch_type_for_algorithm(h: SketchAlgorithm) -> Option { - match h { - SketchAlgorithm::DDSketch => Some(SketchType::DDSketch), - SketchAlgorithm::Kll => Some(SketchType::KLL), - SketchAlgorithm::Hll => Some(SketchType::HLL), - SketchAlgorithm::CountSketch | SketchAlgorithm::CountSketchWithHeap => { - Some(SketchType::CountSketch) - } - SketchAlgorithm::Cms | SketchAlgorithm::CmsWithHeap => Some(SketchType::CountMinSketch), - SketchAlgorithm::UnivMon | SketchAlgorithm::Kmv | SketchAlgorithm::Theta => None, - } -} - -/// The sketch families that satisfy a required [`Capability`]. -/// -/// A concrete handle pins exactly one family; `Any` expands to the capability -/// class's full candidate set. `ExactAgg` returns an empty vec — it is served -/// by the exact-aggregation / archive path, so "allocate a sketch for it" is a -/// no-op (the caller routes it to cold/exact instead). -pub fn sketch_families_for_capability(cap: &Capability) -> Vec { - match cap { - Capability::QuantileApprox(h) => concrete_or(h, &[SketchType::DDSketch, SketchType::KLL]), - Capability::CardinalityApprox => vec![SketchType::HLL], - Capability::FrequencyEstimate(h) | Capability::FrequencyTopk(h) => { - concrete_or(h, &[SketchType::CountSketch, SketchType::CountMinSketch]) - } - Capability::ExactAgg(_) => Vec::new(), - } -} - -/// Union the sketch families required by a set of capabilities, de-duplicated -/// and order-stable (first appearance wins). This is what a planner calls -/// after lowering a query set to capabilities: the result is the set of -/// sketches that, allocated together, answer every query in the set. -pub fn required_sketches_for_capabilities<'a, I>(caps: I) -> Vec -where - I: IntoIterator, -{ - let mut out: Vec = Vec::new(); - for cap in caps { - for fam in sketch_families_for_capability(cap) { - if !out.contains(&fam) { - out.push(fam); - } - } - } - out -} - -fn concrete_or(h: &Option, any_set: &[SketchType]) -> Vec { - match h.clone().and_then(sketch_type_for_algorithm) { - Some(t) => vec![t], - None => any_set.to_vec(), - } -} - -#[cfg(test)] -mod tests { - use super::*; - use asap_types::AggregationType; - - #[test] - fn quantile_any_expands_to_ddsketch_and_kll() { - let fams = sketch_families_for_capability(&Capability::QuantileApprox(None)); - assert_eq!(fams, vec![SketchType::DDSketch, SketchType::KLL]); - } - - #[test] - fn quantile_concrete_handle_pins_one_family() { - assert_eq!( - sketch_families_for_capability(&Capability::QuantileApprox(Some( - SketchAlgorithm::DDSketch - ))), - vec![SketchType::DDSketch] - ); - assert_eq!( - sketch_families_for_capability(&Capability::QuantileApprox(Some(SketchAlgorithm::Kll))), - vec![SketchType::KLL] - ); - } - - #[test] - fn cardinality_is_hll() { - assert_eq!( - sketch_families_for_capability(&Capability::CardinalityApprox), - vec![SketchType::HLL] - ); - } - - #[test] - fn frequency_families_and_heap_collapse() { - // bare frequency: Any -> both matrix families - assert_eq!( - sketch_families_for_capability(&Capability::FrequencyEstimate(None)), - vec![SketchType::CountSketch, SketchType::CountMinSketch] - ); - // heap-bearing handles collapse to their matrix family - assert_eq!( - sketch_families_for_capability(&Capability::FrequencyTopk(Some( - SketchAlgorithm::CmsWithHeap - ))), - vec![SketchType::CountMinSketch] - ); - assert_eq!( - sketch_families_for_capability(&Capability::FrequencyTopk(Some( - SketchAlgorithm::CountSketchWithHeap - ))), - vec![SketchType::CountSketch] - ); - } - - #[test] - fn exact_agg_allocates_no_sketch() { - assert!( - sketch_families_for_capability(&Capability::ExactAgg(AggregationType::Sum)).is_empty() - ); - } - - #[test] - fn union_dedups_and_preserves_order() { - // a query set needing {quantile, cardinality, quantile-again} -> - // DDSketch, KLL, HLL with no duplicate DDSketch/KLL. - let caps = vec![ - Capability::QuantileApprox(None), - Capability::CardinalityApprox, - Capability::QuantileApprox(Some(SketchAlgorithm::DDSketch)), - ]; - let got = required_sketches_for_capabilities(&caps); - assert_eq!( - got, - vec![SketchType::DDSketch, SketchType::KLL, SketchType::HLL] - ); - } -} From b91a9aafe07ef4fa7284eec44ca4b6630ad47cf8 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 15:04:12 -0600 Subject: [PATCH 2/5] style: rustfmt --- control_plane/src/epsilon_alloc.rs | 19 ------------------- control_plane/src/main.rs | 10 ---------- 2 files changed, 29 deletions(-) diff --git a/control_plane/src/epsilon_alloc.rs b/control_plane/src/epsilon_alloc.rs index 308db7c95..7f322fabd 100644 --- a/control_plane/src/epsilon_alloc.rs +++ b/control_plane/src/epsilon_alloc.rs @@ -19,9 +19,6 @@ //! control_plane has no dependency on data_plane, so the canonical one-liner is //! re-stated here — keep the two in sync.) - - - /// Admission sampling probability under the unified ε-floor law, /// `p = 1/(1 + ε²·rate)`. /// @@ -35,7 +32,6 @@ pub fn derive_sample_p(epsilon: f64, rate: f64) -> f64 { 1.0 / (1.0 + epsilon * epsilon * rate) } - /// GOS budget split (design §7, Layer B) — **exact linear peel**. /// /// The staleness term `ε_st` is a *deterministic worst-case* bound, so it adds @@ -74,15 +70,6 @@ pub fn split_budget(eps_total: f64, eps_sketch: f64, w_edge: f64, w_comm: f64) - (eps_sa, eps_st) } - - - - - - - - - #[cfg(test)] mod tests { use super::*; @@ -127,10 +114,4 @@ mod tests { assert!(derive_sample_p(0.05, 100.0) > derive_sample_p(0.05, 10_000.0)); assert!(derive_sample_p(0.01, 1000.0) > derive_sample_p(0.10, 1000.0)); } - - - - - - } diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 20b864a67..a6ffdfb7b 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -1348,12 +1348,6 @@ async fn handle_plan(State(st): State, Json(spec): Json) -> .into_response() } - - - - - - async fn handle_get_plan( State(st): State, Path(metric): Path, @@ -1395,9 +1389,6 @@ async fn handle_get_plan( .into_response() } - - - /// Bootstrap YAML config for agent collectors. /// /// Collectors start with: @@ -1716,7 +1707,6 @@ async fn emit_bootstrap_typed( .with_context(|| format!("emit_for_runtime failed for `{metric}`")) } - /// Returns the current EMA cost model state — blended benchmark + observed costs /// per sketch type. Useful for diagnosing whether the online cost model has /// received sufficient observations to meaningfully influence plan selection. From 6d35d85e82e0f6b0e587e7f4a48df8293c5b2cba Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 15:13:06 -0600 Subject: [PATCH 3/5] style: drop the doc comment orphaned by the removed module --- control_plane/src/lib.rs | 3 --- 1 file changed, 3 deletions(-) diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs index da3261d7e..e93d3c0f4 100644 --- a/control_plane/src/lib.rs +++ b/control_plane/src/lib.rs @@ -35,9 +35,6 @@ pub mod store; pub mod types; pub mod workload; -/// PromQL → ASAPPlanner's canonical post-ASAP plan via -/// `asap_aware_mapping::bind::implement_tree`. - /// Crate-wide test-only utilities for serialising access to process-global /// state (environment variables). /// From 5c053918bd0c3af85c7c2c449879f027d19d1457 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 16:00:49 -0600 Subject: [PATCH 4/5] test: drop the tests for the removed endpoints --- control_plane/src/main.rs | 108 -------------------------------------- 1 file changed, 108 deletions(-) diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index a6ffdfb7b..02cd4b017 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -2129,85 +2129,11 @@ mod api_tests { // ── POST /api/v1/plan/:metric/rollback ──────────────────────────────────── - #[tokio::test] - async fn rollback_no_previous_returns_400() { - let (st, app) = test_app(); - // Seed one plan directly. - use crate::physical::workload_planner::DeploymentPlanCompiler; - let wl = crate::types::QueryWorkload { - metric_name: "m".into(), - label_filters: std::collections::HashMap::new(), - group_by_labels: vec![], - aggregations: vec![crate::types::AggType::Quantile], - time_window: std::time::Duration::from_secs(300), - repeat_every: None, - accuracy_sla: 0.01, - accuracy: crate::types::AccuracyTarget::Epsilon(0.01), - latency_sla: None, - sketch_type_override: None, - exact_required: false, - quantiles: vec![], - }; - st.store.set( - "m", - control_plane::workload::AggRole::Quantile, - DeploymentPlanCompiler::new().plan(&wl), - ); - let req = Request::builder() - .method("POST") - .uri("/api/v1/plan/m/rollback") - .body(Body::empty()) - .unwrap(); - let resp = app.oneshot(req).await.unwrap(); - assert_eq!(resp.status(), StatusCode::BAD_REQUEST); - } - #[tokio::test] - async fn rollback_not_found_returns_400() { - let (_, app) = test_app(); - let req = Request::builder() - .method("POST") - .uri("/api/v1/plan/ghost/rollback") - .body(Body::empty()) - .unwrap(); - let resp = app.oneshot(req).await.unwrap(); - assert_eq!(resp.status(), StatusCode::BAD_REQUEST); - } // ── GET /api/v1/plan/:metric/diff ───────────────────────────────────────── - #[tokio::test] - async fn diff_no_previous_returns_has_diff_false() { - let (_, app) = test_app(); - // POST a plan once. - let req = Request::builder() - .method("POST") - .uri("/api/v1/plan") - .header("content-type", "application/json") - .body(Body::from(plan_spec("rtt").to_string())) - .unwrap(); - app.clone().oneshot(req).await.unwrap(); - // Diff should exist but has_diff=false (only one version). - let req = Request::builder() - .uri("/api/v1/plan/rtt/diff") - .body(Body::empty()) - .unwrap(); - let resp = app.oneshot(req).await.unwrap(); - assert_eq!(resp.status(), StatusCode::OK); - let body = body_json(resp).await; - assert_eq!(body["has_diff"], false); - } - #[tokio::test] - async fn diff_not_found_returns_404() { - let (_, app) = test_app(); - let req = Request::builder() - .uri("/api/v1/plan/ghost/diff") - .body(Body::empty()) - .unwrap(); - let resp = app.oneshot(req).await.unwrap(); - assert_eq!(resp.status(), StatusCode::NOT_FOUND); - } // ── GET /api/v1/cost-model ──────────────────────────────────────────────── @@ -2231,43 +2157,9 @@ mod api_tests { // ── POST /api/v1/plan/pareto ────────────────────────────────────────────── - #[tokio::test] - async fn pareto_returns_frontier_for_quantile() { - let (_, app) = test_app(); - let body = serde_json::json!({ - "metric_name": "latency", "aggregations": ["quantile"], - "time_window": "5m", "accuracy_sla": 0.02, - "weights": { "bandwidth": 0.7, "cpu": 0.2, "memory": 0.1 } - }); - let req = Request::builder() - .method("POST") - .uri("/api/v1/plan/pareto") - .header("content-type", "application/json") - .body(Body::from(body.to_string())) - .unwrap(); - let resp = app.oneshot(req).await.unwrap(); - assert_eq!(resp.status(), StatusCode::OK); - let body = body_json(resp).await; - let frontier = body["frontier"].as_array().unwrap(); - assert!(!frontier.is_empty(), "frontier should not be empty"); - assert!(body["best"].as_str().is_some(), "best sketch should be set"); - } // ── GET /api/v1/agents ──────────────────────────────────────────────────── - #[tokio::test] - async fn agents_returns_empty_map_initially() { - let (_, app) = test_app(); - let req = Request::builder() - .uri("/api/v1/agents") - .body(Body::empty()) - .unwrap(); - let resp = app.oneshot(req).await.unwrap(); - assert_eq!(resp.status(), StatusCode::OK); - let body = body_json(resp).await; - // No agents connected → empty object. - assert_eq!(body, serde_json::json!({})); - } // ── POST /api/v1/tco ───────────────────────────────────────────────────── From 7588fbb0f81bbe572da8ef69bf203911c5561e04 Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 13 Sep 2026 16:15:20 -0600 Subject: [PATCH 5/5] style: drop the section headers for the removed endpoints --- control_plane/src/main.rs | 14 -------------- 1 file changed, 14 deletions(-) diff --git a/control_plane/src/main.rs b/control_plane/src/main.rs index 02cd4b017..7e8833e64 100644 --- a/control_plane/src/main.rs +++ b/control_plane/src/main.rs @@ -2127,14 +2127,6 @@ mod api_tests { assert_eq!(body["metric"], "cpu"); } - // ── POST /api/v1/plan/:metric/rollback ──────────────────────────────────── - - - - // ── GET /api/v1/plan/:metric/diff ───────────────────────────────────────── - - - // ── GET /api/v1/cost-model ──────────────────────────────────────────────── #[tokio::test] @@ -2155,12 +2147,6 @@ mod api_tests { } } - // ── POST /api/v1/plan/pareto ────────────────────────────────────────────── - - - // ── GET /api/v1/agents ──────────────────────────────────────────────────── - - // ── POST /api/v1/tco ───────────────────────────────────────────────────── #[tokio::test]