diff --git a/data_plane/src/drivers/query/servers/http.rs b/data_plane/src/drivers/query/servers/http.rs index 37d75e43..5bdd2a4b 100644 --- a/data_plane/src/drivers/query/servers/http.rs +++ b/data_plane/src/drivers/query/servers/http.rs @@ -41,6 +41,40 @@ use asap_types::Statistic; /// unknown value returns 400. pub const ENGINE_OVERRIDE_HEADER: &str = "X-ASAP-Engine"; pub const ENGINE_OVERRIDE_QUERY_PARAM: &str = "engine"; +/// Per-request accuracy contract. `exact` routes ASAP-managed metrics directly +/// to the archive; `approximate` (the default) keeps warm-first failover. +pub const ACCURACY_HEADER: &str = "X-ASAP-Accuracy"; + +fn extract_accuracy(headers: &HeaderMap) -> Result { + let Some(value) = headers.get(ACCURACY_HEADER) else { + return Ok(AccuracyTarget::Epsilon(0.01)); + }; + let value = value + .to_str() + .map_err(|_| format!("{ACCURACY_HEADER} must be valid UTF-8"))? + .trim(); + if value.eq_ignore_ascii_case("exact") { + Ok(AccuracyTarget::Exact) + } else if value.eq_ignore_ascii_case("approximate") { + Ok(AccuracyTarget::Epsilon(0.01)) + } else { + Err(format!( + "invalid {ACCURACY_HEADER} value {value:?}; expected `exact` or `approximate`" + )) + } +} + +fn invalid_accuracy_response(error: String) -> Response { + ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ + "status": "error", + "errorType": "bad_data", + "error": error, + })), + ) + .into_response() +} /// Per-request tenant header (per-tenant `BackendStorageRouting`, /// follow-up to PR #333). @@ -617,6 +651,7 @@ async fn process_query_request( start_time: Instant, headers: HashMap, engine_override: Option, + accuracy: AccuracyTarget, tenant: &str, ) -> Response { // Check if handling is enabled @@ -721,10 +756,12 @@ async fn process_query_request( state.hot_reload_config.is_some(), ); - if matches!(metric_storage, StorageBackend::SketchStore) { + if matches!(metric_storage, StorageBackend::SketchStore) + && !matches!(accuracy, AccuracyTarget::Exact) + { process_via_simple_engine(state, parsed_request, start_time, headers).await } else { - process_via_router(state, parsed_request, start_time, metric_storage).await + process_via_router(state, parsed_request, start_time, metric_storage, accuracy).await } } @@ -1381,16 +1418,9 @@ async fn process_via_named_engine( /// * `metric_storage` — looked up from the hot-reload `StreamingConfig` /// by the caller. /// -/// The remaining two — `Statistic` + `AccuracyTarget` — would -/// normally be derived by parsing the PromQL AST. Pre-Phase-6 we don't -/// have an HTTP-side parser wired in; the router's -/// [`compatible_storage_backends`] consults them only for the -/// `DoubleWrite` head-selection heuristic (other deploy shapes -/// degenerate to a fixed list keyed only by `metric_storage`), so -/// defaulting to `(Sum, Approximate)` is safe for `GorillaObjectStore`- -/// only deploys. A follow-up will thread -/// the real values through once the Phase-6 query-tracker exposes -/// them per request. +/// `Statistic` remains a shape-derived follow-up. `AccuracyTarget` is +/// supplied by the caller through [`ACCURACY_HEADER`] and defaults to +/// approximate when the header is absent. /// /// Map `EngineRouterError` variants to HTTP statuses: /// * `NoEngineRegistered` → 503 (configuration bug — restart with the @@ -1404,6 +1434,7 @@ async fn process_via_router( parsed_request: &ParsedQueryRequest, start_time: Instant, metric_storage: StorageBackend, + accuracy: AccuracyTarget, ) -> Response { use crate::drivers::query::adapters::QueryExecutionResult; use crate::query_engines::EngineError; @@ -1414,20 +1445,16 @@ async fn process_via_router( parsed_request.query, metric_storage, ); - // Default `(Sum, Approximate)` — see fn doc above. The router's - // capability table only consults these axes for `DoubleWrite` - // metrics; for `GorillaObjectStore`-only deploys the dispatch - // is a function of `metric_storage` alone. + // Statistic is not currently used by the routing policy. Accuracy is + // the validated request contract threaded in by the HTTP handler. let stat = Statistic::Sum; - let accuracy = AccuracyTarget::Epsilon(0.01); - let router_result = state .query_router - .execute(&parsed_request.query, stat, accuracy, metric_storage) + .execute_routed(&parsed_request.query, stat, accuracy, metric_storage) .await; match router_result { - Ok(query_result) => { + Ok((query_result, data_source_id)) => { let query_duration = query_start_time.elapsed(); debug!( "EngineRouter dispatch took: {:.2}ms; result: {:?}", @@ -1459,9 +1486,7 @@ async fn process_via_router( .format_success_response(&execution_result) .await { - Ok(response) => { - annotate_data_source(response, metric_storage.data_source_id()).await - } + Ok(response) => annotate_data_source(response, data_source_id).await, Err(status) => status.into_response(), } } @@ -1565,6 +1590,10 @@ async fn handle_instant_query( // strips the param out (it doesn't, but reading the source of // truth keeps this robust to adapter changes). let engine_override = extract_engine_override(&headers, &query_params.0); + let accuracy = match extract_accuracy(&headers) { + Ok(accuracy) => accuracy, + Err(error) => return invalid_accuracy_response(error), + }; // Per-tenant `BackendStorageRouting`: read the tenant id from the // `X-ASAP-Tenant` header (default `"default"`). Captured before // `parse_get_request` consumes `query_params`. @@ -1598,6 +1627,7 @@ async fn handle_instant_query( start_time, forwarding_headers, engine_override, + accuracy, &tenant, ) .await; @@ -1641,6 +1671,10 @@ async fn handle_instant_query_post( debug!("=== INCOMING POST REQUEST ==="); let forwarding_headers = extract_fallback_headers(&headers); + let accuracy = match extract_accuracy(&headers) { + Ok(accuracy) => accuracy, + Err(error) => return invalid_accuracy_response(error), + }; // Check content type to determine how to parse the body let content_type = headers @@ -1761,6 +1795,7 @@ async fn handle_instant_query_post( start_time, forwarding_headers, engine_override, + accuracy, &tenant, ) .await; @@ -1907,6 +1942,7 @@ async fn process_range_query_request( parsed_request: &ParsedRangeQueryRequest, start_time: Instant, forwarding_headers: HashMap, + accuracy: AccuracyTarget, tenant: &str, ) -> Response { // Check if handling is enabled @@ -1952,14 +1988,9 @@ async fn process_range_query_request( let metric_storage = resolve_metric_storage(state, &parsed_request.query, tenant); - // Match the instant path's hardcoding of `(Sum, Approximate)`. - // TODO: derive `accuracy` (and `stat`) from the request rather than - // pinning Approximate — once the request carries an accuracy hint, - // an `Exact` range query will route straight to the archive via the - // shared policy table. + // Statistic is not currently used by the routing policy. Accuracy is + // the validated request contract threaded in by the HTTP handler. let stat = Statistic::Sum; - let accuracy = AccuracyTarget::Epsilon(0.01); - // Warm-vs-archive routing fix: split by the warm-retention boundary // rather than "archive-on ⇒ everything to archive". When the // requested `[start_ms, end_ms]` window lies entirely inside the @@ -2104,6 +2135,10 @@ async fn handle_range_query( debug!("=== INCOMING RANGE QUERY GET REQUEST ==="); debug!("Raw query params: {:?}", query_params.0); + let accuracy = match extract_accuracy(&headers) { + Ok(accuracy) => accuracy, + Err(error) => return invalid_accuracy_response(error), + }; let parsed_request = match state.adapter.parse_range_get_request(query_params).await { Ok(req) => { debug!( @@ -2131,6 +2166,7 @@ async fn handle_range_query( &parsed_request, start_time, extract_fallback_headers(&headers), + accuracy, &tenant, ) .await; @@ -2146,6 +2182,10 @@ async fn handle_range_query_post( let _timer = srv_metrics::start_query_timer(srv_metrics::QUERY_TYPE_RANGE); let start_time = Instant::now(); debug!("=== INCOMING RANGE QUERY POST REQUEST ==="); + let accuracy = match extract_accuracy(&headers) { + Ok(accuracy) => accuracy, + Err(error) => return invalid_accuracy_response(error), + }; // Parse the body as form data let body_str = match String::from_utf8(body.to_vec()) { @@ -2204,6 +2244,7 @@ async fn handle_range_query_post( &parsed_request, start_time, extract_fallback_headers(&headers), + accuracy, &tenant, ) .await; @@ -3747,6 +3788,23 @@ aggregations: fn capabilities(&self) -> EngineCapabilities { self.caps } + + async fn execute_range( + &self, + _query: &str, + _start_ms: u64, + _end_ms: u64, + _step_ms: u64, + ) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + match self.outcome { + MockOutcome::OkEmpty => Ok(QueryResult::matrix(Vec::new())), + MockOutcome::Backend => Err(EngineError::backend( + self.caps.data_source_id, + "simulated backend failure", + )), + } + } } /// Build an `HttpServer` whose router holds the supplied set of @@ -3867,6 +3925,90 @@ aggregations: ); } + #[test] + fn accuracy_header_parses_explicit_contract_and_rejects_typos() { + let mut headers = HeaderMap::new(); + assert!(matches!( + extract_accuracy(&headers), + Ok(AccuracyTarget::Epsilon(_)) + )); + headers.insert(ACCURACY_HEADER, "exact".parse().unwrap()); + assert!(matches!( + extract_accuracy(&headers), + Ok(AccuracyTarget::Exact) + )); + headers.insert(ACCURACY_HEADER, "approximate".parse().unwrap()); + assert!(matches!( + extract_accuracy(&headers), + Ok(AccuracyTarget::Epsilon(_)) + )); + headers.insert(ACCURACY_HEADER, "best-effort".parse().unwrap()); + assert!(extract_accuracy(&headers).is_err()); + } + + #[tokio::test] + async fn exact_accuracy_header_bypasses_warm_tier_for_archive() { + let (archive, archive_calls) = + MockQueryEngine::new(StorageBackend::GorillaObjectStore, MockOutcome::OkEmpty); + let server_port = setup_test_server_with_router( + StorageBackend::SketchStore, + vec![archive as Arc], + ) + .await; + let response = Client::new() + .get(format!("http://127.0.0.1:{server_port}/api/v1/query")) + .header(ACCURACY_HEADER, "exact") + .query(&[("query", "sum_over_time(foo[5m])"), ("time", "1700000000")]) + .send() + .await + .unwrap(); + assert!(response.status().is_success()); + let body: serde_json::Value = response.json().await.unwrap(); + assert_data_source(&body, StorageBackend::GorillaObjectStore.data_source_id()); + assert_eq!(archive_calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn exact_accuracy_header_routes_range_query_to_archive() { + let (archive, archive_calls) = + MockQueryEngine::new(StorageBackend::GorillaObjectStore, MockOutcome::OkEmpty); + let server_port = setup_test_server_with_router( + StorageBackend::SketchStore, + vec![archive as Arc], + ) + .await; + let response = Client::new() + .get(format!("http://127.0.0.1:{server_port}/api/v1/query_range")) + .header(ACCURACY_HEADER, "exact") + .query(&[ + ("query", "sum_over_time(foo[5m])"), + ("start", "1699999940"), + ("end", "1700000000"), + ("step", "10"), + ]) + .send() + .await + .unwrap(); + assert!(response.status().is_success()); + let body: serde_json::Value = response.json().await.unwrap(); + assert_data_source(&body, StorageBackend::GorillaObjectStore.data_source_id()); + assert_eq!(archive_calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn invalid_accuracy_header_returns_bad_request() { + let server_port = + setup_test_server_with_router(StorageBackend::SketchStore, Vec::new()).await; + let response = Client::new() + .get(format!("http://127.0.0.1:{server_port}/api/v1/query")) + .header(ACCURACY_HEADER, "best-effort") + .query(&[("query", "sum_over_time(foo[5m])")]) + .send() + .await + .unwrap(); + assert_eq!(response.status(), reqwest::StatusCode::BAD_REQUEST); + } + #[tokio::test] async fn http_routes_asap_tier_metric_to_simple_engine() { // Default (no hot-reload) → `SketchStore`. The handler diff --git a/data_plane/src/query_engines/routing/query_engine_routing.rs b/data_plane/src/query_engines/routing/query_engine_routing.rs index 96a21636..dd5a5047 100644 --- a/data_plane/src/query_engines/routing/query_engine_routing.rs +++ b/data_plane/src/query_engines/routing/query_engine_routing.rs @@ -231,6 +231,21 @@ impl EngineRouter { accuracy: AccuracyTarget, metric_storage: StorageBackend, ) -> Result { + self.execute_routed(query, stat, accuracy, metric_storage) + .await + .map(|(result, _)| result) + } + + /// Instant dispatch that also returns the canonical id of the engine that + /// answered. HTTP callers use this to report truthful provenance after a + /// warm-to-archive routing decision. + pub async fn execute_routed( + &self, + query: &str, + stat: Statistic, + accuracy: AccuracyTarget, + metric_storage: StorageBackend, + ) -> Result<(QueryResult, &'static str), EngineRouterError> { let backends = compatible_storage_backends(stat, &accuracy, metric_storage); debug!( query = query, @@ -261,7 +276,7 @@ impl EngineRouter { backend = ?backend, "router: dispatch succeeded", ); - return Ok(result); + return Ok((result, id)); } Err(e) => { warn!(