From 1765d145f08fe6a180333315b78c3a40e753436c Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Sat, 9 May 2026 15:50:01 -0400 Subject: [PATCH] mvp #46: default archive engine + precompute /jobs API alignment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two narrow follow-ons to the controller↔backend wiring work; v7 RoutingTarget flatten was dropped after audit found the controller **does** emit shape-based multi-target routing (cold S3 fallback for ad-hoc count/topk on `http_requests_total{service="payments"}` relies on it). Multi-target stays. ## Default archive engine When neither `ASAP_THANOS_QUERY_URL` nor the `ASAP_GORILLA_S3_*` env-var family is configured, the backend used to leave the cold-tier slot empty → archive queries returned `503 NoEngineRegistered`. New `engines/no_data_archive.rs` registers a `NoDataArchiveEngine` stub under `data_source_id = "no_data_archive"`, aliased onto `gorilla_archive` so the existing `compatible_storage_backends` failover sequence finds it. Returns an empty instant vector for any query, with one warn-log at construction. `ASAP_REQUIRE_ARCHIVE_ENGINE=1` opts back into the original fail-loud `503` for environments where missing config should be treated as a deploy error rather than soft-empty. ## Precompute /jobs API The controller's `PrecomputeClient` (controller/src/config/ precompute.rs) registers + cancels precompute jobs via: POST /api/v1/precompute/jobs register DELETE /api/v1/precompute/jobs/:job_id cancel The backend only routed `POST /api/v1/precompute` (different shape). Adds the controller's `/jobs` routes alongside the existing route: * `PrecomputeJobSpec` matches the controller's `JobRequest` byte- for-byte (`{query, granularity, source, sketch_type, store_path}`). * `PrecomputeJobRegistry` is an in-memory `Arc>>` on `HttpServer`/`AppState`. Registration returns `{job_id (UUID), status: "created", created_at}`; DELETE returns 204 on hit, 404 on miss. Spec is tracked in memory; precompute scheduling is a later wiring. The legacy `POST /api/v1/precompute` route stays untouched. Verification: - `cargo build` clean. - `cargo test --lib --no-fail-fast`: **881 passed / 32 failed / 11 ignored**. 32 failures are pre-existing datafusion + schema- timeline-dispatch failures unrelated to this PR. New tests pass: `engines::no_data_archive::tests::execute_returns_empty_instant_vector`, `capabilities_use_no_data_archive_id`, `drivers::query::servers::http::tests::http_precompute_jobs_register_then_delete_roundtrip`, `http_precompute_jobs_delete_unknown_id_returns_404`. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../src/drivers/query/servers/http.rs | 228 +++++++++++++++++- asap-query-engine/src/engines/mod.rs | 2 + .../src/engines/no_data_archive.rs | 104 ++++++++ asap-query-engine/src/main.rs | 46 +++- 4 files changed, 368 insertions(+), 12 deletions(-) create mode 100644 asap-query-engine/src/engines/no_data_archive.rs diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index 2c37e3bf..49045f3a 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -9,7 +9,7 @@ use axum::{ }; use serde_json::Value; use std::collections::HashMap; -use std::sync::Arc; +use std::sync::{Arc, RwLock}; use std::time::Instant; use tokio::net::TcpListener; use tracing::{debug, info, warn}; @@ -24,6 +24,73 @@ use crate::stores::Store; use asap_types::{AccuracyTarget, StorageBackend}; use promql_utilities::query_logics::enums::Statistic; +// ─── Controller-pushed precompute job registry ──────────────────────────── +// +// The controller's `PrecomputeClient` (controller/src/config/precompute.rs) +// registers / cancels precompute jobs via: +// +// * `POST /api/v1/precompute/jobs` — body +// `{query, granularity, source, sketch_type, store_path}`; response +// `{job_id, status, created_at}`. +// * `DELETE /api/v1/precompute/jobs/{job_id}` — 204 on success, 404 +// when the id is unknown. +// +// The handler is intentionally minimal: it tracks the spec in an +// in-memory map and acknowledges the call. No precompute work is +// scheduled — that's a later wiring. The goal is to make the +// controller's calls succeed instead of 404 so the controller can +// progress its plan-push loop end-to-end. + +/// Body shape posted by the controller's `PrecomputeClient::register`. +/// Matches `controller/src/config/precompute.rs::JobRequest`. +#[derive(Clone, Debug, serde::Deserialize, serde::Serialize)] +pub struct PrecomputeJobSpec { + pub query: String, + pub granularity: String, + pub source: String, + pub sketch_type: String, + pub store_path: String, +} + +/// In-memory map of `job_id → spec`, populated by the +/// `POST /api/v1/precompute/jobs` handler and drained by the matching +/// DELETE handler. Wrapped in an `Arc>` so the axum state +/// can clone freely; the lock is held briefly per request and is +/// uncontended in practice (job count is small). +#[derive(Clone, Default)] +pub struct PrecomputeJobRegistry { + inner: Arc>>, +} + +impl PrecomputeJobRegistry { + pub fn new() -> Self { + Self::default() + } + + /// Insert a new job with a fresh UUID. Returns the generated id. + pub fn insert(&self, spec: PrecomputeJobSpec) -> String { + let job_id = uuid::Uuid::new_v4().to_string(); + let mut guard = self.inner.write().expect("poisoned"); + guard.insert(job_id.clone(), spec); + job_id + } + + /// Remove a job. Returns `true` when the id was present. + pub fn remove(&self, job_id: &str) -> bool { + let mut guard = self.inner.write().expect("poisoned"); + guard.remove(job_id).is_some() + } +} + +impl std::fmt::Debug for PrecomputeJobRegistry { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let n = self.inner.read().map(|g| g.len()).unwrap_or(0); + f.debug_struct("PrecomputeJobRegistry") + .field("jobs", &n) + .finish() + } +} + /// Per-query engine override header (Phase-6 accuracy reducer). /// /// When the client sets `X-ASAP-Engine: ` (or the @@ -146,6 +213,12 @@ pub struct HttpServer { /// normal routing-table path (which goes to Thanos for the cold /// archive and observes the 60–90 s flush gap). probe_cache: Option>, + /// In-memory job-spec map populated by the controller's + /// `POST /api/v1/precompute/jobs` calls. Always present (an empty + /// `Default` registry is fine for binaries that never wire the + /// controller). Future PRs will plumb this into the precompute + /// engine; today the handler just acks the call. + precompute_jobs: PrecomputeJobRegistry, } #[derive(Clone)] @@ -173,6 +246,8 @@ struct AppState { data_retention_ms: Option, /// See [`HttpServer::probe_cache`]. probe_cache: Option>, + /// See [`HttpServer::precompute_jobs`]. + precompute_jobs: PrecomputeJobRegistry, } impl HttpServer { @@ -199,6 +274,7 @@ impl HttpServer { backfill: None, data_retention_ms: None, probe_cache: None, + precompute_jobs: PrecomputeJobRegistry::new(), } } @@ -345,6 +421,15 @@ impl HttpServer { self } + /// Attach a [`PrecomputeJobRegistry`] used by the controller-pushed + /// `POST /api/v1/precompute/jobs` and matching `DELETE` endpoints. + /// Callers that don't override this share the per-server default + /// (an empty in-memory map populated by the registration handler). + pub fn with_precompute_jobs(mut self, registry: PrecomputeJobRegistry) -> Self { + self.precompute_jobs = registry; + self + } + pub async fn run(self) -> Result<(), Box> { srv_metrics::register_all(); @@ -373,6 +458,7 @@ impl HttpServer { backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, probe_cache: self.probe_cache.clone(), + precompute_jobs: self.precompute_jobs.clone(), }; let range_query_endpoint = adapter.get_range_query_endpoint(); @@ -391,6 +477,18 @@ impl HttpServer { .route("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/internal/s3_cost.csv", get(handle_s3_cost_csv)) // Controller integration endpoints .route("/api/v1/precompute", post(handle_precompute_job)) + // Controller's `PrecomputeClient` (controller/src/config/precompute.rs) + // posts to `/jobs` and DELETEs by job_id. Tracks the spec + // in memory; the `/api/v1/precompute` route stays for + // legacy `{query_expr, granularity_secs, start, end}` callers. + .route( + "/api/v1/precompute/jobs", + post(handle_post_precompute_job_register), + ) + .route( + "/api/v1/precompute/jobs/:job_id", + axum::routing::delete(handle_delete_precompute_job), + ) .route("/api/v1/health", get(handle_health)) .route("/api/v1/store/metrics", get(handle_store_metrics)) .route( @@ -452,6 +550,7 @@ impl HttpServer { backfill: self.backfill.clone(), data_retention_ms: self.data_retention_ms, probe_cache: self.probe_cache.clone(), + precompute_jobs: self.precompute_jobs.clone(), }; let range_query_endpoint = adapter.get_range_query_endpoint(); @@ -489,6 +588,14 @@ impl HttpServer { "/api/v1/db/backfill/jobs/:job_id", get(handle_get_backfill_job).delete(handle_delete_backfill_job), ) + .route( + "/api/v1/precompute/jobs", + post(handle_post_precompute_job_register), + ) + .route( + "/api/v1/precompute/jobs/:job_id", + axum::routing::delete(handle_delete_precompute_job), + ) .with_state(app_state); let listener = TcpListener::bind("127.0.0.1:0").await?; @@ -4478,6 +4585,87 @@ aggregations: saw probe value {now_ms} leaked into response: {body}", ); } + + // ── Controller-pushed precompute job registry tests ─────────────────── + + /// `POST /api/v1/precompute/jobs` returns 200 + a `job_id`; the + /// matching DELETE returns 204 the first time and 404 on the + /// second call. + #[tokio::test] + async fn http_precompute_jobs_register_then_delete_roundtrip() { + let port = setup_test_server_with_router( + StorageBackend::SketchWarmTier, + Vec::new(), + ) + .await; + let client = Client::new(); + + let body = serde_json::json!({ + "query": "quantile_over_time(0.99, http_requests_total[5m])", + "granularity": "60s", + "source": "backend:4317", + "sketch_type": "ddsketch", + "store_path": "precomputed/http_requests_total/p99/5m", + }); + let resp = client + .post(format!("http://127.0.0.1:{port}/api/v1/precompute/jobs")) + .json(&body) + .send() + .await + .expect("send ok"); + assert_eq!(resp.status().as_u16(), 200, "register must 200"); + let resp_body: serde_json::Value = resp.json().await.expect("json"); + let job_id = resp_body["job_id"].as_str().expect("job_id").to_string(); + assert!(!job_id.is_empty(), "job_id must be non-empty"); + assert_eq!(resp_body["status"], "created"); + + let resp = client + .delete(format!( + "http://127.0.0.1:{port}/api/v1/precompute/jobs/{job_id}" + )) + .send() + .await + .expect("send ok"); + assert_eq!( + resp.status().as_u16(), + 204, + "first delete must 204, got {}", + resp.status() + ); + + let resp = client + .delete(format!( + "http://127.0.0.1:{port}/api/v1/precompute/jobs/{job_id}" + )) + .send() + .await + .expect("send ok"); + assert_eq!( + resp.status().as_u16(), + 404, + "second delete must 404 (job already removed)", + ); + } + + /// `DELETE /api/v1/precompute/jobs/{unknown}` returns 404. + #[tokio::test] + async fn http_precompute_jobs_delete_unknown_id_returns_404() { + let port = setup_test_server_with_router( + StorageBackend::SketchWarmTier, + Vec::new(), + ) + .await; + let client = Client::new(); + let resp = client + .delete(format!( + "http://127.0.0.1:{port}/api/v1/precompute/jobs/{}", + "never-registered" + )) + .send() + .await + .expect("send ok"); + assert_eq!(resp.status().as_u16(), 404); + } } // ── Controller integration: PrecomputeJob execution ────────────────────────── @@ -4554,6 +4742,44 @@ async fn handle_precompute_job( } } +/// `POST /api/v1/precompute/jobs` — register a controller-pushed +/// precompute job. Body matches `controller/src/config/precompute.rs`'s +/// `JobRequest` (`{query, granularity, source, sketch_type, store_path}`). +/// Returns `200 OK` with `{job_id, status, created_at}`. +/// +/// The handler is intentionally minimal: it stores the spec in an +/// in-memory map and acknowledges. No precompute work is scheduled — +/// future PRs will plumb this into the precompute engine. +async fn handle_post_precompute_job_register( + State(state): State, + axum::Json(spec): axum::Json, +) -> Response { + let job_id = state.precompute_jobs.insert(spec); + let body = serde_json::json!({ + "job_id": job_id, + "status": "created", + "created_at": chrono::Utc::now().to_rfc3339(), + }); + (StatusCode::OK, axum::Json(body)).into_response() +} + +/// `DELETE /api/v1/precompute/jobs/:job_id` — drop a registered job. +/// Returns `204 No Content` on success and `404` when unknown. +async fn handle_delete_precompute_job( + State(state): State, + axum::extract::Path(job_id): axum::extract::Path, +) -> Response { + if state.precompute_jobs.remove(&job_id) { + (StatusCode::NO_CONTENT, ()).into_response() + } else { + let body = serde_json::json!({ + "status": "error", + "error": format!("precompute job '{job_id}' not found"), + }); + (StatusCode::NOT_FOUND, axum::Json(body)).into_response() + } +} + /// Health check endpoint for DataCollector controller to verify backend is alive. async fn handle_health() -> &'static str { "ok" diff --git a/asap-query-engine/src/engines/mod.rs b/asap-query-engine/src/engines/mod.rs index e056f6d5..0dc307c2 100644 --- a/asap-query-engine/src/engines/mod.rs +++ b/asap-query-engine/src/engines/mod.rs @@ -25,6 +25,7 @@ pub mod gorilla; pub mod logical; +pub mod no_data_archive; pub mod physical; pub mod prometheus; pub mod query_result; @@ -35,6 +36,7 @@ pub mod window_merger; pub use gorilla::{ EngineError as GorillaEngineError, GorillaEngineConfig, GorillaQueryEngine, }; +pub use no_data_archive::{NoDataArchiveEngine, DATA_SOURCE_ID_NO_DATA_ARCHIVE}; pub use prometheus::{PrometheusForwardConfig, PrometheusForwardEngine, PrometheusForwardError}; pub use query_result::{InstantVector, QueryResult, RangeVector, RangeVectorElement, Sample}; pub use simple::SimpleEngine; diff --git a/asap-query-engine/src/engines/no_data_archive.rs b/asap-query-engine/src/engines/no_data_archive.rs new file mode 100644 index 00000000..06608201 --- /dev/null +++ b/asap-query-engine/src/engines/no_data_archive.rs @@ -0,0 +1,104 @@ +//! `NoDataArchiveEngine` — a stub archive engine that always answers +//! with an empty result set. +//! +//! ## Why this exists +//! +//! When a deploy is configured with the warm-tier sketch path only and +//! has neither `ASAP_THANOS_QUERY_URL` nor `ASAP_GORILLA_S3_*` env +//! vars set, no archive engine is registered on the +//! [`crate::routing::EngineRouter`]. Cold queries (queries the +//! per-metric routing table sends to `gorilla_archive` / +//! `thanos_archive`) then surface as +//! `503 NoEngineRegistered` from the HTTP handler. +//! +//! Treating "no archive configured" as a 503 trips up dashboards and +//! freshness probes that just want a degraded but successful answer. +//! This engine flips that default: when the archive env vars are +//! unset, the binary registers a [`NoDataArchiveEngine`] under the +//! `gorilla_archive` slot. Cold queries return an **empty result +//! set** with `data_source_id = "no_data_archive"` so the wire +//! response carries enough signal for operators to notice the +//! misconfig without breaking the request path. +//! +//! Operators that want the original "fail-loud" behaviour can opt +//! back in by setting `ASAP_REQUIRE_ARCHIVE_ENGINE=1` (the binary +//! reads this env var in `main.rs` and skips the no-data registration +//! when set). + +use async_trait::async_trait; +use tracing::info; + +use asap_types::StorageBackend; + +use crate::engines::{EngineError, QueryResult}; +use crate::routing::{EngineCapabilities, QueryEngine}; + +/// Stable engine id for the no-data fallback. Reported in +/// `data_source_id` on the wire response for cold queries when the +/// archive env vars are unset. +pub const DATA_SOURCE_ID_NO_DATA_ARCHIVE: &str = "no_data_archive"; + +/// Engine that answers every query with an empty instant vector. Used +/// by the binary as a stand-in for a real archive engine when neither +/// `ASAP_THANOS_QUERY_URL` nor `ASAP_GORILLA_S3_*` is configured. +#[derive(Debug, Default)] +pub struct NoDataArchiveEngine; + +impl NoDataArchiveEngine { + /// Build the stub. Logs once at construction so the misconfig is + /// visible in the binary's startup log. + pub fn new() -> Self { + info!( + "no archive backend configured (set ASAP_THANOS_QUERY_URL or \ + ASAP_GORILLA_S3_* to enable); cold queries will return empty results" + ); + Self + } +} + +#[async_trait] +impl QueryEngine for NoDataArchiveEngine { + async fn execute(&self, _query: &str) -> Result { + // Empty instant vector at t=0. The HTTP handler annotates the + // wire response with `data_source: no_data_archive` so the + // caller can distinguish "engine missing" from "real archive + // had nothing". + Ok(QueryResult::vector(Vec::new(), 0)) + } + + fn capabilities(&self) -> EngineCapabilities { + EngineCapabilities { + data_source_id: DATA_SOURCE_ID_NO_DATA_ARCHIVE, + // Register under the archive slot. The binary aliases this + // engine onto the `gorilla_archive` id so the routing + // table's archive entries dispatch here transparently. + storage_backend: StorageBackend::GorillaS3Archive, + supports_streams_above_bytes: 0, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn execute_returns_empty_instant_vector() { + let engine = NoDataArchiveEngine::new(); + let result = engine.execute("count(any_metric)").await.expect("ok"); + match result { + QueryResult::Vector(iv) => { + assert!(iv.values.is_empty(), "expected empty vector"); + } + QueryResult::Matrix(_) => panic!("expected instant vector, got matrix"), + } + } + + #[test] + fn capabilities_use_no_data_archive_id() { + let engine = NoDataArchiveEngine::new(); + let caps = engine.capabilities(); + assert_eq!(caps.data_source_id, "no_data_archive"); + assert_eq!(caps.storage_backend, StorageBackend::GorillaS3Archive); + } +} diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs index 01a213b3..22768793 100644 --- a/asap-query-engine/src/main.rs +++ b/asap-query-engine/src/main.rs @@ -704,10 +704,13 @@ async fn main() -> Result<()> { // 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. + // When neither env-var family is configured the binary registers + // a `NoDataArchiveEngine` stub under the `gorilla_archive` alias + // so cold queries succeed with an empty result instead of + // surfacing as `503 NoEngineRegistered`. Operators that want the + // original fail-loud behaviour can opt back in by setting + // `ASAP_REQUIRE_ARCHIVE_ENGINE=1`. + let mut archive_registered = false; match query_engine_rust::engines::gorilla::thanos_engine_from_env() { Ok(Some(thanos)) => { use query_engine_rust::engines::gorilla::DATA_SOURCE_THANOS_ARCHIVE_ID; @@ -716,11 +719,6 @@ async fn main() -> Result<()> { 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( @@ -731,10 +729,9 @@ async fn main() -> Result<()> { asap_types::StorageBackend::GorillaS3Archive.data_source_id(), thanos_arc, ); + archive_registered = true; } 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) { @@ -749,6 +746,7 @@ async fn main() -> Result<()> { "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); + archive_registered = true; } Err(e) => { warn!( @@ -771,6 +769,32 @@ async fn main() -> Result<()> { } } + // No archive engine configured — register a `NoDataArchiveEngine` + // stub under the `gorilla_archive` alias so cold queries succeed + // with an empty result. `ASAP_REQUIRE_ARCHIVE_ENGINE=1` opts back + // into the original fail-loud (`503 NoEngineRegistered`) behaviour. + if !archive_registered { + let require_archive = std::env::var("ASAP_REQUIRE_ARCHIVE_ENGINE") + .map(|v| matches!(v.as_str(), "1" | "true" | "yes" | "on")) + .unwrap_or(false); + if require_archive { + warn!( + "ASAP_REQUIRE_ARCHIVE_ENGINE=1 set and no archive engine configured — cold queries will return 503 NoEngineRegistered", + ); + } else { + use query_engine_rust::engines::NoDataArchiveEngine; + use query_engine_rust::routing::QueryEngine; + info!( + "Registering NoDataArchiveEngine stub on the archive slot (data_source_id=no_data_archive, alias=gorilla_archive); set ASAP_REQUIRE_ARCHIVE_ENGINE=1 to disable", + ); + let stub: Arc = Arc::new(NoDataArchiveEngine::new()); + server = server.with_query_engine_aliased( + asap_types::StorageBackend::GorillaS3Archive.data_source_id(), + stub, + ); + } + } + // Phase ε.2: register a `PrometheusForwardEngine` under the // `prometheus_remote` engine id when `ASAP_PROMETHEUS_QUERY_URL` // is set. The controller's Mode 3 (`RawAtEdgePrometheusArchive`)