diff --git a/Cargo.lock b/Cargo.lock index 5f155d4e7..52ff81618 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -859,6 +859,7 @@ dependencies = [ "parking_lot", "prometheus", "promql-parser 0.8.0", + "promql_utilities", "prost", "prost-build", "reqwest 0.12.28", diff --git a/control_plane/Cargo.toml b/control_plane/Cargo.toml index 62ba06279..b81c24371 100644 --- a/control_plane/Cargo.toml +++ b/control_plane/Cargo.toml @@ -33,6 +33,7 @@ parking_lot = "0.12" prometheus = { version = "0.13", default-features = false, features = ["process"] } tonic = { version = "0.12", features = ["gzip"] } tokio-stream = { version = "0.1", features = ["net"] } +promql_utilities.workspace = true [dev-dependencies] tokio = { version = "1", features = ["full", "test-util"] } diff --git a/control_plane/src/sketch_algebra/capability.rs b/control_plane/src/sketch_algebra/capability.rs index 1d4240f50..409eed81b 100644 --- a/control_plane/src/sketch_algebra/capability.rs +++ b/control_plane/src/sketch_algebra/capability.rs @@ -46,6 +46,7 @@ use serde::{Deserialize, Serialize}; use crate::intent_algebra::agg_intent::AggIntent; use crate::sketch_algebra::params::SketchKind; use crate::types_v2::AccuracyTarget; +use promql_utilities::query_logics::enums::AggregationType; // ── Query-side capability tag ──────────────────────────────────────────────── @@ -84,6 +85,31 @@ pub enum Capability { /// wire format can answer this. `Any` required matches either /// `CmsWithHeap` or `CountSketchWithHeap`. FrequencyTopk(SketchKindHandle), + /// Exact-aggregation warm-tier state — Sum / Count / MinMax / Avg / + /// Rate / Increase / SetAggregator etc. Backed by a per-accumulator + /// payload (`AggPayload::ExactAgg` in the data plane). One variant + /// per [`AggregationType`] — the inner enum names the concrete + /// accumulator family. + /// + /// Distinct from the `*Approx` variants above: the `*Approx` + /// capabilities serve approximate sketch-bound intents; `ExactAgg` + /// serves the warm-tier exact-aggregation path (the data plane's + /// `AggKind::ExactAgg`-backed sids). Routing an analyzer candidate + /// at `Capability::ExactAgg(Sum)` to a sid whose `agg_kind` is + /// `AggKind::ExactAgg { agg_type: Sum, .. }` is what closes the gap + /// between the control plane's vocabulary and the data plane's + /// exact-aggregation state. + /// + /// PR 6 introduces this variant + the matching machinery. The + /// `capability_for(&AggIntent)` lookup deliberately does NOT route + /// `Sum` / `Min` / `Max` / `Rate` / `Increase` / exact-accuracy + /// intents to `ExactAgg` yet — that re-routing is a behavior + /// change deferred to a follow-up. The variant is dormant on the + /// analyzer side until then; the matching half (`is_satisfied_by`) + /// is wired so that sids whose stored `Capability` is + /// `ExactAgg(...)` can be filtered against an `ExactAgg(...)` + /// required capability once callers start populating it. + ExactAgg(AggregationType), } /// Compact, hashable handle for sketch implementation choice. Mirrors @@ -149,6 +175,12 @@ impl Capability { (Capability::FrequencyEstimate(req), Capability::FrequencyTopk(have)) => { is_heap_bearing(*have) && handles_compatible(*req, *have) } + // Exact-aggregation family: the agg_type must match exactly. + // There is no `Any` wildcard for ExactAgg — a Sum sid does + // not satisfy a MinMax requirement and vice versa. If a + // future PR introduces a wildcard semantic (e.g. "any + // single-population accumulator"), extend the match here. + (Capability::ExactAgg(req), Capability::ExactAgg(have)) => req == have, _ => false, } } @@ -856,6 +888,101 @@ mod tests { assert!(!required.is_satisfied_by(&bad)); } + // ── Capability::ExactAgg — matching ────────────────────────────────── + + #[test] + fn is_satisfied_by_exact_agg_same_type_matches() { + // Sum required, Sum indexed → match. Same for every concrete + // AggregationType — the equality check is structural. + let required = Capability::ExactAgg(AggregationType::Sum); + let indexed = Capability::ExactAgg(AggregationType::Sum); + assert!(required.is_satisfied_by(&indexed)); + } + + #[test] + fn is_satisfied_by_exact_agg_different_types_do_not_match() { + // Sum required, MinMax indexed → no match. No wildcard for + // ExactAgg — every agg_type stands on its own. + let required = Capability::ExactAgg(AggregationType::Sum); + let indexed = Capability::ExactAgg(AggregationType::MinMax); + assert!(!required.is_satisfied_by(&indexed)); + } + + #[test] + fn is_satisfied_by_exact_agg_does_not_match_other_families() { + // ExactAgg is its own family — no cross-family satisfaction + // with QuantileApprox / CardinalityApprox / FrequencyEstimate / + // FrequencyTopk. + let required = Capability::ExactAgg(AggregationType::Sum); + assert!(!required + .is_satisfied_by(&Capability::QuantileApprox(SketchKindHandle::DDSketch))); + assert!(!required.is_satisfied_by(&Capability::CardinalityApprox)); + assert!(!required + .is_satisfied_by(&Capability::FrequencyEstimate(SketchKindHandle::CountMin))); + assert!(!required + .is_satisfied_by(&Capability::FrequencyTopk(SketchKindHandle::CmsWithHeap))); + + // And the reverse — a sketch-family required capability must + // not match an ExactAgg-backed sid. + let sketch_required = Capability::QuantileApprox(SketchKindHandle::Any); + let exact_indexed = Capability::ExactAgg(AggregationType::DatasketchesKLL); + assert!(!sketch_required.is_satisfied_by(&exact_indexed)); + } + + #[test] + fn exact_agg_covers_each_canonical_agg_type() { + // Spot-check the full AggregationType surface — each variant + // round-trips through Capability::ExactAgg without losing + // information. Documents the intended coverage of the new + // variant. If a future PR adds an AggregationType variant, this + // test (combined with the exhaustive match in `is_satisfied_by`'s + // `req == have` form) will not require code changes — equality + // is structural. + let cases = [ + AggregationType::Sum, + AggregationType::Increase, + AggregationType::MinMax, + AggregationType::DatasketchesKLL, + AggregationType::MultipleSum, + AggregationType::MultipleIncrease, + AggregationType::MultipleMinMax, + AggregationType::HydraKLL, + AggregationType::CountMinSketch, + AggregationType::CountMinSketchWithHeap, + AggregationType::CountSketch, + AggregationType::SetAggregator, + AggregationType::DeltaSetAggregator, + AggregationType::HLL, + AggregationType::DDSketch, + ]; + for t in cases { + let cap = Capability::ExactAgg(t); + assert!( + cap.is_satisfied_by(&Capability::ExactAgg(t)), + "ExactAgg({t:?}) should satisfy itself" + ); + } + } + + // ── capability_for: ExactAgg dormancy ──────────────────────────────── + + #[test] + fn capability_for_sum_still_returns_none_after_exact_agg_landing() { + // PR 6 explicitly does NOT change `capability_for` for the + // intents that today return `None` (Sum / Min / Max / Avg / + // Rate / Increase / archive-only). The `Capability::ExactAgg` + // variant is wired into `is_satisfied_by` but the analyzer's + // intent → capability bridge stays as it was — re-routing + // those intents to warm-tier ExactAgg is a follow-up that + // requires populating `SketchInstanceMetadata.capability` with + // `Some(Capability::ExactAgg(_))` for the ExactAgg-backed sids + // first. + assert_eq!(capability_for(&AggIntent::Sum), None); + // Min / Max are intentionally NOT in this dormancy list — they + // already route to QuantileApprox (DDSketch / KLL answer them + // via quantile(0) / quantile(1)) and that path is unchanged. + } + #[test] fn count_sketch_with_heap_handle_round_trips() { // `CountSketchWithHeap` is the CountSketch counterpart to diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index 93eca4dd8..a028b25c0 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -1,6 +1,8 @@ pub mod aggregation_config; pub mod capability_matching; pub mod enums; +pub mod policy_fingerprint; +pub mod policy_registry; pub mod query_requirements; pub mod streaming_config; pub mod traits; @@ -13,5 +15,7 @@ pub use capability_matching::{ ENGINE_ID_THANOS_QUERY, }; pub use enums::*; +pub use policy_fingerprint::PolicyFingerprint; +pub use policy_registry::PolicyRegistry; pub use query_requirements::*; pub use streaming_config::*; diff --git a/crates/asap_types/src/policy_fingerprint.rs b/crates/asap_types/src/policy_fingerprint.rs new file mode 100644 index 000000000..2df36a813 --- /dev/null +++ b/crates/asap_types/src/policy_fingerprint.rs @@ -0,0 +1,320 @@ +//! Content-addressed policy identity. +//! +//! `PolicyFingerprint` is the merged-sid-identity-chain replacement for +//! the controller-allocated `aggregation_id: u64`. Where `aggregation_id` +//! is a counter the control plane mints and ships in the streaming-config +//! YAML, `PolicyFingerprint` is derived deterministically from the +//! `AggregationConfig`'s content — so two control planes producing the +//! same policy independently produce the same fingerprint, and the data +//! plane can index without a separate id allocation. +//! +//! ## Identity contract +//! +//! `PolicyFingerprint = h(metric, agg_type, sub_type, parameters, +//! grouping_labels, aggregated_labels, rollup_labels, window_size, +//! slide_interval, window_type, spatial_filter_normalized)` +//! +//! The hash includes **every** field of `AggregationConfig` that +//! determines what the policy does — sketch / exact-agg shape, +//! group-by + rollup layout, window cadence, spatial filter. Two +//! configs that compare equal on these dimensions produce the same +//! fingerprint; two that differ produce different fingerprints. +//! +//! Fields *excluded* from the fingerprint: +//! - `aggregation_id` itself (the thing we're replacing — it's a +//! downstream label, not part of identity). +//! - `original_yaml` (incidental serialization artifact). +//! - `num_aggregates_to_retain` (retention policy, not aggregation +//! semantics — two policies with the same shape but different +//! retention are *the same policy* for ingest/query routing +//! purposes; retention is a separate concern). +//! - `table_name` / `value_column` (SQL-mode wire shape; folded into +//! `metric` upstream for time-series mode). +//! +//! ## Hash function +//! +//! `xxh64` keyed at 0, matching the existing `compute_agg_config_id` +//! helper this replaces. 64-bit gives ~4B-policy birthday bound +//! (collision probability ~10⁻¹¹ at 100K live policies); ample for +//! foreseeable workloads. Bump to sha256 if the control plane ever +//! manages >10⁶ live policies and we want deterministic uniqueness. +//! +//! The fingerprint is **stable across hosts and versions**: the byte +//! layout this module produces is the contract. Don't reorder fields, +//! don't change separator bytes — any such change invalidates every +//! deployed fingerprint and forces a cold-start rebuild. + +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; +use xxhash_rust::xxh64::xxh64; + +use crate::aggregation_config::AggregationConfig; + +/// Stable, content-addressed handle for an `AggregationConfig`. +/// +/// Wrap a `u64` so callers can't accidentally swap a `PolicyFingerprint` +/// with an `aggregation_id` — they're both u64-shaped but they index +/// different things (content-addressed vs. controller-allocated). +#[derive( + Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize, +)] +#[serde(transparent)] +pub struct PolicyFingerprint(pub u64); + +impl PolicyFingerprint { + /// Compute the fingerprint of an [`AggregationConfig`]. + /// + /// Hash inputs are concatenated with `\0` byte separators and + /// canonicalized so that map/iteration order can't affect the + /// outcome. Parameter values are rendered via `serde_json::to_string` + /// for nested-shape determinism (matches the existing + /// `parameters_canonical` form used in `AggKind::ExactAgg`). + pub fn from_config(cfg: &AggregationConfig) -> Self { + let mut buf: Vec = Vec::with_capacity(512); + + // 1. metric name + buf.extend_from_slice(cfg.metric.as_bytes()); + buf.push(0); + + // 2. aggregation_type (Serialize impl is the stable form) + buf.extend_from_slice( + serde_json::to_string(&cfg.aggregation_type) + .unwrap_or_default() + .as_bytes(), + ); + buf.push(0); + + // 3. aggregation_sub_type + buf.extend_from_slice(cfg.aggregation_sub_type.as_bytes()); + buf.push(0); + + // 4. parameters — canonicalized (sorted keys, JSON-rendered values) + let sorted: BTreeMap<&String, &serde_json::Value> = cfg.parameters.iter().collect(); + for (k, v) in sorted { + buf.extend_from_slice(k.as_bytes()); + buf.push(b'='); + buf.extend_from_slice( + serde_json::to_string(v).unwrap_or_default().as_bytes(), + ); + buf.push(b';'); + } + buf.push(0); + + // 5. grouping_labels (already sorted at construction per + // KeyByLabelNames invariant; encode as `,`-joined list) + for l in &cfg.grouping_labels.labels { + buf.extend_from_slice(l.as_bytes()); + buf.push(b','); + } + buf.push(0); + + // 6. aggregated_labels + for l in &cfg.aggregated_labels.labels { + buf.extend_from_slice(l.as_bytes()); + buf.push(b','); + } + buf.push(0); + + // 7. rollup_labels + for l in &cfg.rollup_labels.labels { + buf.extend_from_slice(l.as_bytes()); + buf.push(b','); + } + buf.push(0); + + // 8. window_size + slide_interval + window_type (cadence) + buf.extend_from_slice(&cfg.window_size.to_le_bytes()); + buf.push(0); + buf.extend_from_slice(&cfg.slide_interval.to_le_bytes()); + buf.push(0); + buf.extend_from_slice( + serde_json::to_string(&cfg.window_type) + .unwrap_or_default() + .as_bytes(), + ); + buf.push(0); + + // 9. spatial_filter_normalized — canonicalized predicate + buf.extend_from_slice(cfg.spatial_filter_normalized.as_bytes()); + + Self(xxh64(&buf, 0)) + } + + /// The raw u64. Use sparingly — prefer comparing `PolicyFingerprint` + /// values directly. + pub fn as_u64(self) -> u64 { + self.0 + } +} + +impl std::fmt::Display for PolicyFingerprint { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // Hex form so logs distinguish a fingerprint from a decimal + // counter id at a glance. + write!(f, "policy_fp:{:016x}", self.0) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::enums::WindowType; + use promql_utilities::data_model::KeyByLabelNames; + use promql_utilities::query_logics::enums::AggregationType; + use std::collections::HashMap; + + fn cfg( + metric: &str, + agg_type: AggregationType, + params: HashMap, + group_by: Vec<&str>, + window_size: u64, + spatial_filter: &str, + ) -> AggregationConfig { + AggregationConfig::new( + 0, + agg_type, + String::new(), + params, + KeyByLabelNames::new(group_by.into_iter().map(|s| s.to_string()).collect()), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + window_size, + window_size, + WindowType::Tumbling, + spatial_filter.to_string(), + metric.to_string(), + None, + None, + None, + ) + } + + #[test] + fn same_config_yields_same_fingerprint() { + let a = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec!["zone"], 60, ""); + let b = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec!["zone"], 60, ""); + assert_eq!(PolicyFingerprint::from_config(&a), PolicyFingerprint::from_config(&b)); + } + + #[test] + fn different_metric_yields_different_fingerprint() { + let a = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec![], 60, ""); + let b = cfg("cpu_pct", AggregationType::Sum, HashMap::new(), vec![], 60, ""); + assert_ne!(PolicyFingerprint::from_config(&a), PolicyFingerprint::from_config(&b)); + } + + #[test] + fn different_window_yields_different_fingerprint() { + let a = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec![], 60, ""); + let b = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec![], 300, ""); + assert_ne!(PolicyFingerprint::from_config(&a), PolicyFingerprint::from_config(&b)); + } + + #[test] + fn different_spatial_filter_yields_different_fingerprint() { + let a = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec![], 60, ""); + let b = cfg( + "http_lat", + AggregationType::Sum, + HashMap::new(), + vec![], + 60, + r#"status="200""#, + ); + assert_ne!(PolicyFingerprint::from_config(&a), PolicyFingerprint::from_config(&b)); + } + + #[test] + fn different_group_by_yields_different_fingerprint() { + let a = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec!["zone"], 60, ""); + let b = cfg( + "http_lat", + AggregationType::Sum, + HashMap::new(), + vec!["zone", "service"], + 60, + "", + ); + assert_ne!(PolicyFingerprint::from_config(&a), PolicyFingerprint::from_config(&b)); + } + + #[test] + fn aggregation_id_does_not_affect_fingerprint() { + let mut a = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec![], 60, ""); + let mut b = a.clone(); + a.aggregation_id = 7; + b.aggregation_id = 42; + assert_eq!( + PolicyFingerprint::from_config(&a), + PolicyFingerprint::from_config(&b), + "aggregation_id is incidental, not part of policy identity" + ); + } + + #[test] + fn num_aggregates_to_retain_does_not_affect_fingerprint() { + let mut a = cfg("http_lat", AggregationType::Sum, HashMap::new(), vec![], 60, ""); + let mut b = a.clone(); + a.num_aggregates_to_retain = Some(100); + b.num_aggregates_to_retain = Some(500); + assert_eq!( + PolicyFingerprint::from_config(&a), + PolicyFingerprint::from_config(&b), + "retention is a separate concern from policy identity" + ); + } + + #[test] + fn parameter_map_order_does_not_affect_fingerprint() { + // HashMap iteration order is non-deterministic; the fingerprint + // must be order-independent. + let mut p1 = HashMap::new(); + p1.insert("a".into(), serde_json::json!(1)); + p1.insert("b".into(), serde_json::json!(2)); + let mut p2 = HashMap::new(); + p2.insert("b".into(), serde_json::json!(2)); + p2.insert("a".into(), serde_json::json!(1)); + let a = cfg("http_lat", AggregationType::Sum, p1, vec![], 60, ""); + let b = cfg("http_lat", AggregationType::Sum, p2, vec![], 60, ""); + assert_eq!( + PolicyFingerprint::from_config(&a), + PolicyFingerprint::from_config(&b) + ); + } + + #[test] + fn spatial_filter_canonicalization_drives_fingerprint() { + // Two filters that differ only in matcher ordering produce the + // SAME normalized form, hence the SAME fingerprint. The + // canonicalization step in `AggregationConfig::new` (via + // `normalize_spatial_filter`) sorts matchers by key. + let a = cfg( + "http_lat", + AggregationType::Sum, + HashMap::new(), + vec![], + 60, + r#"status="200",zone="us""#, + ); + let b = cfg( + "http_lat", + AggregationType::Sum, + HashMap::new(), + vec![], + 60, + r#"zone="us",status="200""#, + ); + assert_eq!( + PolicyFingerprint::from_config(&a), + PolicyFingerprint::from_config(&b) + ); + } + + #[test] + fn display_format_is_hex_with_prefix() { + let fp = PolicyFingerprint(0xdeadbeef); + assert_eq!(format!("{}", fp), "policy_fp:00000000deadbeef"); + } +} diff --git a/crates/asap_types/src/policy_registry.rs b/crates/asap_types/src/policy_registry.rs new file mode 100644 index 000000000..4c3df334a --- /dev/null +++ b/crates/asap_types/src/policy_registry.rs @@ -0,0 +1,195 @@ +//! Content-addressed policy registry. +//! +//! Derived view over a `StreamingConfig` that maps +//! [`PolicyFingerprint`] → [`AggregationConfig`]. This is the +//! merged-sid-identity-chain replacement for the controller-allocated +//! `aggregation_id`-keyed `HashMap` that today's `StreamingConfig` +//! carries. +//! +//! ## Dual-keyed transition +//! +//! PR 3 (where this lives): the registry exists alongside the +//! `aggregation_id`-keyed map. Callers can opt into either lookup. +//! Construction is `O(N)` over the source configs; computation is +//! pure (no I/O, no mutation of the source). +//! +//! Subsequent PRs migrate one set of callers at a time off +//! `aggregation_id` → `PolicyFingerprint`, until the final PR can +//! delete the legacy index. +//! +//! ## Identity invariants +//! +//! Two `AggregationConfig`s that produce the same `PolicyFingerprint` +//! ARE the same policy. The registry treats this as a *deduplication* +//! invariant — if two distinct entries in the source `aggregation_configs` +//! map produce the same fingerprint, the later one wins (last-write +//! semantics). In practice the source should never contain duplicates; +//! if it does, that's a control-plane bug worth surfacing in telemetry +//! (see `PolicyRegistry::from_streaming_config_with_collisions`). + +use std::collections::HashMap; + +use crate::aggregation_config::AggregationConfig; +use crate::policy_fingerprint::PolicyFingerprint; +use crate::streaming_config::StreamingConfig; + +/// Content-addressed lookup table for active aggregation policies. +#[derive(Debug, Clone, Default)] +pub struct PolicyRegistry { + policies: HashMap, +} + +impl PolicyRegistry { + /// Construct from a list of configs. Duplicates (same fingerprint) + /// collapse to the last entry; use + /// [`Self::from_configs_with_collisions`] when you want to detect + /// them. + pub fn from_configs(configs: I) -> Self + where + I: IntoIterator, + { + let mut policies = HashMap::new(); + for cfg in configs { + policies.insert(PolicyFingerprint::from_config(&cfg), cfg); + } + Self { policies } + } + + /// Construct + report the count of duplicate fingerprints (entries + /// where the source contained two configs producing the same + /// fingerprint and the later one displaced the earlier). Zero in + /// the well-formed case; non-zero is a control-plane bug worth + /// surfacing. + pub fn from_configs_with_collisions(configs: I) -> (Self, usize) + where + I: IntoIterator, + { + let mut policies = HashMap::new(); + let mut collisions = 0usize; + for cfg in configs { + let fp = PolicyFingerprint::from_config(&cfg); + if policies.insert(fp, cfg).is_some() { + collisions += 1; + } + } + (Self { policies }, collisions) + } + + /// Build from a `StreamingConfig`. Sugar over `from_configs` — + /// keeps callers from needing to walk the legacy map themselves. + pub fn from_streaming_config(cfg: &StreamingConfig) -> Self { + Self::from_configs(cfg.aggregation_configs.values().cloned()) + } + + /// `from_streaming_config` + collision count. + pub fn from_streaming_config_with_collisions(cfg: &StreamingConfig) -> (Self, usize) { + Self::from_configs_with_collisions(cfg.aggregation_configs.values().cloned()) + } + + /// Look up the config for a fingerprint. + pub fn get(&self, fp: PolicyFingerprint) -> Option<&AggregationConfig> { + self.policies.get(&fp) + } + + /// Iterate fingerprint → config pairs. + pub fn iter(&self) -> impl Iterator { + self.policies.iter() + } + + /// Live policy count. + pub fn len(&self) -> usize { + self.policies.len() + } + + pub fn is_empty(&self) -> bool { + self.policies.is_empty() + } + + /// All fingerprints currently registered. Useful for diffing two + /// registries during a hot-reload swap. + pub fn fingerprints(&self) -> impl Iterator + '_ { + self.policies.keys().copied() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::enums::WindowType; + use promql_utilities::data_model::KeyByLabelNames; + use promql_utilities::query_logics::enums::AggregationType; + use std::collections::HashMap as StdHashMap; + + fn cfg(id: u64, metric: &str) -> AggregationConfig { + AggregationConfig::new( + id, + AggregationType::Sum, + String::new(), + StdHashMap::new(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + KeyByLabelNames::empty(), + String::new(), + 60, + 60, + WindowType::Tumbling, + String::new(), + metric.to_string(), + None, + None, + None, + ) + } + + #[test] + fn from_configs_round_trips_lookup() { + let a = cfg(1, "http_lat"); + let b = cfg(2, "cpu_pct"); + let fp_a = PolicyFingerprint::from_config(&a); + let fp_b = PolicyFingerprint::from_config(&b); + let reg = PolicyRegistry::from_configs(vec![a.clone(), b.clone()]); + assert_eq!(reg.len(), 2); + assert_eq!(reg.get(fp_a).unwrap().metric, "http_lat"); + assert_eq!(reg.get(fp_b).unwrap().metric, "cpu_pct"); + } + + #[test] + fn aggregation_id_does_not_affect_indexing() { + // Two configs with different aggregation_ids but otherwise + // identical content collapse to ONE entry (because their + // fingerprints are equal). This is the content-addressing + // contract. + let a = cfg(1, "http_lat"); + let b = cfg(99, "http_lat"); + let (reg, collisions) = PolicyRegistry::from_configs_with_collisions(vec![a, b]); + assert_eq!(reg.len(), 1); + assert_eq!(collisions, 1); + } + + #[test] + fn distinct_policies_keep_distinct_entries() { + let a = cfg(1, "http_lat"); + let b = cfg(1, "cpu_pct"); // same id, different metric → distinct policies + let (reg, collisions) = PolicyRegistry::from_configs_with_collisions(vec![a, b]); + assert_eq!(reg.len(), 2); + assert_eq!(collisions, 0); + } + + #[test] + fn from_streaming_config_walks_the_map() { + let mut map = StdHashMap::new(); + map.insert(1, cfg(1, "http_lat")); + map.insert(2, cfg(2, "cpu_pct")); + let sc = StreamingConfig::new(map); + let reg = PolicyRegistry::from_streaming_config(&sc); + assert_eq!(reg.len(), 2); + } + + #[test] + fn empty_streaming_config_yields_empty_registry() { + let sc = StreamingConfig::new(StdHashMap::new()); + let reg = PolicyRegistry::from_streaming_config(&sc); + assert!(reg.is_empty()); + assert_eq!(reg.len(), 0); + } +} diff --git a/crates/asap_types/src/streaming_config.rs b/crates/asap_types/src/streaming_config.rs index 9a427b9e6..23fc21cd6 100644 --- a/crates/asap_types/src/streaming_config.rs +++ b/crates/asap_types/src/streaming_config.rs @@ -10,6 +10,7 @@ use crate::aggregation_config::{AggregationConfig, AggregationIdInfo}; use crate::capability_matching::find_compatible_aggregation as common_find_compatible; use crate::capability_matching::StorageBackend; use crate::enums::QueryLanguage; +use crate::policy_registry::PolicyRegistry; use crate::query_requirements::QueryRequirements; #[derive(Debug, Clone, Serialize, Deserialize)] @@ -64,6 +65,20 @@ impl StreamingConfig { self.aggregation_configs.contains_key(&aggregation_id) } + /// Derived content-addressed view. Builds a [`PolicyRegistry`] keyed + /// on [`crate::PolicyFingerprint`] — the merged-sid-identity-chain + /// replacement for the `aggregation_id`-keyed lookup. Cheap (O(N) + /// over `aggregation_configs.len()`); call at swap time, not per + /// query, if it shows up in hot-path profiles. + /// + /// Dual-keyed transition: this method exists alongside the legacy + /// `get_aggregation_config(aggregation_id)` so callers can migrate + /// one at a time. The two views are derived from the same source — + /// they can never disagree. + pub fn policy_registry(&self) -> PolicyRegistry { + PolicyRegistry::from_streaming_config(self) + } + pub fn from_yaml_file(yaml_file: &str) -> Result { let file = File::open(yaml_file)?; let reader = BufReader::new(file); diff --git a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs index 747341dbe..e41e62ebc 100644 --- a/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs +++ b/data_plane/src/storage_engines/sketch_db/query/sketch_reducer.rs @@ -234,13 +234,23 @@ impl<'a> SketchReducer<'a> { /// each `WarmTierCandidate` a `required_capability`, and the /// reducer picks a family without ever matching on the PromQL /// function-name string. + /// + /// Returns `None` for `Capability::ExactAgg(_)` — the warm-tier + /// sketch reducer only handles sketch-backed sids. Exact-aggregation + /// state is read through `SketchStore::query_precomputes_by_agg` + /// (a parallel code path), so an ExactAgg capability has no + /// `QueryFamily` mapping here. #[allow(dead_code)] - pub(crate) fn capability_to_family(cap: &Capability) -> QueryFamily { + pub(crate) fn capability_to_family(cap: &Capability) -> Option { match cap { - Capability::QuantileApprox(_) => QueryFamily::Quantile, - Capability::CardinalityApprox => QueryFamily::Cardinality, - Capability::FrequencyTopk(_) => QueryFamily::FrequencyTopk, - Capability::FrequencyEstimate(_) => QueryFamily::FrequencyEstimate, + Capability::QuantileApprox(_) => Some(QueryFamily::Quantile), + Capability::CardinalityApprox => Some(QueryFamily::Cardinality), + Capability::FrequencyTopk(_) => Some(QueryFamily::FrequencyTopk), + Capability::FrequencyEstimate(_) => Some(QueryFamily::FrequencyEstimate), + // ExactAgg sids are served by the precompute query path, not + // the sketch reducer. Callers that hand ExactAgg to this + // helper should branch to the precompute path instead. + Capability::ExactAgg(_) => None, } }