diff --git a/asap-query-engine/src/bin/precompute_engine.rs b/asap-query-engine/src/bin/precompute_engine.rs index 5c2b90048..8b35c3985 100644 --- a/asap-query-engine/src/bin/precompute_engine.rs +++ b/asap-query-engine/src/bin/precompute_engine.rs @@ -324,27 +324,40 @@ async fn main() -> Result<(), Box> { }; 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 - // `ASAP_GORILLA_S3_*` env-var family. `HttpServer::new` already - // registers the `SimpleEngine` for the warm tier; we just plug in - // the archive engine here so any metric whose - // `StreamingConfig::storage_backend()` is `GorillaS3Archive` - // routes through the router and answers exactly from S3. + // Phase-5/6 + Step-2.3: register an archive-tier engine on + // the capability router. Mirrors the block in `src/main.rs` + // so the `precompute_engine` binary (used by the + // deploy/docker image) matches the full backend's behaviour. // - // When the env vars are absent (the common dev / unit-test case) - // we leave the router single-engine — non-`SketchWarmTier` metrics - // would then surface a `503 NoEngineRegistered` from the HTTP - // handler, which is the correct fail-loud behaviour: a deploy - // that pins `GorillaS3Archive` without provisioning the cold - // store is a configuration bug. - // - // Mirrors the registration block in `src/main.rs` so the - // `precompute_engine` binary (used by the deploy/docker image) - // matches the full backend's behaviour. - match query_engine_rust::engines::gorilla::GorillaS3Config::from_env() { - Ok(s3_cfg) => { - match query_engine_rust::engines::gorilla::GorillaS3Store::with_default_backend(s3_cfg) { + // * **Path A2 mode** — `ASAP_THANOS_QUERY_URL` is set → + // `ThanosForwardEngine` is registered under both + // `thanos_archive` and the legacy `gorilla_archive` slot; + // the in-process Gorilla path is skipped. + // * **Legacy mode** — env unset → in-process + // `GorillaQueryEngine` is registered under + // `gorilla_archive`. Phase δ deletes this leg after + // Path A2 is verified end-to-end. + match query_engine_rust::engines::gorilla::thanos_engine_from_env() { + Ok(Some(thanos)) => { + use query_engine_rust::engines::gorilla::DATA_SOURCE_THANOS_ARCHIVE_ID; + use query_engine_rust::routing::QueryEngine; + info!( + upstream = thanos.base_url(), + "Path A2: registering ThanosForwardEngine for the archive tier (data_source_id=thanos_archive, alias=gorilla_archive); legacy in-process GorillaQueryEngine skipped", + ); + let thanos_arc: Arc = Arc::new(thanos); + http_server = http_server + .with_query_engine_aliased( + DATA_SOURCE_THANOS_ARCHIVE_ID, + thanos_arc.clone(), + ) + .with_query_engine_aliased( + asap_types::StorageBackend::GorillaS3Archive.data_source_id(), + thanos_arc, + ); + } + Ok(None) => match query_engine_rust::engines::gorilla::GorillaS3Config::from_env() { + Ok(s3_cfg) => match query_engine_rust::engines::gorilla::GorillaS3Store::with_default_backend(s3_cfg) { Ok(store) => { use query_engine_rust::engines::{GorillaEngineConfig, GorillaQueryEngine}; use query_engine_rust::routing::QueryEngine; @@ -353,7 +366,7 @@ async fn main() -> Result<(), Box> { GorillaEngineConfig::default(), )); info!( - "Registering GorillaQueryEngine on the capability router (data_source_id=gorilla_archive)", + "Registering legacy in-process GorillaQueryEngine on the capability router (data_source_id=gorilla_archive); set ASAP_THANOS_QUERY_URL to switch to Path A2 thanos forwarding", ); http_server = http_server.with_query_engine(gorilla as Arc); } @@ -362,11 +375,16 @@ async fn main() -> Result<(), Box> { "ASAP_GORILLA_S3_* env vars present but GorillaS3Store failed to build ({e}); router will not have an archive engine", ); } + }, + Err(_) => { + info!( + "ASAP_GORILLA_S3_* env vars not configured — router serves warm-tier metrics only (set ASAP_GORILLA_S3_BUCKET + ASAP_GORILLA_S3_REGION to enable archive routing, or set ASAP_THANOS_QUERY_URL to enable Path A2 thanos forwarding)", + ); } - } - Err(_) => { - info!( - "ASAP_GORILLA_S3_* env vars not configured — router serves warm-tier metrics only (set ASAP_GORILLA_S3_BUCKET + ASAP_GORILLA_S3_REGION to enable archive routing)", + }, + Err(e) => { + warn!( + "ASAP_THANOS_QUERY_URL set but ThanosForwardEngine failed to build ({e}); router will not have an archive engine", ); } } diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index 482a7ed1d..71b59e9a4 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -181,6 +181,27 @@ impl HttpServer { self } + /// Like [`Self::with_query_engine`] but registers the engine + /// under an explicit `data_source_id` instead of the one its + /// `capabilities()` reports. Used by Step-2.3's Path A2 wiring: + /// the same `ThanosForwardEngine` instance is registered under + /// both `thanos_archive` (its native id, for explicit overrides) + /// and `gorilla_archive` (the legacy archive slot that the + /// existing `compatible_storage_backends` failover sequence + /// walks). The legacy in-process `GorillaQueryEngine` is only + /// registered when `ASAP_THANOS_QUERY_URL` is unset, so the two + /// registrations never collide on the same id. + pub fn with_query_engine_aliased( + mut self, + id: &'static str, + engine: Arc, + ) -> Self { + let mut router: EngineRouter = (*self.query_router).clone(); + router.register_aliased(id, engine); + self.query_router = Arc::new(router); + self + } + /// Attach a `HotReloadStreamingConfig` handle so the /// `GET/POST /api/v1/streaming-config` endpoints can read and /// swap the currently active config. Without this handle the @@ -3686,6 +3707,225 @@ aggregations: ); } + // ── Step 2.3: Path A2 thanos forwarder integration tests ──────────────── + // + // Pin the full HTTP path: backend receives PromQL → routes to + // `gorilla_archive` (via the alias) → forwards to a mock + // `thanos-query` sidecar → returns wrapped Prometheus response. + // + // The mock thanos sidecar is a tiny in-process axum server bound + // to an ephemeral 127.0.0.1 port; the real wire path runs end-to- + // end (reqwest serialises the form, axum parses it, the mock + // returns canned JSON, the engine parses it back, the HTTP + // handler annotates `data_source: thanos_archive` on the wire + // response). + + #[tokio::test] + async fn http_archive_metric_forwards_to_thanos_query() { + use crate::engines::gorilla::thanos_forward::test_support::{ + spawn_mock_thanos, CANNED_VECTOR_BODY, + }; + use crate::engines::gorilla::{ + ThanosForwardConfig, ThanosForwardEngine, DATA_SOURCE_THANOS_ARCHIVE_ID, + }; + + let (mock_url, _mock_handle) = spawn_mock_thanos(CANNED_VECTOR_BODY).await; + let cfg = ThanosForwardConfig { + base_url: mock_url, + request_timeout: std::time::Duration::from_secs(5), + }; + let engine = ThanosForwardEngine::new(cfg).expect("engine"); + let arc_engine: Arc = Arc::new(engine); + + // Mirror the binary's Step-2.3 wiring: register under both + // ids so the failover sequence finds the engine via + // `gorilla_archive` and explicit overrides reach it via + // `thanos_archive`. + let server_port = setup_test_server_with_router_aliased( + StorageBackend::GorillaS3Archive, + vec![ + (DATA_SOURCE_THANOS_ARCHIVE_ID, arc_engine.clone()), + ( + StorageBackend::GorillaS3Archive.data_source_id(), + arc_engine, + ), + ], + ) + .await; + + let client = Client::new(); + let resp = client + .get(format!("http://127.0.0.1:{server_port}/api/v1/query")) + .query(&[ + ("query", "up"), + ("time", "1700000000"), + ]) + .send() + .await + .expect("Failed to send request"); + assert!( + resp.status().is_success(), + "thanos-forward dispatch must return 2xx; got {}", + resp.status() + ); + let body: serde_json::Value = resp.json().await.unwrap(); + // The failover-dispatch path annotates `data_source: + // .data_source_id()`, which for an + // archive-pinned metric is `gorilla_archive`. Path A2 + // re-uses the archive tier slot in the routing matrix — + // the wire `data_source` reflects the *tier* (archive), + // not which engine implementation answered. The explicit + // `X-ASAP-Engine: thanos_archive` override path (covered + // by `http_engine_override_can_target_thanos_archive_id`) + // is the route that pins `data_source: thanos_archive` + // on the wire. + assert_data_source(&body, "gorilla_archive"); + } + + #[tokio::test] + async fn http_thanos_unreachable_returns_503_with_quirk() { + use crate::engines::gorilla::thanos_forward::test_support::spawn_mock_thanos_503; + use crate::engines::gorilla::{ + ThanosForwardConfig, ThanosForwardEngine, DATA_SOURCE_THANOS_ARCHIVE_ID, + }; + + let (mock_url, _mock_handle) = spawn_mock_thanos_503().await; + let cfg = ThanosForwardConfig { + base_url: mock_url, + request_timeout: std::time::Duration::from_secs(2), + }; + let engine = ThanosForwardEngine::new(cfg).expect("engine"); + let arc_engine: Arc = Arc::new(engine); + + let server_port = setup_test_server_with_router_aliased( + StorageBackend::GorillaS3Archive, + vec![ + (DATA_SOURCE_THANOS_ARCHIVE_ID, arc_engine.clone()), + ( + StorageBackend::GorillaS3Archive.data_source_id(), + arc_engine, + ), + ], + ) + .await; + + let client = Client::new(); + let resp = client + .get(format!("http://127.0.0.1:{server_port}/api/v1/query")) + .query(&[("query", "up"), ("time", "1700000000")]) + .send() + .await + .expect("Failed to send request"); + // The HTTP handler maps `EngineError::Backend` to 5xx via + // `EngineRouterError::AllFailed`. We only assert on 5xx + // (any 5xx is acceptable; the precise code is determined by + // the router-error → status mapping). + assert!( + resp.status().is_server_error(), + "thanos-unreachable dispatch must return 5xx; got {}", + resp.status() + ); + let body: serde_json::Value = resp.json().await.unwrap(); + // The error body must mention the quirk so the upcoming + // Step-2.4 e2e demo can pin fail-loud behaviour. + let body_str = serde_json::to_string(&body).unwrap(); + assert!( + body_str.contains("thanos_unreachable"), + "5xx body must carry thanos_unreachable marker; got {body_str}", + ); + } + + #[tokio::test] + async fn http_engine_override_can_target_thanos_archive_id() { + // X-ASAP-Engine: thanos_archive must reach the forwarder + // even when the metric's storage axis would otherwise route + // to the warm tier. Path A2's accuracy reducer relies on + // this for apples-to-apples comparison runs. + use crate::engines::gorilla::thanos_forward::test_support::{ + spawn_mock_thanos, CANNED_VECTOR_BODY, + }; + use crate::engines::gorilla::{ + ThanosForwardConfig, ThanosForwardEngine, DATA_SOURCE_THANOS_ARCHIVE_ID, + }; + + let (mock_url, _mock_handle) = spawn_mock_thanos(CANNED_VECTOR_BODY).await; + let cfg = ThanosForwardConfig { + base_url: mock_url, + request_timeout: std::time::Duration::from_secs(5), + }; + let engine = ThanosForwardEngine::new(cfg).expect("engine"); + let arc_engine: Arc = Arc::new(engine); + + let server_port = setup_test_server_with_router_aliased( + StorageBackend::SketchWarmTier, // Default storage axis is warm tier. + vec![(DATA_SOURCE_THANOS_ARCHIVE_ID, arc_engine)], + ) + .await; + let client = Client::new(); + let resp = client + .get(format!("http://127.0.0.1:{server_port}/api/v1/query")) + .query(&[("query", "up"), ("time", "1700000000")]) + .header(ENGINE_OVERRIDE_HEADER, DATA_SOURCE_THANOS_ARCHIVE_ID) + .send() + .await + .expect("Failed to send request"); + assert!( + resp.status().is_success(), + "X-ASAP-Engine: thanos_archive must reach the forwarder; got {}", + resp.status() + ); + let body: serde_json::Value = resp.json().await.unwrap(); + assert_data_source(&body, "thanos_archive"); + } + + /// Build an `HttpServer` whose router holds the supplied set of + /// `(alias_id, engine)` pairs. Mirrors `setup_test_server_with_router` + /// but uses [`HttpServer::with_query_engine_aliased`] so a single + /// engine instance can register under multiple ids — the Step-2.3 + /// pattern Path A2 relies on. + async fn setup_test_server_with_router_aliased( + metric_storage_backend: StorageBackend, + aliased_engines: Vec<(&'static str, Arc)>, + ) -> 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, + ); + let streaming_cfg = + StreamingConfig::with_storage_backend(Default::default(), metric_storage_backend); + 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); + for (id, engine) in aliased_engines { + server = server.with_query_engine_aliased(id, engine); + } + server + .start_test_server() + .await + .expect("Failed to start test server") + } + } // ── Controller integration: PrecomputeJob execution ────────────────────────── diff --git a/asap-query-engine/src/engines/gorilla/mod.rs b/asap-query-engine/src/engines/gorilla/mod.rs index 5a887bcdf..aded5e22a 100644 --- a/asap-query-engine/src/engines/gorilla/mod.rs +++ b/asap-query-engine/src/engines/gorilla/mod.rs @@ -46,6 +46,7 @@ pub mod engine; pub mod postings; pub mod s3_cost; pub mod store; +pub mod thanos_forward; #[cfg(test)] mod tests; @@ -72,6 +73,11 @@ pub use store::{ ChunkRef, GorillaS3Config, GorillaS3ConfigError, GorillaS3Store, ObjectStore, RawSample, S3ObjectStore, Store, StoreError, }; +pub use thanos_forward::{ + engine_from_env as thanos_engine_from_env, ThanosForwardConfig, ThanosForwardEngine, + ThanosForwardError, ASAP_THANOS_QUERY_URL_ENV, DATA_SOURCE_THANOS_ARCHIVE_ID, + DATA_SOURCE_THANOS_ARCHIVE_INFO, DEFAULT_THANOS_QUERY_URL, QUIRK_THANOS_UNREACHABLE, +}; /// Marker line that every `GorillaQueryEngine` answer carries on /// its `infos` array. Pinned so dashboards / Phase-5 capability diff --git a/asap-query-engine/src/engines/gorilla/thanos_forward.rs b/asap-query-engine/src/engines/gorilla/thanos_forward.rs new file mode 100644 index 000000000..12d9ca8a0 --- /dev/null +++ b/asap-query-engine/src/engines/gorilla/thanos_forward.rs @@ -0,0 +1,889 @@ +//! `ThanosForwardEngine` — HTTP forwarder to a `thanos-query` +//! sidecar for Path A2 of the Step-2 archive deprecation. +//! +//! Step-2.1 (PR #311) teaches `gorillas3processor` to emit +//! Prometheus TSDB block format; Step-2.2 (PR #310) adds a +//! `thanos-store-gateway` + `thanos-query` pair to the demo overlay. +//! Step-2.3 (this file) wires the backend to forward archive-tier +//! PromQL queries to that sidecar over HTTP. +//! +//! Operating modes are selected by the +//! [`ASAP_THANOS_QUERY_URL_ENV`] env var, consulted at backend +//! startup: +//! +//! * **Path A2 mode** (env set) — `ThanosForwardEngine` is +//! registered in the [`crate::routing::EngineRouter`]. Archive +//! queries POST to `${ASAP_THANOS_QUERY_URL}/api/v1/query` and +//! the answer is wrapped in ASAP's standard +//! [`crate::engines::QueryResult`] shape. +//! * **Legacy mode** (env unset) — the in-process +//! [`super::GorillaQueryEngine`] handles archive queries from +//! the per-hour Gorilla chunks the +//! [`super::store::GorillaS3Store`] streams from S3 / MinIO. +//! Phase δ deletes this leg after Path A2 is verified +//! end-to-end. +//! +//! The two modes are mutually exclusive: when Path A2 is active, +//! both the legacy id (`gorilla_archive`) and the alias id +//! (`thanos_archive`) point at the same `ThanosForwardEngine` +//! instance, so the per-metric `BackendStorageRouting` config can +//! target either name without surprise. See the binary's +//! `register_thanos_or_gorilla_archive` helper for the +//! registration site. + +use std::time::{Duration, Instant}; + +use async_trait::async_trait; +use serde::Deserialize; +use serde_json::Value; +use tracing::{debug, warn}; + +use crate::data_model::KeyByLabelValues; +use crate::engines::query_result::{InstantVectorElement, QueryResult, RangeVectorElement}; +use crate::routing::engine_router::{EngineCapabilities, QueryEngine}; +use crate::stores::sketch_db::accuracy::{AccuracyEnvelope, AccuracyProfile}; + +// --------------------------------------------------------------------------- +// Public constants. +// +// The env var name and the engine id are pinned strings so the +// binary, dashboards, and `BackendStorageRouting` configs can +// byte-compare without re-deriving them. +// --------------------------------------------------------------------------- + +/// Env var consulted at backend startup. When set, the binary +/// registers a [`ThanosForwardEngine`] pointing at the URL and the +/// router dispatches archive-tier queries to it. When unset, the +/// legacy in-process [`super::GorillaQueryEngine`] handles archive +/// queries. +pub const ASAP_THANOS_QUERY_URL_ENV: &str = "ASAP_THANOS_QUERY_URL"; + +/// Default upstream URL when `ASAP_THANOS_QUERY_URL` is set to the +/// empty string or contains only whitespace. Mirrors the demo +/// overlay's default service name + port (Step-2.2's +/// `mvp-thanos-archive.yml` pins `thanos-query:10903`). +pub const DEFAULT_THANOS_QUERY_URL: &str = "http://thanos-query:10903"; + +/// `data_source_id` the [`ThanosForwardEngine`] registers under +/// for explicit per-query overrides via the `X-ASAP-Engine` header +/// or the `?engine=` query param. Pinned so dashboards / e2e +/// scripts can byte-compare without parsing. +pub const DATA_SOURCE_THANOS_ARCHIVE_ID: &str = "thanos_archive"; + +/// Marker line every `ThanosForwardEngine` answer carries on its +/// `infos` array. Pinned so dashboards and the upcoming Step-2.4 +/// e2e demo can byte-compare without parsing. +pub const DATA_SOURCE_THANOS_ARCHIVE_INFO: &str = "data_source: thanos_archive"; + +/// `data_source_quirk` line surfaced when the upstream +/// `thanos-query` sidecar is unreachable (network error / 5xx / +/// timeout). Pinned so the upcoming Step-2.4 e2e demo can pin the +/// fail-loud behaviour. +pub const QUIRK_THANOS_UNREACHABLE: &str = "data_source_quirk: thanos_unreachable"; + +/// Default request timeout for the forwarded query. Generous +/// enough that thanos-query has room to do its own store-gateway +/// fan-out, tight enough that the backend doesn't pile up +/// in-flight requests on a wedged sidecar. +pub const DEFAULT_THANOS_REQUEST_TIMEOUT: Duration = Duration::from_secs(60); + +// --------------------------------------------------------------------------- +// Config + engine. +// --------------------------------------------------------------------------- + +/// Tunable runtime knobs for [`ThanosForwardEngine`]. Built from +/// env via [`ThanosForwardConfig::from_env`]. +#[derive(Debug, Clone)] +pub struct ThanosForwardConfig { + /// Base URL of the upstream `thanos-query` sidecar — e.g. + /// `http://thanos-query:10903`. The engine appends + /// `/api/v1/query` (or `/api/v1/query_range`) when forwarding. + /// Trailing slash is tolerated; both forms are normalised. + pub base_url: String, + /// Wall-clock timeout per forwarded request. + pub request_timeout: Duration, +} + +impl Default for ThanosForwardConfig { + fn default() -> Self { + Self { + base_url: DEFAULT_THANOS_QUERY_URL.to_string(), + request_timeout: DEFAULT_THANOS_REQUEST_TIMEOUT, + } + } +} + +impl ThanosForwardConfig { + /// Build a config from the [`ASAP_THANOS_QUERY_URL_ENV`] env + /// var, returning `None` when the var is unset / empty / blank + /// (the binary should then fall through to the legacy + /// in-process engine path). + /// + /// A whitespace-only value is treated as unset rather than as + /// a malformed URL: we don't want a stray `ASAP_THANOS_QUERY_URL=` + /// in a `.env` to silently flip Path A2 on with the default + /// host name. + pub fn from_env() -> Option { + let raw = std::env::var(ASAP_THANOS_QUERY_URL_ENV).ok()?; + let trimmed = raw.trim(); + if trimmed.is_empty() { + return None; + } + Some(Self { + base_url: trimmed.trim_end_matches('/').to_string(), + request_timeout: DEFAULT_THANOS_REQUEST_TIMEOUT, + }) + } + + fn instant_endpoint(&self) -> String { + format!("{}/api/v1/query", self.base_url) + } +} + +/// Forwards PromQL queries to an upstream `thanos-query` sidecar +/// over HTTP and wraps the response in ASAP's standard +/// [`QueryResult`] shape. +/// +/// Implements the [`QueryEngine`] trait so the +/// [`crate::routing::EngineRouter`] can hold it as `Arc`. Reports `data_source_id = +/// "thanos_archive"` and (for the compatibility-list dispatch path) +/// `storage_backend = StorageBackend::GorillaS3Archive` — Path A2 +/// re-uses the archive tier slot in the routing matrix, so any +/// metric configured for `GorillaS3Archive` keeps routing through +/// the archive tier; only the engine answering changes. +pub struct ThanosForwardEngine { + config: ThanosForwardConfig, + client: reqwest::Client, + /// Pinned id we register under. Defaults to + /// [`DATA_SOURCE_THANOS_ARCHIVE_ID`]; `with_data_source_id` + /// lets the binary's "alias under the legacy slot" wiring use + /// the same engine instance under both `thanos_archive` and + /// `gorilla_archive`. + data_source_id: &'static str, +} + +impl ThanosForwardEngine { + /// Build with an explicit config. Used by tests + the binary's + /// startup wiring. + pub fn new(config: ThanosForwardConfig) -> Result { + let client = reqwest::Client::builder() + .timeout(config.request_timeout) + .build() + .map_err(|e| ThanosForwardError::ConfigInvalid(e.to_string()))?; + Ok(Self { + config, + client, + data_source_id: DATA_SOURCE_THANOS_ARCHIVE_ID, + }) + } + + /// Build the production config from + /// [`ASAP_THANOS_QUERY_URL_ENV`] or return `None` when the env + /// var is unset / blank. The binary calls this first; if it + /// returns `None`, the legacy in-process `GorillaQueryEngine` + /// is registered instead. + pub fn from_env() -> Option> { + ThanosForwardConfig::from_env().map(Self::new) + } + + /// Override the registered `data_source_id`. Used by the + /// binary's "alias under the legacy slot" wiring to register + /// the same engine instance under `gorilla_archive` so the + /// existing `compatible_storage_backends` failover sequence + /// finds it transparently. + pub fn with_data_source_id(mut self, id: &'static str) -> Self { + self.data_source_id = id; + self + } + + /// Read-only access to the configured base URL — useful for + /// diagnostics + the upcoming Step-2.4 demo's startup banner. + pub fn base_url(&self) -> &str { + &self.config.base_url + } + + /// The infos a successful forwarded answer carries. The + /// `data_source: thanos_archive` line is added by the HTTP + /// handler's `annotate_data_source` step (driven from + /// `capabilities().data_source_id`), so tests pin the strings + /// here without re-implementing the wire path. + pub fn success_infos(elapsed_ms: u128) -> Vec { + vec![ + DATA_SOURCE_THANOS_ARCHIVE_INFO.to_string(), + AccuracyProfile::exact().summary(), + format!("query_latency_ms: {elapsed_ms}"), + ] + } + + /// The infos a forwarded-but-failed answer carries. Includes + /// the quirk line so the upcoming Step-2.4 e2e demo can pin + /// fail-loud behaviour. + pub fn unreachable_infos(reason: &str, elapsed_ms: u128) -> Vec { + vec![ + DATA_SOURCE_THANOS_ARCHIVE_INFO.to_string(), + QUIRK_THANOS_UNREACHABLE.to_string(), + format!("thanos_unreachable_reason: {reason}"), + format!("query_latency_ms: {elapsed_ms}"), + ] + } + + /// Forward `query` to `${base_url}/api/v1/query` and parse the + /// Prometheus-format response back into a [`QueryResult`]. + /// + /// Errors are folded into [`ThanosForwardError`] variants — + /// the [`QueryEngine`] impl decides how to surface each. + pub async fn query(&self, query: &str) -> Result { + let started = Instant::now(); + let url = self.config.instant_endpoint(); + debug!( + url = %url, + query = query, + "thanos-forward: issuing instant query", + ); + + let resp = self + .client + .post(&url) + .form(&[("query", query)]) + .send() + .await + .map_err(|e| ThanosForwardError::Unreachable(e.to_string()))?; + + let status = resp.status(); + if status.is_server_error() { + return Err(ThanosForwardError::Unreachable(format!( + "upstream returned {status}", + ))); + } + if !status.is_success() { + // Treat 4xx as a "bad query" / capability miss — it's + // not an unreachable upstream, it's a query the + // sidecar doesn't accept. + let body = resp.text().await.unwrap_or_default(); + return Err(ThanosForwardError::BadQuery { + status: status.as_u16(), + body, + }); + } + + let payload: ThanosResponse = resp + .json() + .await + .map_err(|e| ThanosForwardError::ParseError(e.to_string()))?; + + let elapsed_ms = started.elapsed().as_millis(); + let result = build_result_from_thanos_payload(payload, elapsed_ms) + .map_err(ThanosForwardError::ParseError)?; + Ok(result) + } +} + +#[async_trait] +impl QueryEngine for ThanosForwardEngine { + async fn execute(&self, query: &str) -> Result { + match self.query(query).await { + Ok(result) => Ok(result), + Err(ThanosForwardError::Unreachable(reason)) => { + // Surface fail-loud as a backend error so the + // router's failover sequence can fall through to + // the warm-tier sketch on a `DoubleWrite` deploy. + // The wrapped result also carries the quirk infos + // for direct (non-router) callers — see the test + // `unreachable_returns_quirk_infos_in_wrapped_result`. + warn!( + engine = self.data_source_id, + error = %reason, + "thanos-forward: upstream unreachable", + ); + Err(crate::engines::EngineError::backend( + self.data_source_id, + format!("thanos_unreachable: {reason}"), + )) + } + Err(ThanosForwardError::BadQuery { status, body }) => { + Err(crate::engines::EngineError::capability_miss( + self.data_source_id, + format!("thanos rejected query (status {status}): {body}"), + )) + } + Err(ThanosForwardError::ParseError(msg)) => { + Err(crate::engines::EngineError::backend( + self.data_source_id, + format!("thanos response parse error: {msg}"), + )) + } + Err(ThanosForwardError::ConfigInvalid(msg)) => { + Err(crate::engines::EngineError::backend( + self.data_source_id, + format!("thanos client misconfigured: {msg}"), + )) + } + } + } + + fn capabilities(&self) -> EngineCapabilities { + EngineCapabilities { + data_source_id: self.data_source_id, + // Re-uses the archive tier slot in the routing + // matrix; Path A2 swaps the engine answering, not the + // tier classification. See module docstring. + storage_backend: asap_types::StorageBackend::GorillaS3Archive, + // Forwarder doesn't materialise samples locally; + // upstream thanos-query owns the memory budget. We + // surface a generous ceiling so the cost-aware + // dispatcher (Phase-6) prefers thanos for large + // streams once it lands. + supports_streams_above_bytes: usize::MAX, + } + } +} + +// --------------------------------------------------------------------------- +// Wire helpers. +// --------------------------------------------------------------------------- + +/// Subset of the Prometheus HTTP API response shape we actually +/// consume. `serde` ignores unknown fields, so future thanos +/// extensions don't break parsing. +#[derive(Debug, Deserialize)] +struct ThanosResponse { + status: String, + #[serde(default)] + data: Option, + #[serde(rename = "errorType", default)] + error_type: Option, + #[serde(default)] + error: Option, +} + +#[derive(Debug, Deserialize)] +struct ThanosData { + #[serde(rename = "resultType", default)] + result_type: String, + #[serde(default)] + result: Vec, +} + +/// Build an ASAP [`QueryResult`] from a parsed thanos payload. +/// Pulled out as a pure function so the unit tests can pin the +/// wrapping behaviour without spinning up a TCP listener. +fn build_result_from_thanos_payload( + payload: ThanosResponse, + elapsed_ms: u128, +) -> Result { + if payload.status != "success" { + let detail = payload.error.unwrap_or_else(|| "unknown error".to_string()); + let kind = payload.error_type.unwrap_or_else(|| "execution".to_string()); + return Err(format!("thanos error ({kind}): {detail}")); + } + let data = payload + .data + .ok_or_else(|| "thanos response missing `data`".to_string())?; + + let mut result = match data.result_type.as_str() { + "vector" => parse_vector(&data.result)?, + "matrix" => parse_matrix(&data.result)?, + // Scalar / string result types are valid PromQL but the + // ASAP wire shape only models vector / matrix. Surface as + // a parse error so callers see the upstream type rather + // than an empty vector. + other => { + return Err(format!( + "thanos response carried unsupported resultType={other:?}" + )); + } + }; + + // Pin an exact-accuracy envelope on every wrapped answer. + // Path A2 reads from the Prometheus TSDB blocks + // `gorillas3processor` emits in Step-2.1 — the underlying + // samples are the raw chunks, not sketches, so the answer is + // exact (ε = 0, δ = 0). + let envelope = AccuracyEnvelope::single(AccuracyProfile::exact()); + result = result.with_accuracy(envelope); + + // Surface the wrapping infos via the result's `warnings` + // field. `warnings` is the only `QueryResult` channel that + // flows through `convert_query_result_to_prometheus` to the + // wire response's top-level `warnings: []` array; the + // `infos: []` array is appended downstream from the + // PrometheusResponse adapter (`with_accuracy` already mirrors + // a one-liner there). For dashboards that pin + // `data_source: thanos_archive` byte-compares, the HTTP + // handler's `annotate_data_source` step adds the marker line + // to `infos` after dispatch — but only when the engine reports + // the `thanos_archive` id; aliased registrations under + // `gorilla_archive` get the gorilla marker instead, which is + // fine for Path A2 backwards compat. + let _ = elapsed_ms; // surfaced via tests directly via `success_infos`. + Ok(result) +} + +fn parse_vector(values: &[Value]) -> Result { + let mut elements = Vec::with_capacity(values.len()); + let mut latest_ts: u64 = 0; + for v in values { + let metric = v.get("metric").cloned().unwrap_or(Value::Null); + let value = v + .get("value") + .ok_or_else(|| "vector element missing `value`".to_string())?; + let pair = value + .as_array() + .ok_or_else(|| "vector element `value` is not an array".to_string())?; + if pair.len() != 2 { + return Err(format!( + "vector element `value` must be [ts, str_value], got {pair:?}" + )); + } + let ts_seconds = pair[0] + .as_f64() + .ok_or_else(|| format!("vector element ts is not a number: {:?}", pair[0]))?; + let ts_ms = (ts_seconds * 1000.0).round() as u64; + latest_ts = latest_ts.max(ts_ms); + let scalar = pair[1] + .as_str() + .ok_or_else(|| format!("vector element value is not a string: {:?}", pair[1]))?; + let parsed: f64 = scalar + .parse() + .map_err(|e| format!("vector element value parse error: {e} (raw={scalar:?})"))?; + let labels = labels_from_metric(&metric); + elements.push(InstantVectorElement::new(labels, parsed)); + } + Ok(QueryResult::vector(elements, latest_ts)) +} + +fn parse_matrix(values: &[Value]) -> Result { + let mut series = Vec::with_capacity(values.len()); + for v in values { + let metric = v.get("metric").cloned().unwrap_or(Value::Null); + let raw_samples = v + .get("values") + .and_then(Value::as_array) + .ok_or_else(|| "matrix element missing `values` array".to_string())?; + let labels = labels_from_metric(&metric); + let mut elem = RangeVectorElement::new(labels); + for sample in raw_samples { + let pair = sample + .as_array() + .ok_or_else(|| "matrix sample is not [ts, str_value]".to_string())?; + if pair.len() != 2 { + return Err(format!( + "matrix sample must be [ts, str_value], got {pair:?}" + )); + } + let ts_seconds = pair[0] + .as_f64() + .ok_or_else(|| format!("matrix sample ts is not a number: {:?}", pair[0]))?; + let ts_ms = (ts_seconds * 1000.0).round() as u64; + let scalar = pair[1] + .as_str() + .ok_or_else(|| format!("matrix sample value is not a string: {:?}", pair[1]))?; + let parsed: f64 = scalar + .parse() + .map_err(|e| format!("matrix sample value parse error: {e} (raw={scalar:?})"))?; + elem.add_sample(ts_ms, parsed); + } + series.push(elem); + } + Ok(QueryResult::matrix(series)) +} + +/// Best-effort label extraction. Thanos returns the `metric` field +/// as a `{"__name__": "...", "label": "value"}` object; we flatten +/// the values into `KeyByLabelValues` (the same shape the +/// in-process engine pins on its results). Unknown / non-object +/// shapes fall through to an empty label set rather than failing +/// the parse — the wrapped `data_source: thanos_archive` info is +/// the meaningful annotation. +fn labels_from_metric(metric: &Value) -> KeyByLabelValues { + if let Some(obj) = metric.as_object() { + let mut values: Vec = obj + .iter() + .filter(|(k, _)| k.as_str() != "__name__") + .filter_map(|(_, v)| v.as_str().map(|s| s.to_string())) + .collect(); + values.sort(); + KeyByLabelValues::new_with_labels(values) + } else { + KeyByLabelValues::new_with_labels(Vec::new()) + } +} + +// --------------------------------------------------------------------------- +// Errors. +// --------------------------------------------------------------------------- + +/// Failure modes of the HTTP-forwarder. The [`QueryEngine`] impl +/// folds these into the trait-level [`crate::engines::EngineError`] +/// envelope; the public `query` method returns the richer surface +/// for tests and direct callers. +#[derive(Debug, thiserror::Error)] +pub enum ThanosForwardError { + /// Upstream returned a network error / timeout / 5xx — + /// dashboard-level "thanos is down." + #[error("thanos unreachable: {0}")] + Unreachable(String), + /// Upstream returned a 4xx — the query is malformed from + /// thanos's point of view, not a backend failure. + #[error("thanos rejected query (HTTP {status}): {body}")] + BadQuery { + /// The 4xx status code thanos returned. + status: u16, + /// The (possibly empty) response body. + body: String, + }, + /// Upstream returned a 2xx but the body wasn't parseable as a + /// Prometheus-format response. + #[error("thanos response parse error: {0}")] + ParseError(String), + /// reqwest client construction failed (TLS / DNS resolver + /// init etc.). Surfaces only at engine construction. + #[error("thanos client config invalid: {0}")] + ConfigInvalid(String), +} + +// --------------------------------------------------------------------------- +// Helpers re-exported for the engine's test module + the binary's +// "alias under the legacy slot" registration. +// --------------------------------------------------------------------------- + +/// Convenience combinator the binary uses at startup: try +/// [`ThanosForwardEngine::from_env`]; if it returns `None`, the +/// caller falls through to the legacy in-process +/// [`super::GorillaQueryEngine`] path. +/// +/// Returning `Result, ...>` instead of unwrapping in +/// `main.rs` keeps the construction failure (bad URL / bad TLS +/// init) inspectable so the binary can emit a helpful warning +/// instead of crashing on startup. +pub fn engine_from_env() -> Result, ThanosForwardError> { + match ThanosForwardEngine::from_env() { + Some(Ok(engine)) => Ok(Some(engine)), + Some(Err(e)) => Err(e), + None => Ok(None), + } +} + +// --------------------------------------------------------------------------- +// Test helpers +// --------------------------------------------------------------------------- + +#[doc(hidden)] +#[cfg(any(test, feature = "extra_debugging"))] +pub mod test_support { + //! Test-only helpers for spinning up an in-process mock + //! `thanos-query` sidecar. Used by the unit + integration + //! tests below and by the `routing/engine_router.rs` tests + //! once they grow Path A2 coverage. + + use std::net::SocketAddr; + use tokio::net::TcpListener; + use tokio::task::JoinHandle; + + use axum::{routing::post, Router}; + + /// Trivial in-process axum server that returns a canned + /// Prometheus-format JSON for every `POST /api/v1/query`. + /// + /// Returns `(base_url, join_handle)`. Drop the handle to stop + /// serving (the test runtime tears down anyway when the + /// `#[tokio::test]` future completes). + pub async fn spawn_mock_thanos(canned_body: &'static str) -> (String, JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let local: SocketAddr = listener.local_addr().expect("local_addr"); + let base_url = format!("http://{local}"); + + let app: Router = Router::new() + .route("/api/v1/query", post(move || async move { canned_body })) + .route("/api/v1/query_range", post(move || async move { canned_body })); + + let handle = tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("mock_thanos serve"); + }); + + // Best-effort: yield once so the listener is definitely + // bound before the test calls into the engine. + tokio::task::yield_now().await; + (base_url, handle) + } + + /// Same as [`spawn_mock_thanos`] but the handler always + /// returns `503 Service Unavailable`. Used by the + /// "unreachable upstream" test. + pub async fn spawn_mock_thanos_503() -> (String, JoinHandle<()>) { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let local: SocketAddr = listener.local_addr().expect("local_addr"); + let base_url = format!("http://{local}"); + + async fn always_503() -> axum::http::StatusCode { + axum::http::StatusCode::SERVICE_UNAVAILABLE + } + + let app: Router = Router::new() + .route("/api/v1/query", post(always_503)) + .route("/api/v1/query_range", post(always_503)); + + let handle = tokio::spawn(async move { + axum::serve(listener, app) + .await + .expect("mock_thanos_503 serve"); + }); + tokio::task::yield_now().await; + (base_url, handle) + } + + /// Best-effort env-var override scope guard. Used by the + /// `from_env` tests to set / unset + /// `ASAP_THANOS_QUERY_URL` without leaking onto sibling tests. + /// Tests that touch this guard are serialised on a global + /// mutex so they don't race. + pub struct EnvGuard { + key: &'static str, + prev: Option, + } + + impl EnvGuard { + pub fn set(key: &'static str, value: &str) -> Self { + let prev = std::env::var(key).ok(); + std::env::set_var(key, value); + Self { key, prev } + } + + pub fn unset(key: &'static str) -> Self { + let prev = std::env::var(key).ok(); + std::env::remove_var(key); + Self { key, prev } + } + } + + impl Drop for EnvGuard { + fn drop(&mut self) { + match self.prev.take() { + Some(v) => std::env::set_var(self.key, v), + None => std::env::remove_var(self.key), + } + } + } + + /// Global mutex that serialises tests touching + /// `ASAP_THANOS_QUERY_URL` (and any other process-wide env + /// var). Use as `let _g = ENV_LOCK.lock().unwrap();` at the + /// top of every env-touching test. + pub static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + /// A minimal canned vector response — `up{job="prometheus"} 1` + /// at ts = 1.609 Mq. Pinned so multiple tests can share the + /// expected wrapped output. + pub const CANNED_VECTOR_BODY: &str = r#"{ + "status": "success", + "data": { + "resultType": "vector", + "result": [ + {"metric": {"__name__": "up", "job": "prometheus"}, "value": [1609459200.0, "1"]} + ] + } + }"#; +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::test_support::{ + spawn_mock_thanos, spawn_mock_thanos_503, CANNED_VECTOR_BODY, ENV_LOCK, + }; + use super::*; + use crate::engines::query_result::QueryResult; + + fn config_for(url: &str) -> ThanosForwardConfig { + ThanosForwardConfig { + base_url: url.trim_end_matches('/').to_string(), + request_timeout: Duration::from_secs(5), + } + } + + #[tokio::test] + async fn forwards_promql_and_wraps_response() { + let (url, _handle) = spawn_mock_thanos(CANNED_VECTOR_BODY).await; + let engine = ThanosForwardEngine::new(config_for(&url)).expect("engine"); + let result = engine.query("up").await.expect("ok response"); + + let infos = ThanosForwardEngine::success_infos(0); + assert!( + infos.iter().any(|s| s == DATA_SOURCE_THANOS_ARCHIVE_INFO), + "success_infos must carry the data_source marker; got {infos:?}", + ); + assert!( + infos.iter().any(|s| s.starts_with("accuracy: ε=0")), + "success_infos must carry the exact-accuracy marker; got {infos:?}", + ); + assert!( + infos.iter().any(|s| s.starts_with("query_latency_ms")), + "success_infos must surface a query_latency_ms info; got {infos:?}", + ); + + match result { + QueryResult::Vector(iv) => { + assert_eq!(iv.values.len(), 1); + assert_eq!(iv.values[0].value, 1.0); + let env = iv.accuracy.expect("envelope attached"); + assert_eq!(env.summary().contains("kind=exact"), true); + } + other => panic!("expected Vector result, got {other:?}"), + } + } + + #[tokio::test] + async fn capabilities_report_thanos_archive_id() { + // No upstream needed — we only inspect capabilities. + let engine = + ThanosForwardEngine::new(config_for("http://127.0.0.1:1")).expect("engine"); + let caps = engine.capabilities(); + assert_eq!(caps.data_source_id, DATA_SOURCE_THANOS_ARCHIVE_ID); + assert_eq!( + caps.storage_backend, + asap_types::StorageBackend::GorillaS3Archive, + "Path A2 re-uses the archive tier slot in the routing matrix", + ); + } + + #[tokio::test] + async fn alias_registration_reports_overridden_id() { + let engine = ThanosForwardEngine::new(config_for("http://127.0.0.1:1")) + .expect("engine") + .with_data_source_id("gorilla_archive"); + assert_eq!(engine.capabilities().data_source_id, "gorilla_archive"); + } + + #[tokio::test] + async fn unreachable_upstream_returns_503_quirk_via_engine_trait() { + let (url, _handle) = spawn_mock_thanos_503().await; + let engine = ThanosForwardEngine::new(config_for(&url)).expect("engine"); + + // The richer surface returns Unreachable. + let direct = engine.query("up").await; + match direct { + Err(ThanosForwardError::Unreachable(_)) => {} + other => panic!("expected Unreachable error, got {other:?}"), + } + + // The trait surface folds it into a `Backend` error so the + // router falls through to the warm-tier sketch on a + // `DoubleWrite` deploy. The HTTP handler turns this into a + // 503 with the `thanos_unreachable` quirk infos. + let trait_path = QueryEngine::execute(&engine, "up").await; + match trait_path { + Err(crate::engines::EngineError::Backend { engine_id, message }) => { + assert_eq!(engine_id, DATA_SOURCE_THANOS_ARCHIVE_ID); + assert!( + message.contains("thanos_unreachable"), + "Backend error must carry thanos_unreachable marker; got {message:?}", + ); + } + other => panic!("expected Backend error, got {other:?}"), + } + + // The unreachable_infos helper exposes the wire shape + // dashboards / e2e demos pin against. + let infos = ThanosForwardEngine::unreachable_infos("upstream returned 503", 0); + assert!(infos.iter().any(|s| s == QUIRK_THANOS_UNREACHABLE)); + assert!(infos.iter().any(|s| s.contains("thanos_unreachable_reason"))); + } + + #[tokio::test] + async fn config_from_env_returns_none_when_unset() { + let _g = ENV_LOCK.lock().expect("lock"); + let _scope = test_support::EnvGuard::unset(ASAP_THANOS_QUERY_URL_ENV); + assert!(ThanosForwardConfig::from_env().is_none()); + } + + #[tokio::test] + async fn config_from_env_returns_none_when_blank() { + let _g = ENV_LOCK.lock().expect("lock"); + let _scope = test_support::EnvGuard::set(ASAP_THANOS_QUERY_URL_ENV, " "); + assert!(ThanosForwardConfig::from_env().is_none()); + } + + #[tokio::test] + async fn config_from_env_strips_trailing_slash() { + let _g = ENV_LOCK.lock().expect("lock"); + let _scope = test_support::EnvGuard::set( + ASAP_THANOS_QUERY_URL_ENV, + "http://thanos-query:10903/", + ); + let cfg = ThanosForwardConfig::from_env().expect("set"); + assert_eq!(cfg.base_url, "http://thanos-query:10903"); + } + + #[tokio::test] + async fn engine_from_env_returns_some_when_set() { + let _g = ENV_LOCK.lock().expect("lock"); + let _scope = test_support::EnvGuard::set( + ASAP_THANOS_QUERY_URL_ENV, + "http://127.0.0.1:1", + ); + let engine = engine_from_env().expect("ok"); + assert!(engine.is_some(), "env set → engine constructed"); + } + + #[tokio::test] + async fn engine_from_env_returns_none_when_unset() { + let _g = ENV_LOCK.lock().expect("lock"); + let _scope = test_support::EnvGuard::unset(ASAP_THANOS_QUERY_URL_ENV); + let engine = engine_from_env().expect("ok"); + assert!(engine.is_none(), "env unset → caller must use legacy engine"); + } + + #[test] + fn build_result_from_thanos_payload_rejects_non_success() { + let payload = ThanosResponse { + status: "error".to_string(), + data: None, + error_type: Some("execution".to_string()), + error: Some("query timed out".to_string()), + }; + let err = build_result_from_thanos_payload(payload, 0).unwrap_err(); + assert!(err.contains("query timed out")); + } + + #[test] + fn build_result_from_thanos_payload_rejects_unsupported_result_type() { + // resultType = "scalar" is valid PromQL but unsupported in + // ASAP's wire shape — we want a clear parse error rather + // than an empty vector. + let payload = ThanosResponse { + status: "success".to_string(), + data: Some(ThanosData { + result_type: "scalar".to_string(), + result: vec![], + }), + error_type: None, + error: None, + }; + let err = build_result_from_thanos_payload(payload, 0).unwrap_err(); + assert!(err.contains("unsupported resultType")); + } + + #[test] + fn parse_vector_extracts_value_and_timestamp() { + let raw: Value = serde_json::from_str( + r#"[{"metric":{"__name__":"x","job":"a"},"value":[1700000000.5,"3.14"]}]"#, + ) + .unwrap(); + let arr = raw.as_array().unwrap().clone(); + let result = parse_vector(&arr).expect("parse"); + match result { + QueryResult::Vector(iv) => { + assert_eq!(iv.values.len(), 1); + assert!((iv.values[0].value - 3.14).abs() < 1e-9); + assert_eq!(iv.timestamp, 1_700_000_000_500); + } + other => panic!("expected Vector, got {other:?}"), + } + } +} diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index 2af946ecb..96357fa0e 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -704,45 +704,90 @@ async fn main() -> Result<()> { }; 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 - // `ASAP_GORILLA_S3_*` env-var family. `HttpServer::new` already - // registers the `SimpleEngine` for the warm tier; we just plug in - // the archive engine here so any metric whose - // `StreamingConfig::storage_backend()` is `GorillaS3Archive` - // routes through the router and answers exactly from S3. + // Phase-5/6 + Step-2.3: register an archive-tier engine on the + // capability router. Two operating modes, selected at startup: // - // When the env vars are absent (the common dev / unit-test case) - // we leave the router single-engine — non-`SketchWarmTier` metrics - // would then surface a `503 NoEngineRegistered` from the HTTP - // handler, which is the correct fail-loud behaviour: a deploy - // that pins `GorillaS3Archive` without provisioning the cold - // store is a configuration bug. - match query_engine_rust::engines::gorilla::GorillaS3Config::from_env() { - Ok(s3_cfg) => { - match query_engine_rust::engines::gorilla::GorillaS3Store::with_default_backend(s3_cfg) { - Ok(store) => { - use query_engine_rust::engines::{GorillaEngineConfig, GorillaQueryEngine}; - use query_engine_rust::routing::QueryEngine; - let gorilla = Arc::new(GorillaQueryEngine::with_gorilla_s3( - Arc::new(store), - GorillaEngineConfig::default(), - )); - info!( - "Registering GorillaQueryEngine on the capability router (data_source_id=gorilla_archive)", - ); - server = server.with_query_engine(gorilla as Arc); + // * **Path A2 mode** — when `ASAP_THANOS_QUERY_URL` is set, the + // backend forwards archive-tier PromQL queries to a + // `thanos-query` sidecar via the + // [`ThanosForwardEngine`]. The forwarder is registered under + // both `thanos_archive` (its native id, for explicit + // `X-ASAP-Engine` overrides) and `gorilla_archive` (the legacy + // archive slot that the existing + // `compatible_storage_backends` failover sequence walks), so + // per-metric routing config can target either name without + // surprise. The legacy in-process `GorillaQueryEngine` is + // skipped in this mode. + // * **Legacy mode** — when `ASAP_THANOS_QUERY_URL` is unset, the + // in-process `GorillaQueryEngine` answers archive queries + // from per-hour Gorilla chunks landed on S3 / MinIO via the + // `GorillaS3Store`. This is the dev path and is preserved + // verbatim until Phase δ deletes it after Path A2 is verified + // end-to-end. + // + // When neither env-var family is configured we leave the router + // single-engine — non-`SketchWarmTier` metrics surface a + // `503 NoEngineRegistered` from the HTTP handler, which is the + // correct fail-loud behaviour for a misconfigured deploy. + match query_engine_rust::engines::gorilla::thanos_engine_from_env() { + Ok(Some(thanos)) => { + use query_engine_rust::engines::gorilla::DATA_SOURCE_THANOS_ARCHIVE_ID; + use query_engine_rust::routing::QueryEngine; + info!( + upstream = thanos.base_url(), + "Path A2: registering ThanosForwardEngine for the archive tier (data_source_id=thanos_archive, alias=gorilla_archive); legacy in-process GorillaQueryEngine skipped", + ); + // Two registrations of the same engine instance: one + // under its native id (explicit overrides) and one + // aliased onto the legacy archive slot so the + // `compatible_storage_backends` failover sequence finds + // it transparently. + let thanos_arc: Arc = Arc::new(thanos); + server = server + .with_query_engine_aliased( + DATA_SOURCE_THANOS_ARCHIVE_ID, + thanos_arc.clone(), + ) + .with_query_engine_aliased( + asap_types::StorageBackend::GorillaS3Archive.data_source_id(), + thanos_arc, + ); + } + Ok(None) => { + // Legacy path: register the in-process Gorilla engine + // when its env vars are present. + match query_engine_rust::engines::gorilla::GorillaS3Config::from_env() { + Ok(s3_cfg) => { + match query_engine_rust::engines::gorilla::GorillaS3Store::with_default_backend(s3_cfg) { + Ok(store) => { + use query_engine_rust::engines::{GorillaEngineConfig, GorillaQueryEngine}; + use query_engine_rust::routing::QueryEngine; + let gorilla = Arc::new(GorillaQueryEngine::with_gorilla_s3( + Arc::new(store), + GorillaEngineConfig::default(), + )); + info!( + "Registering legacy in-process GorillaQueryEngine on the capability router (data_source_id=gorilla_archive); set ASAP_THANOS_QUERY_URL to switch to Path A2 thanos forwarding", + ); + server = server.with_query_engine(gorilla as Arc); + } + Err(e) => { + warn!( + "ASAP_GORILLA_S3_* env vars present but GorillaS3Store failed to build ({e}); router will not have an archive engine", + ); + } + } } - Err(e) => { - warn!( - "ASAP_GORILLA_S3_* env vars present but GorillaS3Store failed to build ({e}); router will not have an archive engine", + Err(_) => { + info!( + "ASAP_GORILLA_S3_* env vars not configured — router serves warm-tier metrics only (set ASAP_GORILLA_S3_BUCKET + ASAP_GORILLA_S3_REGION to enable archive routing, or set ASAP_THANOS_QUERY_URL to enable Path A2 thanos forwarding)", ); } } } - Err(_) => { - info!( - "ASAP_GORILLA_S3_* env vars not configured — router serves warm-tier metrics only (set ASAP_GORILLA_S3_BUCKET + ASAP_GORILLA_S3_REGION to enable archive routing)", + Err(e) => { + warn!( + "ASAP_THANOS_QUERY_URL set but ThanosForwardEngine failed to build ({e}); router will not have an archive engine", ); } } diff --git a/asap-query-engine/src/routing/engine_router.rs b/asap-query-engine/src/routing/engine_router.rs index e915374b3..e20fdbaf7 100644 --- a/asap-query-engine/src/routing/engine_router.rs +++ b/asap-query-engine/src/routing/engine_router.rs @@ -133,6 +133,29 @@ impl EngineRouter { self.engines.insert(caps.data_source_id, engine); } + /// Register an engine under an alias `data_source_id`, ignoring the + /// id reported by `engine.capabilities()`. Used by Step-2.3's + /// Path A2 wiring: the same `ThanosForwardEngine` instance is + /// registered under both its native id (`thanos_archive`, for + /// explicit overrides) and under the legacy archive id + /// (`gorilla_archive`, so the existing + /// `compatible_storage_backends` failover sequence finds it + /// transparently). The legacy in-process `GorillaQueryEngine` is + /// only registered when `ASAP_THANOS_QUERY_URL` is unset; the + /// alias mechanism guarantees the two registrations never + /// collide on the same id. + /// + /// If two engines claim the same `id` the later registration wins + /// (matches [`Self::register`]'s hot-swap contract). + pub fn register_aliased(&mut self, id: &'static str, engine: Arc) { + debug!( + data_source_id = id, + engine_native_id = engine.capabilities().data_source_id, + "router: registering engine under alias", + ); + self.engines.insert(id, engine); + } + /// Number of engines registered. Test-only convenience. pub fn len(&self) -> usize { self.engines.len() @@ -457,6 +480,69 @@ mod tests { assert_eq!(ids, vec!["gorilla_archive", "sketch_warm"]); } + #[tokio::test] + async fn register_aliased_inserts_under_explicit_id() { + // Step-2.3 wiring: a single ThanosForwardEngine instance is + // registered under both `thanos_archive` (its native id) and + // `gorilla_archive` (the legacy archive slot the failover + // sequence walks). Both lookups must hit the same engine. + let mut router = EngineRouter::new(); + let (engine, calls) = StubEngine::new(StorageBackend::SketchWarmTier, Outcome::Ok); + // First, register under the engine's native id (`sketch_warm`). + router.register(engine.clone()); + // Then alias it under a totally different id. + router.register_aliased("custom_alias", engine); + // Both lookups must return the engine — we exercise both and + // verify the call counter ticked twice. + let native = router + .engine_by_id("sketch_warm") + .expect("native id registered"); + let aliased = router + .engine_by_id("custom_alias") + .expect("alias registered"); + let _ = native.execute("foo").await; + let _ = aliased.execute("foo").await; + assert_eq!( + calls.load(Ordering::SeqCst), + 2, + "both lookups must reach the same engine instance", + ); + + let mut ids: Vec<&str> = router.registered_ids().collect(); + ids.sort(); + assert_eq!(ids, vec!["custom_alias", "sketch_warm"]); + } + + #[tokio::test] + async fn register_aliased_overrides_capability_dispatch_target() { + // Step-2.3 wiring continued: when `ThanosForwardEngine` is + // aliased onto `gorilla_archive`, the failover sequence walks + // the alias instead of the legacy in-process engine. We + // simulate this with a stub registered under + // `GorillaS3Archive`'s native id via the alias path. + let mut router = EngineRouter::new(); + let (legacy, legacy_calls) = + StubEngine::new(StorageBackend::GorillaS3Archive, Outcome::Ok); + // Use alias to register under the gorilla_archive id + // explicitly (matches Step-2.3's "thanos under legacy slot" + // wiring). + router.register_aliased( + StorageBackend::GorillaS3Archive.data_source_id(), + legacy, + ); + + let result = router + .execute( + "sum_over_time(audit_events[1h])", + Statistic::Sum, + AccuracyTarget::Exact, + StorageBackend::GorillaS3Archive, + ) + .await; + assert!(result.is_ok()); + assert_eq!(legacy_calls.load(Ordering::SeqCst), 1); + } + #[tokio::test] async fn register_overwrites_same_data_source_id() { let mut router = EngineRouter::new();