From 4b50ae597e0be5c5162ddea1f7efbe29ee566a7b Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Mon, 20 Apr 2026 12:19:13 -0400 Subject: [PATCH] =?UTF-8?q?feat(http):=20surface=20=C2=A77=20Partial=20res?= =?UTF-8?q?ults=20on=20Prometheus=20`warnings`?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Task #34 gap #2 of 3. #48's dispatcher activated per-segment execution for combinable stats (Count/Sum/Min/Max) but refused to run for non-combinable stats (Quantile/Topk/Cardinality/ Rate/Increase) — those still fell through to the single-agg path and saw the same data cliff the dispatcher was written to prevent. Reason: there was no way to tell the HTTP caller "this answer is Partial because the query spans a reconfigure boundary and the statistic can't be combined scalarly." Changes: - **`QueryResult::vector_with_warnings` + `warnings()` accessor.** `InstantVector` and `RangeVector` gain a `#[serde(default, skip_serializing_if = "Vec::is_empty")]` `warnings: Vec` field. Default constructors (`QueryResult::vector`, `QueryResult::matrix`) stay wire- compatible — they still serialise without the field when empty, so every existing caller and snapshot test holds. - **Prometheus adapter** adds a top-level `warnings: []` on `PrometheusResponse` (also skip-if-empty), matching upstream's native API. `format_success_response` + `format_range_success_response` thread through any warnings the engine populated; absent → no field. - **Dispatcher** drops the combinable-only early return. The full loop now runs for every statistic whenever the timeline has ≥2 segments. `combine_statistic` still returns `Full(v)` for cleanly-combinable inputs and `Partial { covered, missing }` everywhere else — on Partial we accumulate the `covered` scalar (when present), flag `any_partial`, and build a human-readable warnings list with the metric, range, statistic, dropped-group count, and up to three unresolved segments (agg_id, clipped range, status, coverage). Over-three are summarised; the full set is still inspectable via `GET /api/v1/db/timeline`. 5 new unit tests: 3 in `engines::query_result::tests` covering the default-empty / with-warnings / matrix wire contract; 2 in the Prometheus adapter covering the response-side serialisation contract. 734 lib tests pass (+5), clippy clean, fmt clean. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../drivers/query/adapters/prometheus_http.rs | 68 +++++++++- asap-query-engine/src/engines/query_result.rs | 88 ++++++++++++- .../src/engines/simple_engine.rs | 122 +++++++++++++----- 3 files changed, 239 insertions(+), 39 deletions(-) diff --git a/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs b/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs index a52ad686..69e9d811 100644 --- a/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs +++ b/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs @@ -24,6 +24,13 @@ pub struct PrometheusResponse { pub error_type: Option, #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, + /// Non-error advisories — maps to Prometheus's top-level + /// `warnings: []` field. Phase 3b-2-b uses this to surface + /// partial results from the §7 schema-timeline dispatcher + /// (query spans a reconfigure boundary with a non-combinable + /// statistic or a Purged segment). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub warnings: Vec, } impl PrometheusResponse { @@ -33,6 +40,19 @@ impl PrometheusResponse { data: Some(data), error_type: None, error: None, + warnings: Vec::new(), + } + } + + /// `success` + a non-empty `warnings` list attached. Used by the + /// query adapters when the engine returned a `CombinedResult::Partial`. + pub fn success_with_warnings(data: Value, warnings: Vec) -> Self { + Self { + status: "success".to_string(), + data: Some(data), + error_type: None, + error: None, + warnings, } } @@ -42,6 +62,7 @@ impl PrometheusResponse { data: None, error_type: Some(error_type.to_string()), error: Some(error.to_string()), + warnings: Vec::new(), } } } @@ -200,7 +221,15 @@ impl QueryResponseAdapter for PrometheusHttpAdapter { StatusCode::INTERNAL_SERVER_ERROR })?; - let response = PrometheusResponse::success(prometheus_data); + // Thread through any Phase 3 timeline-dispatch warnings so + // they land on the top-level `warnings` field, matching + // Prometheus's native API. + let warnings = result.query_result.warnings().to_vec(); + let response = if warnings.is_empty() { + PrometheusResponse::success(prometheus_data) + } else { + PrometheusResponse::success_with_warnings(prometheus_data, warnings) + }; Ok(Json(serde_json::to_value(response).unwrap()).into_response()) } @@ -215,7 +244,12 @@ impl QueryResponseAdapter for PrometheusHttpAdapter { error!("Failed to convert range result: {}", e); StatusCode::INTERNAL_SERVER_ERROR })?; - let response = PrometheusResponse::success(prometheus_data); + let warnings = result.warnings().to_vec(); + let response = if warnings.is_empty() { + PrometheusResponse::success(prometheus_data) + } else { + PrometheusResponse::success_with_warnings(prometheus_data, warnings) + }; Ok(Json(serde_json::to_value(response).unwrap()).into_response()) } @@ -512,4 +546,34 @@ mod tests { let parsed = result.unwrap(); assert_eq!(parsed.query, "sum(metric)"); } + + #[test] + fn success_response_without_warnings_omits_field() { + let r = PrometheusResponse::success(json!({"resultType": "vector", "result": []})); + let s = serde_json::to_string(&r).unwrap(); + assert!(s.contains("\"status\":\"success\"")); + assert!( + !s.contains("\"warnings\""), + "empty warnings must be skip-serialised for wire compatibility" + ); + } + + #[test] + fn success_response_with_warnings_serialises_the_top_level_field() { + // This is the Phase 3b-2-b contract: a Partial result coming + // out of the §7 timeline dispatcher lands on Prometheus's + // native `warnings: []` field at the top of the response, + // matching upstream behaviour for warning-carrying queries. + let r = PrometheusResponse::success_with_warnings( + json!({"resultType": "vector", "result": []}), + vec![ + "partial result: query spans 2 schemas".to_string(), + "1 group(s) dropped".to_string(), + ], + ); + let s = serde_json::to_string(&r).unwrap(); + assert!(s.contains("\"warnings\":[")); + assert!(s.contains("partial result: query spans 2 schemas")); + assert!(s.contains("1 group(s) dropped")); + } } diff --git a/asap-query-engine/src/engines/query_result.rs b/asap-query-engine/src/engines/query_result.rs index 24be3e8e..5a7437c1 100644 --- a/asap-query-engine/src/engines/query_result.rs +++ b/asap-query-engine/src/engines/query_result.rs @@ -19,11 +19,47 @@ impl QueryResult { } pub fn vector(values: Vec, timestamp: u64) -> Self { - QueryResult::Vector(InstantVector { values, timestamp }) + QueryResult::Vector(InstantVector { + values, + timestamp, + warnings: Vec::new(), + }) + } + + /// Phase 3b-2-b: construct an instant vector with a non-empty + /// warnings list. Used by the timeline dispatcher when the query + /// spans a reconfigure boundary and one or more segments could + /// not contribute to the answer (non-combinable statistic, purged + /// data, or agg_id removed from config mid-flight). Prometheus's + /// native JSON surface carries these back to the caller via the + /// top-level `warnings` field, matching the upstream contract. + pub fn vector_with_warnings( + values: Vec, + timestamp: u64, + warnings: Vec, + ) -> Self { + QueryResult::Vector(InstantVector { + values, + timestamp, + warnings, + }) } pub fn matrix(values: Vec) -> Self { - QueryResult::Matrix(RangeVector { values }) + QueryResult::Matrix(RangeVector { + values, + warnings: Vec::new(), + }) + } + + /// Accumulated per-result warnings. Empty for single-schema + /// queries; populated by the §7 timeline dispatcher when a query + /// spans a reconfigure boundary. + pub fn warnings(&self) -> &[String] { + match self { + QueryResult::Vector(iv) => &iv.warnings, + QueryResult::Matrix(m) => &m.warnings, + } } } @@ -32,6 +68,15 @@ impl QueryResult { pub struct InstantVector { pub values: Vec, pub timestamp: u64, + /// Non-error advisories attached to this result, surfaced on + /// Prometheus's top-level `warnings` field. Empty for + /// single-schema queries; populated by the Phase 3b-2-b timeline + /// dispatcher when one or more segments produced a + /// [`crate::engines::timeline_dispatch::CombinedResult::Partial`] + /// (non-combinable statistic, purged coverage, or agg_id + /// missing from the current config). + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub warnings: Vec, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -50,6 +95,9 @@ impl InstantVectorElement { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct RangeVector { pub values: Vec, + /// See [`InstantVector::warnings`]. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub warnings: Vec, } /// Individual element in a range vector @@ -235,4 +283,40 @@ mod tests { assert_eq!(sample.timestamp, 12345); assert_eq!(sample.value, 99.9); } + + #[test] + fn vector_without_warnings_returns_empty_slice_and_omits_field_in_json() { + let labels = create_test_labels(); + let el = InstantVectorElement::new(labels, 1.0); + let qr = QueryResult::vector(vec![el], 100); + assert!(qr.warnings().is_empty()); + let json = serde_json::to_string(&qr).unwrap(); + assert!( + !json.contains("\"warnings\""), + "default-empty warnings must be skip-serialised for wire compatibility" + ); + } + + #[test] + fn vector_with_warnings_round_trips_through_serde() { + let labels = create_test_labels(); + let el = InstantVectorElement::new(labels, 1.0); + let warnings = vec!["partial result: 2 schemas".to_string()]; + let qr = QueryResult::vector_with_warnings(vec![el], 100, warnings.clone()); + assert_eq!(qr.warnings(), warnings.as_slice()); + + let json = serde_json::to_string(&qr).unwrap(); + assert!(json.contains("\"warnings\"")); + let back: QueryResult = serde_json::from_str(&json).unwrap(); + assert_eq!(back.warnings(), warnings.as_slice()); + } + + #[test] + fn matrix_warnings_default_empty_and_skip_serialised() { + let el = RangeVectorElement::new(create_test_labels()); + let qr = QueryResult::matrix(vec![el]); + assert!(qr.warnings().is_empty()); + let json = serde_json::to_string(&qr).unwrap(); + assert!(!json.contains("\"warnings\"")); + } } diff --git a/asap-query-engine/src/engines/simple_engine.rs b/asap-query-engine/src/engines/simple_engine.rs index 4114ce1c..927600d3 100644 --- a/asap-query-engine/src/engines/simple_engine.rs +++ b/asap-query-engine/src/engines/simple_engine.rs @@ -3007,30 +3007,38 @@ impl SimpleEngine { } /// Phase 3b-2-b: per-segment dispatch across the §7 schema - /// timeline for combinable statistics. + /// timeline. /// - /// Returns `Some(result)` when: - /// * `SchemaRegistry::timeline_for_metric` yields two or more - /// segments for the query's metric within its time range - /// (i.e. the query spans a reconfigure boundary), AND - /// * the query's statistic is one of Count / Sum / Min / Max, - /// which `timeline_dispatch::combine_statistic` can stitch - /// cleanly at the scalar level. + /// Returns `Some(result)` when `SchemaRegistry::timeline_for_metric` + /// yields two or more segments for the query's metric within its + /// time range (i.e. the query spans a reconfigure boundary). + /// Returns `None` otherwise (single-schema range, unparseable + /// query, unresolved probe aggregation) so the caller falls back + /// to the default single-agg path — that path is still correct + /// whenever the timeline doesn't actually span a boundary. /// - /// Returns `None` otherwise (single-schema range, non-combinable - /// statistic, unparseable query, unresolved probe aggregation). - /// The caller falls back to the default single-agg path — that - /// path is still correct whenever the timeline doesn't actually - /// span a boundary. Non-combinable statistics (quantile / topk / - /// cardinality / rate / increase) are routed through the - /// default path here too; PR B2 will surface - /// [`crate::engines::timeline_dispatch::CombinedResult::Partial`] - /// to the HTTP response so users can see "covered" + "missing" - /// segments explicitly instead of the single-agg data cliff. + /// ## Combinable vs non-combinable statistics + /// + /// For combinable scalar statistics (Count / Sum / Min / Max) + /// every segment contributes and the result is a clean `Full` + /// value the user can trust without caveat. + /// + /// For non-combinable statistics (Quantile / Topk / Cardinality / + /// Rate / Increase) — or for any combinable run that includes a + /// `Purged` / config-missing segment — `combine_statistic` + /// returns `Partial`. This method surfaces Partial on the + /// Prometheus HTTP response's `warnings` field: the top-level + /// result carries whatever combinable prefix we could compute + /// (for additive stats) or an empty vector (for non-combinable), + /// plus one or more `warnings` strings explaining the schema + /// boundary, the dropped groups, and the unresolved segments. /// /// Delivers the user-visible Phase 3 outcome documented in /// `docs/design-sketch-db.md` §7: queries spanning a reconfigure - /// boundary no longer see a data cliff for additive statistics. + /// boundary no longer see a silent data cliff — additive stats + /// get the combined answer, and non-combinable stats get an + /// explicit Partial notice instead of the arbitrary single-agg + /// single-segment result. fn try_handle_query_promql_via_timeline( &self, query: &str, @@ -3070,15 +3078,6 @@ impl SimpleEngine { return None; } - // Phase 3: only activate for combinable statistics. See the - // module doc on `timeline_dispatch` §7.3 combinability table. - if !matches!( - stat, - Statistic::Count | Statistic::Sum | Statistic::Min | Statistic::Max - ) { - return None; - } - debug!( metric = %metric_name, segments = segments.len(), @@ -3156,7 +3155,14 @@ impl SimpleEngine { // combiner folds per-segment scalars into one final scalar // per group. Groups that only appear in `unresolved` (no // segment ever produced a value for them) are skipped. + // + // `any_partial` tracks whether any group came back non-`Full` + // — drives the Prometheus `warnings` surface below so the + // caller sees "this answer is partial" explicitly instead of + // a silent cliff. let mut output: Vec = Vec::new(); + let mut any_partial = false; + let mut groups_with_no_value = 0usize; for (label_key, segment_values) in per_group { match combine_statistic(stat, &segment_values, &unresolved) { CombinedResult::Full(v) => { @@ -3165,22 +3171,68 @@ impl SimpleEngine { CombinedResult::Partial { covered: Some(v), .. } => { - // Best-effort: emit `covered` for combinable - // stats so the user sees the partial sum. PR B2 - // will add a first-class Partial response surface - // carrying the `missing` list. + // Emit `covered` for combinable stats so the user + // sees the partial sum. The warning below tells + // them not to trust the scalar as a full range + // answer. output.push(InstantVectorElement::new(label_key.unwrap_or_default(), v)); + any_partial = true; } CombinedResult::Partial { covered: None, .. } => { - // No segment produced a value for this group — - // drop it rather than emit a misleading 0. + // Non-combinable stat (quantile / topk / rate / + // increase / cardinality) OR a group the + // combiner couldn't reduce. Drop the group — + // there is no meaningful scalar to show — but + // flag the whole response partial. + any_partial = true; + groups_with_no_value += 1; } } } + // Phase 6: build Prometheus `warnings` when the combiner + // returned any Partial. One line summarising the schema + // boundary, plus up-to-three per-segment lines with agg_id + // + clipped range so operators can correlate against the + // `GET /api/v1/db/timeline` surface. We cap at three to + // keep responses bounded; the full set is still inspectable + // via the timeline endpoint. + let warnings = if any_partial { + let mut w = Vec::with_capacity(2 + unresolved.len().min(3)); + w.push(format!( + "partial result: query spans {} schemas for metric '{}' over [{}, {}] and the requested statistic {:?} is not cleanly combinable across schema boundaries — see `GET /api/v1/db/timeline?metric={}&start_ms={}&end_ms={}` for the full segment map", + segments.len(), + metric_name, + t1, + t2, + stat, + metric_name, + t1, + t2, + )); + if groups_with_no_value > 0 { + w.push(format!( + "{} group(s) dropped because no segment could answer the statistic", + groups_with_no_value, + )); + } + for seg in unresolved.iter().take(3) { + w.push(format!( + "segment agg_id={} [{}, {}) status={:?} coverage={:?} unresolved", + seg.agg_id, seg.start_ms, seg.end_ms, seg.status, seg.coverage, + )); + } + if unresolved.len() > 3 { + w.push(format!("... and {} more", unresolved.len() - 3)); + } + w + } else { + Vec::new() + }; + Some(( probe_context.metadata.query_output_labels, - QueryResult::vector(output, probe_context.query_time), + QueryResult::vector_with_warnings(output, probe_context.query_time, warnings), )) }