diff --git a/control_plane/src/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs index 563d3e27..78688a94 100644 --- a/control_plane/src/asap_tier_analysis.rs +++ b/control_plane/src/asap_tier_analysis.rs @@ -812,40 +812,52 @@ pub fn find_policy_by_content( hit } -/// Find every policy in `registry` whose contents satisfy `candidate`. +/// Find every policy in `index` whose contents satisfy `candidate` — the +/// "whole-query resolution" mode of +/// `control_plane/docs/design-backend-plan-wire-format.md` §4's +/// `RoutingIndex` (family-level `Capability` match, returns every +/// surviving candidate rather than ranking down to one — see that +/// design doc's note on why this differs from its own originally-sketched +/// "pick one winner" framing: this function's actual, tested behavior is +/// "union every match," and the caller's own sid-level Hit/Ghost +/// classification does the real narrowing downstream). +/// /// The result is empty when no policy fits — caller routes the query /// to the archive engine (cold tier) in that case. Multiple matches /// are valid (different windows / different sketch families all /// serving the same intent); the caller can pick the cheapest via the /// cost model or fan out to all of them and combine. /// +/// `index.candidates_for_metric(&candidate.metric_name)` (Tier 2) already +/// narrows to this metric's own policies before any predicate below runs +/// — no per-candidate metric-name check needed here anymore. +/// /// Matching predicate: -/// 1. `policy.metric == candidate.metric_name` -/// 2. `candidate.group_by_keys ⊆ policy.grouping_labels.labels` — +/// 1. `candidate.group_by_keys ⊆ policy.grouping_labels.labels` — /// the policy's group-by must cover every key the candidate names /// (extra group-by keys on the policy are fine; the query can /// re-aggregate down to its required projection). -/// 3. `policy_capability(policy)` is `Some(c)` and +/// 2. `policy_capability(policy)` is `Some(c)` and /// `candidate.required_capability.is_satisfied_by(&c)`. -/// 4. `policy.window_size ≤ candidate.range_seconds` — finer windows +/// 3. `policy.window_size ≤ candidate.range_seconds` — finer windows /// can answer coarser queries by merging; the reverse isn't true. /// When `candidate.range_seconds == 0` (instant-vector query), /// any policy window passes. -/// 5. `policy.spatial_filter_normalized == candidate.spatial_filter_canonical` +/// 4. `policy.spatial_filter_normalized == candidate.spatial_filter_canonical` /// — exact match on the canonical filter form. Empty matches empty /// (the unfiltered case); non-empty must be byte-identical (both /// sides come from `asap_types::utils::normalize_spatial_filter`, /// which sorts matchers, so the comparison is independent of the /// user's source ordering). pub fn find_matching_policies( - registry: &asap_types::PolicyRegistry, + index: &asap_types::RoutingIndex, candidate: &ASAPTierCandidate, ) -> Vec { let mut out = Vec::new(); - for (fp, cfg) in registry.iter() { - if cfg.metric != candidate.metric_name { - continue; - } + for fp in index.candidates_for_metric(&candidate.metric_name) { + let cfg = index + .get(*fp) + .expect("fp came from this index's own metric bucket"); let policy_keys: BTreeSet = cfg.grouping_labels.labels.iter().cloned().collect(); if !candidate.group_by_keys.is_subset(&policy_keys) { continue; @@ -1574,7 +1586,7 @@ mod tests { use super::super::*; use asap_types::AggregationType; use asap_types::KeyByLabelNames; - use asap_types::{AggregationConfig, PolicyFingerprint, PolicyRegistry}; + use asap_types::{AggregationConfig, PolicyFingerprint, PolicyRegistry, RoutingIndex}; use std::collections::HashMap; fn cfg( @@ -1673,7 +1685,7 @@ mod tests { 60, "", )]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &[], @@ -1692,7 +1704,7 @@ mod tests { 60, "", )]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); // Query asks for per-zone increase; policy keeps {zone, // service} (superset). let cand = candidate( @@ -1711,7 +1723,7 @@ mod tests { // check rejects this even though capabilities would // structurally satisfy. let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &["zone"], @@ -1724,7 +1736,7 @@ mod tests { #[test] fn matches_exact_metric_and_capability() { let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &[], @@ -1739,7 +1751,7 @@ mod tests { #[test] fn does_not_match_different_metric() { let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "cpu_pct", &[], @@ -1754,7 +1766,7 @@ mod tests { // Policy is Sum (ExactAgg); candidate asks for QuantileApprox. use crate::sketch_algebra::capability::SketchKindHandle; let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &[], @@ -1775,7 +1787,7 @@ mod tests { 60, "", )]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &["zone"], @@ -1790,7 +1802,7 @@ mod tests { // Policy keeps {zone}; candidate asks for {zone, service}. // That's NOT covered — policy already projected service away. let policies = vec![cfg("http_lat", AggregationType::Sum, vec!["zone"], 60, "")]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &["zone", "service"], @@ -1805,7 +1817,7 @@ mod tests { // Policy emits 60s windows; candidate wants 300s range. // Finer can answer coarser via merge. let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &[], @@ -1820,7 +1832,7 @@ mod tests { // Policy emits 300s windows; candidate wants 60s range. // Can't downsample 300s into 60s. let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 300, "")]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &[], @@ -1834,7 +1846,7 @@ mod tests { fn zero_range_query_accepts_any_window() { // Instant-vector queries (range_seconds=0) match any policy. let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 300, "")]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &[], @@ -1856,7 +1868,7 @@ mod tests { 60, r#"status="200""#, )]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate_with_filter( "http_lat", &[], @@ -1876,7 +1888,7 @@ mod tests { 60, r#"status="200""#, )]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &[], @@ -1889,7 +1901,7 @@ mod tests { #[test] fn unfiltered_policy_does_not_match_filtered_candidate() { let policies = vec![cfg("http_lat", AggregationType::Sum, vec![], 60, "")]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate_with_filter( "http_lat", &[], @@ -1909,7 +1921,7 @@ mod tests { 60, r#"status="200""#, )]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate_with_filter( "http_lat", &[], @@ -1928,7 +1940,7 @@ mod tests { cfg("http_lat", AggregationType::Sum, vec![], 60, ""), cfg("http_lat", AggregationType::Sum, vec![], 30, ""), ]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &[], @@ -1952,7 +1964,7 @@ mod tests { 60, "", )]; - let registry = PolicyRegistry::from_configs(policies); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(policies)); let cand = candidate( "http_lat", &["zone"], @@ -1964,7 +1976,7 @@ mod tests { #[test] fn empty_registry_yields_empty_matches() { - let registry = PolicyRegistry::from_configs(Vec::::new()); + let registry = RoutingIndex::build(PolicyRegistry::from_configs(Vec::::new())); let cand = candidate( "http_lat", &[], diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index 1a867c19..8b14c401 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -7,6 +7,7 @@ pub mod monitor_spec; pub mod policy_fingerprint; pub mod policy_registry; pub mod query_requirements; +pub mod routing_index; pub mod traits; pub mod utils; @@ -19,3 +20,4 @@ pub use monitor_spec::MonitorSpec; pub use policy_fingerprint::PolicyFingerprint; pub use policy_registry::PolicyRegistry; pub use query_requirements::*; +pub use routing_index::RoutingIndex; diff --git a/crates/asap_types/src/routing_index.rs b/crates/asap_types/src/routing_index.rs new file mode 100644 index 00000000..2ad2d09d --- /dev/null +++ b/crates/asap_types/src/routing_index.rs @@ -0,0 +1,189 @@ +//! `RoutingIndex` — a metric-bucketed structural index over a +//! [`PolicyRegistry`], per `control_plane/docs/design-backend-plan-wire-format.md` +//! §4's Tier-1/Tier-2 design (currently sourced from `PolicyRegistry` — the +//! content-addressed view over `StreamingConfig`'s `AggregationConfig`s, +//! which is genuinely "what control_plane planned" today, not a +//! reconstruction from ingest-side-effects — pending that doc's `BackendPlan` +//! wire format actually existing). +//! +//! **Tier 1** (exact `PolicyFingerprint` → config) is [`PolicyRegistry::get`] +//! itself — already O(1), nothing to add here. +//! +//! **Tier 2** (structural match: "every policy registered for this metric") +//! is what this type adds. Before this existed, +//! `control_plane::asap_tier_analysis::find_matching_policies` scanned +//! *every* policy in the registry for every candidate, checking each one's +//! metric name first — i.e. it paid for every OTHER metric's policies on +//! every lookup. `RoutingIndex` buckets by metric once, at construction +//! time, so a lookup only ever touches the policies that could possibly +//! match. +//! +//! ## Lifecycle +//! +//! Built fresh from a `PolicyRegistry` snapshot — construction is `O(N)` +//! over the registry, same order as `PolicyRegistry::from_configs` itself. +//! Callers that build one per query (mirroring today's +//! `streaming_snap.policy_registry()` call) still get the Tier-2 win for +//! any query with more than one candidate sharing the same snapshot +//! (composed PromQL shapes routinely do). Building it once per +//! `StreamingConfig` hot-reload swap instead of once per query — the same +//! "cheap, but call at swap time not per query, if it shows up in +//! profiles" note `StreamingConfig::policy_registry`'s own doc comment +//! already flags — is a further, larger change (it means threading a +//! cached derived value through `HotReloadStreamingConfig`'s swap path) +//! and is not done by this type on its own. + +use std::collections::HashMap; + +use crate::aggregation_config::AggregationConfig; +use crate::policy_fingerprint::PolicyFingerprint; +use crate::policy_registry::PolicyRegistry; + +/// See module docs. +#[derive(Debug, Clone, Default)] +pub struct RoutingIndex { + registry: PolicyRegistry, + by_metric: HashMap>, +} + +impl RoutingIndex { + /// Build from a `PolicyRegistry` snapshot. Takes ownership rather than + /// borrowing — callers that still need their own `PolicyRegistry` + /// handle after this should `.clone()` it first (cheap-ish, but real; + /// most callers don't need the raw registry once they have the index, + /// since [`Self::get`] delegates straight through). + pub fn build(registry: PolicyRegistry) -> Self { + let mut by_metric: HashMap> = HashMap::new(); + for (fp, cfg) in registry.iter() { + by_metric.entry(cfg.metric.clone()).or_default().push(*fp); + } + Self { registry, by_metric } + } + + /// Tier 1 — exact fingerprint lookup. Delegates to the underlying + /// registry; see [`PolicyRegistry::get`]. + pub fn get(&self, fp: PolicyFingerprint) -> Option<&AggregationConfig> { + self.registry.get(fp) + } + + /// Tier 2 — every policy fingerprint registered for `metric`, in + /// registration order. Empty slice (not an `Option`/error) when + /// nothing is registered for this metric — callers already treat "no + /// candidates" as a normal, expected outcome (capability miss → + /// archive fallback), not a failure to report. + pub fn candidates_for_metric(&self, metric: &str) -> &[PolicyFingerprint] { + self.by_metric + .get(metric) + .map(Vec::as_slice) + .unwrap_or(&[]) + } + + /// Live policy count (same as the underlying registry's). + pub fn len(&self) -> usize { + self.registry.len() + } + + pub fn is_empty(&self) -> bool { + self.registry.is_empty() + } + + /// All fingerprints currently registered. Delegates to the underlying + /// registry — see [`PolicyRegistry::fingerprints`]. + pub fn fingerprints(&self) -> impl Iterator + '_ { + self.registry.fingerprints() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::enums::WindowKind; + use crate::AggregationType; + use crate::KeyByLabelNames; + use std::collections::HashMap as StdHashMap; + + fn cfg(metric: &str) -> AggregationConfig { + AggregationConfig::new( + AggregationType::Sum, + String::new(), + StdHashMap::new(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 60, + 60, + WindowKind::Tumbling, + String::new(), + metric.to_string(), + None, + None, + None, + ) + } + + #[test] + fn buckets_by_metric() { + let a1 = cfg("http_lat"); + let a2 = cfg("cpu_pct"); + let fp_a1 = PolicyFingerprint::from_config(&a1); + let fp_a2 = PolicyFingerprint::from_config(&a2); + let idx = RoutingIndex::build(PolicyRegistry::from_configs(vec![a1, a2])); + + assert_eq!(idx.candidates_for_metric("http_lat"), &[fp_a1]); + assert_eq!(idx.candidates_for_metric("cpu_pct"), &[fp_a2]); + } + + #[test] + fn multiple_policies_for_the_same_metric_all_bucket_together() { + // Same metric, distinct group-by shapes -> distinct fingerprints, + // same bucket. + let a = AggregationConfig::new( + AggregationType::Sum, + String::new(), + StdHashMap::new(), + KeyByLabelNames::new(vec!["zone".to_string()]), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 60, + 60, + WindowKind::Tumbling, + String::new(), + "http_lat".to_string(), + None, + None, + None, + ); + let b = cfg("http_lat"); + assert_ne!(PolicyFingerprint::from_config(&a), PolicyFingerprint::from_config(&b)); + let idx = RoutingIndex::build(PolicyRegistry::from_configs(vec![a, b])); + assert_eq!(idx.candidates_for_metric("http_lat").len(), 2); + } + + #[test] + fn unknown_metric_returns_empty_slice_not_missing() { + let idx = RoutingIndex::build(PolicyRegistry::from_configs(vec![cfg("http_lat")])); + assert!(idx.candidates_for_metric("no_such_metric").is_empty()); + } + + #[test] + fn get_delegates_to_underlying_registry() { + let a = cfg("http_lat"); + let fp = PolicyFingerprint::from_config(&a); + let idx = RoutingIndex::build(PolicyRegistry::from_configs(vec![a])); + assert_eq!(idx.get(fp).map(|c| c.metric.clone()), Some("http_lat".to_string())); + assert!(idx.get(PolicyFingerprint::from_config(&cfg("nope"))).is_none()); + } + + #[test] + fn len_and_is_empty_match_registry() { + let idx = RoutingIndex::build(PolicyRegistry::from_configs(Vec::::new())); + assert!(idx.is_empty()); + assert_eq!(idx.len(), 0); + + let idx = RoutingIndex::build(PolicyRegistry::from_configs(vec![cfg("m")])); + assert!(!idx.is_empty()); + assert_eq!(idx.len(), 1); + } +} diff --git a/data_plane/src/query_engines/asap_query_engine/engine.rs b/data_plane/src/query_engines/asap_query_engine/engine.rs index 791a00fc..c0a3cc5e 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -333,7 +333,7 @@ impl ASAPQueryEngine { } let streaming_snap = self.streaming_config_snapshot(); - let policy_registry = streaming_snap.policy_registry(); + let routing_index = asap_types::RoutingIndex::build(streaming_snap.policy_registry()); let mut combined_result: Option = None; // Resilience fix -- see the instant-query `execute(&str)` path's @@ -365,7 +365,7 @@ impl ASAPQueryEngine { // populated `policy_fp`) but its result is unioned with // the catalog-walk result so we don't miss the sketches. let policy_fps = control_plane::asap_tier_analysis::find_matching_policies( - &policy_registry, + &routing_index, candidate, ); let mut sids: std::collections::BTreeSet = std::collections::BTreeSet::new(); @@ -843,7 +843,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // policy lookups. Hot-reload swaps the underlying Arc; the // snapshot pins one revision for the duration. let streaming_snap = self.streaming_config_snapshot(); - let policy_registry = streaming_snap.policy_registry(); + let routing_index = asap_types::RoutingIndex::build(streaming_snap.policy_registry()); // Resilience fix (design-target-architecture.md Part B, // completing the analyzer-side fix in @@ -883,7 +883,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // empty for the MVP demo workload — see issue #271 / // tracking #272. let policy_fps = control_plane::asap_tier_analysis::find_matching_policies( - &policy_registry, + &routing_index, candidate, ); let mut sids: std::collections::BTreeSet = std::collections::BTreeSet::new();