mvp step 2.3: backend HTTP-forward to thanos-query for Path A2 archive

asap-query-engine/src/bin/precompute_engine.rs

@@ -324,27 +324,40 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         };
         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<dyn QueryEngine> = 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<dyn std::error::Error + Send + Sync>> {
                             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<dyn QueryEngine>);
                     }
@@ -362,11 +375,16 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                             "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",
                 );
             }
         }

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<dyn QueryEngine>,
+    ) -> 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<dyn QueryEngine> = 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:
+        // <metric_storage>.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<dyn QueryEngine> = 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<dyn QueryEngine> = 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<dyn QueryEngine>)>,
+    ) -> 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 ──────────────────────────

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