From 6d2cb538286d76ae59f76212811a4f82998cccdd Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 6 May 2026 14:19:33 -0400 Subject: [PATCH] fix(http): consult per-metric BackendStorageRouting on every query MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Phase-5/6 EngineRouter wiring (PR #87) sourced the per-metric StorageBackend axis from `StreamingConfig::storage_backend()` — a single field that applies to the entire streaming config. In production deploys the YAML loader (`StreamingConfig::from_yaml_data`) constructs via `Self::new(...)` and always defaults `storage_backend` to `SketchWarmTier`, so the HTTP handler always took the `SimpleEngine`-direct-dispatch branch and the EngineRouter was effectively bypassed for every query — `data_source: gorilla_archive` never landed on cold-archive responses (issue #46 v2 criterion 5 PARTIAL). Per the design correction: the streaming engine never sees Gorilla data on its OTLP-ingest path (the agent's `gorillas3processor` writes chunks directly to S3), so there is nothing for `StreamingConfig::from_yaml_data` to learn. The fix lives in a separate per-metric routing layer: - New `BackendStorageRouting` data type (`{metric_name: StorageBackend}` map) loaded once at startup from `--backend-storage-routing` YAML (or its `ASAP_BACKEND_STORAGE_ROUTING` env-var alias). Wired on both the legacy `query_engine_rust` binary and the deployed `precompute_engine` binary so the Docker image picks it up. - HTTP handler's `process_query_request` extracts the metric name from the PromQL AST (via `promql_parser`) and consults the routing table; falls back to the streaming-config single axis only when no routing table is wired (preserves pre-Phase-5 behaviour). - Three new unit tests exercise the **production code path** (routing table loaded, streaming-config default unchanged) — distinct from the existing tests that mock the dispatch by pinning `streaming_cfg.storage_backend` directly. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/bin/precompute_engine.rs | 48 +++ .../src/data_model/backend_storage_routing.rs | 241 +++++++++++++ asap-query-engine/src/data_model/mod.rs | 2 + .../src/drivers/query/servers/http.rs | 330 +++++++++++++++++- asap-query-engine/src/main.rs | 43 +++ 5 files changed, 648 insertions(+), 16 deletions(-) create mode 100644 asap-query-engine/src/data_model/backend_storage_routing.rs diff --git a/asap-query-engine/src/bin/precompute_engine.rs b/asap-query-engine/src/bin/precompute_engine.rs index 4bee0af0..c012b0ee 100644 --- a/asap-query-engine/src/bin/precompute_engine.rs +++ b/asap-query-engine/src/bin/precompute_engine.rs @@ -157,6 +157,21 @@ struct Args { /// `--enable-otel-ingest` is set). #[arg(long, default_value_t = 4318)] otel_http_port: u16, + + /// Path to the per-metric backend storage routing YAML + /// (`{metric_name: storage_backend}` map). Loaded once at + /// startup; the HTTP query handler consults it on every PromQL + /// query to decide which engine should answer (warm-tier + /// SimpleEngine vs cold-archive GorillaQueryEngine vs JSONL + /// fallback). Without this flag the handler falls back to the + /// streaming-config single axis (which always defaults to + /// `SketchWarmTier`), so cold-archive metrics never reach the + /// `EngineRouter` — the bug issue #46 v2's `MVP_REPORT.md` flagged. + /// Reads from `ASAP_BACKEND_STORAGE_ROUTING` so containerised + /// deploys can wire it via env (matches the backend Docker + /// image's pattern in `deploy/docker-compose/base.yml`). + #[arg(long, env = "ASAP_BACKEND_STORAGE_ROUTING")] + backend_storage_routing: Option, } #[tokio::main] @@ -277,6 +292,39 @@ async fn main() -> Result<(), Box> { }; let mut http_server = HttpServer::new(http_config, query_engine, store.clone(), None); + // Per-metric storage-backend routing table (issue #46 + // criterion ⑤). When provided, the HTTP handler consults this + // table on every PromQL query — extracting the metric name + // from the AST and looking up its `StorageBackend`. Without + // it the handler falls back to the streaming-config single + // axis (which always defaults to `SketchWarmTier`) and the + // `EngineRouter` is effectively bypassed. + if let Some(routing_path) = args.backend_storage_routing.as_deref() { + match query_engine_rust::data_model::BackendStorageRouting::from_yaml_file( + routing_path, + ) { + Ok(routing) => { + info!( + "Loaded backend-storage-routing from {:?}: default={:?}, entries={}", + routing_path, + routing.default_backend(), + routing.len(), + ); + http_server = http_server.with_backend_storage_routing(Arc::new(routing)); + } + Err(e) => { + warn!( + "Failed to load backend-storage-routing from {:?}: {} — falling back to streaming-config single axis", + routing_path, e, + ); + } + } + } else { + info!( + "--backend-storage-routing not set — every query routes per the streaming-config single axis (typically `sketch_warm`)", + ); + } + // Phase-5/6: register a `GorillaQueryEngine` for the cold // archive tier when the operator has provisioned one via the // `ASAP_GORILLA_S3_*` env-var family. `HttpServer::new` already diff --git a/asap-query-engine/src/data_model/backend_storage_routing.rs b/asap-query-engine/src/data_model/backend_storage_routing.rs new file mode 100644 index 00000000..c6ff9d40 --- /dev/null +++ b/asap-query-engine/src/data_model/backend_storage_routing.rs @@ -0,0 +1,241 @@ +//! Per-metric storage-backend routing table consulted by the HTTP query +//! handler at request time. +//! +//! ## Why this exists +//! +//! Phase-5 (PR #87) wired the `EngineRouter` into the HTTP query handler, +//! but the per-metric `StorageBackend` axis was sourced from +//! `StreamingConfig::storage_backend()` — a single field that applies to +//! the entire streaming config. In production deploys (the +//! `precompute_engine` binary loading `backend-streaming.yaml`) the +//! field decodes via `Self::new(...)` which always defaults to +//! `SketchWarmTier`, so the handler always took the +//! `SimpleEngine`-direct-dispatch branch and the `EngineRouter` was +//! effectively bypassed for every query — the `data_source: +//! gorilla_archive` info-line never landed on cold-archive responses +//! even when the chunks were on disk in MinIO. +//! +//! The fix lives **outside** the streaming pipeline: the streaming +//! engine on the OTLP-ingest path never sees Gorilla data (the +//! `gorillas3processor` writes chunks directly to S3), so there is +//! nothing for `StreamingConfig::from_yaml_data` to learn. What we +//! actually need is a tiny standalone routing table — one entry per +//! metric whose storage backend differs from the default — that the +//! HTTP handler consults to pick the right engine for each query. +//! +//! ## Schema +//! +//! ```yaml +//! # deploy/configs/backend-storage-routing.yaml +//! default: sketch_warm_tier # StorageBackend; optional +//! metrics: +//! http_requests_total: gorilla_s3_archive +//! audit_events: gorilla_s3_archive +//! foo_count: cold_jsonl_fallback +//! ``` +//! +//! Valid `StorageBackend` values mirror the snake-cased serde tags on +//! `asap_types::StorageBackend`: `sketch_warm_tier`, +//! `gorilla_s3_archive`, `cold_jsonl_fallback`, `double_write`. +//! +//! Loaded once at backend startup (CLI flag `--backend-storage-routing` +//! on `precompute_engine`) and stored in `AppState`. Lookup is +//! O(metric-name-hash); a query that doesn't match any entry falls back +//! to `default` (which itself falls back to `SketchWarmTier`). +//! +//! ## Out of scope +//! +//! * 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. +//! * Per-`(metric, statistic, accuracy)` granularity — `StorageBackend` +//! already encodes the `DoubleWrite` axis the cost-aware dispatcher +//! uses to pick warm-vs-archive per query. + +use std::collections::HashMap; +use std::path::Path; + +use anyhow::{Context, Result}; +use asap_types::StorageBackend; +use serde::{Deserialize, Serialize}; +use tracing::{debug, info}; + +/// 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`. + #[serde(default)] + default: StorageBackend, + /// Per-metric overrides keyed by the bare metric name (no labels). + #[serde(default)] + metrics: HashMap, +} + +/// In-memory routing table consulted by the HTTP handler at request +/// time. Build via [`Self::from_yaml_file`] / [`Self::from_yaml_str`] +/// or [`Self::empty`] (everything routes to `SketchWarmTier`). +#[derive(Debug, Clone)] +pub struct BackendStorageRouting { + default: StorageBackend, + metrics: HashMap, +} + +impl BackendStorageRouting { + /// Build an empty router — every metric resolves to + /// `SketchWarmTier`. Equivalent to "no routing config at all" and + /// preserves pre-Phase-5 dispatch (`SimpleEngine` direct path). + pub fn empty() -> Self { + Self { + default: StorageBackend::default(), + metrics: HashMap::new(), + } + } + + /// Construct directly. Used by the YAML loader and tests; production + /// callers go through [`Self::from_yaml_file`]. + pub fn new(default: StorageBackend, metrics: HashMap) -> Self { + Self { default, metrics } + } + + /// Parse YAML text. See module docs for the schema. + pub fn from_yaml_str(text: &str) -> Result { + let parsed: BackendStorageRoutingYaml = + serde_yaml::from_str(text).context("failed to parse backend-storage-routing YAML")?; + Ok(Self { + default: parsed.default, + metrics: parsed.metrics, + }) + } + + /// Read + parse a YAML file. Returns the populated router on + /// success. The caller is expected to log the entry count at + /// startup so operators can spot-check the deployment. + pub fn from_yaml_file(path: &Path) -> Result { + let text = std::fs::read_to_string(path) + .with_context(|| format!("failed to read backend-storage-routing YAML: {:?}", path))?; + let routing = Self::from_yaml_str(&text)?; + info!( + path = %path.display(), + default = ?routing.default, + entries = routing.metrics.len(), + "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. + pub fn lookup(&self, metric_name: &str) -> StorageBackend { + match self.metrics.get(metric_name).copied() { + Some(backend) => { + debug!( + metric = metric_name, + backend = ?backend, + "backend-storage-routing: per-metric override", + ); + backend + } + None => { + debug!( + metric = metric_name, + default = ?self.default, + "backend-storage-routing: no override, using default", + ); + self.default + } + } + } + + /// Read-only view of the configured default. Tests use this; the + /// HTTP handler goes through `lookup`. + pub fn default_backend(&self) -> StorageBackend { + self.default + } + + /// Number of explicit per-metric overrides. Tests / operator + /// tooling. + pub fn len(&self) -> usize { + self.metrics.len() + } + + /// `true` iff no per-metric overrides are configured. Equivalent to + /// `Self::empty()` but preserves a custom `default`. + pub fn is_empty(&self) -> bool { + self.metrics.is_empty() + } +} + +impl Default for BackendStorageRouting { + fn default() -> Self { + Self::empty() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_router_routes_everything_to_warm_tier() { + let r = BackendStorageRouting::empty(); + assert_eq!(r.lookup("anything"), StorageBackend::SketchWarmTier); + assert_eq!(r.lookup("http_requests_total"), StorageBackend::SketchWarmTier); + } + + #[test] + fn yaml_with_per_metric_override_routes_correctly() { + let yaml = r#" +default: sketch_warm_tier +metrics: + http_requests_total: gorilla_s3_archive + audit_events: cold_jsonl_fallback +"#; + let r = BackendStorageRouting::from_yaml_str(yaml).expect("parse"); + assert_eq!( + r.lookup("http_requests_total"), + StorageBackend::GorillaS3Archive + ); + assert_eq!( + r.lookup("audit_events"), + StorageBackend::ColdJsonlFallback + ); + assert_eq!(r.lookup("unlisted"), StorageBackend::SketchWarmTier); + assert_eq!(r.len(), 2); + } + + #[test] + fn yaml_default_only_routes_all_metrics_to_default() { + let yaml = "default: gorilla_s3_archive\n"; + let r = BackendStorageRouting::from_yaml_str(yaml).expect("parse"); + assert_eq!(r.lookup("anything"), StorageBackend::GorillaS3Archive); + assert!(r.is_empty()); + assert_eq!(r.default_backend(), StorageBackend::GorillaS3Archive); + } + + #[test] + fn yaml_omitted_default_falls_back_to_sketch_warm() { + let yaml = "metrics:\n foo: gorilla_s3_archive\n"; + let r = BackendStorageRouting::from_yaml_str(yaml).expect("parse"); + assert_eq!(r.lookup("foo"), StorageBackend::GorillaS3Archive); + assert_eq!(r.lookup("bar"), StorageBackend::SketchWarmTier); + } + + #[test] + fn empty_yaml_is_valid_and_empty() { + let r = BackendStorageRouting::from_yaml_str("").expect("empty parse"); + assert_eq!(r.lookup("foo"), StorageBackend::SketchWarmTier); + assert!(r.is_empty()); + } + + #[test] + fn invalid_yaml_returns_error() { + let yaml = "metrics: not-a-map\n"; + assert!(BackendStorageRouting::from_yaml_str(yaml).is_err()); + } +} diff --git a/asap-query-engine/src/data_model/mod.rs b/asap-query-engine/src/data_model/mod.rs index 0145b2b0..9444aadc 100644 --- a/asap-query-engine/src/data_model/mod.rs +++ b/asap-query-engine/src/data_model/mod.rs @@ -1,5 +1,6 @@ pub mod aggregation_config; pub mod aggregation_reference; +pub mod backend_storage_routing; pub mod enums; pub mod hot_reload_config; pub mod inference_config; @@ -13,6 +14,7 @@ pub mod traits; pub use aggregation_config::*; pub use aggregation_reference::*; +pub use backend_storage_routing::*; pub use enums::*; pub use hot_reload_config::*; pub use inference_config::*; diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index 5090ce00..bc0f9c51 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -49,6 +49,17 @@ pub struct HttpServer { /// Hot-reloadable `StreamingConfig` source. `None` when hot-reload /// is not wired up by the caller (unit tests, legacy binaries). hot_reload_config: Option, + /// Per-metric storage-backend routing table consulted by the HTTP + /// instant-query handler at request time. When `Some(..)` and the + /// query parses, the handler extracts the metric name from the + /// PromQL AST, consults this table, and dispatches through + /// `EngineRouter` for any per-metric override. When `None` the + /// handler falls back to the pre-Phase-5 behaviour of consulting + /// the streaming-config's single `storage_backend()` axis (which + /// itself defaults to `SketchWarmTier`). Wired by the binary via + /// [`Self::with_backend_storage_routing`]; production deploys + /// load `deploy/configs/backend-storage-routing.yaml`. + backend_storage_routing: Option>, /// Per-`agg_id` schema registry (sketch DB §6). `None` when the /// caller hasn't wired the precompute engine into the HTTP /// server — in that case the `POST /api/v1/streaming-config` @@ -80,6 +91,8 @@ struct AppState { adapter: Arc, fallback: Option>, hot_reload_config: Option, + /// See [`HttpServer::backend_storage_routing`]. + backend_storage_routing: Option>, /// Per-`agg_id` schema registry (sketch DB §6). Phase 2b wires /// `POST /api/v1/streaming-config` to call `schemas.reconcile()` /// on every swap so schema lifecycle transitions happen @@ -114,6 +127,7 @@ impl HttpServer { store, query_tracker, hot_reload_config: None, + backend_storage_routing: None, schemas: None, backfill: None, data_retention_ms: None, @@ -150,6 +164,25 @@ impl HttpServer { self } + /// Attach a per-metric storage-backend routing table loaded from + /// `backend-storage-routing.yaml`. When attached, every instant + /// query consults this table (after extracting the metric name + /// from the PromQL AST) and dispatches through `EngineRouter` for + /// any per-metric override. Without this handle the handler falls + /// back to the pre-Phase-5 single-axis behaviour driven by + /// `StreamingConfig::storage_backend()`. + /// + /// This is the bridge from "warm-tier-only deploy" to + /// "cold-archive-routed metrics" until the controller's plan-push + /// pipeline lands per-metric `StorageBackend` updates. + pub fn with_backend_storage_routing( + mut self, + routing: Arc, + ) -> Self { + self.backend_storage_routing = Some(routing); + self + } + /// Attach the `SchemaRegistry` that the precompute engine's /// `IngestState` also holds. When attached, the /// `POST /api/v1/streaming-config` handler calls @@ -211,6 +244,7 @@ impl HttpServer { adapter: adapter.clone(), fallback: self.config.adapter_config.fallback.clone(), hot_reload_config: self.hot_reload_config.clone(), + backend_storage_routing: self.backend_storage_routing.clone(), schemas: self.schemas.clone(), backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, @@ -278,6 +312,7 @@ impl HttpServer { adapter: adapter.clone(), fallback: self.config.adapter_config.fallback.clone(), hot_reload_config: self.hot_reload_config.clone(), + backend_storage_routing: self.backend_storage_routing.clone(), schemas: self.schemas.clone(), backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, @@ -370,24 +405,33 @@ async fn process_query_request( // Step 2: Pick a dispatch path based on the metric's pinned // storage backend (Phase-5 capability routing). // - // - `SketchWarmTier` (default for legacy / unconfigured deploys) - // keeps the direct `SimpleEngine::handle_query` path: it returns - // a `KeyByLabelNames` Prometheus needs to populate the `metric` - // map, which the trait surface (`router.execute → QueryResult` - // only) cannot thread through. - // - Anything else (`GorillaS3Archive`, `DoubleWrite`, - // `ColdJsonlFallback`) goes through the `EngineRouter`. Phase-6 - // (Gorilla MVP) returns a scalar with empty labels, so dropping - // `KeyByLabelNames` is acceptable; the response carries - // `accuracy` + `data_source` via the wire-extension annotations. - let metric_storage = state - .hot_reload_config - .as_ref() - .map(|h| h.snapshot().storage_backend()) - .unwrap_or_default(); + // Routing precedence: + // (a) Per-metric `BackendStorageRouting` table (loaded from + // `backend-storage-routing.yaml` at startup). The PromQL + // query is parsed; the metric name is extracted from the + // AST and looked up in the table. This is the production + // path the issue-46 MVP relies on so cold-archive metrics + // (e.g. `http_requests_total` → `gorilla_archive`) actually + // route through the `EngineRouter`. + // (b) Single-axis `StreamingConfig::storage_backend()` from the + // hot-reload config (the pre-Phase-5 fallback). Pre-controller + // deploys ride this path; it always lands on `SketchWarmTier` + // unless the YAML was hand-patched. + // (c) Default — `SketchWarmTier`. Keeps the direct + // `SimpleEngine::handle_query` path so the response carries + // the `KeyByLabelNames` the Prometheus adapter needs to + // populate the `metric` map. + // + // For non-`SketchWarmTier` axes the dispatch goes through the + // `EngineRouter`. Phase-6 (Gorilla MVP) returns a scalar with + // empty labels, so dropping `KeyByLabelNames` is acceptable; the + // response carries `accuracy` + `data_source` via the + // wire-extension annotations. + let metric_storage = resolve_metric_storage(state, &parsed_request.query); debug!( - "Dispatch axis: metric_storage={:?} (from hot-reload config: {})", + "Dispatch axis: metric_storage={:?} (from backend-storage-routing: {}, hot-reload: {})", metric_storage, + state.backend_storage_routing.is_some(), state.hot_reload_config.is_some(), ); @@ -398,6 +442,72 @@ async fn process_query_request( } } +/// Resolve the [`StorageBackend`] that should handle this query. +/// +/// Routing precedence (see `process_query_request` for context): +/// 1. Per-metric `BackendStorageRouting` table — parse the PromQL, +/// pull the metric name out of the AST, look it up in the table. +/// This is the path issue #46's MVP demo relies on. +/// 2. Streaming-config single-axis fallback — preserves pre-Phase-5 +/// behaviour for deploys that haven't loaded a routing table. +/// 3. Default `SketchWarmTier`. +/// +/// 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). +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); + debug!( + "resolve_metric_storage: routing-table hit for metric={} → {:?}", + metric_name, backend, + ); + return backend; + } + debug!( + "resolve_metric_storage: PromQL parsed but no metric name found in AST; falling back to streaming-config axis", + ); + } + Err(e) => { + debug!( + "resolve_metric_storage: PromQL parse failed ({}); falling back to streaming-config axis", + e, + ); + } + } + } + + state + .hot_reload_config + .as_ref() + .map(|h| h.snapshot().storage_backend()) + .unwrap_or_default() +} + +/// Walk a PromQL AST and return the first metric name we encounter. +/// Used by the routing-table lookup to pick a key. PromQL queries that +/// reference multiple metrics (e.g. `a / on(x) b`) are not currently +/// supported by the routing table — the first-encountered metric wins. +/// In practice the issue-46 demo replay queries each touch exactly one +/// metric, so this heuristic is correct for the MVP. +fn first_metric_name(expr: &promql_parser::parser::Expr) -> Option { + use promql_parser::parser::Expr; + match expr { + Expr::VectorSelector(vs) => vs.name.clone(), + Expr::MatrixSelector(ms) => ms.vs.name.clone(), + Expr::Call(call) => call.args.args.iter().find_map(|a| first_metric_name(a)), + Expr::Aggregate(agg) => first_metric_name(&agg.expr), + Expr::Binary(bin) => first_metric_name(&bin.lhs).or_else(|| first_metric_name(&bin.rhs)), + Expr::Subquery(sq) => first_metric_name(&sq.expr), + Expr::Paren(p) => first_metric_name(&p.expr), + Expr::Unary(u) => first_metric_name(&u.expr), + _ => None, + } +} + /// Direct `SimpleEngine::handle_query` dispatch — preserves the /// `KeyByLabelNames` the Prometheus adapter needs to fill in the /// `metric` map. Used for warm-tier metrics (the default) so the @@ -2272,6 +2382,63 @@ aggregations: .expect("Failed to start test server") } + /// Build an `HttpServer` wired with a per-metric + /// `BackendStorageRouting` table — the **production path** the + /// issue-46 MVP relies on. The streaming-config single axis stays + /// at `SketchWarmTier` (the realistic deploy state); the routing + /// table is what flips per-metric dispatch over to the + /// `EngineRouter`. This proves the production code path + /// (`process_query_request → resolve_metric_storage → routing + /// table lookup`), as opposed to the + /// `setup_test_server_with_router` helper above which mocks the + /// resolution by pinning `streaming_cfg.storage_backend` directly. + async fn setup_test_server_with_routing_table( + routing: crate::data_model::BackendStorageRouting, + extra_engines: Vec>, + ) -> u16 { + let adapter_config = AdapterConfig::prometheus_promql( + "http://127.0.0.1:9999".to_string(), + false, + ); + let config = HttpServerConfig { + port: 0, + handle_http_requests: true, + adapter_config, + }; + let inference_config = InferenceConfig::new( + crate::data_model::QueryLanguage::promql, + crate::data_model::CleanupPolicy::NoCleanup, + ); + // Streaming-config stays on the default `SketchWarmTier` axis + // — exactly what the production deploy looks like (the YAML + // loader doesn't parse `storage_backend`). All routing + // decisions must come from the per-metric routing table. + let streaming_cfg = StreamingConfig::default(); + let streaming_arc = Arc::new(streaming_cfg); + let hot_reload = HotReloadStreamingConfig::from_arc(streaming_arc.clone()); + let store = Arc::new(SimpleMapStore::new( + streaming_arc.clone(), + crate::data_model::CleanupPolicy::NoCleanup, + )); + let query_engine = Arc::new(SimpleEngine::new( + store.clone(), + inference_config, + streaming_arc, + 15000, + crate::data_model::QueryLanguage::promql, + )); + let mut server = HttpServer::new(config, query_engine, store, None) + .with_hot_reload_config(hot_reload) + .with_backend_storage_routing(Arc::new(routing)); + for engine in extra_engines { + server = server.with_query_engine(engine); + } + server + .start_test_server() + .await + .expect("Failed to start test server") + } + /// Build a server whose `EngineRouter` has zero registered /// engines. We can't reach this through the public API /// (`HttpServer::new` always registers `SimpleEngine`), so the @@ -2536,6 +2703,137 @@ aggregations: assert_eq!(jsonl_calls.load(Ordering::SeqCst), 1); } + // ── Issue #46 production-path coverage: BackendStorageRouting ───────────── + // + // The tests above (e.g. `http_routes_archive_metric_to_gorilla_engine`) + // mock the routing decision by pinning `streaming_cfg.storage_backend + // = GorillaS3Archive` directly. That proves the dispatch BRANCH is + // wired, but not the production code path — in real deploys the + // streaming-config YAML loader drops `storage_backend` (it always + // defaults to `SketchWarmTier`), so the issue-46 v2 demo's queries + // never reached the EngineRouter. The tests below exercise the + // **production path** end-to-end: streaming config stays default, + // a per-metric `BackendStorageRouting` table is loaded at startup + // (mirroring `--backend-storage-routing` on `precompute_engine`), + // and the handler must consult the table on every request. + + #[tokio::test] + async fn http_production_path_routes_archive_metric_via_routing_table() { + // Production path: streaming-config single axis stays on + // `SketchWarmTier` (the YAML loader's default), but the + // per-metric routing table flips `http_requests_total` to + // `gorilla_archive`. The handler must extract the metric name + // from the PromQL AST, look it up, and dispatch through the + // EngineRouter — landing the `data_source: gorilla_archive` + // info-line on the response. + let mut metrics = std::collections::HashMap::new(); + metrics.insert( + "http_requests_total".to_string(), + StorageBackend::GorillaS3Archive, + ); + let routing = crate::data_model::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)"), + ("time", "1700000000"), + ]) + .send() + .await + .expect("Failed to send request"); + assert!( + resp.status().is_success(), + "production-path archive dispatch must 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, + "GorillaQueryEngine must be hit exactly once on the production path", + ); + } + + #[tokio::test] + async fn http_production_path_unlisted_metric_falls_back_to_warm_tier() { + // The same routing table only overrides `http_requests_total`; + // a query against a different metric must take the warm-tier + // direct-dispatch path (no `EngineRouter` round-trip). + let mut metrics = std::collections::HashMap::new(); + metrics.insert( + "http_requests_total".to_string(), + StorageBackend::GorillaS3Archive, + ); + let routing = crate::data_model::BackendStorageRouting::new( + StorageBackend::SketchWarmTier, + metrics, + ); + let server_port = + setup_test_server_with_routing_table(routing, Vec::new()).await; + let client = Client::new(); + let resp = client + .get(format!("http://127.0.0.1:{server_port}/api/v1/query")) + .query(&[ + ("query", "sum_over_time(some_other_metric[5m])"), + ("time", "1700000000"), + ]) + .send() + .await + .expect("Failed to send request"); + assert!( + resp.status().is_success(), + "warm-tier fallback must return 2xx; got {}", + resp.status(), + ); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_data_source(&body, "sketch_warm"); + } + + #[tokio::test] + async fn http_production_path_default_axis_routes_all_metrics() { + // Routing table with no per-metric overrides but a non-default + // 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( + StorageBackend::GorillaS3Archive, + std::collections::HashMap::new(), + ); + 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(any_metric_at_all)"), + ("time", "1700000000"), + ]) + .send() + .await + .expect("Failed to send request"); + assert!(resp.status().is_success()); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_data_source(&body, "gorilla_archive"); + assert_eq!(gorilla_calls.load(Ordering::SeqCst), 1); + } + } // ── Controller integration: PrecomputeJob execution ────────────────────────── diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index 7a3ac00b..e7b0204b 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -277,6 +277,19 @@ struct Args { /// Defaults to `min(10% * memory_limit_mb, 512)`. #[arg(long)] persistence_part_cache_mb: Option, + + /// Path to the per-metric backend storage routing YAML + /// (`{metric_name: storage_backend}` map). Loaded at startup and + /// consulted by the HTTP query handler on every PromQL request to + /// pick the right engine (`SimpleEngine` for warm-tier sketches, + /// `GorillaQueryEngine` for the cold archive, etc.). Without + /// this flag the handler falls back to the streaming-config + /// single axis (always `SketchWarmTier`) and the EngineRouter is + /// effectively bypassed — the issue-46 v2 demo's criterion ⑤ + /// failure mode. Mirrors the `precompute_engine` binary's flag + /// of the same name. + #[arg(long, env = "ASAP_BACKEND_STORAGE_ROUTING")] + backend_storage_routing: Option, } #[tokio::main] @@ -670,6 +683,36 @@ async fn main() -> Result<()> { let mut server = HttpServer::new(http_config, engine, store.clone(), query_tracker) .with_hot_reload_config(hot_reload_config.clone()); + // Per-metric storage-backend routing table (issue #46 + // criterion ⑤). Mirror the `precompute_engine` binary: load it + // from `--backend-storage-routing` (or its env-var alias) so the + // HTTP handler consults a per-metric `StorageBackend` map on + // every PromQL query instead of bypassing the EngineRouter when + // the streaming-config single axis defaults to `SketchWarmTier`. + if let Some(routing_path) = args.backend_storage_routing.as_deref() { + match query_engine_rust::data_model::BackendStorageRouting::from_yaml_file(routing_path) { + Ok(routing) => { + info!( + "Loaded backend-storage-routing from {:?}: default={:?}, entries={}", + routing_path, + routing.default_backend(), + routing.len(), + ); + server = server.with_backend_storage_routing(Arc::new(routing)); + } + Err(e) => { + warn!( + "Failed to load backend-storage-routing from {:?}: {} — falling back to streaming-config single axis", + routing_path, e, + ); + } + } + } else { + info!( + "--backend-storage-routing not set — every query routes per the streaming-config single axis (typically `sketch_warm`)", + ); + } + // Phase-5/6: register a `GorillaQueryEngine` for the cold // archive tier when the operator has provisioned one via the // `ASAP_GORILLA_S3_*` env-var family. `HttpServer::new` already