From a730a82bd1a732a07f6c057099cf04e84bc5f5c7 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Thu, 7 May 2026 16:57:54 -0400 Subject: [PATCH] =?UTF-8?q?mvp=20phase=20=CE=B1:=20backend=20hot-loads=20c?= =?UTF-8?q?ontroller-emitted=20BackendStorageRouting?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pairs with ASAPCollector#314 (controller emitter). Today the per-metric `BackendStorageRouting` table is hand-authored YAML that the backend reads at startup; after Phase α the controller emits it as part of its plan and the backend hot-loads it on push, closing the controller → backend feedback loop. Concretely: * `BackendStorageRouting::from_json_payload(&JsonValue)` — parses the controller-emitted JSON document (schema mirrors `controller/src/config/stage_config.rs::emit_backend_storage_routing`). Maps engine-name strings — `sketch_warm_tier`, `thanos_archive` (alias for `gorilla_s3_archive`), `gorilla_s3_archive`, `double_write` — into the existing `StorageBackend` variants. Unknown query-shape strings map to `QueryShape::Other` for forward-compat. Side fields (`warm_tier_native_shapes`) are ignored. * `QueryShape` extended with `HistogramQuantile`, `Delta`, `Deriv`, `Absent` — the controller's archive-eligible shape vocabulary. `classify_query_shape` updated to recognise the matching PromQL function names (`histogram_quantile`, `delta`/`increase`, `deriv`, `absent`/`absent_over_time`). * `HotReloadBackendStorageRouting` — `ArcSwap`-backed wrapper mirroring `HotReloadStreamingConfig`. Lets the swap handler atomically replace the routing table at runtime without restart. Cloneable; clones share the underlying `ArcSwap` so all holders see the same swaps. * `routing_table_hash(&BackendStorageRouting)` — stable short hash for the swap-response body. Lets the controller verify the backend installed exactly the bytes it pushed. * `POST /api/v1/storage_routing` HTTP endpoint — accepts the JSON document, validates via `from_json_payload`, atomically swaps the table via the hot-reload handle, returns 200 with the new table's hash + entry count. 400 on invalid JSON / schema; 503 when the backend wasn't built with a routing handle. * `GET /api/v1/storage_routing` — JSON snapshot for operator diagnostics (default engine, metrics count, table hash). * `HttpServer::with_backend_storage_routing` — now wraps the `Arc` it receives in a hot-reload handle internally, so existing call sites keep their signature. The internal field type changed from `Option>` to `Option`. The single read site in `resolve_metric_storage` snapshots the handle once per request, giving the dispatcher torn-read-free access to the current table. * `HttpServer::with_hot_reload_backend_storage_routing` — sibling builder that takes a pre-built handle, for callers that want to share it with other subsystems. * Bootstrap: `main.rs` and `bin/precompute_engine.rs` now always install a hot-reload routing handle — bootstrap from `--backend-storage-routing` YAML when set, an empty table otherwise. Operators can still hand-author the YAML for dev / standalone deployments; controller pushes overwrite it. Tests: * 14 new unit tests in `routing::backend_storage_routing` — `from_json_payload` happy path / unknown shape / invalid engine / empty targets / missing metrics / optional default / back-compat alias; `replace` in-place swap; hot-reload wrapper visibility + concurrent-reader torn-state guard; routing-table-hash stability. Plus `HistogramQuantile` / `Delta` / `Absent` classifier tests. * 4 new HTTP integration tests in `drivers::query::servers::http` — POST swaps atomically; POST rejects invalid JSON without partial swap; GET returns a snapshot whose hash matches the live table; swap is observed by subsequent dispatch lookups. Backend lib tests: 875 pass / 33 pre-existing fail / 9 ignored (vs origin/main: 856 pass / 33 fail / 9 ignored — +19 new passing tests, 0 new failures). Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/bin/precompute_engine.rs | 19 +- asap-query-engine/src/data_model/mod.rs | 3 +- .../src/drivers/query/servers/http.rs | 394 +++++++++++- asap-query-engine/src/main.rs | 20 +- .../src/routing/backend_storage_routing.rs | 606 +++++++++++++++++- asap-query-engine/src/routing/mod.rs | 3 +- 6 files changed, 1017 insertions(+), 28 deletions(-) diff --git a/asap-query-engine/src/bin/precompute_engine.rs b/asap-query-engine/src/bin/precompute_engine.rs index aa415b8ab..5c2b90048 100644 --- a/asap-query-engine/src/bin/precompute_engine.rs +++ b/asap-query-engine/src/bin/precompute_engine.rs @@ -289,7 +289,13 @@ async fn main() -> Result<(), Box> { // 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() { + // Phase α (MVP): always install a hot-reload routing handle — + // bootstrap from YAML when available, an empty table otherwise. + // The `POST /api/v1/storage_routing` endpoint can then swap in + // a controller-emitted table at runtime without restart. + let bootstrap_routing = if let Some(routing_path) = + args.backend_storage_routing.as_deref() + { match query_engine_rust::data_model::BackendStorageRouting::from_yaml_file( routing_path, ) { @@ -300,20 +306,23 @@ async fn main() -> Result<(), Box> { routing.default_backend(), routing.len(), ); - http_server = http_server.with_backend_storage_routing(Arc::new(routing)); + routing } Err(e) => { warn!( - "Failed to load backend-storage-routing from {:?}: {} — falling back to streaming-config single axis", + "Failed to load backend-storage-routing from {:?}: {} — installing an empty routing table; the controller's first POST /api/v1/storage_routing push will fill it", routing_path, e, ); + query_engine_rust::data_model::BackendStorageRouting::empty() } } } else { info!( - "--backend-storage-routing not set — every query routes per the streaming-config single axis (typically `sketch_warm`)", + "--backend-storage-routing not set — installing an empty routing table; the controller's first POST /api/v1/storage_routing push will fill it", ); - } + query_engine_rust::data_model::BackendStorageRouting::empty() + }; + http_server = http_server.with_backend_storage_routing(Arc::new(bootstrap_routing)); // Phase-5/6: register a `GorillaQueryEngine` for the cold // archive tier when the operator has provisioned one via the diff --git a/asap-query-engine/src/data_model/mod.rs b/asap-query-engine/src/data_model/mod.rs index 8527e081c..3b1e26e4c 100644 --- a/asap-query-engine/src/data_model/mod.rs +++ b/asap-query-engine/src/data_model/mod.rs @@ -30,5 +30,6 @@ pub use traits::*; // `crate::data_model::BackendStorageRouting` compiling for any // transitive caller that hasn't been migrated yet. pub use crate::routing::{ - classify_query_shape, BackendStorageRouting, QueryShape, RoutingTarget, + classify_query_shape, BackendStorageRouting, HotReloadBackendStorageRouting, + QueryShape, RoutingTarget, }; diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index 6b4700183..482a7ed1d 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -78,8 +78,17 @@ pub struct HttpServer { /// 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>, + /// bootstrap from `deploy/configs/backend-storage-routing.yaml` + /// (legacy form) or the controller's first + /// `POST /api/v1/storage_routing` push (Phase α). + /// + /// Phase α: this field is now a `HotReloadBackendStorageRouting` + /// — an `ArcSwap`-backed wrapper that supports atomic at-runtime + /// swap from the `POST /api/v1/storage_routing` endpoint. The + /// existing read path snapshots the wrapper once per request + /// (`handle.snapshot().lookup_with_shape(...)`); swap is observed + /// by the next request without restart. + 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` @@ -112,7 +121,7 @@ struct AppState { fallback: Option>, hot_reload_config: Option, /// See [`HttpServer::backend_storage_routing`]. - backend_storage_routing: Option>, + 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 @@ -184,22 +193,44 @@ 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()`. + /// Attach a per-metric storage-backend routing table. The table is + /// wrapped in a hot-reload handle internally so the + /// `POST /api/v1/storage_routing` endpoint (Phase α) can swap it + /// atomically without restart. + /// + /// Bootstrap typically comes from + /// `BackendStorageRouting::from_yaml_file(...)` for legacy / dev + /// deploys, or from `BackendStorageRouting::empty()` when the + /// controller will push the first table — the controller's first + /// `POST /api/v1/storage_routing` then fills in all the entries. /// - /// 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. + /// When the wrapper is attached, every instant query consults the + /// snapshot (after extracting the metric name from the PromQL AST) + /// and dispatches through `EngineRouter` for any per-metric + /// override. Without the wrapper the handler falls back to the + /// pre-Phase-5 single-axis behaviour driven by + /// `StreamingConfig::storage_backend()`. pub fn with_backend_storage_routing( mut self, routing: Arc, ) -> Self { - self.backend_storage_routing = Some(routing); + self.backend_storage_routing = Some( + crate::routing::HotReloadBackendStorageRouting::from_arc(routing), + ); + self + } + + /// Phase α (MVP): attach a pre-built hot-reload routing handle. + /// Used by callers that want to share the same handle with other + /// subsystems (e.g. the query-router for diagnostics) — the + /// `with_backend_storage_routing` builder is the simpler entry + /// point that wraps an `Arc` for callers + /// that don't. + pub fn with_hot_reload_backend_storage_routing( + mut self, + handle: crate::routing::HotReloadBackendStorageRouting, + ) -> Self { + self.backend_storage_routing = Some(handle); self } @@ -292,6 +323,13 @@ impl HttpServer { "/api/v1/streaming-config", get(handle_get_streaming_config).post(handle_post_streaming_config), ) + // Phase α (MVP): controller-pushed `BackendStorageRouting` + // table. POST replaces the current table atomically; GET + // returns a JSON snapshot for operator diagnostics. + .route( + "/api/v1/storage_routing", + get(handle_get_storage_routing).post(handle_post_storage_routing), + ) .route("/api/v1/db/schemas", get(handle_get_schemas)) .route( "/api/v1/db/schemas/:agg_id/retire", @@ -354,6 +392,13 @@ impl HttpServer { "/api/v1/streaming-config", get(handle_get_streaming_config).post(handle_post_streaming_config), ) + // Phase α (MVP): controller-pushed `BackendStorageRouting` + // table. POST replaces the current table atomically; GET + // returns a JSON snapshot for operator diagnostics. + .route( + "/api/v1/storage_routing", + get(handle_get_storage_routing).post(handle_post_storage_routing), + ) .route("/api/v1/db/schemas", get(handle_get_schemas)) .route( "/api/v1/db/schemas/:agg_id/retire", @@ -509,7 +554,12 @@ async fn process_query_request( /// 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() { + if let Some(routing_handle) = state.backend_storage_routing.as_ref() { + // Phase α: snapshot the hot-reload handle once per request. + // Concurrent swaps from `POST /api/v1/storage_routing` produce + // a fresh `Arc`; this snapshot remains valid for the rest of + // the dispatch (no torn read). + let routing = routing_handle.snapshot(); match promql_parser::parser::parse(query) { Ok(expr) => { if let Some(metric_name) = first_metric_name(&expr) { @@ -3434,6 +3484,208 @@ aggregations: assert_eq!(gorilla_calls.load(Ordering::SeqCst), 1); } + // ── Phase α: BackendStorageRouting hot-reload HTTP integration ──── + + /// Standard test wiring for the `/api/v1/storage_routing` endpoint: + /// install an empty hot-reload routing handle, hold the handle so + /// the test can introspect the swap result. + async fn setup_test_server_for_storage_routing() -> (u16, crate::routing::HotReloadBackendStorageRouting) { + use crate::data_model::{HotReloadStreamingConfig, StreamingConfig}; + use crate::routing::HotReloadBackendStorageRouting; + + 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, + ); + 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 routing_handle = HotReloadBackendStorageRouting::empty(); + let server = HttpServer::new(config, query_engine, store, None) + .with_hot_reload_config(hot_reload) + .with_hot_reload_backend_storage_routing(routing_handle.clone()); + let port = server.start_test_server().await.expect("start ok"); + (port, routing_handle) + } + + fn fixture_routing_json() -> serde_json::Value { + serde_json::json!({ + "default_engine": "sketch_warm_tier", + "metrics": [ + { + "name": "http_requests_total", + "targets": [ + { "engine": "sketch_warm_tier" }, + { + "engine": "thanos_archive", + "applies_to_query_shape": [ + "histogram_quantile", "delta", "absent", + "rate_post_hoc", "count" + ] + } + ] + } + ] + }) + } + + #[tokio::test] + async fn storage_routing_post_swaps_table_atomically() { + let (port, handle) = setup_test_server_for_storage_routing().await; + // Initial table is empty. + assert_eq!(handle.snapshot().len(), 0); + + let client = Client::new(); + let resp = client + .post(format!("http://127.0.0.1:{port}/api/v1/storage_routing")) + .header("Content-Type", "application/json") + .body(fixture_routing_json().to_string()) + .send() + .await + .expect("send ok"); + + assert!(resp.status().is_success(), "swap must 2xx; got {}", resp.status()); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["status"], "success"); + assert_eq!(body["metrics_count"], 1); + let returned_hash = body["table_hash"].as_str().unwrap().to_string(); + assert!(!returned_hash.is_empty(), "hash must be non-empty"); + + // Snapshot now reflects the new table — and the hash matches + // what the response advertised. + let snap = handle.snapshot(); + assert_eq!(snap.len(), 1); + let live_hash = crate::routing::routing_table_hash(snap.as_ref()); + assert_eq!(live_hash, returned_hash, "live hash must match advertised"); + } + + #[tokio::test] + async fn storage_routing_post_rejects_invalid_json() { + let (port, handle) = setup_test_server_for_storage_routing().await; + let client = Client::new(); + + // Garbage body — not even valid JSON. + let resp = client + .post(format!("http://127.0.0.1:{port}/api/v1/storage_routing")) + .header("Content-Type", "application/json") + .body("not json {{") + .send() + .await + .expect("send ok"); + assert_eq!(resp.status().as_u16(), 400, "garbage body must 400"); + + // Valid JSON but invalid schema (unknown engine). + let bad = serde_json::json!({ + "default_engine": "sketch_warm_tier", + "metrics": [{ + "name": "x", + "targets": [{ "engine": "not_a_real_engine" }] + }] + }); + let resp = client + .post(format!("http://127.0.0.1:{port}/api/v1/storage_routing")) + .header("Content-Type", "application/json") + .body(bad.to_string()) + .send() + .await + .expect("send ok"); + assert_eq!( + resp.status().as_u16(), + 400, + "schema-invalid body must 400; got {}", + resp.status(), + ); + + // Confirm the table was NOT swapped (still empty). + assert_eq!(handle.snapshot().len(), 0); + } + + #[tokio::test] + async fn storage_routing_get_returns_current_snapshot() { + let (port, handle) = setup_test_server_for_storage_routing().await; + // Pre-load the table. + let new = crate::data_model::BackendStorageRouting::from_json_payload( + &fixture_routing_json(), + ) + .expect("parse"); + handle.swap(new); + + let client = Client::new(); + let resp = client + .get(format!("http://127.0.0.1:{port}/api/v1/storage_routing")) + .send() + .await + .expect("send ok"); + assert!(resp.status().is_success()); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_eq!(body["status"], "success"); + assert_eq!(body["default_engine"], "sketch_warm"); + assert_eq!(body["metrics_count"], 1); + let snap_hash = body["table_hash"].as_str().unwrap(); + let live_hash = crate::routing::routing_table_hash(handle.snapshot().as_ref()); + assert_eq!(snap_hash, live_hash); + } + + #[tokio::test] + async fn storage_routing_swap_observed_by_subsequent_query_dispatch() { + // End-to-end production-path test: POST a routing table, then + // issue a `count(http_requests_total)` query. The handler must + // see the freshly-swapped table and route the query through + // the EngineRouter. We can't easily assert the response engine + // without setting up a Gorilla mock, but we can verify the + // swap landed by GETting the hash — that's the contract the + // controller relies on. + let (port, handle) = setup_test_server_for_storage_routing().await; + let client = Client::new(); + + let resp = client + .post(format!("http://127.0.0.1:{port}/api/v1/storage_routing")) + .header("Content-Type", "application/json") + .body(fixture_routing_json().to_string()) + .send() + .await + .expect("send ok"); + assert!(resp.status().is_success()); + + // `lookup_with_shape` must reflect the swapped contents on + // the very next read. + let snap = handle.snapshot(); + assert_eq!( + snap.lookup_with_shape( + "http_requests_total", + crate::data_model::QueryShape::Count, + ), + StorageBackend::GorillaS3Archive, + ); + assert_eq!( + snap.lookup_with_shape( + "http_requests_total", + crate::data_model::QueryShape::Quantile, + ), + StorageBackend::SketchWarmTier, + ); + } + } // ── Controller integration: PrecomputeJob execution ────────────────────────── @@ -3668,6 +3920,118 @@ async fn handle_post_streaming_config( (StatusCode::OK, axum::Json(body)).into_response() } +// ── Phase α: BackendStorageRouting hot-reload endpoints ──────────── + +/// `GET /api/v1/storage_routing` — return a JSON snapshot of the +/// currently-active per-metric `BackendStorageRouting` table. +/// +/// Useful for operators to confirm a controller push landed with the +/// expected entries. Returns 503 when the backend wasn't built with a +/// routing-table handle (legacy deploys that loaded the YAML directly +/// can still hit `/api/v1/streaming-config` — this endpoint is for +/// the Phase α JSON path). +async fn handle_get_storage_routing(State(state): State) -> axum::response::Response { + use axum::http::StatusCode; + use axum::response::IntoResponse; + + let Some(handle) = state.backend_storage_routing.as_ref() else { + let body = serde_json::json!({ + "status": "error", + "error": "routing handle not attached; backend was built without HttpServer::with_backend_storage_routing", + }); + return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response(); + }; + let snap = handle.snapshot(); + let body = serde_json::json!({ + "status": "success", + "default_engine": snap.default_backend().data_source_id(), + "metrics_count": snap.len(), + "table_hash": crate::routing::routing_table_hash(snap.as_ref()), + }); + (StatusCode::OK, axum::Json(body)).into_response() +} + +/// `POST /api/v1/storage_routing` — replace the per-metric routing +/// table from a controller-emitted JSON document. +/// +/// Body shape — see +/// `controller/src/config/stage_config.rs::emit_backend_storage_routing` +/// (or `BackendStorageRouting::from_json_payload` in this crate for +/// the matching parser). On 2xx the response body carries the new +/// table's hash and entry count so the controller can verify the +/// installed bytes match what it pushed. +/// +/// Errors: +/// * 400 — body is not valid UTF-8, not valid JSON, or the JSON +/// fails the schema check (unknown engine / empty targets / etc.). +/// * 503 — backend wasn't built with a routing-table handle. +/// +/// The swap is atomic: in-flight queries either see the entire old +/// table or the entire new table; never a half-applied state. Mirrors +/// the existing `POST /api/v1/streaming-config` swap contract. +async fn handle_post_storage_routing( + State(state): State, + body: axum::body::Bytes, +) -> axum::response::Response { + use axum::http::StatusCode; + use axum::response::IntoResponse; + + let Some(handle) = state.backend_storage_routing.as_ref() else { + let body = serde_json::json!({ + "status": "error", + "error": "routing handle not attached; backend was built without HttpServer::with_backend_storage_routing", + }); + return (StatusCode::SERVICE_UNAVAILABLE, axum::Json(body)).into_response(); + }; + + let json_text = match std::str::from_utf8(&body) { + Ok(s) => s, + Err(e) => { + let body = serde_json::json!({ + "status": "error", + "error": format!("request body is not valid UTF-8: {e}"), + }); + return (StatusCode::BAD_REQUEST, axum::Json(body)).into_response(); + } + }; + let json_value: serde_json::Value = match serde_json::from_str(json_text) { + Ok(v) => v, + Err(e) => { + let body = serde_json::json!({ + "status": "error", + "error": format!("JSON parse error: {e}"), + }); + return (StatusCode::BAD_REQUEST, axum::Json(body)).into_response(); + } + }; + let new_table = match crate::data_model::BackendStorageRouting::from_json_payload(&json_value) { + Ok(t) => t, + Err(e) => { + let body = serde_json::json!({ + "status": "error", + "error": format!("BackendStorageRouting build error: {e:#}"), + }); + return (StatusCode::BAD_REQUEST, axum::Json(body)).into_response(); + } + }; + + let entries = new_table.len(); + let hash = crate::routing::routing_table_hash(&new_table); + let _old = handle.swap(new_table); + info!( + entries, + table_hash = %hash, + "storage-routing JSON hot-reload swap completed", + ); + + let body = serde_json::json!({ + "status": "success", + "metrics_count": entries, + "table_hash": hash, + }); + (StatusCode::OK, axum::Json(body)).into_response() +} + /// §15.2 of the sketch DB design: expose the `SchemaRegistry` over /// HTTP so operators and the controller can inspect agg lifecycle /// state without attaching a debugger. Filter by `?status=` — diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index 99873d3ab..2af946ecb 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -670,7 +670,14 @@ async fn main() -> Result<()> { // 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() { + // + // Phase α (MVP): even when no static YAML is loaded, install an + // empty hot-reload handle so the controller's first + // `POST /api/v1/storage_routing` push lands without first-call 503 + // lossage. Operators can still hand-author the YAML for + // dev / standalone — the YAML supplies the bootstrap, controller + // pushes overwrite it. + let bootstrap_routing = 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!( @@ -679,20 +686,23 @@ async fn main() -> Result<()> { routing.default_backend(), routing.len(), ); - server = server.with_backend_storage_routing(Arc::new(routing)); + routing } Err(e) => { warn!( - "Failed to load backend-storage-routing from {:?}: {} — falling back to streaming-config single axis", + "Failed to load backend-storage-routing from {:?}: {} — installing an empty routing table; the controller's first POST /api/v1/storage_routing push will fill it", routing_path, e, ); + query_engine_rust::data_model::BackendStorageRouting::empty() } } } else { info!( - "--backend-storage-routing not set — every query routes per the streaming-config single axis (typically `sketch_warm`)", + "--backend-storage-routing not set — installing an empty routing table; the controller's first POST /api/v1/storage_routing push will fill it", ); - } + query_engine_rust::data_model::BackendStorageRouting::empty() + }; + server = server.with_backend_storage_routing(Arc::new(bootstrap_routing)); // Phase-5/6: register a `GorillaQueryEngine` for the cold // archive tier when the operator has provisioned one via the diff --git a/asap-query-engine/src/routing/backend_storage_routing.rs b/asap-query-engine/src/routing/backend_storage_routing.rs index ee3557635..161d06104 100644 --- a/asap-query-engine/src/routing/backend_storage_routing.rs +++ b/asap-query-engine/src/routing/backend_storage_routing.rs @@ -99,6 +99,7 @@ use std::path::Path; use anyhow::{Context, Result}; use asap_types::StorageBackend; use serde::{Deserialize, Serialize}; +use serde_json::Value as JsonValue; use tracing::{debug, info}; // --------------------------------------------------------------------------- @@ -115,6 +116,11 @@ use tracing::{debug, info}; /// 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). +/// +/// Phase α (controller-emitted routing tables) adds `HistogramQuantile` +/// / `Delta` / `Deriv` / `Absent` — these are PromQL shapes no warm-tier +/// sketch can serve and the controller's emitter reliably routes them +/// to the archive. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum QueryShape { @@ -146,6 +152,20 @@ pub enum QueryShape { /// accumulator (Change B), archive serves by selecting the /// most-recent sample in each chunk. LastOverTime, + /// `histogram_quantile(φ, ...)` — Prometheus-native histogram + /// readout. No warm-tier sketch fits the bucket-vector input + /// shape; archive serves via post-hoc bucket scan. + HistogramQuantile, + /// `delta([])` — first-difference over a range. + /// Archive-only. + Delta, + /// `deriv([])` — least-squares slope of a gauge. + /// Archive-only. + Deriv, + /// `absent({...})` — 1 if no series match, vacuous + /// vector otherwise. Archive answers natively from the postings + /// index; warm-tier sketch has no compatible aggregation. + Absent, /// 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 @@ -154,7 +174,11 @@ pub enum QueryShape { } impl QueryShape { - /// Stable string tag used in YAML. + /// Stable string tag used in YAML / JSON. Mirrors the controller's + /// `config::stage_config::emit_backend_storage_routing` shape + /// vocabulary — the wire form is the canonical PromQL function + /// name (or a `_` prefixed variant for shapes without a single + /// canonical name, e.g. `rate_post_hoc`). pub fn as_str(self) -> &'static str { match self { QueryShape::Count => "count", @@ -163,6 +187,10 @@ impl QueryShape { QueryShape::Quantile => "quantile", QueryShape::Sum => "sum", QueryShape::LastOverTime => "last_over_time", + QueryShape::HistogramQuantile => "histogram_quantile", + QueryShape::Delta => "delta", + QueryShape::Deriv => "deriv", + QueryShape::Absent => "absent", QueryShape::Other => "other", } } @@ -220,6 +248,14 @@ pub fn classify_query_shape(expr: &promql_parser::parser::Expr) -> QueryShape { QueryShape::Count } else if name == "last_over_time" { QueryShape::LastOverTime + } else if name == "histogram_quantile" { + QueryShape::HistogramQuantile + } else if name == "delta" || name == "increase" { + QueryShape::Delta + } else if name == "deriv" { + QueryShape::Deriv + } else if name == "absent" || name == "absent_over_time" { + QueryShape::Absent } else { QueryShape::Other } @@ -425,6 +461,143 @@ impl BackendStorageRouting { Ok(routing) } + /// Phase α (MVP): parse a controller-emitted JSON document into a + /// fresh routing table. The schema mirrors + /// `controller/src/config/stage_config.rs::emit_backend_storage_routing`: + /// + /// ```json + /// { + /// "default_engine": "sketch_warm_tier", + /// "metrics": [ + /// { "name": "http_requests_total", + /// "targets": [ + /// { "engine": "sketch_warm_tier" }, + /// { "engine": "thanos_archive", + /// "applies_to_query_shape": ["count", "topk", "rate_post_hoc", + /// "histogram_quantile", "delta", "absent"] } + /// ] + /// } + /// ] + /// } + /// ``` + /// + /// Engine-name compatibility (controller → backend `StorageBackend`): + /// + /// * `sketch_warm_tier` → `SketchWarmTier` + /// * `thanos_archive` → `GorillaS3Archive` (Phase α uses the existing + /// archive engine; future phases may register a real Thanos engine). + /// * `gorilla_s3_archive` → `GorillaS3Archive` (back-compat alias). + /// * `double_write` → `DoubleWrite`. + /// + /// Unknown query-shape strings are mapped to [`QueryShape::Other`] + /// rather than failing the parse — the controller's vocabulary may + /// drift forward of the backend's. Empty `targets` arrays are + /// rejected (same contract as `from_yaml_str`). + /// + /// Side fields (e.g. `warm_tier_native_shapes`) the controller emits + /// for operator inspection are ignored — the JSON parser pulls only + /// `default_engine` and `metrics:[...]`. + pub fn from_json_payload(value: &JsonValue) -> Result { + let default_engine = value + .get("default_engine") + .and_then(|v| v.as_str()) + .unwrap_or("sketch_warm_tier"); + let default = parse_engine_string(default_engine).with_context(|| { + format!( + "backend-storage-routing JSON: invalid default_engine '{}'", + default_engine + ) + })?; + + let metrics_arr = value + .get("metrics") + .and_then(|v| v.as_array()) + .ok_or_else(|| { + anyhow::anyhow!("backend-storage-routing JSON: missing 'metrics' array") + })?; + + let mut metrics: HashMap> = HashMap::new(); + for (i, entry) in metrics_arr.iter().enumerate() { + let name = entry + .get("name") + .and_then(|v| v.as_str()) + .ok_or_else(|| { + anyhow::anyhow!( + "backend-storage-routing JSON: metrics[{}] missing 'name'", + i + ) + })? + .to_string(); + let targets_arr = entry + .get("targets") + .and_then(|v| v.as_array()) + .ok_or_else(|| { + anyhow::anyhow!( + "backend-storage-routing JSON: metric '{}' missing 'targets' array", + name, + ) + })?; + if targets_arr.is_empty() { + anyhow::bail!( + "backend-storage-routing JSON: metric '{}' has empty targets list", + name, + ); + } + let mut targets: Vec = Vec::with_capacity(targets_arr.len()); + for (j, t) in targets_arr.iter().enumerate() { + let engine_str = t + .get("engine") + .and_then(|v| v.as_str()) + .ok_or_else(|| { + anyhow::anyhow!( + "backend-storage-routing JSON: metric '{}' targets[{}] missing 'engine'", + name, j, + ) + })?; + let backend = parse_engine_string(engine_str).with_context(|| { + format!( + "backend-storage-routing JSON: metric '{}' targets[{}] invalid engine '{}'", + name, j, engine_str, + ) + })?; + let applies_to_query_shape = + t.get("applies_to_query_shape").and_then(|v| v.as_array()).map(|arr| { + arr.iter() + .filter_map(|s| s.as_str()) + .map(parse_query_shape_string) + .collect::>() + }); + targets.push(RoutingTarget { + backend, + applies_to_query_shape, + }); + } + metrics.insert(name, targets); + } + + Ok(Self { default, metrics }) + } + + /// Atomically replace this routing table's contents with `new_table`. + /// Used by the `POST /api/v1/storage_routing` swap handler — the + /// HTTP layer wraps the table in `arc_swap::ArcSwap` so the swap is + /// observed atomically by all in-flight queries; this method is the + /// in-place form that callers without an `ArcSwap` wrapper can use + /// (e.g. tests, single-threaded shadow-mode evaluation). Production + /// deployments should go through [`HotReloadBackendStorageRouting`] + /// instead. + pub fn replace(&mut self, new_table: BackendStorageRouting) { + let added = new_table.metrics.len(); + let removed = self.metrics.len(); + *self = new_table; + info!( + added, + removed, + new_default = ?self.default, + "BackendStorageRouting: replaced (in-place)", + ); + } + /// 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 @@ -571,6 +744,172 @@ impl Default for BackendStorageRouting { } } +/// Map a JSON `engine` string into a backend `StorageBackend` variant. +/// Phase α accepts both the controller's vocabulary (`thanos_archive`) +/// and the existing YAML's vocabulary (`gorilla_s3_archive`) — both +/// resolve to `StorageBackend::GorillaS3Archive` because the cold-archive +/// engine registered today serves both via `GorillaQueryEngine`. +/// `unknown_engine` returns an error so a typo doesn't silently turn +/// into a default-routing footgun. +fn parse_engine_string(s: &str) -> Result { + match s { + "sketch_warm_tier" | "sketch_warm" => Ok(StorageBackend::SketchWarmTier), + // `thanos_archive` is the controller-emitted name; the backend + // currently registers the Gorilla-S3 cold archive under + // `gorilla_archive` / `gorilla_s3_archive`. They map to the + // same `StorageBackend` variant for Phase α — when a real + // Thanos engine lands the parser can split the two. + "thanos_archive" | "gorilla_s3_archive" | "gorilla_archive" => { + Ok(StorageBackend::GorillaS3Archive) + } + "double_write" => Ok(StorageBackend::DoubleWrite), + other => Err(anyhow::anyhow!( + "unknown engine '{}': expected one of \ + [sketch_warm_tier, thanos_archive, gorilla_s3_archive, double_write]", + other, + )), + } +} + +/// Map a JSON `applies_to_query_shape` string into a backend +/// `QueryShape`. Unknown shapes are mapped to [`QueryShape::Other`] — +/// the controller's vocabulary may emit shape names a backend revision +/// doesn't yet understand, and `Other` is the safe fall-through (the +/// archive's claim list typically includes `Other` so unknowns still +/// route to the archive). +fn parse_query_shape_string(s: &str) -> QueryShape { + match s { + "count" => QueryShape::Count, + "topk" => QueryShape::Topk, + "rate_post_hoc" | "rate" | "irate" => QueryShape::RatePostHoc, + "quantile" | "quantile_over_time" => QueryShape::Quantile, + "sum" | "sum_over_time" => QueryShape::Sum, + "last_over_time" => QueryShape::LastOverTime, + "histogram_quantile" => QueryShape::HistogramQuantile, + "delta" | "increase" => QueryShape::Delta, + "deriv" => QueryShape::Deriv, + "absent" | "absent_over_time" => QueryShape::Absent, + _ => QueryShape::Other, + } +} + +/// Compute a stable, short hash of a `BackendStorageRouting` table for +/// the swap handler's response. The controller uses this to verify the +/// backend installed exactly the bytes it pushed (cheap drift check on +/// every plan emit). +/// +/// The hash is computed over the routing table's data fields — default +/// + sorted metric → sorted target list. We sort to make the hash +/// reproducible across `HashMap` iteration orders. +pub fn routing_table_hash(table: &BackendStorageRouting) -> String { + use std::collections::hash_map::DefaultHasher; + use std::hash::{Hash, Hasher}; + + let mut h = DefaultHasher::new(); + // Default tag. + table.default.data_source_id().hash(&mut h); + // Sort metric names for determinism. + let mut names: Vec<&String> = table.metrics.keys().collect(); + names.sort(); + for name in names { + name.hash(&mut h); + let targets = &table.metrics[name]; + for t in targets { + t.backend.data_source_id().hash(&mut h); + if let Some(shapes) = &t.applies_to_query_shape { + for s in shapes { + s.as_str().hash(&mut h); + } + "}".hash(&mut h); + } else { + "*".hash(&mut h); + } + } + } + format!("{:016x}", h.finish()) +} + +// --------------------------------------------------------------------------- +// Phase α: HotReloadBackendStorageRouting — atomic-swap wrapper +// --------------------------------------------------------------------------- + +/// Atomic-swap wrapper around `BackendStorageRouting`, mirroring +/// [`crate::data_model::HotReloadStreamingConfig`]. Lets the +/// `POST /api/v1/storage_routing` HTTP handler swap the table at +/// runtime without restarting the backend. Cloneable; clones share the +/// underlying `ArcSwap` so all holders see the same swaps. +/// +/// ## Read path +/// +/// HTTP query handler calls [`Self::snapshot`] once per request to get +/// a stable `Arc` it can call +/// `lookup_with_shape` on. The snapshot is cheap (single atomic load) +/// and lock-free; concurrent swaps don't block readers. +/// +/// ## Write path +/// +/// The swap handler calls [`Self::swap`] with the new table parsed from +/// the controller's JSON. The previous `Arc` is dropped when the last +/// in-flight reader goes out of scope. +/// +/// ## Bootstrap +/// +/// Built at backend startup from either: +/// * `Self::from_yaml_file(path)` — load the static +/// `deploy/configs/backend-storage-routing.yaml` (legacy +/// bootstrap; preserved for dev / standalone deployments). +/// * `Self::empty()` — start with an empty table; the controller's +/// first push fills it. +#[derive(Clone)] +pub struct HotReloadBackendStorageRouting { + inner: std::sync::Arc>, +} + +impl HotReloadBackendStorageRouting { + /// Construct with an initial routing table. + pub fn new(initial: BackendStorageRouting) -> Self { + Self { + inner: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new(initial))), + } + } + + /// Construct with an empty table — every metric resolves to + /// `SketchWarmTier` until the first push lands. + pub fn empty() -> Self { + Self::new(BackendStorageRouting::empty()) + } + + /// Construct from a pre-built `Arc` — + /// avoids a redundant clone when the caller already holds one. + pub fn from_arc(initial: std::sync::Arc) -> Self { + Self { + inner: std::sync::Arc::new(arc_swap::ArcSwap::new(initial)), + } + } + + /// Cheap, cloneable snapshot of the current table. Stable for the + /// caller's lifetime; concurrent swaps don't invalidate it. + pub fn snapshot(&self) -> std::sync::Arc { + self.inner.load_full() + } + + /// Atomically replace the current table. Returns the `Arc` that + /// was just replaced for callers that want to log the diff. + pub fn swap(&self, new: BackendStorageRouting) -> std::sync::Arc { + self.inner.swap(std::sync::Arc::new(new)) + } +} + +impl std::fmt::Debug for HotReloadBackendStorageRouting { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let snap = self.snapshot(); + f.debug_struct("HotReloadBackendStorageRouting") + .field("entries", &snap.metrics.len()) + .field("default", &snap.default) + .finish() + } +} + #[cfg(test)] mod tests { use super::*; @@ -853,4 +1192,269 @@ routes: let e = parse("http_requests_total"); assert_eq!(classify_query_shape(&e), QueryShape::Other); } + + // ── Phase α: from_json_payload + replace + hot-reload tests ──────── + + fn fixture_json() -> serde_json::Value { + serde_json::json!({ + "default_engine": "sketch_warm_tier", + "metrics": [ + { + "name": "http_requests_total", + "targets": [ + { "engine": "sketch_warm_tier" }, + { + "engine": "thanos_archive", + "applies_to_query_shape": [ + "histogram_quantile", "delta", "deriv", + "absent", "rate_post_hoc", "count" + ] + } + ], + "warm_tier_native_shapes": ["topk", "rate", "sum"] + }, + { + "name": "request_latency_seconds", + "targets": [ + { "engine": "sketch_warm_tier" }, + { + "engine": "thanos_archive", + "applies_to_query_shape": [ + "histogram_quantile", "delta", "absent", + "rate_post_hoc", "topk", "count" + ] + } + ] + } + ] + }) + } + + #[test] + fn json_payload_parses_controller_fixture() { + let r = BackendStorageRouting::from_json_payload(&fixture_json()).expect("parse"); + assert_eq!(r.default_backend(), StorageBackend::SketchWarmTier); + assert_eq!(r.len(), 2); + + // http_requests_total: histogram_quantile / delta / etc → archive, + // quantile / sum / topk → warm. + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::HistogramQuantile), + StorageBackend::GorillaS3Archive, + ); + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Delta), + StorageBackend::GorillaS3Archive, + ); + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Count), + StorageBackend::GorillaS3Archive, + ); + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Quantile), + StorageBackend::SketchWarmTier, + ); + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::Topk), + StorageBackend::SketchWarmTier, + ); + // LastOverTime not in the archive's filter list → falls + // through to the default (warm) slot. + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::LastOverTime), + StorageBackend::SketchWarmTier, + ); + } + + #[test] + fn json_payload_unknown_shape_defaults_to_other() { + let value = serde_json::json!({ + "default_engine": "sketch_warm_tier", + "metrics": [ + { + "name": "x", + "targets": [ + { "engine": "sketch_warm_tier" }, + { + "engine": "thanos_archive", + "applies_to_query_shape": ["some_future_shape", "count"] + } + ] + } + ] + }); + let r = BackendStorageRouting::from_json_payload(&value).expect("parse"); + // count still routes to archive; the unknown shape was mapped + // to QueryShape::Other (silently — forward-compat). + assert_eq!( + r.lookup_with_shape("x", QueryShape::Count), + StorageBackend::GorillaS3Archive, + ); + assert_eq!( + r.lookup_with_shape("x", QueryShape::Other), + StorageBackend::GorillaS3Archive, + ); + } + + #[test] + fn json_payload_invalid_engine_errors() { + let value = serde_json::json!({ + "default_engine": "sketch_warm_tier", + "metrics": [{ + "name": "x", + "targets": [{ "engine": "not_a_real_engine" }] + }] + }); + let err = BackendStorageRouting::from_json_payload(&value).expect_err("must reject"); + // The bad engine name appears somewhere in the error chain + // (the parser wraps the inner `parse_engine_string` error in + // a context that mentions the metric/target index). + let chain_str = format!("{err:#}"); + assert!( + chain_str.contains("not_a_real_engine"), + "error chain must mention the bad engine: {chain_str}" + ); + } + + #[test] + fn json_payload_empty_targets_errors() { + let value = serde_json::json!({ + "default_engine": "sketch_warm_tier", + "metrics": [{ "name": "x", "targets": [] }] + }); + let err = BackendStorageRouting::from_json_payload(&value).expect_err("must reject"); + assert!(err.to_string().contains("empty targets")); + } + + #[test] + fn json_payload_missing_metrics_errors() { + let value = serde_json::json!({ "default_engine": "sketch_warm_tier" }); + let err = BackendStorageRouting::from_json_payload(&value).expect_err("must reject"); + assert!(err.to_string().contains("metrics")); + } + + #[test] + fn json_payload_default_engine_optional_falls_back_to_warm() { + let value = serde_json::json!({ + "metrics": [ + { "name": "x", "targets": [{ "engine": "sketch_warm_tier" }] } + ] + }); + let r = BackendStorageRouting::from_json_payload(&value).expect("parse"); + assert_eq!(r.default_backend(), StorageBackend::SketchWarmTier); + } + + #[test] + fn json_payload_back_compat_gorilla_s3_archive_alias() { + // An older deploy might emit the YAML's vocabulary instead of + // `thanos_archive`; both must parse and resolve the same. + let value = serde_json::json!({ + "default_engine": "sketch_warm_tier", + "metrics": [{ + "name": "audit_events", + "targets": [ + { "engine": "gorilla_s3_archive" } + ] + }] + }); + let r = BackendStorageRouting::from_json_payload(&value).expect("parse"); + assert_eq!(r.lookup("audit_events"), StorageBackend::GorillaS3Archive); + } + + #[test] + fn replace_swaps_table_in_place() { + let mut r = BackendStorageRouting::new_from_single_targets( + StorageBackend::SketchWarmTier, + HashMap::from([( + "old_metric".to_string(), + StorageBackend::GorillaS3Archive, + )]), + ); + let new = BackendStorageRouting::from_json_payload(&fixture_json()).expect("parse"); + r.replace(new); + // Old metric is gone; new metrics are visible. + assert_eq!(r.lookup("old_metric"), StorageBackend::SketchWarmTier); + assert_eq!( + r.lookup_with_shape("http_requests_total", QueryShape::HistogramQuantile), + StorageBackend::GorillaS3Archive, + ); + } + + #[test] + fn hot_reload_wrapper_swap_is_observed_by_clones() { + let hr = HotReloadBackendStorageRouting::empty(); + let hr_writer = hr.clone(); + + let new = BackendStorageRouting::from_json_payload(&fixture_json()).expect("parse"); + hr_writer.swap(new); + + let snap = hr.snapshot(); + assert_eq!(snap.len(), 2); + assert_eq!( + snap.lookup_with_shape("http_requests_total", QueryShape::Delta), + StorageBackend::GorillaS3Archive, + ); + } + + #[test] + fn hot_reload_wrapper_concurrent_readers_see_no_torn_state() { + use std::thread; + let hr = HotReloadBackendStorageRouting::empty(); + let writer_hr = hr.clone(); + let writer = thread::spawn(move || { + for i in 0..50 { + let mut metrics = HashMap::new(); + metrics.insert( + format!("metric_{i}"), + vec![RoutingTarget::always(StorageBackend::GorillaS3Archive)], + ); + let new = BackendStorageRouting::new(StorageBackend::SketchWarmTier, metrics); + writer_hr.swap(new); + } + }); + let reader_hr = hr.clone(); + let reader = thread::spawn(move || { + for _ in 0..200 { + let snap = reader_hr.snapshot(); + // Snapshot must always be internally consistent — + // either empty (initial) or one-entry (post-swap). + let n = snap.len(); + assert!(n == 0 || n == 1, "torn snapshot: {n} entries"); + } + }); + writer.join().unwrap(); + reader.join().unwrap(); + } + + #[test] + fn routing_table_hash_is_stable_across_runs() { + let r1 = BackendStorageRouting::from_json_payload(&fixture_json()).expect("parse 1"); + let r2 = BackendStorageRouting::from_json_payload(&fixture_json()).expect("parse 2"); + assert_eq!(routing_table_hash(&r1), routing_table_hash(&r2)); + } + + #[test] + fn routing_table_hash_differs_when_table_differs() { + let r1 = BackendStorageRouting::from_json_payload(&fixture_json()).expect("parse"); + let r2 = BackendStorageRouting::empty(); + assert_ne!(routing_table_hash(&r1), routing_table_hash(&r2)); + } + + #[test] + fn classifies_histogram_quantile_correctly() { + let e = parse("histogram_quantile(0.99, sum by (le) (rate(http_request_duration_bucket[5m])))"); + assert_eq!(classify_query_shape(&e), QueryShape::HistogramQuantile); + } + + #[test] + fn classifies_delta_correctly() { + let e = parse("delta(http_requests_total[5m])"); + assert_eq!(classify_query_shape(&e), QueryShape::Delta); + } + + #[test] + fn classifies_absent_correctly() { + let e = parse("absent(http_requests_total{job=\"x\"})"); + assert_eq!(classify_query_shape(&e), QueryShape::Absent); + } } diff --git a/asap-query-engine/src/routing/mod.rs b/asap-query-engine/src/routing/mod.rs index 9a099367f..1cfc1c33e 100644 --- a/asap-query-engine/src/routing/mod.rs +++ b/asap-query-engine/src/routing/mod.rs @@ -26,7 +26,8 @@ pub mod backend_storage_routing; pub mod engine_router; pub use backend_storage_routing::{ - classify_query_shape, BackendStorageRouting, QueryShape, RoutingTarget, + classify_query_shape, routing_table_hash, BackendStorageRouting, + HotReloadBackendStorageRouting, QueryShape, RoutingTarget, }; pub use engine_router::{ EngineCapabilities, EngineRouter, EngineRouterError, QueryEngine,