diff --git a/control_plane/src/emit/monitor.rs b/control_plane/src/emit/monitor.rs index ceb6e73f..d93f6f69 100644 --- a/control_plane/src/emit/monitor.rs +++ b/control_plane/src/emit/monitor.rs @@ -215,7 +215,7 @@ mod tests { fn streaming_entry_deserializes_as_monitor_spec() { // The emitted JSON must round-trip into the backend's MonitorSpec. let entry = streaming_config_monitor_entry(&sum_intent()); - let spec: asap_types::streaming_config::MonitorSpec = + let spec: asap_types::MonitorSpec = serde_json::from_value(entry).expect("MonitorSpec deserialize"); assert_eq!(spec.agg_id, agg_id_for_metric("bytes_sent")); assert_eq!(spec.tau, 100.0); diff --git a/crates/asap_types/src/capability_matching.rs b/crates/asap_types/src/capability_matching.rs deleted file mode 100644 index 91218ea8..00000000 --- a/crates/asap_types/src/capability_matching.rs +++ /dev/null @@ -1,535 +0,0 @@ -use crate::Statistic; -use serde::{Deserialize, Serialize}; - -use crate::AggregationType; - -pub const ENGINE_ID_ASAP_QUERY: &str = "asap_query"; -pub const ENGINE_ID_THANOS_QUERY: &str = "thanos_query"; - -pub const CANONICAL_QUERY_ENGINE_IDS: &[&str] = &[ENGINE_ID_ASAP_QUERY, ENGINE_ID_THANOS_QUERY]; - -// --------------------------------------------------------------------------- -// Phase-5: storage-backend capability axis -// -// Matching on `(metric, statistic, sub_type, window_size, grouping_labels, -// spatial_filter)` alone has no axis for "which storage tier serves this -// query." The Phase-5 `GorillaQueryEngine` (PR #85) introduces a parallel -// exact tier; the planner / router needs to disambiguate between ASAP-tier -// sketches and Gorilla-S3 chunks. See `docs/design-gorilla-s3-cold-engine.md` -// §8. -// --------------------------------------------------------------------------- - -/// Which physical storage tier a query (or a metric configuration) routes to. -/// -/// `SketchStore` is the default — every existing `AggregationConfig` and -/// `StreamingConfig` decodes into this variant via `#[serde(default)]`, so -/// pre-Phase-5 deploys keep dispatching to `ASAPQueryEngine` unchanged. -/// -/// **Step-1 of the JSONL deprecation refactor** removed the -/// `ColdJsonlFallback` variant. The legacy local-FS JSONL leg -/// (`LocalFsColdStore`, `parse_jsonl`, the §5.2 raw-store -/// fallback) was deleted at the same commit; the surviving -/// failover surface is ASAP-tier sketch ↔ Thanos archive. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)] -#[serde(rename_all = "snake_case")] -pub enum StorageBackend { - /// Warm-tier sketch DB (today's `SketchStore` + accumulators). - /// Served by `ASAPQueryEngine`. Default for unconfigured metrics. - #[default] - SketchStore, - - /// Thanos archive over MinIO/S3. The enum name is kept for - /// serde/back-compat with existing configs, but its canonical - /// query-engine identity is `thanos_query`. Gorilla is an - /// archive chunk format/storage detail, not a public query engine. - GorillaObjectStore, - - /// Double-write: the metric is written to both ASAP-tier sketches AND the - /// Gorilla-S3 archive. Capability matching surfaces both options and the - /// cost-aware dispatcher picks per query (typically ASAP-tier for low- - /// latency approximate, archive for exact). - DoubleWrite, - - /// Prometheus-remote: the metric's data is shipped raw to a - /// Prometheus instance via the native OTLP receiver. Phase ε.2 - /// registers a `PrometheusForwardEngine` (HTTP-forwarder to - /// Prometheus's `/api/v1/query`) under this slot so the - /// controller's `RawAtEdgePrometheusArchive` mode can route a - /// metric's queries to Prometheus directly. Mirrors the - /// `GorillaObjectStore` slot's "single backend, no failover" - /// semantics — there is no ASAP-tier sketch to fall back on for a - /// Prometheus-remote metric. - PrometheusRemote, -} - -impl StorageBackend { - /// Canonical string tag pinned for byte-comparable dispatch on the wire (mirrors - /// the `data_source: ` info-line on `QueryResult`). Engines - /// register themselves under these IDs in the router. - pub const fn data_source_id(self) -> &'static str { - match self { - StorageBackend::SketchStore => ENGINE_ID_ASAP_QUERY, - StorageBackend::GorillaObjectStore => ENGINE_ID_THANOS_QUERY, - StorageBackend::DoubleWrite => "double_write", - StorageBackend::PrometheusRemote => "prometheus_remote", - } - } -} - -pub fn parse_storage_backend_engine_id(s: &str) -> Option { - match s { - ENGINE_ID_ASAP_QUERY => Some(StorageBackend::SketchStore), - ENGINE_ID_THANOS_QUERY => Some(StorageBackend::GorillaObjectStore), - "double_write" => Some(StorageBackend::DoubleWrite), - "prometheus_remote" => Some(StorageBackend::PrometheusRemote), - _ => None, - } -} - -/// Accuracy hint pushed by the controller at intent-binding time -/// (`controller/docs/design.md` §6 `core::workload`). The Phase-5 capability -/// router consults this to decide whether a metric configured for both warm- -/// tier and Gorilla-S3 should answer from the archive (Exact) or the -/// approximate ASAP-tier sketch. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)] -#[serde(rename_all = "snake_case")] -pub enum AccuracyTarget { - /// Caller demands an exact answer; ASAP-tier sketches are not eligible - /// unless they happen to be exact accumulators (Sum, MinMax, Increase). - Exact, - /// Caller accepts ε/δ-bounded approximate answers. Default. - #[default] - Approximate, -} - -// --------------------------------------------------------------------------- -// Pure compatibility helpers -// --------------------------------------------------------------------------- - -/// Returns the aggregation types that can serve this statistic. -/// -/// This list is the **superset of compatibility** and, as of the -/// `promql_utilities` retirement, the **single source of truth** for it — -/// there used to be a second, independently-maintained table -/// (`promql_utilities::query_logics::logics::map_statistic_to_precompute_operator`, -/// the planner's own canonical map) that this one had to agree with. That -/// table was dead code (a Python-planner relic) and was deleted; this is -/// now the only table. -/// -/// `QueryTreatmentType` is not consulted here — this list intentionally -/// enumerates *every* type that could serve the statistic, treatment-agnostic. -/// Selection between e.g. `Sum` (exact) and `CountMinSketch` (approximate) -/// for `Statistic::Sum` is made downstream by the caller. -pub fn compatible_agg_types(stat: Statistic) -> &'static [AggregationType] { - match stat { - // Sum: exact via Sum / MultipleSum; approximate via CountMinSketch. - // Pre-fix this list omitted CountMinSketch, so a `sum_over_time(...)` - // query against a CMS-only config fell through capability matching - // and onto the cold tier. - Statistic::Sum => &[ - AggregationType::Sum, - AggregationType::MultipleSum, - AggregationType::CountMinSketch, - // Counters: ASAP-tier ingest stores counter metrics - // (OTel `Sum` / monotonic=true) as Increase / - // MultipleIncrease accumulators, whose `query` - // implementation answers `Statistic::Sum` with the - // latest cumulative value per series — matching - // Prometheus' instant `sum()` semantics. - // Without these here, `sum by (zone) (http_requests_total)` - // capability-misses (issue ProjectASAP/ASAPCollector#46; - // diagnosis in PR #108). - AggregationType::Increase, - AggregationType::MultipleIncrease, - ], - // Count: exact via MultipleSum (the planner's canonical pick for - // Count-Exact uses `MultipleSum` with sub_type="count"); approximate - // via CountMinSketch / CountMinSketchWithHeap. - // - // HLL is also valid here: the ASAP-tier MVP demo - // (ProjectASAP/ASAPCollector#46) plans `unique_users_per_min` - // as an HLL agg and the replay client queries it with - // `count(unique_users_per_min)`. `HllSketchAccumulator` - // answers `Statistic::Count` as a cardinality alias — - // see `precompute_operators/hll_sketch_accumulator.rs:220`. - // Without HLL listed here the warm engine returns `status=error` - // for every count-of-HLL replay row. - Statistic::Count => &[ - AggregationType::MultipleSum, - AggregationType::CountMinSketch, - AggregationType::CountMinSketchWithHeap, - AggregationType::HLL, - ], - Statistic::Min | Statistic::Max => { - &[AggregationType::MinMax, AggregationType::MultipleMinMax] - } - // Quantile: KLL (planner-emitted canonical pick) plus the - // sketch types whose accumulators answer `Statistic::Quantile` - // natively but that the planner's canonical map does not - // emit. The MVP demo's controller (`ASAPCollector/controller`) - // plans `http_requests_total_latency_ms` as a `DDSketch` - // directly from `mvp-workload.yaml` and routes the resulting - // delta payloads through the modified-OTLP wire format - // (DDSketch state landing in `DDSketchAccumulator`, which - // supports `Statistic::Quantile` — see - // `precompute_operators/dd_sketch_accumulator.rs`). Without - // DDSketch enumerated here, capability matching for an - // out-of-YAML query like `quantile_over_time(0.99, - // http_requests_total_latency_ms[1m])` would miss and the - // ASAP-tier engine returns `EngineError::CapabilityMiss`. - Statistic::Quantile => &[ - AggregationType::DatasketchesKLL, - AggregationType::HydraKLL, - AggregationType::DDSketch, - ], - // Rate / Increase: the canonical exact accumulators are the - // counter-shaped Increase / MultipleIncrease, but `rate(...)` - // and `increase(...)` over a CountMinSketch-backed agg are - // also valid — CMS records every insert and answers - // `Statistic::Rate` natively (events / range_ms when the - // engine passes `range_ms` in query_kwargs; raw event count - // as a units-of-events/window fallback otherwise — see - // `precompute_operators/count_min_sketch_accumulator.rs`). - // Without CMS / CMSWithHeap listed here, `rate(metric[5m])` - // against a CMS-only config — the canonical MVP demo - // CountMin path — capability-misses and the warm engine - // returns `status=error`. Closes the PR #111 honest-gap - // call-out for `Statistic::Rate` not implemented. - Statistic::Rate | Statistic::Increase => &[ - AggregationType::Increase, - AggregationType::MultipleIncrease, - AggregationType::CountMinSketch, - AggregationType::CountMinSketchWithHeap, - ], - // Cardinality: HLL is the approximate cardinality estimator - // (wired in via modified-OTLP from the agent processors) — - // see `precompute_operators/hll_sketch_accumulator.rs`. The - // historical exact-key trackers (`SetAggregator` / - // `DeltaSetAggregator`) were retired wholesale; HLL is the - // sole cardinality answerer today. - Statistic::Cardinality => &[AggregationType::HLL], - // Topk: `CountMinSketchWithHeap` is the canonical CMS-Heap - // pattern. CountSketch is the second-tier reservoir-style - // approximator the MVP demo's controller plans for - // `top_endpoint_qps` (median-of-row estimator over a - // signed-counter matrix). `CountSketchAccumulator` answers - // `Statistic::Topk` directly — see - // `precompute_operators/count_sketch_accumulator.rs:284`. - // `CountSketchWithHeap` is the explicit heap-bearing variant - // that also satisfies Topk through the heap directly - // (parallel to `CountMinSketchWithHeap`); the analyzer's - // `topk(...)` candidate returns `FrequencyTopk(Any)` so - // either heap-bearing variant matches. - // Without CountSketch / CountSketchWithHeap listed here, - // `topk(K, top_endpoint_qps)` capability-misses and the - // warm engine returns `status=error`. - Statistic::Topk => &[ - AggregationType::CountMinSketchWithHeap, - AggregationType::CountSketch, - AggregationType::CountSketchWithHeap, - ], - } -} - -/// Returns the storage backends that can serve a `(statistic, accuracy)` -/// query when the metric is configured for `metric_storage_config`. -/// -/// The returned list is **ordered by preference**: the router walks it in -/// order and dispatches to the first backend whose engine is registered, -/// falling through on `CapabilityMiss` / `Backend` to the next entry. -/// -/// **ASAP-first centralization refactor**: the decision tree is now -/// owned here (and consumed identically by `EngineRouter::execute` and -/// `EngineRouter::execute_range`) so the HTTP transport layer never -/// re-derives routing. The policy is: -/// -/// * `accuracy == Exact` → archive only `[GorillaObjectStore]` (served -/// by the `thanos_query` engine). The caller demands an exact answer, -/// so the ε/δ-bounded ASAP-tier sketches are not eligible — go -/// straight to the archive regardless of where the metric is stored. -/// * `accuracy == Approximate` (the default) → ASAP-first failover -/// `[SketchStore, GorillaObjectStore]` for any metric stored in an -/// ASAP-managed tier (`SketchStore`, `GorillaObjectStore`, or -/// `DoubleWrite`): try the warm sketch (`asap_query`) first and fall -/// back to the archive (`thanos_query`) on a capability miss. This -/// collapses the old per-`metric_storage` sequences into one shared -/// ASAP-first-then-archive contract. -/// * `PrometheusRemote` keeps its own single-backend sequence -/// `[PrometheusRemote]` (Phase ε.2): the metric's raw samples never -/// landed in ASAP-managed storage, so there is no ASAP-tier sketch to -/// fall back on and the accuracy hint does not apply. A missing -/// engine surfaces as a `NoEngineRegistered` 503 from the HTTP -/// handler — the correct fail-loud behaviour for a misconfigured -/// deploy. -pub fn compatible_storage_backends( - _stat: Statistic, - accuracy: AccuracyTarget, - metric_storage_config: StorageBackend, -) -> Vec { - match metric_storage_config { - // Prometheus-remote owns its own storage; the accuracy hint does - // not apply and there is no ASAP-tier sketch to fall back on. - StorageBackend::PrometheusRemote => vec![StorageBackend::PrometheusRemote], - - // Every ASAP-managed tier shares the same ASAP-first policy, - // gated only on the accuracy target. - StorageBackend::SketchStore - | StorageBackend::GorillaObjectStore - | StorageBackend::DoubleWrite => match accuracy { - // Exact: archive only — the warm sketches are ε/δ-bounded. - AccuracyTarget::Exact => vec![StorageBackend::GorillaObjectStore], - // Approximate: ASAP-tier first, archive (Thanos) fallback. - AccuracyTarget::Approximate => vec![ - StorageBackend::SketchStore, - StorageBackend::GorillaObjectStore, - ], - }, - } -} - -/// Returns the required aggregation_sub_type for this statistic, if any. -/// `Min` requires `"min"`, `Max` requires `"max"`. All others are unconstrained. -pub fn required_sub_type(stat: Statistic) -> Option<&'static str> { - match stat { - Statistic::Min => Some("min"), - Statistic::Max => Some("max"), - _ => None, - } -} - -// --------------------------------------------------------------------------- -// Tests -// --------------------------------------------------------------------------- - -#[cfg(test)] -mod tests { - use super::*; - - /// Pin the canonical-approximator picks driving the ASAP-tier query path - /// (the "five sketch types" CMS / KLL / HLL / DDSketch / CountSketch - /// canonical statistic table from PROGRESS.md). HLL / DDSketch / - /// CountSketch route via the modified-OTLP wire format and are not in the - /// planner's canonical map; KLL covers Quantile, CMS covers Sum + Count, - /// CMSWithHeap covers Topk. Each must appear in its `Statistic`'s compat - /// list — this is the bug fix that motivated this PR. - #[test] - fn five_sketch_canonical_statistics_in_compat_list() { - // KLL → Quantile - assert!( - compatible_agg_types(Statistic::Quantile).contains(&AggregationType::DatasketchesKLL), - "KLL must be a compatible type for Quantile", - ); - // DDSketch → Quantile (Phase-3.1 fix). Required so the MVP demo's - // `quantile_over_time(0.99, http_requests_total_latency_ms[1m])` - // — which routes a DDSketch agg from `mvp-workload.yaml` and may - // miss the inference-YAML exact-string match — still resolves - // through capability matching instead of returning a 404. - assert!( - compatible_agg_types(Statistic::Quantile).contains(&AggregationType::DDSketch), - "DDSketch must be a compatible type for Quantile (Phase-3.1 fix)", - ); - // HLL → Cardinality (Phase-3.1 fix). HLL accumulators answer - // `Statistic::Cardinality` natively (and `Statistic::Count` as a - // cardinality alias); enumerating them here lets capability - // matching pick up an HLL-only deploy. - assert!( - compatible_agg_types(Statistic::Cardinality).contains(&AggregationType::HLL), - "HLL must be a compatible type for Cardinality (Phase-3.1 fix)", - ); - // CMS → Sum (the headline bug fix that motivated this PR) - assert!( - compatible_agg_types(Statistic::Sum).contains(&AggregationType::CountMinSketch), - "CountMinSketch must be a compatible type for Sum (PR fix)", - ); - // CMS → Count - assert!( - compatible_agg_types(Statistic::Count).contains(&AggregationType::CountMinSketch), - "CountMinSketch must be a compatible type for Count", - ); - // CMSWithHeap → Topk - assert!( - compatible_agg_types(Statistic::Topk) - .contains(&AggregationType::CountMinSketchWithHeap), - "CountMinSketchWithHeap must be a compatible type for Topk", - ); - // CountSketch → Topk (warm-engine-error-on-replay-queries fix). - // Required so the MVP demo's `topk(5, top_endpoint_qps)` — - // which routes through the agent's `countsketchprocessor` - // and lands as a CountSketch-only config — resolves - // through capability matching. Without this, the warm - // engine returned `status=error` for every topk replay row. - assert!( - compatible_agg_types(Statistic::Topk).contains(&AggregationType::CountSketch), - "CountSketch must be a compatible type for Topk (warm-engine-error fix)", - ); - // HLL → Count (warm-engine-error-on-replay-queries fix). The - // MVP demo's `count(unique_users_per_min)` is structurally a - // PromQL `Statistic::Count` (the AggregationOperator::Count - // → Statistic::Count mapping in - // `promql_utilities::query_logics::enums`); the - // `HllSketchAccumulator` answers it as a cardinality alias - // (`hll_sketch_accumulator.rs:220`). Without HLL listed - // here, capability matching missed and the warm engine - // returned `status=error` for every count-of-HLL replay row. - assert!( - compatible_agg_types(Statistic::Count).contains(&AggregationType::HLL), - "HLL must be a compatible type for Count (warm-engine-error fix)", - ); - } - // ----------------------------------------------------------------------- - // Phase-5: storage-backend routing - // - // The Phase-5 `EngineRouter` (see `asap-query-engine/src/routing/query_engine_routing.rs`) - // consults `compatible_storage_backends(stat, accuracy, metric_storage)` - // to pick a backend. These tests pin the routing matrix so the dispatcher - // stays in lock-step with the design doc §8. - // ----------------------------------------------------------------------- - - #[test] - fn exact_accuracy_routes_to_archive_only() { - // ASAP-first refactor: `Exact` goes straight to the archive - // (Thanos via the GorillaObjectStore slot) regardless of where - // the metric is stored — the warm sketches are ε/δ-bounded. - for cfg in [ - StorageBackend::GorillaObjectStore, - StorageBackend::SketchStore, - StorageBackend::DoubleWrite, - ] { - let backends = compatible_storage_backends(Statistic::Sum, AccuracyTarget::Exact, cfg); - assert_eq!( - backends, - vec![StorageBackend::GorillaObjectStore], - "Exact accuracy must route to archive only for {cfg:?}", - ); - } - } - - #[test] - fn approximate_accuracy_is_asap_first_with_archive_fallback() { - // ASAP-first refactor: every ASAP-managed tier shares the same - // `[SketchStore, GorillaObjectStore]` sequence for `Approximate` - // — try the warm sketch first, fall back to the Thanos archive - // on a capability miss. - for cfg in [ - StorageBackend::SketchStore, - StorageBackend::GorillaObjectStore, - StorageBackend::DoubleWrite, - ] { - let backends = - compatible_storage_backends(Statistic::Quantile, AccuracyTarget::Approximate, cfg); - assert_eq!( - backends, - vec![ - StorageBackend::SketchStore, - StorageBackend::GorillaObjectStore, - ], - "Approximate accuracy must be ASAP-first then archive for {cfg:?}", - ); - } - } - - #[test] - fn storage_backend_default_is_asap_tier() { - // `#[serde(default)]` on `StreamingConfig.storage_backend` (and on - // `StorageBackend::default()`) MUST be `SketchStore` so pre-Phase-5 - // configs decode without bumping deploys onto the archive. - assert_eq!(StorageBackend::default(), StorageBackend::SketchStore); - } - - #[test] - fn storage_backend_data_source_id_is_pinned() { - // The router registers engines by these strings; dashboards - // byte-compare them. Pin to catch accidental rename. - assert_eq!( - StorageBackend::SketchStore.data_source_id(), - ENGINE_ID_ASAP_QUERY - ); - assert_eq!( - StorageBackend::GorillaObjectStore.data_source_id(), - ENGINE_ID_THANOS_QUERY, - ); - assert_eq!(StorageBackend::DoubleWrite.data_source_id(), "double_write",); - assert_eq!( - StorageBackend::PrometheusRemote.data_source_id(), - "prometheus_remote", - ); - } - - #[test] - fn storage_backend_engine_id_parser_accepts_only_canonical_query_engines() { - assert_eq!( - parse_storage_backend_engine_id(ENGINE_ID_ASAP_QUERY), - Some(StorageBackend::SketchStore), - ); - assert_eq!( - parse_storage_backend_engine_id(ENGINE_ID_THANOS_QUERY), - Some(StorageBackend::GorillaObjectStore), - ); - assert_eq!(parse_storage_backend_engine_id("not_an_engine"), None); - } - - /// Source-of-truth agreement check for the storage axis. - /// - /// For every `(Statistic, AccuracyTarget, StorageBackend)` triple - /// the returned backend list must be non-empty and its head must - /// match the routing matrix in `compatible_storage_backends`'s - /// docstring. - #[test] - fn capability_storage_backend_agreement() { - let stats = [ - Statistic::Count, - Statistic::Sum, - Statistic::Cardinality, - Statistic::Increase, - Statistic::Rate, - Statistic::Min, - Statistic::Max, - Statistic::Quantile, - Statistic::Topk, - ]; - let accuracies = [AccuracyTarget::Exact, AccuracyTarget::Approximate]; - let configs = [ - StorageBackend::SketchStore, - StorageBackend::GorillaObjectStore, - StorageBackend::DoubleWrite, - StorageBackend::PrometheusRemote, - ]; - - for &stat in &stats { - for &acc in &accuracies { - for &cfg in &configs { - let backends = compatible_storage_backends(stat, acc, cfg); - assert!( - !backends.is_empty(), - "compatible_storage_backends({stat:?}, {acc:?}, {cfg:?}) returned empty \ - — every metric configuration must route to at least one backend", - ); - let last = *backends.last().unwrap(); - assert!( - last == StorageBackend::SketchStore - || last == StorageBackend::GorillaObjectStore - || last == StorageBackend::PrometheusRemote, - "backend list for ({stat:?}, {acc:?}, {cfg:?}) must terminate in a \ - dispatchable failover (SketchStore, GorillaObjectStore, or \ - PrometheusRemote); got {last:?}", - ); - // ASAP-first refactor: the head is determined by - // `(metric_storage_config, accuracy)`. PrometheusRemote - // keeps its single-backend slot; every ASAP-managed tier - // goes archive-only on `Exact` and ASAP-tier-first on - // `Approximate`. - let expected_head = match (cfg, acc) { - (StorageBackend::PrometheusRemote, _) => StorageBackend::PrometheusRemote, - (_, AccuracyTarget::Exact) => StorageBackend::GorillaObjectStore, - (_, AccuracyTarget::Approximate) => StorageBackend::SketchStore, - }; - assert_eq!( - backends[0], expected_head, - "head mismatch for ({stat:?}, {acc:?}, {cfg:?}): expected {expected_head:?}, got {:?}", - backends[0], - ); - } - } - } - } -} diff --git a/crates/asap_types/src/lib.rs b/crates/asap_types/src/lib.rs index 412369d7..701b5c0a 100644 --- a/crates/asap_types/src/lib.rs +++ b/crates/asap_types/src/lib.rs @@ -1,24 +1,19 @@ pub mod aggregation_config; pub mod aggregation_type; -pub mod capability_matching; pub mod enums; pub mod key_by_label_names; +pub mod monitor_spec; pub mod policy_fingerprint; pub mod policy_registry; pub mod query_requirements; -pub mod streaming_config; pub mod traits; pub mod utils; pub use aggregation_config::*; pub use aggregation_type::AggregationType; -pub use capability_matching::{ - compatible_storage_backends, parse_storage_backend_engine_id, AccuracyTarget, StorageBackend, - CANONICAL_QUERY_ENGINE_IDS, ENGINE_ID_ASAP_QUERY, ENGINE_ID_THANOS_QUERY, -}; pub use enums::*; pub use key_by_label_names::KeyByLabelNames; +pub use monitor_spec::MonitorSpec; 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/monitor_spec.rs b/crates/asap_types/src/monitor_spec.rs new file mode 100644 index 00000000..12aaf676 --- /dev/null +++ b/crates/asap_types/src/monitor_spec.rs @@ -0,0 +1,53 @@ +use serde::{Deserialize, Serialize}; + +/// One continuous-monitoring (CDM) threshold spec. The data-plane monitor +/// coordinator owns the AUTHORITATIVE `tau`/`epsilon`/`window_ms` (the edge +/// copy is advisory), keyed by the same content-addressed `agg_id` the edge and +/// coordinator share. `key` is the CMS point-frequency key for point monitors +/// (empty for Sum / whole-stream). See +/// `ASAPCollector/docs/continuous-monitoring-tumbling-cost-analysis.md`. +/// +/// Stays here (unlike `data_plane::storage_engines::types::StreamingConfig`, +/// which holds a `Vec` field) because `control_plane` genuinely +/// needs it: `emit/monitor.rs` builds the `StreamingConfig.monitors[]` JSON +/// entry by hand and has a regression test asserting that JSON deserializes +/// into this exact type. `control_plane` cannot depend on `data_plane` (the +/// dependency runs the other way), so this type has to live somewhere both +/// sides can reach — same reasoning as `AggregationConfig`/`PolicyFingerprint`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MonitorSpec { + pub agg_id: u64, + /// Additive readout the edge reports: "sum" (default), "cms_point", "f2". + /// Pass-through metadata so the edge can auto-learn its reporting mode from + /// the pushed config; the coordinator allocation is value-driven and does not + /// branch on it (p_i ∝ √(value/rate) is the F2 allocation when value=‖f‖²). + #[serde(default)] + pub functional: String, + /// CMS point-frequency key x; empty (default) for Sum / whole-stream / F2. + #[serde(default)] + pub key: String, + /// Threshold τ (authoritative here, not at the edge). + pub tau: f64, + /// Relative tolerance ε; the alert fires when the estimate reaches (1−ε)τ. + #[serde(default = "default_monitor_epsilon")] + pub epsilon: f64, + /// Tumbling epoch length in ms; MUST match the edge window for this agg_id. + pub window_ms: u64, + /// Count-Sketch depth (rows) for whole-sketch `functional="f2"` monitors. + /// 0 (default) for scalar monitors; MUST match the edge's Count-Sketch for + /// this agg when F2 (both sides square/merge the same cell matrix). + #[serde(default)] + pub d: usize, + /// Count-Sketch width (buckets/row) for F2 monitors; 0 for scalar. + #[serde(default)] + pub w: usize, + /// F2 monitoring variant: "distributed" (default, ship every window) or + /// "geometric" (Sharfman–Schuster–Keren safe-zone, ship on local violation). + /// Ignored by scalar monitors. + #[serde(default)] + pub mode: String, +} + +fn default_monitor_epsilon() -> f64 { + 0.05 +} diff --git a/crates/asap_types/src/policy_registry.rs b/crates/asap_types/src/policy_registry.rs index a10f0548..945556d2 100644 --- a/crates/asap_types/src/policy_registry.rs +++ b/crates/asap_types/src/policy_registry.rs @@ -1,10 +1,11 @@ //! Content-addressed policy registry. //! -//! Derived view over a `StreamingConfig` that maps +//! Derived view over a collection of `AggregationConfig`s 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. +//! `aggregation_id`-keyed `HashMap` that `data_plane`'s `StreamingConfig` +//! carries (see `data_plane::storage_engines::types::streaming_config`'s +//! module doc for why that type lives there, not here). //! //! ## Dual-keyed transition //! @@ -25,13 +26,12 @@ //! 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`). +//! (see [`PolicyRegistry::from_configs_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)] @@ -75,17 +75,6 @@ impl PolicyRegistry { (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) @@ -176,22 +165,4 @@ mod tests { 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/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index eba747b2..0a6fbb6c 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -17,11 +17,11 @@ use tracing::{debug, info, warn}; use crate::drivers::query::adapters::{create_http_adapter, AdapterConfig, HttpProtocolAdapter}; use crate::drivers::query::servers::metrics as srv_metrics; use crate::query_engines::routing::{ - EngineRouter, EngineRouterError, FreshnessProbeCache, QueryEngine, + AccuracyTarget, EngineRouter, EngineRouterError, FreshnessProbeCache, QueryEngine, }; use crate::query_engines::ASAPQueryEngine; +use crate::storage_engines::types::StorageBackend; use asap_types::Statistic; -use asap_types::{AccuracyTarget, StorageBackend}; // ─── Control-plane-pushed precompute job registry ──────────────────────────── // @@ -5330,7 +5330,7 @@ async fn handle_post_streaming_config( } }; let new_config = - match asap_types::streaming_config::StreamingConfig::from_yaml_data(&yaml_value) { + match crate::storage_engines::types::StreamingConfig::from_yaml_data(&yaml_value) { Ok(c) => c, Err(e) => { let body = serde_json::json!({ 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 c983274c..e6b88970 100644 --- a/data_plane/src/query_engines/asap_query_engine/engine.rs +++ b/data_plane/src/query_engines/asap_query_engine/engine.rs @@ -487,7 +487,7 @@ impl ASAPQueryEngine { ) .map_err(|e| { crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore topk-over-rate fallback reducer failed for `{query}`: \ {e:?} — failing over to archive" @@ -661,7 +661,7 @@ impl ASAPQueryEngine { { let Some(idx) = self.sketch_index.as_ref() else { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!("ASAPQueryEngine: no sketch index for `{query}` — failing over"), )); }; @@ -670,7 +670,7 @@ impl ASAPQueryEngine { if let Some(reason) = &analysis.unsupported { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore analyzer rejected `{query}` for range query: \ {reason:?} — failing over to archive" @@ -679,7 +679,7 @@ impl ASAPQueryEngine { } if analysis.candidates.is_empty() { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore analyzer produced no ASAP-tier candidates for \ `{query}` — failing over to archive" @@ -724,7 +724,7 @@ impl ASAPQueryEngine { sids.extend(idx.instances_matching(&candidate.metric_name, &candidate.group_by_keys)); if sids.is_empty() { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore has no policy for metric `{}` satisfying \ capability {:?} — failing over to archive", @@ -755,7 +755,7 @@ impl ASAPQueryEngine { } if hit_sids.is_empty() { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore has no sid satisfying capability {:?} for \ metric `{}` — failing over to archive", @@ -794,7 +794,8 @@ impl ASAPQueryEngine { ); if is_exact_sum_family && candidate.outer_fn == OuterFn::SumOverTime { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore + .data_source_id(), format!( "SketchStore cannot answer `sum_over_time` over counter \ deltas for `{query}` (issue #301) — failing over to archive" @@ -816,7 +817,7 @@ impl ASAPQueryEngine { ) .map_err(|e| { crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore exact-agg rate reducer failed for `{query}` over \ [{start_ms}, {end_ms}]: {e:?} — failing over to archive" @@ -835,7 +836,8 @@ impl ASAPQueryEngine { ) .map_err(|e| { crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore + .data_source_id(), format!( "SketchStore exact-agg reducer failed for `{query}` over \ [{start_ms}, {end_ms}]: {e:?} — failing over to archive" @@ -865,7 +867,8 @@ impl ASAPQueryEngine { ) .map_err(|e| { crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore + .data_source_id(), format!( "SketchStore reducer failed for `{query}` over \ [{start_ms}, {end_ms}]: {e:?} — failing over to archive" @@ -888,7 +891,7 @@ impl ASAPQueryEngine { let result = combined_result.ok_or_else(|| { crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!("SketchStore reducer produced no result for `{query}`"), ) })?; @@ -1328,7 +1331,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // Branch 1 — the control plane analyzer rejects the shape. if let Some(reason) = &analysis.unsupported { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore analyzer rejected `{query}`: {reason:?} — \ failing over to archive" @@ -1341,7 +1344,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu // when `candidates.is_empty()` but we keep the // belt-and-braces miss-path for safety. return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore analyzer produced no ASAP-tier candidates for \ `{query}` — failing over to archive" @@ -1429,7 +1432,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu &req, ); return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore has no policy for metric `{}` \ with group_by_keys ⊇ {:?} satisfying capability \ @@ -1536,7 +1539,8 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu &req, ); return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore + .data_source_id(), format!( "SketchStore has no sid satisfying capability \ {:?} for metric `{}` — failing over to archive", @@ -1612,7 +1616,8 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu &req, ); return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore + .data_source_id(), format!( "SketchStore FrequencyEstimate sid for metric `{}` cannot \ answer the per-item selector `{}` (CMS/CountSketch return \ @@ -1671,7 +1676,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu &req, ); return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore cannot answer `sum_over_time` over counter \ deltas for metric `{}` (issue #301: Σ-of-cumulative-samples \ @@ -1780,7 +1785,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu ), ) => { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore reducer does not support function `{name}` \ — failing over to archive" @@ -1791,7 +1796,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu function, capability}) => { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore reducer cannot answer `{function}` against \ capability {capability:?} — failing over to archive" @@ -1803,7 +1808,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu encoding, reason}) => { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore reducer failed to decode sketch for sid \ {sid} (encoding={encoding:?}): {reason} — failing over \ @@ -1814,7 +1819,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu Err(crate::storage_engines::sketch_db::query::ASAPTierError::NoData { metric_name: m}) => { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore reducer found no samples for metric \ `{m}` in window — failing over to archive" @@ -1825,7 +1830,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu sid, sketch_kind}) => { return Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!( "SketchStore reducer cannot enumerate top-k for sid \ {sid} (sketch_kind={sketch_kind:?}, no heap) — \ @@ -1917,7 +1922,7 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu ); } Err(crate::query_engines::EngineError::capability_miss( - asap_types::StorageBackend::SketchStore.data_source_id(), + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id(), format!("ASAPQueryEngine: no sketch index for `{query}` — failing over to archive"), )) } @@ -1946,8 +1951,9 @@ impl crate::query_engines::routing::query_engine_routing::QueryEngine for ASAPQu &self, ) -> crate::query_engines::routing::query_engine_routing::EngineCapabilities { crate::query_engines::routing::query_engine_routing::EngineCapabilities { - data_source_id: asap_types::StorageBackend::SketchStore.data_source_id(), - storage_backend: asap_types::StorageBackend::SketchStore, + data_source_id: crate::storage_engines::types::StorageBackend::SketchStore + .data_source_id(), + storage_backend: crate::storage_engines::types::StorageBackend::SketchStore, // Warm-tier sketches are O(sketch-size); call it 16 MiB ceiling // for buffered ops (KLL with k=200 is well below this). supports_streams_above_bytes: 16 * 1024 * 1024, @@ -2627,7 +2633,7 @@ mod asap_tier_classify_tests { EngineError::CapabilityMiss { engine_id, .. } => { assert_eq!( engine_id, - asap_types::StorageBackend::SketchStore.data_source_id() + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id() ); } other => panic!("expected CapabilityMiss, got {other:?}"), @@ -2657,7 +2663,7 @@ mod asap_tier_classify_tests { EngineError::CapabilityMiss { engine_id, .. } => { assert_eq!( engine_id, - asap_types::StorageBackend::SketchStore.data_source_id() + crate::storage_engines::types::StorageBackend::SketchStore.data_source_id() ); } other => panic!("expected CapabilityMiss, got {other:?}"), @@ -4661,8 +4667,9 @@ mod range_stitch_tests { } fn capabilities(&self) -> EngineCapabilities { EngineCapabilities { - data_source_id: asap_types::StorageBackend::GorillaObjectStore.data_source_id(), - storage_backend: asap_types::StorageBackend::GorillaObjectStore, + data_source_id: crate::storage_engines::types::StorageBackend::GorillaObjectStore + .data_source_id(), + storage_backend: crate::storage_engines::types::StorageBackend::GorillaObjectStore, supports_streams_above_bytes: usize::MAX, } } diff --git a/data_plane/src/query_engines/no_data_archive.rs b/data_plane/src/query_engines/no_data_archive.rs index 6ceba7f8..6728ff79 100644 --- a/data_plane/src/query_engines/no_data_archive.rs +++ b/data_plane/src/query_engines/no_data_archive.rs @@ -27,7 +27,7 @@ use async_trait::async_trait; use tracing::info; -use asap_types::StorageBackend; +use crate::storage_engines::types::StorageBackend; use crate::query_engines::routing::{EngineCapabilities, QueryEngine}; use crate::query_engines::{EngineError, QueryResult}; @@ -67,7 +67,7 @@ impl QueryEngine for NoDataArchiveEngine { fn capabilities(&self) -> EngineCapabilities { EngineCapabilities { - data_source_id: asap_types::ENGINE_ID_THANOS_QUERY, + data_source_id: crate::storage_engines::types::ENGINE_ID_THANOS_QUERY, // Register under the canonical archive query-engine id so // archive entries dispatch here transparently when no real // ThanosQueryEngine is configured. @@ -97,7 +97,10 @@ mod tests { fn capabilities_use_no_data_archive_id() { let engine = NoDataArchiveEngine::new(); let caps = engine.capabilities(); - assert_eq!(caps.data_source_id, asap_types::ENGINE_ID_THANOS_QUERY); + assert_eq!( + caps.data_source_id, + crate::storage_engines::types::ENGINE_ID_THANOS_QUERY + ); assert_eq!(caps.storage_backend, StorageBackend::GorillaObjectStore); } } diff --git a/data_plane/src/query_engines/routing/backend_storage_routing.rs b/data_plane/src/query_engines/routing/backend_storage_routing.rs index 833cb9fd..3e267449 100644 --- a/data_plane/src/query_engines/routing/backend_storage_routing.rs +++ b/data_plane/src/query_engines/routing/backend_storage_routing.rs @@ -75,7 +75,7 @@ //! in BOTH wins from `routes:` (multi-target overrides single-target). //! //! Valid `StorageBackend` values mirror the snake-cased serde tags on -//! `asap_types::StorageBackend`: `sketch_store`, +//! `crate::storage_engines::types::StorageBackend`: `sketch_store`, //! `gorilla_object_store`, `double_write`, `prometheus_remote`. (Step-1 //! of the JSONL deprecation refactor removed the `cold_jsonl_fallback` tag.) //! @@ -97,11 +97,14 @@ use std::collections::HashMap; use std::path::Path; use anyhow::{Context, Result}; -use asap_types::{parse_storage_backend_engine_id, StorageBackend, CANONICAL_QUERY_ENGINE_IDS}; use serde::{Deserialize, Serialize}; use serde_json::Value as JsonValue; use tracing::{debug, info}; +use crate::storage_engines::types::{ + parse_storage_backend_engine_id, StorageBackend, CANONICAL_QUERY_ENGINE_IDS, +}; + // --------------------------------------------------------------------------- // Query-shape taxonomy // --------------------------------------------------------------------------- diff --git a/data_plane/src/query_engines/routing/capability_matching.rs b/data_plane/src/query_engines/routing/capability_matching.rs new file mode 100644 index 00000000..af19b0cd --- /dev/null +++ b/data_plane/src/query_engines/routing/capability_matching.rs @@ -0,0 +1,212 @@ +//! Storage-backend routing policy: given a `(statistic, accuracy)` query +//! and the storage tier a metric is configured for, decide which backends +//! can serve it and in what preference order. +//! +//! Split out of `asap_types`'s former `capability_matching` module (see +//! `scratchpad/artifacts/enum-unification-plan.md`). `StorageBackend` and +//! the `StreamingConfig` wire format it's a field of both turned out to +//! have zero real `control_plane` dependency either — see +//! [`crate::storage_engines::types::storage_backend`]'s module doc — so +//! both moved into this crate; only `AccuracyTarget` and the routing +//! *decision* below were ever split out separately, since neither has a +//! shared-struct-field reason to exist. Exercised only by this crate's +//! own [`super::query_engine_routing`]. + +use asap_types::Statistic; +use serde::{Deserialize, Serialize}; + +use crate::storage_engines::types::StorageBackend; + +/// Accuracy hint pushed by the controller at intent-binding time +/// (`controller/docs/design.md` §6 `core::workload`). The Phase-5 capability +/// router consults this to decide whether a metric configured for both warm- +/// tier and Gorilla-S3 should answer from the archive (Exact) or the +/// approximate ASAP-tier sketch. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub enum AccuracyTarget { + /// Caller demands an exact answer; ASAP-tier sketches are not eligible + /// unless they happen to be exact accumulators (Sum, MinMax, Increase). + Exact, + /// Caller accepts ε/δ-bounded approximate answers. Default. + #[default] + Approximate, +} + +/// Returns the storage backends that can serve a `(statistic, accuracy)` +/// query when the metric is configured for `metric_storage_config`. +/// +/// The returned list is **ordered by preference**: the router walks it in +/// order and dispatches to the first backend whose engine is registered, +/// falling through on `CapabilityMiss` / `Backend` to the next entry. +/// +/// **ASAP-first centralization refactor**: the decision tree is now +/// owned here (and consumed identically by `EngineRouter::execute` and +/// `EngineRouter::execute_range`) so the HTTP transport layer never +/// re-derives routing. The policy is: +/// +/// * `accuracy == Exact` → archive only `[GorillaObjectStore]` (served +/// by the `thanos_query` engine). The caller demands an exact answer, +/// so the ε/δ-bounded ASAP-tier sketches are not eligible — go +/// straight to the archive regardless of where the metric is stored. +/// * `accuracy == Approximate` (the default) → ASAP-first failover +/// `[SketchStore, GorillaObjectStore]` for any metric stored in an +/// ASAP-managed tier (`SketchStore`, `GorillaObjectStore`, or +/// `DoubleWrite`): try the warm sketch (`asap_query`) first and fall +/// back to the archive (`thanos_query`) on a capability miss. This +/// collapses the old per-`metric_storage` sequences into one shared +/// ASAP-first-then-archive contract. +/// * `PrometheusRemote` keeps its own single-backend sequence +/// `[PrometheusRemote]` (Phase ε.2): the metric's raw samples never +/// landed in ASAP-managed storage, so there is no ASAP-tier sketch to +/// fall back on and the accuracy hint does not apply. A missing +/// engine surfaces as a `NoEngineRegistered` 503 from the HTTP +/// handler — the correct fail-loud behaviour for a misconfigured +/// deploy. +pub fn compatible_storage_backends( + _stat: Statistic, + accuracy: AccuracyTarget, + metric_storage_config: StorageBackend, +) -> Vec { + match metric_storage_config { + // Prometheus-remote owns its own storage; the accuracy hint does + // not apply and there is no ASAP-tier sketch to fall back on. + StorageBackend::PrometheusRemote => vec![StorageBackend::PrometheusRemote], + + // Every ASAP-managed tier shares the same ASAP-first policy, + // gated only on the accuracy target. + StorageBackend::SketchStore + | StorageBackend::GorillaObjectStore + | StorageBackend::DoubleWrite => match accuracy { + // Exact: archive only — the warm sketches are ε/δ-bounded. + AccuracyTarget::Exact => vec![StorageBackend::GorillaObjectStore], + // Approximate: ASAP-tier first, archive (Thanos) fallback. + AccuracyTarget::Approximate => vec![ + StorageBackend::SketchStore, + StorageBackend::GorillaObjectStore, + ], + }, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + // ----------------------------------------------------------------------- + // Phase-5: storage-backend routing + // + // The Phase-5 `EngineRouter` (see `query_engine_routing.rs`) consults + // `compatible_storage_backends(stat, accuracy, metric_storage)` to pick + // a backend. These tests pin the routing matrix so the dispatcher stays + // in lock-step with the design doc §8. + // ----------------------------------------------------------------------- + + #[test] + fn exact_accuracy_routes_to_archive_only() { + // ASAP-first refactor: `Exact` goes straight to the archive + // (Thanos via the GorillaObjectStore slot) regardless of where + // the metric is stored — the warm sketches are ε/δ-bounded. + for cfg in [ + StorageBackend::GorillaObjectStore, + StorageBackend::SketchStore, + StorageBackend::DoubleWrite, + ] { + let backends = compatible_storage_backends(Statistic::Sum, AccuracyTarget::Exact, cfg); + assert_eq!( + backends, + vec![StorageBackend::GorillaObjectStore], + "Exact accuracy must route to archive only for {cfg:?}", + ); + } + } + + #[test] + fn approximate_accuracy_is_asap_first_with_archive_fallback() { + // ASAP-first refactor: every ASAP-managed tier shares the same + // `[SketchStore, GorillaObjectStore]` sequence for `Approximate` + // — try the warm sketch first, fall back to the Thanos archive + // on a capability miss. + for cfg in [ + StorageBackend::SketchStore, + StorageBackend::GorillaObjectStore, + StorageBackend::DoubleWrite, + ] { + let backends = + compatible_storage_backends(Statistic::Quantile, AccuracyTarget::Approximate, cfg); + assert_eq!( + backends, + vec![ + StorageBackend::SketchStore, + StorageBackend::GorillaObjectStore, + ], + "Approximate accuracy must be ASAP-first then archive for {cfg:?}", + ); + } + } + + /// Source-of-truth agreement check for the storage axis. + /// + /// For every `(Statistic, AccuracyTarget, StorageBackend)` triple + /// the returned backend list must be non-empty and its head must + /// match the routing matrix in `compatible_storage_backends`'s + /// docstring. + #[test] + fn capability_storage_backend_agreement() { + let stats = [ + Statistic::Count, + Statistic::Sum, + Statistic::Cardinality, + Statistic::Increase, + Statistic::Rate, + Statistic::Min, + Statistic::Max, + Statistic::Quantile, + Statistic::Topk, + ]; + let accuracies = [AccuracyTarget::Exact, AccuracyTarget::Approximate]; + let configs = [ + StorageBackend::SketchStore, + StorageBackend::GorillaObjectStore, + StorageBackend::DoubleWrite, + StorageBackend::PrometheusRemote, + ]; + + for &stat in &stats { + for &acc in &accuracies { + for &cfg in &configs { + let backends = compatible_storage_backends(stat, acc, cfg); + assert!( + !backends.is_empty(), + "compatible_storage_backends({stat:?}, {acc:?}, {cfg:?}) returned empty \ + — every metric configuration must route to at least one backend", + ); + let last = *backends.last().unwrap(); + assert!( + last == StorageBackend::SketchStore + || last == StorageBackend::GorillaObjectStore + || last == StorageBackend::PrometheusRemote, + "backend list for ({stat:?}, {acc:?}, {cfg:?}) must terminate in a \ + dispatchable failover (SketchStore, GorillaObjectStore, or \ + PrometheusRemote); got {last:?}", + ); + // ASAP-first refactor: the head is determined by + // `(metric_storage_config, accuracy)`. PrometheusRemote + // keeps its single-backend slot; every ASAP-managed tier + // goes archive-only on `Exact` and ASAP-tier-first on + // `Approximate`. + let expected_head = match (cfg, acc) { + (StorageBackend::PrometheusRemote, _) => StorageBackend::PrometheusRemote, + (_, AccuracyTarget::Exact) => StorageBackend::GorillaObjectStore, + (_, AccuracyTarget::Approximate) => StorageBackend::SketchStore, + }; + assert_eq!( + backends[0], expected_head, + "head mismatch for ({stat:?}, {acc:?}, {cfg:?}): expected {expected_head:?}, got {:?}", + backends[0], + ); + } + } + } + } +} diff --git a/data_plane/src/query_engines/routing/mod.rs b/data_plane/src/query_engines/routing/mod.rs index b199e6d7..b0ac5884 100644 --- a/data_plane/src/query_engines/routing/mod.rs +++ b/data_plane/src/query_engines/routing/mod.rs @@ -13,8 +13,14 @@ //! * [`query_engine_routing`] — the engine dispatcher. Holds a small map //! of `data_source_id → Arc` and walks the //! compatibility list returned by -//! [`asap_types::compatible_storage_backends`] to pick which +//! [`capability_matching::compatible_storage_backends`] to pick which //! engine answers a given `(query, metric_storage)` pair. +//! * [`capability_matching`] — the storage-backend routing policy itself +//! (`AccuracyTarget`, `compatible_storage_backends`). Split out of +//! `asap_types`'s former `capability_matching` module. `StorageBackend` +//! itself later moved into this crate too, alongside `StreamingConfig` +//! (see `crate::storage_engines::types::storage_backend`'s module doc) — +//! `control_plane` turned out to have zero real dependency on either. //! //! Step-1 of the JSONL deprecation refactor lifted these out of //! `data_model/backend_storage_routing.rs` and `query-engines/router.rs` @@ -23,9 +29,12 @@ //! instead of straddling two unrelated module trees. pub mod backend_storage_routing; +pub mod capability_matching; pub mod freshness_probe_cache; pub mod query_engine_routing; +pub use capability_matching::{compatible_storage_backends, AccuracyTarget}; + pub use backend_storage_routing::{ classify_query_shape, routing_table_hash, BackendStorageRouting, HotReloadBackendStorageRouting, QueryShape, RoutingTarget, DEFAULT_TENANT, diff --git a/data_plane/src/query_engines/routing/query_engine_routing.rs b/data_plane/src/query_engines/routing/query_engine_routing.rs index ce69147e..010abe9d 100644 --- a/data_plane/src/query_engines/routing/query_engine_routing.rs +++ b/data_plane/src/query_engines/routing/query_engine_routing.rs @@ -2,9 +2,9 @@ //! to the engine that owns the chosen storage tier. //! //! The router holds a small map keyed by -//! [`asap_types::StorageBackend::data_source_id`] +//! [`crate::storage_engines::types::StorageBackend::data_source_id`] //! and walks the ordered backend list returned by -//! [`asap_types::compatible_storage_backends`]. The first registered +//! [`super::capability_matching::compatible_storage_backends`]. The first registered //! engine answers; on a recoverable backend failure (`EngineError::Backend`), //! the router falls through to the next compatible backend if the list //! still has options. A hard capability miss in the head engine likewise @@ -20,9 +20,10 @@ use async_trait::async_trait; use thiserror::Error; use tracing::{debug, warn}; +use crate::storage_engines::types::StorageBackend; use asap_types::Statistic; -use asap_types::{compatible_storage_backends, AccuracyTarget, StorageBackend}; +use super::capability_matching::{compatible_storage_backends, AccuracyTarget}; use crate::query_engines::{EngineError, QueryResult}; // --------------------------------------------------------------------------- diff --git a/data_plane/src/query_engines/thanos_query_engine/forward.rs b/data_plane/src/query_engines/thanos_query_engine/forward.rs index 6814cf85..e33da377 100644 --- a/data_plane/src/query_engines/thanos_query_engine/forward.rs +++ b/data_plane/src/query_engines/thanos_query_engine/forward.rs @@ -59,7 +59,7 @@ pub const DEFAULT_THANOS_QUERY_URL: &str = "http://thanos-query:10903"; /// for explicit per-query overrides via the `X-ASAP-Engine` header /// or the `?engine=` query param. Pinned so dashboards / e2e /// scripts can byte-compare without parsing. -pub const DATA_SOURCE_THANOS_QUERY_ID: &str = asap_types::ENGINE_ID_THANOS_QUERY; +pub const DATA_SOURCE_THANOS_QUERY_ID: &str = crate::storage_engines::types::ENGINE_ID_THANOS_QUERY; /// Marker line every `ThanosQueryEngine` answer carries on its /// `infos` array. Pinned so dashboards and the upcoming Step-2.4 @@ -411,7 +411,7 @@ impl QueryEngine for ThanosQueryEngine { // Re-uses the archive tier slot in the routing // matrix; Path A2 swaps the engine answering, not the // tier classification. See module docstring. - storage_backend: asap_types::StorageBackend::GorillaObjectStore, + storage_backend: crate::storage_engines::types::StorageBackend::GorillaObjectStore, // Forwarder doesn't materialise samples locally; // upstream thanos-query owns the memory budget. We // surface a generous ceiling so the cost-aware @@ -933,7 +933,7 @@ mod tests { assert_eq!(caps.data_source_id, DATA_SOURCE_THANOS_QUERY_ID); assert_eq!( caps.storage_backend, - asap_types::StorageBackend::GorillaObjectStore, + crate::storage_engines::types::StorageBackend::GorillaObjectStore, "Path A2 re-uses the archive tier slot in the routing matrix", ); } diff --git a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs index 59ed4bdd..d5c9ec55 100644 --- a/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs +++ b/data_plane/src/storage_engines/sketch_db/lifecycle/reconcile.rs @@ -35,8 +35,8 @@ use std::collections::{BTreeSet, HashSet}; use std::sync::Arc; use std::time::Duration; +use crate::storage_engines::types::StreamingConfig; use asap_types::aggregation_config::AggregationConfig; -use asap_types::streaming_config::StreamingConfig; use crate::storage_engines::sketch_db::data::{canonical_parameters, AggKind}; use crate::storage_engines::sketch_db::index::SketchStore; diff --git a/data_plane/src/storage_engines/types/mod.rs b/data_plane/src/storage_engines/types/mod.rs index 69b21d0e..185ab78b 100644 --- a/data_plane/src/storage_engines/types/mod.rs +++ b/data_plane/src/storage_engines/types/mod.rs @@ -11,6 +11,8 @@ pub mod hot_reload_config; pub mod key_by_label_values; pub mod measurement; pub mod precomputed_output; +pub mod storage_backend; +pub mod streaming_config; pub mod traits; pub use enums::*; @@ -18,13 +20,14 @@ pub use hot_reload_config::*; pub use key_by_label_values::*; pub use measurement::*; pub use precomputed_output::*; +pub use storage_backend::*; +pub use streaming_config::*; pub use traits::*; -// Cross-module re-exports of asap_types data types so callers can -// write `crate::storage_engines::types::StreamingConfig` instead of reaching -// across crates. +// Cross-module re-export of asap_types data types so callers can +// write `crate::storage_engines::types::AggregationConfig` instead of +// reaching across crates. pub use asap_types::aggregation_config::*; -pub use asap_types::streaming_config::*; // Re-export the query-side routing surface so existing call sites // like `crate::storage_engines::types::BackendStorageRouting` keep compiling. diff --git a/data_plane/src/storage_engines/types/storage_backend.rs b/data_plane/src/storage_engines/types/storage_backend.rs new file mode 100644 index 00000000..e7f4075d --- /dev/null +++ b/data_plane/src/storage_engines/types/storage_backend.rs @@ -0,0 +1,140 @@ +use serde::{Deserialize, Serialize}; + +pub const ENGINE_ID_ASAP_QUERY: &str = "asap_query"; +pub const ENGINE_ID_THANOS_QUERY: &str = "thanos_query"; + +pub const CANONICAL_QUERY_ENGINE_IDS: &[&str] = &[ENGINE_ID_ASAP_QUERY, ENGINE_ID_THANOS_QUERY]; + +// --------------------------------------------------------------------------- +// Phase-5: storage-backend capability axis +// +// Matching on `(metric, statistic, sub_type, window_size, grouping_labels, +// spatial_filter)` alone has no axis for "which storage tier serves this +// query." The Phase-5 `GorillaQueryEngine` (PR #85) introduces a parallel +// exact tier; the planner / router needs to disambiguate between ASAP-tier +// sketches and Gorilla-S3 chunks. See `docs/design-gorilla-s3-cold-engine.md` +// §8. +// --------------------------------------------------------------------------- + +/// Which physical storage tier a query (or a metric configuration) routes to. +/// +/// `SketchStore` is the default — every existing `AggregationConfig` and +/// `StreamingConfig` decodes into this variant via `#[serde(default)]`, so +/// pre-Phase-5 deploys keep dispatching to `ASAPQueryEngine` unchanged. +/// +/// **Step-1 of the JSONL deprecation refactor** removed the +/// `ColdJsonlFallback` variant. The legacy local-FS JSONL leg +/// (`LocalFsColdStore`, `parse_jsonl`, the §5.2 raw-store +/// fallback) was deleted at the same commit; the surviving +/// failover surface is ASAP-tier sketch ↔ Thanos archive. +/// +/// Formerly `asap_types::capability_matching::StorageBackend` (then +/// `asap_types::storage_backend::StorageBackend`). Moved here alongside +/// `StreamingConfig` (see `scratchpad/artifacts/enum-unification-plan.md`) +/// once auditing real call sites showed `control_plane` never actually +/// depends on this type or `StreamingConfig` — it emits wire-compatible +/// JSON by hand via its own `StreamingConfigEmitter`, never importing +/// either. See [`super::streaming_config`]'s module doc for the fuller +/// story. The routing *policy* (`AccuracyTarget`, +/// `compatible_storage_backends`) already lived in +/// `data_plane::query_engines::routing::capability_matching`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize, Default)] +#[serde(rename_all = "snake_case")] +pub enum StorageBackend { + /// Warm-tier sketch DB (today's `SketchStore` + accumulators). + /// Served by `ASAPQueryEngine`. Default for unconfigured metrics. + #[default] + SketchStore, + + /// Thanos archive over MinIO/S3. The enum name is kept for + /// serde/back-compat with existing configs, but its canonical + /// query-engine identity is `thanos_query`. Gorilla is an + /// archive chunk format/storage detail, not a public query engine. + GorillaObjectStore, + + /// Double-write: the metric is written to both ASAP-tier sketches AND the + /// Gorilla-S3 archive. Capability matching surfaces both options and the + /// cost-aware dispatcher picks per query (typically ASAP-tier for low- + /// latency approximate, archive for exact). + DoubleWrite, + + /// Prometheus-remote: the metric's data is shipped raw to a + /// Prometheus instance via the native OTLP receiver. Phase ε.2 + /// registers a `PrometheusForwardEngine` (HTTP-forwarder to + /// Prometheus's `/api/v1/query`) under this slot so the + /// controller's `RawAtEdgePrometheusArchive` mode can route a + /// metric's queries to Prometheus directly. Mirrors the + /// `GorillaObjectStore` slot's "single backend, no failover" + /// semantics — there is no ASAP-tier sketch to fall back on for a + /// Prometheus-remote metric. + PrometheusRemote, +} + +impl StorageBackend { + /// Canonical string tag pinned for byte-comparable dispatch on the wire (mirrors + /// the `data_source: ` info-line on `QueryResult`). Engines + /// register themselves under these IDs in the router. + pub const fn data_source_id(self) -> &'static str { + match self { + StorageBackend::SketchStore => ENGINE_ID_ASAP_QUERY, + StorageBackend::GorillaObjectStore => ENGINE_ID_THANOS_QUERY, + StorageBackend::DoubleWrite => "double_write", + StorageBackend::PrometheusRemote => "prometheus_remote", + } + } +} + +pub fn parse_storage_backend_engine_id(s: &str) -> Option { + match s { + ENGINE_ID_ASAP_QUERY => Some(StorageBackend::SketchStore), + ENGINE_ID_THANOS_QUERY => Some(StorageBackend::GorillaObjectStore), + "double_write" => Some(StorageBackend::DoubleWrite), + "prometheus_remote" => Some(StorageBackend::PrometheusRemote), + _ => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn storage_backend_default_is_asap_tier() { + // `#[serde(default)]` on `StreamingConfig.storage_backend` (and on + // `StorageBackend::default()`) MUST be `SketchStore` so pre-Phase-5 + // configs decode without bumping deploys onto the archive. + assert_eq!(StorageBackend::default(), StorageBackend::SketchStore); + } + + #[test] + fn storage_backend_data_source_id_is_pinned() { + // The router registers engines by these strings; dashboards + // byte-compare them. Pin to catch accidental rename. + assert_eq!( + StorageBackend::SketchStore.data_source_id(), + ENGINE_ID_ASAP_QUERY + ); + assert_eq!( + StorageBackend::GorillaObjectStore.data_source_id(), + ENGINE_ID_THANOS_QUERY, + ); + assert_eq!(StorageBackend::DoubleWrite.data_source_id(), "double_write",); + assert_eq!( + StorageBackend::PrometheusRemote.data_source_id(), + "prometheus_remote", + ); + } + + #[test] + fn storage_backend_engine_id_parser_accepts_only_canonical_query_engines() { + assert_eq!( + parse_storage_backend_engine_id(ENGINE_ID_ASAP_QUERY), + Some(StorageBackend::SketchStore), + ); + assert_eq!( + parse_storage_backend_engine_id(ENGINE_ID_THANOS_QUERY), + Some(StorageBackend::GorillaObjectStore), + ); + assert_eq!(parse_storage_backend_engine_id("not_an_engine"), None); + } +} diff --git a/crates/asap_types/src/streaming_config.rs b/data_plane/src/storage_engines/types/streaming_config.rs similarity index 82% rename from crates/asap_types/src/streaming_config.rs rename to data_plane/src/storage_engines/types/streaming_config.rs index 9eb59b98..a390b8fa 100644 --- a/crates/asap_types/src/streaming_config.rs +++ b/data_plane/src/storage_engines/types/streaming_config.rs @@ -6,55 +6,24 @@ use std::fs::File; use std::io::BufReader; use std::ops::Index; -use crate::aggregation_config::AggregationConfig; -use crate::capability_matching::StorageBackend; -use crate::enums::QueryLanguage; -use crate::policy_registry::PolicyRegistry; +use asap_types::enums::QueryLanguage; +use asap_types::{AggregationConfig, MonitorSpec, PolicyRegistry}; -/// One continuous-monitoring (CDM) threshold spec. The data-plane monitor -/// coordinator owns the AUTHORITATIVE `tau`/`epsilon`/`window_ms` (the edge -/// copy is advisory), keyed by the same content-addressed `agg_id` the edge and -/// coordinator share. `key` is the CMS point-frequency key for point monitors -/// (empty for Sum / whole-stream). See -/// `ASAPCollector/docs/continuous-monitoring-tumbling-cost-analysis.md`. -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct MonitorSpec { - pub agg_id: u64, - /// Additive readout the edge reports: "sum" (default), "cms_point", "f2". - /// Pass-through metadata so the edge can auto-learn its reporting mode from - /// the pushed config; the coordinator allocation is value-driven and does not - /// branch on it (p_i ∝ √(value/rate) is the F2 allocation when value=‖f‖²). - #[serde(default)] - pub functional: String, - /// CMS point-frequency key x; empty (default) for Sum / whole-stream / F2. - #[serde(default)] - pub key: String, - /// Threshold τ (authoritative here, not at the edge). - pub tau: f64, - /// Relative tolerance ε; the alert fires when the estimate reaches (1−ε)τ. - #[serde(default = "default_monitor_epsilon")] - pub epsilon: f64, - /// Tumbling epoch length in ms; MUST match the edge window for this agg_id. - pub window_ms: u64, - /// Count-Sketch depth (rows) for whole-sketch `functional="f2"` monitors. - /// 0 (default) for scalar monitors; MUST match the edge's Count-Sketch for - /// this agg when F2 (both sides square/merge the same cell matrix). - #[serde(default)] - pub d: usize, - /// Count-Sketch width (buckets/row) for F2 monitors; 0 for scalar. - #[serde(default)] - pub w: usize, - /// F2 monitoring variant: "distributed" (default, ship every window) or - /// "geometric" (Sharfman–Schuster–Keren safe-zone, ship on local violation). - /// Ignored by scalar monitors. - #[serde(default)] - pub mode: String, -} - -fn default_monitor_epsilon() -> f64 { - 0.05 -} +use super::storage_backend::StorageBackend; +/// The backend's active streaming policy config: every `AggregationConfig` +/// currently pushed by the controller, plus the storage-backend pin and CDM +/// monitor specs. +/// +/// Formerly `asap_types::streaming_config::StreamingConfig` — moved here +/// (see `scratchpad/artifacts/enum-unification-plan.md`) because +/// `control_plane` never actually depended on this type: its own +/// `StreamingConfigEmitter` hand-builds wire-compatible JSON independently, +/// and `PolicyRegistry::from_streaming_config` (the only thing that made +/// `asap_types::PolicyRegistry` -- genuinely shared -- look coupled to this +/// type) had exactly one real caller, this struct's own `policy_registry()` +/// method below. `asap_types` keeps the lower-level `PolicyRegistry:: +/// from_configs` primitive this method now calls directly. #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StreamingConfig { pub aggregation_configs: HashMap, @@ -120,7 +89,7 @@ impl StreamingConfig { } /// Derived content-addressed view. Builds a [`PolicyRegistry`] keyed - /// on [`crate::PolicyFingerprint`] — the merged-sid-identity-chain + /// on [`asap_types::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. @@ -130,7 +99,7 @@ impl StreamingConfig { /// 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) + PolicyRegistry::from_configs(self.aggregation_configs.values().cloned()) } pub fn from_yaml_file(yaml_file: &str) -> Result { diff --git a/data_plane/src/tests/capability_miss_http_e2e_tests.rs b/data_plane/src/tests/capability_miss_http_e2e_tests.rs index 1e7c147b..68a78a36 100644 --- a/data_plane/src/tests/capability_miss_http_e2e_tests.rs +++ b/data_plane/src/tests/capability_miss_http_e2e_tests.rs @@ -99,8 +99,8 @@ fn canned_plan_yaml(_agg_id: u64, metric: &str) -> String { fn expected_fp_for(metric: &str) -> u64 { let yaml = canned_plan_yaml(0, metric); let data: serde_yaml::Value = serde_yaml::from_str(&yaml).expect("yaml parses"); - let sc = - asap_types::streaming_config::StreamingConfig::from_yaml_data(&data).expect("yaml decodes"); + let sc = crate::storage_engines::types::StreamingConfig::from_yaml_data(&data) + .expect("yaml decodes"); *sc.aggregation_configs .keys() .next()