From 49e66a3b29e01c52c568bac16a6851a6ee66eaaa Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 3 Apr 2026 23:34:05 -0500 Subject: [PATCH 1/3] feat: precompute job endpoint + controller integration APIs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New endpoints for DataCollector controller integration: POST /api/v1/precompute - Receives PrecomputeJob from DataCollector controller - Executes PromQL query_expr against stored sketches - Returns query result or 404 if not answerable GET /api/v1/health - Health check for controller to verify backend is alive GET /api/v1/store/metrics - Returns list of metrics/aggregation IDs in store - Controller can use this to verify data is flowing Usage with DataCollector controller: Controller creates PrecomputeJob with query_expr (e.g., "topk(10, count_over_time(m{env=\"prod\"}[1m]) by (svc))") → POSTs to backend /api/v1/precompute → Backend evaluates against SimpleMapStore → Returns approximate result Co-Authored-By: Claude Opus 4.6 (1M context) --- .../src/drivers/query/servers/http.rs | 104 ++++++++++++++++++ 1 file changed, 104 insertions(+) diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index 322f560a..dd60efce 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -85,6 +85,10 @@ impl HttpServer { .route(runtime_info_path, get(handle_runtime_info)) .route(runtime_info_path, post(handle_runtime_info)) .route("/metrics", get(handle_metrics)) + // Controller integration endpoints + .route("/api/v1/precompute", post(handle_precompute_job)) + .route("/api/v1/health", get(handle_health)) + .route("/api/v1/store/metrics", get(handle_store_metrics)) .with_state(app_state); let listener = TcpListener::bind(format!("0.0.0.0:{}", self.config.port)).await?; @@ -681,3 +685,103 @@ mod tests { assert!(status.is_success() || status == reqwest::StatusCode::OK); } } + +// ── Controller integration: PrecomputeJob execution ────────────────────────── + +/// Request body from DataCollector controller's PrecomputeJob. +#[derive(serde::Deserialize)] +struct PrecomputeJobRequest { + /// PromQL expression to evaluate against stored sketches. + query_expr: String, + /// Window granularity in seconds. + #[serde(default)] + granularity_secs: u64, + /// Start timestamp (unix seconds). 0 = use earliest available. + #[serde(default)] + start: f64, + /// End timestamp (unix seconds). 0 = use latest available. + #[serde(default)] + end: f64, +} + +/// Execute a precompute job from the DataCollector controller. +/// +/// POST /api/v1/precompute +/// +/// The controller creates PrecomputeJobs when a query's upper sub-tree +/// (e.g., TopK, HistogramQuantile) requires evaluation on merged sketches. +/// This endpoint receives that job and runs it against the SimpleMapStore. +async fn handle_precompute_job( + State(state): State, + axum::Json(req): axum::Json, +) -> axum::response::Response { + use axum::http::StatusCode; + use axum::response::IntoResponse; + + let time = if req.end > 0.0 { req.end } else { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs_f64() + }; + + info!( + query = %req.query_expr, + time = %time, + "Executing precompute job from controller" + ); + + match state.query_engine.handle_query_promql(req.query_expr, time) { + Some((key_by, result)) => { + let body = serde_json::json!({ + "status": "success", + "data": { + "result_type": "precompute", + "key_by": format!("{:?}", key_by), + "result": format!("{:?}", result), + } + }); + (StatusCode::OK, axum::Json(body)).into_response() + } + None => { + // Query not answerable by sketches — return 404 with hint + let body = serde_json::json!({ + "status": "error", + "error": "query not answerable by stored sketches", + "hint": "ensure the metric has been ingested via OTLP/Kafka and a matching query_config exists" + }); + (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" +} + +/// Return list of metrics currently in the store. +async fn handle_store_metrics( + State(state): State, +) -> axum::response::Response { + use axum::http::StatusCode; + use axum::response::IntoResponse; + + match state.store.get_earliest_timestamp_per_aggregation_id() { + Ok(timestamps) => { + let body = serde_json::json!({ + "status": "success", + "aggregation_count": timestamps.len(), + "earliest_timestamps": timestamps, + }); + (StatusCode::OK, axum::Json(body)).into_response() + } + Err(e) => { + let body = serde_json::json!({ + "status": "error", + "error": format!("{}", e), + }); + (StatusCode::INTERNAL_SERVER_ERROR, axum::Json(body)).into_response() + } + } +} From f5f969dcee7219153f67874d8e59b2e35fbef747 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Sun, 12 Apr 2026 21:20:19 -0400 Subject: [PATCH 2/3] style: apply rustfmt to precompute job handlers Co-Authored-By: Claude Opus 4.6 (1M context) --- asap-query-engine/src/drivers/query/servers/http.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index dd60efce..8e753349 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -718,7 +718,9 @@ async fn handle_precompute_job( use axum::http::StatusCode; use axum::response::IntoResponse; - let time = if req.end > 0.0 { req.end } else { + let time = if req.end > 0.0 { + req.end + } else { std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() @@ -761,9 +763,7 @@ async fn handle_health() -> &'static str { } /// Return list of metrics currently in the store. -async fn handle_store_metrics( - State(state): State, -) -> axum::response::Response { +async fn handle_store_metrics(State(state): State) -> axum::response::Response { use axum::http::StatusCode; use axum::response::IntoResponse; From 147aee52436b0675e501d836646107405f303139 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Sun, 12 Apr 2026 21:27:46 -0400 Subject: [PATCH 3/3] fix: log start/end/granularity_secs from precompute request Satisfies clippy dead_code lint on fields previously accepted but unused. These fields are part of the wire format from the DataCollector controller and are now surfaced in the tracing span for observability. Co-Authored-By: Claude Opus 4.6 (1M context) --- asap-query-engine/src/drivers/query/servers/http.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/asap-query-engine/src/drivers/query/servers/http.rs b/asap-query-engine/src/drivers/query/servers/http.rs index 8e753349..712265d1 100644 --- a/asap-query-engine/src/drivers/query/servers/http.rs +++ b/asap-query-engine/src/drivers/query/servers/http.rs @@ -729,6 +729,9 @@ async fn handle_precompute_job( info!( query = %req.query_expr, + start = %req.start, + end = %req.end, + granularity_secs = %req.granularity_secs, time = %time, "Executing precompute job from controller" );