diff --git a/asap-query-engine/src/data_model/backend_storage_routing.rs b/asap-query-engine/src/data_model/backend_storage_routing.rs index c6ff9d40..7b60a1c6 100644 --- a/asap-query-engine/src/data_model/backend_storage_routing.rs +++ b/asap-query-engine/src/data_model/backend_storage_routing.rs @@ -23,17 +23,57 @@ //! metric whose storage backend differs from the default — that the //! HTTP handler consults to pick the right engine for each query. //! +//! ## v7: dual-routing per metric +//! +//! v6.1 surfaced an architectural gap: routing one metric to one engine +//! forces an exclusive trade-off between criterion ④ (warm-tier +//! accuracy) and criterion ⑤ (cold-fallback). Every quantile/sum-by +//! query on `http_requests_total` had to go to either the warm tier +//! (so the accuracy reducer could compute relative error) or the +//! archive (so the `data_source: gorilla_archive` info-line landed on +//! the cold-fallback probe). v7 closes this by letting one metric have +//! multiple targets, each with an optional query-shape filter; the +//! HTTP handler inspects the parsed PromQL and picks the matching +//! target. Predictable / planned queries (quantile, sum_over_time) +//! land on the warm tier; ad-hoc / post-hoc queries +//! (count, topk, rate-post-hoc) route to the cold archive. +//! //! ## Schema //! +//! Two compatible shapes are accepted. The single-target form is +//! preserved verbatim from v6.1 so existing deploys keep working +//! unchanged: +//! //! ```yaml -//! # deploy/configs/backend-storage-routing.yaml -//! default: sketch_warm_tier # StorageBackend; optional +//! # v6.1 form (single-target): +//! default: sketch_warm_tier //! metrics: -//! http_requests_total: gorilla_s3_archive -//! audit_events: gorilla_s3_archive -//! foo_count: cold_jsonl_fallback +//! audit_events: gorilla_s3_archive +//! ``` +//! +//! ```yaml +//! # v7 form (multi-target with query-shape selection): +//! default: sketch_warm_tier +//! routes: +//! - metric: http_requests_total +//! targets: +//! - backend: sketch_warm_tier +//! # default — predictable / planned queries land here +//! - backend: gorilla_s3_archive +//! applies_to_query_shape: [count, topk, rate_post_hoc] +//! - metric: http_freshness_probe_warm +//! targets: +//! - backend: sketch_warm_tier +//! - metric: http_freshness_probe_archive +//! targets: +//! - backend: gorilla_s3_archive //! ``` //! +//! The two shapes can be mixed in the same YAML — metrics under +//! `metrics:` keep the old single-target semantics; metrics under +//! `routes:` use the new list-of-targets semantics. A metric listed +//! in BOTH wins from `routes:` (multi-target overrides single-target). +//! //! Valid `StorageBackend` values mirror the snake-cased serde tags on //! `asap_types::StorageBackend`: `sketch_warm_tier`, //! `gorilla_s3_archive`, `cold_jsonl_fallback`, `double_write`. @@ -47,9 +87,7 @@ //! //! * Hot reload — the controller's plan-push is the long-term answer //! for per-metric routing; this YAML layer is the bridge that -//! unblocks issue #46 criterion ⑤ until the plan-push lands. Adding -//! hot reload is a one-line `ArcSwap` swap; deferred for now to keep -//! the diff small and reviewable. +//! unblocks issue #46 criteria ④/⑤/⑥ until the plan-push lands. //! * Per-`(metric, statistic, accuracy)` granularity — `StorageBackend` //! already encodes the `DoubleWrite` axis the cost-aware dispatcher //! uses to pick warm-vs-archive per query. @@ -62,18 +100,229 @@ use asap_types::StorageBackend; use serde::{Deserialize, Serialize}; use tracing::{debug, info}; +// --------------------------------------------------------------------------- +// Query-shape taxonomy +// --------------------------------------------------------------------------- + +/// A coarse-grained classification of an incoming PromQL query. The +/// HTTP handler extracts this from the parsed AST and consults the +/// routing table's `applies_to_query_shape` filters to pick a target. +/// +/// The shapes intentionally mirror the v7 spec's +/// `[count, topk, rate_post_hoc]` enumeration — each is a PromQL +/// shape the cold-archive engine answers natively, and which the +/// warm-tier sketch path either can't serve at all (count over an +/// approximate sketch is misleading) or serves with worse precision +/// than the archive (rate post-hoc). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum QueryShape { + /// `count({...})` — series count after label predicates. + /// Cold archive serves exactly via postings index; warm tier has + /// no compatible aggregation (a CMS doesn't track per-series + /// existence). + Count, + /// `topk(k, ...)` — top-k by value. Cold archive answers by + /// streaming through chunk samples and tracking k-largest; warm + /// tier needs a `CountMinSketchWithHeap` to answer at all. + Topk, + /// `rate([])` — per-second rate over a range. + /// Marked "post-hoc" because the warm tier's pre-computed + /// `Increase` aggregation answers `rate` natively for known + /// queries, so this shape only kicks in for ad-hoc rate queries + /// the controller didn't pre-plan for. + RatePostHoc, + /// `quantile_over_time(φ, [])` and + /// `quantile(...)` aggregations. Warm tier serves natively via + /// DDSketch / KLL accumulators; cold archive can also serve but + /// at higher latency. + Quantile, + /// `sum_over_time(...)` and `sum(...) by (...)`. Warm tier + /// serves via a sum-typed accumulator. + Sum, + /// `last_over_time([])`. Used by the v6 + /// freshness probes — warm tier serves via a counter-typed + /// accumulator (Change B), archive serves by selecting the + /// most-recent sample in each chunk. + LastOverTime, + /// Anything else — `min/max_over_time`, `count_over_time`, + /// `avg_over_time`, etc. Lets the routing table register + /// targets that catch the long tail without enumerating every + /// PromQL function. + Other, +} + +impl QueryShape { + /// Stable string tag used in YAML. + pub fn as_str(self) -> &'static str { + match self { + QueryShape::Count => "count", + QueryShape::Topk => "topk", + QueryShape::RatePostHoc => "rate_post_hoc", + QueryShape::Quantile => "quantile", + QueryShape::Sum => "sum", + QueryShape::LastOverTime => "last_over_time", + QueryShape::Other => "other", + } + } +} + +/// Classify a parsed PromQL expression into a [`QueryShape`]. Walks +/// the AST and returns the first shape that matches any node — so +/// `topk(5, sum by (zone) (rate(http_requests_total[5m])))` is +/// classified as `Topk` (the outermost shape wins). +/// +/// `RatePostHoc` is conservative: any `rate(...)` call surfaces as +/// `RatePostHoc`. The routing-table consumer can decide whether to +/// honour that or fall through to the default target. +pub fn classify_query_shape(expr: &promql_parser::parser::Expr) -> QueryShape { + use promql_parser::parser::Expr; + match expr { + // Aggregations are the outermost shape — `topk(...)` wins + // over any inner call. We use the operator's Display impl + // (the canonical PromQL keyword: "sum", "count", "topk", + // "quantile", ...) — Debug-formatting the underlying + // `TokenType` returns the numeric token id, not the + // keyword. + Expr::Aggregate(agg) => { + let op = agg.op.to_string().to_lowercase(); + if op == "topk" || op == "bottomk" { + QueryShape::Topk + } else if op == "count" || op == "count_values" { + QueryShape::Count + } else if op == "quantile" { + QueryShape::Quantile + } else if op == "sum" { + // `sum by (...) (...)` — recurse on the inner + // expression; if the inner is a `rate(...)` the + // post-hoc rate path wins. + let inner = classify_query_shape(&agg.expr); + if matches!(inner, QueryShape::RatePostHoc) { + QueryShape::RatePostHoc + } else { + QueryShape::Sum + } + } else { + // min/max/avg/group/stddev/stdvar/... + classify_query_shape(&agg.expr) + } + } + Expr::Call(call) => { + let name = call.func.name.to_lowercase(); + if name == "rate" || name == "irate" { + QueryShape::RatePostHoc + } else if name == "quantile_over_time" { + QueryShape::Quantile + } else if name == "sum_over_time" { + QueryShape::Sum + } else if name == "count_over_time" { + QueryShape::Count + } else if name == "last_over_time" { + QueryShape::LastOverTime + } else { + QueryShape::Other + } + } + Expr::Paren(p) => classify_query_shape(&p.expr), + Expr::Unary(u) => classify_query_shape(&u.expr), + Expr::Binary(bin) => { + // Pick whichever side has the more-specific shape; a + // `rate(...)` on either side is enough to mark + // RatePostHoc. + let l = classify_query_shape(&bin.lhs); + if !matches!(l, QueryShape::Other) { + l + } else { + classify_query_shape(&bin.rhs) + } + } + Expr::Subquery(sq) => classify_query_shape(&sq.expr), + // Bare vector / matrix selectors — no function applied. + _ => QueryShape::Other, + } +} + +// --------------------------------------------------------------------------- +// On-disk YAML schema +// --------------------------------------------------------------------------- + +/// One target in the v7 multi-target form. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +struct RoutingTargetYaml { + backend: StorageBackend, + /// Optional list of query shapes this target applies to. When + /// `None`, the target is the default — it catches any shape the + /// other targets didn't claim. When `Some(list)`, the target + /// only fires when the incoming query's shape is in `list`. + #[serde(default, skip_serializing_if = "Option::is_none")] + applies_to_query_shape: Option>, +} + +/// One row in the v7 multi-target form. +#[derive(Debug, Clone, Serialize, Deserialize)] +struct RouteYaml { + metric: String, + targets: Vec, +} + /// On-disk YAML schema. Public only so the loader / tests can build it /// from literals; runtime callers should go through /// [`BackendStorageRouting`]. #[derive(Debug, Clone, Default, Serialize, Deserialize)] struct BackendStorageRoutingYaml { - /// Fallback storage backend for any metric not explicitly listed in - /// `metrics`. Optional; defaults to `SketchWarmTier`. + /// Fallback storage backend for any metric not explicitly listed. + /// Optional; defaults to `SketchWarmTier`. #[serde(default)] default: StorageBackend, - /// Per-metric overrides keyed by the bare metric name (no labels). + /// v6.1 form — per-metric overrides keyed by the bare metric name + /// (no labels). Each value is a single `StorageBackend`. A metric + /// listed here keeps the v6.1 semantics: every query for the + /// metric routes to the named backend regardless of shape. #[serde(default)] metrics: HashMap, + /// v7 form — per-metric overrides as a list of `(backend, + /// applies_to_query_shape)` targets. The HTTP handler inspects + /// the parsed PromQL, classifies it via [`classify_query_shape`], + /// and picks the first target whose `applies_to_query_shape` + /// either is `None` (default) or contains the query's shape. + /// Falls back to the first target on no match. + #[serde(default)] + routes: Vec, +} + +// --------------------------------------------------------------------------- +// In-memory routing table +// --------------------------------------------------------------------------- + +/// One target in the in-memory routing table — the runtime form of +/// [`RoutingTargetYaml`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RoutingTarget { + /// Storage backend the HTTP handler should dispatch to. + pub backend: StorageBackend, + /// Optional list of query shapes this target claims. `None` = + /// default (always fires); `Some(list)` = this target only fires + /// when the query's classified shape is in `list`. + pub applies_to_query_shape: Option>, +} + +impl RoutingTarget { + /// Build a target that applies to every query shape (the v6.1 + /// single-target form). + pub fn always(backend: StorageBackend) -> Self { + Self { + backend, + applies_to_query_shape: None, + } + } + + /// Build a target that only applies to the listed shapes. + pub fn for_shapes(backend: StorageBackend, shapes: Vec) -> Self { + Self { + backend, + applies_to_query_shape: Some(shapes), + } + } } /// In-memory routing table consulted by the HTTP handler at request @@ -82,7 +331,10 @@ struct BackendStorageRoutingYaml { #[derive(Debug, Clone)] pub struct BackendStorageRouting { default: StorageBackend, - metrics: HashMap, + /// For each metric, the ordered list of targets the HTTP handler + /// walks to pick a backend. v6.1 single-target rows come in as a + /// single-element vec with `applies_to_query_shape: None`. + metrics: HashMap>, } impl BackendStorageRouting { @@ -96,9 +348,24 @@ impl BackendStorageRouting { } } - /// Construct directly. Used by the YAML loader and tests; production + /// Construct directly from an explicit map. Used by tests; production /// callers go through [`Self::from_yaml_file`]. - pub fn new(default: StorageBackend, metrics: HashMap) -> Self { + pub fn new(default: StorageBackend, metrics: HashMap>) -> Self { + Self { default, metrics } + } + + /// Construct from the v6.1 single-target map shape. Each entry + /// maps to a single-target list with no query-shape filter. + /// Preserved for tests and existing call sites; new code should + /// use [`Self::new`] with explicit `RoutingTarget`s. + pub fn new_from_single_targets( + default: StorageBackend, + metrics: HashMap, + ) -> Self { + let metrics = metrics + .into_iter() + .map(|(k, v)| (k, vec![RoutingTarget::always(v)])) + .collect(); Self { default, metrics } } @@ -106,9 +373,37 @@ impl BackendStorageRouting { pub fn from_yaml_str(text: &str) -> Result { let parsed: BackendStorageRoutingYaml = serde_yaml::from_str(text).context("failed to parse backend-storage-routing YAML")?; + + // Start from the v6.1 single-target map. + let mut metrics: HashMap> = parsed + .metrics + .into_iter() + .map(|(k, v)| (k, vec![RoutingTarget::always(v)])) + .collect(); + + // Overlay v7 multi-target rows. A metric in BOTH wins from + // `routes:` (multi-target overrides single-target). + for row in parsed.routes { + if row.targets.is_empty() { + anyhow::bail!( + "backend-storage-routing: metric '{}' has empty targets list", + row.metric, + ); + } + let targets = row + .targets + .into_iter() + .map(|t| RoutingTarget { + backend: t.backend, + applies_to_query_shape: t.applies_to_query_shape, + }) + .collect(); + metrics.insert(row.metric, targets); + } + Ok(Self { default: parsed.default, - metrics: parsed.metrics, + metrics, }) } @@ -123,21 +418,32 @@ impl BackendStorageRouting { path = %path.display(), default = ?routing.default, entries = routing.metrics.len(), + multi_target_entries = routing.metrics.values().filter(|t| t.len() > 1).count(), "Loaded backend-storage-routing YAML", ); Ok(routing) } - /// Look up the storage backend for `metric_name`. Falls back to the - /// table's `default` (which itself defaults to `SketchWarmTier`) - /// when the metric is not listed. + /// Look up the storage backend for `metric_name`, ignoring query + /// shape. Walks the metric's target list and returns the first + /// target's backend (the v6.1 default-target slot). Falls back + /// to the table's `default` when the metric is not listed. + /// + /// Exists for v6.1 callers that haven't been threaded with a + /// parsed PromQL query. New code on the dual-routing path + /// should call [`Self::lookup_with_shape`]. pub fn lookup(&self, metric_name: &str) -> StorageBackend { - match self.metrics.get(metric_name).copied() { - Some(backend) => { + match self.metrics.get(metric_name) { + Some(targets) => { + let backend = targets + .first() + .map(|t| t.backend) + .unwrap_or(self.default); debug!( metric = metric_name, backend = ?backend, - "backend-storage-routing: per-metric override", + target_count = targets.len(), + "backend-storage-routing: lookup (no shape) — picked first target", ); backend } @@ -152,8 +458,88 @@ impl BackendStorageRouting { } } + /// Look up the storage backend for `(metric_name, query_shape)`. + /// + /// Walks the metric's target list and returns the backend of the + /// first target whose `applies_to_query_shape` either is `None` + /// (default) or contains `shape`. If no target matches (empty + /// or all filters miss), returns the first target's backend + /// (matches v6.1's single-target semantics for the default + /// slot). Falls back to `self.default` when the metric is not + /// listed. + /// + /// Selection ordering: v7 puts the *default* target (no filter) + /// first in the YAML, then the shape-specific overrides; this + /// matches v6.1 behaviour for callers that don't supply a + /// shape, but lets shape-specific lookups skip past the default + /// slot if any later target's filter matches `shape`. The + /// implementation walks the list in two passes: + /// 1. First, prefer a target whose filter explicitly includes + /// `shape` — this lets a `[count, topk, rate_post_hoc]` + /// target win over the default warm-tier slot for those + /// shapes. + /// 2. If no shape-specific target matches, fall back to the + /// first target with `applies_to_query_shape: None` + /// (the default slot). + /// 3. If even that's missing, use the first target. + pub fn lookup_with_shape(&self, metric_name: &str, shape: QueryShape) -> StorageBackend { + match self.metrics.get(metric_name) { + Some(targets) => { + // Pass 1: explicit shape match. + for t in targets { + if let Some(list) = &t.applies_to_query_shape { + if list.contains(&shape) { + debug!( + metric = metric_name, + shape = ?shape, + backend = ?t.backend, + "backend-storage-routing: shape-specific match", + ); + return t.backend; + } + } + } + // Pass 2: default slot. + for t in targets { + if t.applies_to_query_shape.is_none() { + debug!( + metric = metric_name, + shape = ?shape, + backend = ?t.backend, + "backend-storage-routing: default-slot fallback", + ); + return t.backend; + } + } + // Pass 3: the first target, regardless of filter + // (only reachable when the metric only has + // shape-specific targets and none matched). + let backend = targets + .first() + .map(|t| t.backend) + .unwrap_or(self.default); + debug!( + metric = metric_name, + shape = ?shape, + backend = ?backend, + "backend-storage-routing: no match; first-target fallback", + ); + backend + } + None => { + debug!( + metric = metric_name, + shape = ?shape, + default = ?self.default, + "backend-storage-routing: metric unlisted, using default", + ); + self.default + } + } + } + /// Read-only view of the configured default. Tests use this; the - /// HTTP handler goes through `lookup`. + /// HTTP handler goes through `lookup` / `lookup_with_shape`. pub fn default_backend(&self) -> StorageBackend { self.default } @@ -169,6 +555,13 @@ impl BackendStorageRouting { pub fn is_empty(&self) -> bool { self.metrics.is_empty() } + + /// Number of targets registered for `metric_name`. Returns 0 for + /// unlisted metrics. Tests use this to assert dual-routing is + /// wired correctly. + pub fn target_count(&self, metric_name: &str) -> usize { + self.metrics.get(metric_name).map(|t| t.len()).unwrap_or(0) + } } impl Default for BackendStorageRouting { @@ -186,10 +579,15 @@ mod tests { let r = BackendStorageRouting::empty(); assert_eq!(r.lookup("anything"), StorageBackend::SketchWarmTier); assert_eq!(r.lookup("http_requests_total"), StorageBackend::SketchWarmTier); + assert_eq!( + r.lookup_with_shape("anything", QueryShape::Count), + StorageBackend::SketchWarmTier + ); } #[test] - fn yaml_with_per_metric_override_routes_correctly() { + fn yaml_with_per_metric_override_routes_correctly_v6_1_form() { + // v6.1 form: `metrics:` map. Each value is a single backend. let yaml = r#" default: sketch_warm_tier metrics: @@ -238,4 +636,220 @@ metrics: let yaml = "metrics: not-a-map\n"; assert!(BackendStorageRouting::from_yaml_str(yaml).is_err()); } + + // ── v7 dual-routing tests ────────────────────────────────────────── + + #[test] + fn yaml_v7_multi_target_routes_count_to_archive_quantile_to_warm() { + // v7 form: `routes:` list. http_requests_total has TWO + // targets — the default warm-tier slot and a cold-archive + // slot scoped to count/topk/rate_post_hoc. + let yaml = r#" +default: sketch_warm_tier +routes: + - metric: http_requests_total + targets: + - backend: sketch_warm_tier + - backend: gorilla_s3_archive + applies_to_query_shape: [count, topk, rate_post_hoc] + - metric: http_freshness_probe_warm + targets: + - backend: sketch_warm_tier + - metric: http_freshness_probe_archive + targets: + - backend: gorilla_s3_archive +"#; + let r = BackendStorageRouting::from_yaml_str(yaml).expect("parse"); + + // Count + topk + rate_post_hoc → archive. + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Count), + StorageBackend::GorillaS3Archive, + ); + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Topk), + StorageBackend::GorillaS3Archive, + ); + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::RatePostHoc), + StorageBackend::GorillaS3Archive, + ); + + // Quantile + sum_over_time + everything else → warm. + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Quantile), + StorageBackend::SketchWarmTier, + ); + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Sum), + StorageBackend::SketchWarmTier, + ); + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Other), + StorageBackend::SketchWarmTier, + ); + + // Single-target metrics keep v6.1 semantics regardless of shape. + assert_eq!( + r.lookup_with_shape("http_freshness_probe_warm", QueryShape::LastOverTime), + StorageBackend::SketchWarmTier, + ); + assert_eq!( + r.lookup_with_shape("http_freshness_probe_archive", QueryShape::LastOverTime), + StorageBackend::GorillaS3Archive, + ); + } + + #[test] + fn yaml_v6_1_single_target_keeps_old_semantics_under_lookup_with_shape() { + // A v6.1 entry with no targets list — every shape resolves + // to the same backend (no dual-routing). + let yaml = r#" +metrics: + audit_events: gorilla_s3_archive +"#; + let r = BackendStorageRouting::from_yaml_str(yaml).expect("parse"); + for shape in [ + QueryShape::Count, + QueryShape::Topk, + QueryShape::Quantile, + QueryShape::Sum, + QueryShape::Other, + ] { + assert_eq!( + r.lookup_with_shape("audit_events", shape), + StorageBackend::GorillaS3Archive, + "shape={shape:?} must resolve to gorilla_s3_archive (single-target)", + ); + } + } + + #[test] + fn yaml_mixed_metrics_and_routes_routes_wins() { + // Both `metrics:` and `routes:` populated; a metric in BOTH + // wins from `routes:` (multi-target overrides single-target). + let yaml = r#" +default: sketch_warm_tier +metrics: + http_requests_total: cold_jsonl_fallback +routes: + - metric: http_requests_total + targets: + - backend: sketch_warm_tier + - backend: gorilla_s3_archive + applies_to_query_shape: [count] +"#; + let r = BackendStorageRouting::from_yaml_str(yaml).expect("parse"); + // Default slot wins for non-count shapes. + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Quantile), + StorageBackend::SketchWarmTier, + ); + // Count → archive. + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Count), + StorageBackend::GorillaS3Archive, + ); + // The `metrics:` entry was overridden — no trace of + // ColdJsonlFallback. + assert_eq!(r.target_count("http_requests_total"), 2); + } + + #[test] + fn empty_targets_list_is_a_parse_error() { + let yaml = r#" +routes: + - metric: http_requests_total + targets: [] +"#; + let err = BackendStorageRouting::from_yaml_str(yaml).expect_err("must reject empty list"); + assert!(err.to_string().contains("empty targets")); + } + + #[test] + fn lookup_falls_back_to_first_target_when_no_shape_matches() { + // Manually-built entry where every target has a filter and + // none matches `Quantile`. The lookup must return *some* + // backend rather than panic. + let mut metrics = HashMap::new(); + metrics.insert( + "x".to_string(), + vec![ + RoutingTarget::for_shapes(StorageBackend::GorillaS3Archive, vec![QueryShape::Count]), + RoutingTarget::for_shapes( + StorageBackend::ColdJsonlFallback, + vec![QueryShape::Topk], + ), + ], + ); + let r = BackendStorageRouting::new(StorageBackend::SketchWarmTier, metrics); + // No filter matches `Quantile`; must return the first target's + // backend. + assert_eq!( + r.lookup_with_shape("x", QueryShape::Quantile), + StorageBackend::GorillaS3Archive, + ); + } + + // ── classify_query_shape unit tests ──────────────────────────────── + + fn parse(query: &str) -> promql_parser::parser::Expr { + promql_parser::parser::parse(query).expect("parse") + } + + #[test] + fn classifies_count_aggregate_as_count() { + let e = parse("count(http_requests_total{service=\"payments\"})"); + assert_eq!(classify_query_shape(&e), QueryShape::Count); + } + + #[test] + fn classifies_topk_as_topk() { + let e = parse("topk(5, sum by (zone) (rate(http_requests_total[5m])))"); + assert_eq!(classify_query_shape(&e), QueryShape::Topk); + } + + #[test] + fn classifies_rate_call_as_rate_post_hoc() { + let e = parse("rate(http_requests_total[5m])"); + assert_eq!(classify_query_shape(&e), QueryShape::RatePostHoc); + } + + #[test] + fn classifies_sum_by_rate_as_rate_post_hoc() { + // `sum by (zone) (rate(...))` — the inner rate makes this + // post-hoc, the outer sum doesn't change that. + let e = parse("sum by (zone) (rate(http_requests_total[5m]))"); + assert_eq!(classify_query_shape(&e), QueryShape::RatePostHoc); + } + + #[test] + fn classifies_quantile_over_time_as_quantile() { + let e = parse("quantile_over_time(0.99, http_requests_total_latency_ms[1m])"); + assert_eq!(classify_query_shape(&e), QueryShape::Quantile); + } + + #[test] + fn classifies_sum_by_as_sum() { + let e = parse("sum by (zone) (http_requests_total)"); + assert_eq!(classify_query_shape(&e), QueryShape::Sum); + } + + #[test] + fn classifies_sum_over_time_as_sum() { + let e = parse("sum_over_time(http_requests_total[1m])"); + assert_eq!(classify_query_shape(&e), QueryShape::Sum); + } + + #[test] + fn classifies_last_over_time_as_last_over_time() { + let e = parse("last_over_time(http_freshness_probe_warm[10s])"); + assert_eq!(classify_query_shape(&e), QueryShape::LastOverTime); + } + + #[test] + fn classifies_bare_selector_as_other() { + let e = parse("http_requests_total"); + assert_eq!(classify_query_shape(&e), QueryShape::Other); + } } diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index 76351a57..0f31b82d 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -459,15 +459,23 @@ async fn process_query_request( /// Parsing failures fall through to (2)/(3) so a malformed PromQL /// doesn't surface as a routing 5xx (the engines themselves will /// reject it with a clearer error). +/// +/// v7: when the routing table has multi-target rows for the metric, +/// the parsed AST is also classified via +/// [`crate::data_model::classify_query_shape`] and the lookup picks +/// the target whose `applies_to_query_shape` matches. v6.1 +/// single-target metrics keep their original semantics — every shape +/// resolves to the one configured backend. fn resolve_metric_storage(state: &AppState, query: &str) -> StorageBackend { if let Some(routing) = state.backend_storage_routing.as_ref() { match promql_parser::parser::parse(query) { Ok(expr) => { if let Some(metric_name) = first_metric_name(&expr) { - let backend = routing.lookup(&metric_name); + let shape = crate::data_model::classify_query_shape(&expr); + let backend = routing.lookup_with_shape(&metric_name, shape); debug!( - "resolve_metric_storage: routing-table hit for metric={} → {:?}", - metric_name, backend, + "resolve_metric_storage: routing-table hit for metric={} shape={:?} → {:?}", + metric_name, shape, backend, ); return backend; } @@ -2758,7 +2766,7 @@ aggregations: "http_requests_total".to_string(), StorageBackend::GorillaS3Archive, ); - let routing = crate::data_model::BackendStorageRouting::new( + let routing = crate::data_model::BackendStorageRouting::new_from_single_targets( StorageBackend::SketchWarmTier, metrics, ); @@ -2803,7 +2811,7 @@ aggregations: "http_requests_total".to_string(), StorageBackend::GorillaS3Archive, ); - let routing = crate::data_model::BackendStorageRouting::new( + let routing = crate::data_model::BackendStorageRouting::new_from_single_targets( StorageBackend::SketchWarmTier, metrics, ); @@ -2834,7 +2842,7 @@ aggregations: // top-level `default: gorilla_s3_archive` — every metric must // route through the router. Pins the §8 "all-metrics-archive" // deploy mode. - let routing = crate::data_model::BackendStorageRouting::new( + let routing = crate::data_model::BackendStorageRouting::new_from_single_targets( StorageBackend::GorillaS3Archive, std::collections::HashMap::new(), ); @@ -2861,6 +2869,127 @@ aggregations: assert_eq!(gorilla_calls.load(Ordering::SeqCst), 1); } + // ── v7 dual-routing production-path coverage ────────────────────────────── + // + // v7 lets one metric fan out to multiple `(backend, + // applies_to_query_shape)` targets. The two tests below mirror + // `http_production_path_routes_archive_metric_via_routing_table` + // — same setup, but the routing table has TWO targets for + // `http_requests_total`: a default warm-tier slot and a + // cold-archive slot scoped to `[count, topk, rate_post_hoc]`. + // A `count(...)` query must land on the archive; a + // `quantile_over_time(...)` query must land on the warm tier. + + #[tokio::test] + async fn http_v7_dual_routing_count_lands_on_archive() { + use crate::data_model::{ + BackendStorageRouting, QueryShape, RoutingTarget, + }; + let mut metrics = std::collections::HashMap::new(); + metrics.insert( + "http_requests_total".to_string(), + vec![ + RoutingTarget::always(StorageBackend::SketchWarmTier), + RoutingTarget::for_shapes( + StorageBackend::GorillaS3Archive, + vec![QueryShape::Count, QueryShape::Topk, QueryShape::RatePostHoc], + ), + ], + ); + let routing = BackendStorageRouting::new(StorageBackend::SketchWarmTier, metrics); + + let (gorilla, gorilla_calls) = + MockQueryEngine::new(StorageBackend::GorillaS3Archive, MockOutcome::OkEmpty); + let server_port = setup_test_server_with_routing_table( + routing, + vec![gorilla as Arc], + ) + .await; + + let client = Client::new(); + let resp = client + .get(format!("http://127.0.0.1:{server_port}/api/v1/query")) + .query(&[ + ("query", "count(http_requests_total{service=\"payments\"})"), + ("time", "1700000000"), + ]) + .send() + .await + .expect("Failed to send request"); + assert!( + resp.status().is_success(), + "v7 dual-routing: count must dispatch and return 2xx; got {}", + resp.status() + ); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_data_source(&body, "gorilla_archive"); + assert_eq!( + gorilla_calls.load(Ordering::SeqCst), + 1, + "v7 dual-routing: count must hit the archive engine", + ); + } + + #[tokio::test] + async fn http_v7_dual_routing_quantile_stays_on_warm_tier() { + use crate::data_model::{ + BackendStorageRouting, QueryShape, RoutingTarget, + }; + let mut metrics = std::collections::HashMap::new(); + metrics.insert( + "http_requests_total".to_string(), + vec![ + RoutingTarget::always(StorageBackend::SketchWarmTier), + RoutingTarget::for_shapes( + StorageBackend::GorillaS3Archive, + vec![QueryShape::Count, QueryShape::Topk, QueryShape::RatePostHoc], + ), + ], + ); + let routing = BackendStorageRouting::new(StorageBackend::SketchWarmTier, metrics); + + // Register a Gorilla mock so a misroute would surface as a + // failed assertion rather than a silent fall-through. The + // mock starts with 0 calls; a quantile must NOT touch it. + let (gorilla, gorilla_calls) = + MockQueryEngine::new(StorageBackend::GorillaS3Archive, MockOutcome::OkEmpty); + let server_port = setup_test_server_with_routing_table( + routing, + vec![gorilla as Arc], + ) + .await; + + let client = Client::new(); + let resp = client + .get(format!("http://127.0.0.1:{server_port}/api/v1/query")) + .query(&[ + ( + "query", + "quantile_over_time(0.99, http_requests_total[1m])", + ), + ("time", "1700000000"), + ]) + .send() + .await + .expect("Failed to send request"); + // Warm tier path returns 2xx with `data_source: sketch_warm` + // (the SimpleEngine returns None for this unconfigured + // metric, but the handler still annotates the wire response + // with the warm-tier source). + assert!( + resp.status().is_success(), + "v7 dual-routing: quantile must dispatch and return 2xx; got {}", + resp.status() + ); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_data_source(&body, "sketch_warm"); + assert_eq!( + gorilla_calls.load(Ordering::SeqCst), + 0, + "v7 dual-routing: quantile must NOT hit the archive engine", + ); + } + } // ── Controller integration: PrecomputeJob execution ────────────────────────── diff --git a/asap-query-engine/src/engines/gorilla_engine/exact_executor.rs b/asap-query-engine/src/engines/gorilla_engine/exact_executor.rs index 60ef04b6..212debcf 100644 --- a/asap-query-engine/src/engines/gorilla_engine/exact_executor.rs +++ b/asap-query-engine/src/engines/gorilla_engine/exact_executor.rs @@ -38,6 +38,12 @@ pub enum AdditiveOp { Increase, /// `(last - first) / range_seconds`. Rate, + /// **v7**: the value of the latest sample in the range. Used by + /// `last_over_time([])` — the freshness probe + /// queries from MVP v6 issue #46 criterion ⑥. The fold tracks + /// `(ts_ms, value)` pairs already; this op just returns `value` + /// of the largest-timestamp sample. + Last, } /// Per-statistic executor. Holds an `Arc` so the @@ -78,6 +84,9 @@ impl ExactExecutor { self.execute_streaming_additive(plan, AdditiveOp::Increase) .await } + QueryStatistic::LastOverTime => { + self.execute_streaming_additive(plan, AdditiveOp::Last).await + } QueryStatistic::QuantileOverTime { phi } => self.execute_quantile(plan, *phi).await, QueryStatistic::TopK { k } => self.execute_topk(plan, *k).await, } @@ -483,6 +492,18 @@ impl AdditiveAccumulator { } _ => f64::NAN, }, + // v7 / issue #46 ⑥: return the value of the + // largest-timestamp sample. Counter-shaped freshness + // probes (http_freshness_probe_*) encode the unix_ts_ms + // of the most recent emission directly in the + // cumulative counter value, so `last_over_time(...)` + // returning that value lets the replay client subtract + // the polled timestamp and get a per-path freshness + // delta. + AdditiveOp::Last => match self.last { + Some((_, lv)) => lv, + None => f64::NAN, + }, } } } diff --git a/asap-query-engine/src/engines/gorilla_engine/query_planner.rs b/asap-query-engine/src/engines/gorilla_engine/query_planner.rs index 80a9d980..694ea5f9 100644 --- a/asap-query-engine/src/engines/gorilla_engine/query_planner.rs +++ b/asap-query-engine/src/engines/gorilla_engine/query_planner.rs @@ -44,6 +44,13 @@ pub enum QueryStatistic { /// sample values in the range — once Phase 5 adds spatial /// grouping the executor will return a per-group vector. TopK { k: usize }, + /// **v7**: `last_over_time(m[range])` — value of the + /// largest-timestamp sample in the range. Used by issue #46 + /// criterion ⑥ freshness probes; counter-shaped probes encode + /// `unix_ts_ms_of_emission` in their cumulative value, and + /// `last_over_time(...)` returns that value so the replay + /// client can compute per-path freshness deltas. + LastOverTime, } impl QueryStatistic { @@ -60,6 +67,7 @@ impl QueryStatistic { | Self::MaxOverTime | Self::Rate | Self::Increase + | Self::LastOverTime ) } } @@ -144,6 +152,7 @@ fn plan_from_call(call: &Call, now_ms: i64) -> Result { | "avg_over_time" | "min_over_time" | "max_over_time" + | "last_over_time" | "rate" | "increase" => { let ms = expect_single_matrix_arg(&call.args, &name)?; @@ -154,6 +163,7 @@ fn plan_from_call(call: &Call, now_ms: i64) -> Result { "avg_over_time" => QueryStatistic::AvgOverTime, "min_over_time" => QueryStatistic::MinOverTime, "max_over_time" => QueryStatistic::MaxOverTime, + "last_over_time" => QueryStatistic::LastOverTime, "rate" => QueryStatistic::Rate, "increase" => QueryStatistic::Increase, _ => unreachable!(), @@ -381,7 +391,20 @@ mod tests { fn streaming_classification() { assert!(QueryStatistic::SumOverTime.is_streaming_additive()); assert!(QueryStatistic::Rate.is_streaming_additive()); + assert!(QueryStatistic::LastOverTime.is_streaming_additive()); assert!(!QueryStatistic::QuantileOverTime { phi: 0.5 }.is_streaming_additive()); assert!(!QueryStatistic::TopK { k: 1 }.is_streaming_additive()); } + + #[test] + fn plans_last_over_time_v7() { + // v7: `last_over_time(...)` translates to the streaming + // additive path, picking the value of the largest-timestamp + // sample in [now-range, now). Issue #46 ⑥ freshness probes + // ride this path. + let plan = plan_query_at("last_over_time(http_freshness_probe_warm[10s])", NOW).unwrap(); + assert_eq!(plan.metric, "http_freshness_probe_warm"); + assert_eq!(plan.statistic, QueryStatistic::LastOverTime); + assert_eq!(plan.time_range_ms, (NOW - 10_000, NOW)); + } } diff --git a/asap-query-engine/src/engines/gorilla_engine/tests.rs b/asap-query-engine/src/engines/gorilla_engine/tests.rs index 11e29d05..e2a9e4dd 100644 --- a/asap-query-engine/src/engines/gorilla_engine/tests.rs +++ b/asap-query-engine/src/engines/gorilla_engine/tests.rs @@ -372,6 +372,79 @@ async fn execute_max_over_time() { } } +#[tokio::test] +async fn execute_last_over_time_v7() { + // v7 / issue #46 ⑥: `last_over_time([])` returns + // the value of the largest-timestamp sample in the window. + // Counter-shaped freshness probes encode `unix_ts_ms` in their + // value; the replay client subtracts (poll_ts - observed_value) + // to get a per-path freshness delta. + // + // Three samples spanning 0..2s with monotonically-increasing + // values 100, 200, 300. The last-stamp sample is at NOW-1s with + // value 300. `last_over_time(...[10s])` must return 300. + let samples = vec![ + raw(NOW_MS - 3_000, 100.0), + raw(NOW_MS - 2_000, 200.0), + raw(NOW_MS - 1_000, 300.0), + ]; + let chunk = ChunkRef { + key: "c".into(), + metric: METRIC.into(), + time_range_ms: (NOW_MS - 3_000, NOW_MS), + label_hash: 0, + sample_count: 3, + size_bytes: 0, + }; + let engine = engine_with(vec![(chunk, samples)]); + let result = engine + .execute_at(&format!("last_over_time({METRIC}[10s])"), NOW_MS) + .await + .unwrap(); + if let QueryResult::Vector(iv) = result { + assert_eq!( + iv.values[0].value, 300.0, + "last_over_time must return the largest-timestamp sample's value", + ); + } else { + panic!("expected Vector"); + } +} + +#[tokio::test] +async fn execute_last_over_time_unordered_samples_picks_largest_ts() { + // Samples arrive with non-monotonic timestamps — the chunk + // claims (start, last_ts+1) but the per-sample observe() must + // still pick the largest ts, not the last-arrived sample. + let samples = vec![ + raw(NOW_MS - 5_000, 50.0), // largest ts is sample[2] + raw(NOW_MS - 8_000, 80.0), // smallest ts but later in vec + raw(NOW_MS - 1_000, 1234.5), // largest ts + raw(NOW_MS - 3_000, 30.0), + ]; + let chunk = ChunkRef { + key: "c".into(), + metric: METRIC.into(), + time_range_ms: (NOW_MS - 8_000, NOW_MS), + label_hash: 0, + sample_count: 4, + size_bytes: 0, + }; + let engine = engine_with(vec![(chunk, samples)]); + let result = engine + .execute_at(&format!("last_over_time({METRIC}[10s])"), NOW_MS) + .await + .unwrap(); + if let QueryResult::Vector(iv) = result { + assert!( + (iv.values[0].value - 1234.5).abs() < 1e-12, + "must return the value of the largest-ts sample, not the last-arrived", + ); + } else { + panic!("expected Vector"); + } +} + #[tokio::test] async fn execute_rate_basic() { // Counter goes from 100 at t=NOW-10s to 200 at t=NOW-1s.