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 aa80a103..534e1301 100644 --- a/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs +++ b/asap-query-engine/src/drivers/query/adapters/prometheus_http.rs @@ -15,7 +15,17 @@ use std::collections::HashMap; use std::sync::Arc; use tracing::{debug, error}; -/// Prometheus-compatible response structure +/// Prometheus-compatible response structure, with two ASAP +/// extensions: +/// +/// * `infos` — Prometheus 3.0-style informational annotations +/// (Grafana 11+ renders these inline). Mirrors a one-liner +/// summary of `accuracy` so older / non-JSON-aware UIs still +/// see the ε/δ bound. +/// * `accuracy` — structured [`AccuracyEnvelope`] carrying ε, δ, +/// kind, and optional `per_segment` for schema-timeline- +/// crossing queries. Standard Prometheus clients ignore +/// unknown top-level fields, so this is a zero-risk extension. #[derive(Debug, Serialize, Deserialize)] pub struct PrometheusResponse { pub status: String, @@ -31,6 +41,19 @@ pub struct PrometheusResponse { /// Purged segment. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub warnings: Vec, + /// Prometheus 3.0 `infos: []`. Grafana 11+ renders each + /// string inline. Used to mirror a human-readable + /// `accuracy: ε=..., δ=..., kind=...` line when the + /// structured `accuracy` field is present. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub infos: Vec, + /// ASAP extension: theoretical accuracy envelope for the + /// answer (§6.4 of docs/design-sketch-db.md). Unknown to + /// standard Prometheus clients (they ignore unknown fields), + /// consumed by Grafana panels / paper artifacts that want + /// the machine-readable (ε, δ) bound. + #[serde(skip_serializing_if = "Option::is_none")] + pub accuracy: Option, } impl PrometheusResponse { @@ -41,6 +64,8 @@ impl PrometheusResponse { error_type: None, error: None, warnings: Vec::new(), + infos: Vec::new(), + accuracy: None, } } @@ -53,9 +78,21 @@ impl PrometheusResponse { error_type: None, error: None, warnings, + infos: Vec::new(), + accuracy: None, } } + /// Attach an accuracy envelope: sets the structured `accuracy` + /// field and mirrors a human-readable one-liner to `infos`. + /// Chainable so both warning + accuracy paths can decorate + /// the same `success(...)` construction. + pub fn with_accuracy(mut self, envelope: crate::stores::sketch_db::AccuracyEnvelope) -> Self { + self.infos.push(envelope.summary()); + self.accuracy = Some(envelope); + self + } + pub fn error(error_type: &str, error: &str) -> Self { Self { status: "error".to_string(), @@ -63,6 +100,8 @@ impl PrometheusResponse { error_type: Some(error_type.to_string()), error: Some(error.to_string()), warnings: Vec::new(), + infos: Vec::new(), + accuracy: None, } } } @@ -225,11 +264,15 @@ impl QueryResponseAdapter for PrometheusHttpAdapter { // 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() { + let accuracy = result.query_result.accuracy().cloned(); + let mut response = if warnings.is_empty() { PrometheusResponse::success(prometheus_data) } else { PrometheusResponse::success_with_warnings(prometheus_data, warnings) }; + if let Some(envelope) = accuracy { + response = response.with_accuracy(envelope); + } Ok(Json(serde_json::to_value(response).unwrap()).into_response()) } @@ -245,11 +288,15 @@ impl QueryResponseAdapter for PrometheusHttpAdapter { StatusCode::INTERNAL_SERVER_ERROR })?; let warnings = result.warnings().to_vec(); - let response = if warnings.is_empty() { + let accuracy = result.accuracy().cloned(); + let mut response = if warnings.is_empty() { PrometheusResponse::success(prometheus_data) } else { PrometheusResponse::success_with_warnings(prometheus_data, warnings) }; + if let Some(envelope) = accuracy { + response = response.with_accuracy(envelope); + } Ok(Json(serde_json::to_value(response).unwrap()).into_response()) } diff --git a/asap-query-engine/src/engines/query_result.rs b/asap-query-engine/src/engines/query_result.rs index 6c8fc72d..3ca4591a 100644 --- a/asap-query-engine/src/engines/query_result.rs +++ b/asap-query-engine/src/engines/query_result.rs @@ -1,4 +1,5 @@ use crate::data_model::KeyByLabelValues; +use crate::stores::sketch_db::AccuracyEnvelope; use serde::{Deserialize, Serialize}; use promql_utilities::query_logics::enums::QueryResultType; @@ -23,6 +24,7 @@ impl QueryResult { values, timestamp, warnings: Vec::new(), + accuracy: None, }) } @@ -42,6 +44,7 @@ impl QueryResult { values, timestamp, warnings, + accuracy: None, }) } @@ -49,6 +52,7 @@ impl QueryResult { QueryResult::Matrix(RangeVector { values, warnings: Vec::new(), + accuracy: None, }) } @@ -61,6 +65,29 @@ impl QueryResult { QueryResult::Matrix(m) => &m.warnings, } } + + /// Theoretical accuracy envelope for this answer (§6.4 of + /// the sketch-DB design). `None` when the engine couldn't + /// resolve a schema — legacy paths that haven't been wired + /// yet, or fallback-produced responses. + pub fn accuracy(&self) -> Option<&AccuracyEnvelope> { + match self { + QueryResult::Vector(iv) => iv.accuracy.as_ref(), + QueryResult::Matrix(m) => m.accuracy.as_ref(), + } + } + + /// Attach an accuracy envelope. Chainable so engine paths + /// can build the bare result first and decorate once the + /// `agg_id → AggregationConfig → AccuracyProfile` lookup + /// has resolved. + pub fn with_accuracy(mut self, envelope: AccuracyEnvelope) -> Self { + match &mut self { + QueryResult::Vector(iv) => iv.accuracy = Some(envelope), + QueryResult::Matrix(m) => m.accuracy = Some(envelope), + } + self + } } /// Instant vector - a set of time series containing a single sample for each time series, all sharing the same timestamp @@ -77,6 +104,12 @@ pub struct InstantVector { /// missing from the current config). #[serde(default, skip_serializing_if = "Vec::is_empty")] pub warnings: Vec, + /// §6.4 accuracy envelope. Attached by the engine once the + /// `agg_id` is resolved; surfaced as `PrometheusResponse`'s + /// top-level `accuracy` field and mirrored to `infos` for + /// Grafana 11+ inline display. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub accuracy: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -98,6 +131,9 @@ pub struct RangeVector { /// See [`InstantVector::warnings`]. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub warnings: Vec, + /// See [`InstantVector::accuracy`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub accuracy: Option, } /// Individual element in a range vector diff --git a/asap-query-engine/src/engines/simple_engine.rs b/asap-query-engine/src/engines/simple_engine.rs index e6f7b454..3c3ae9d0 100644 --- a/asap-query-engine/src/engines/simple_engine.rs +++ b/asap-query-engine/src/engines/simple_engine.rs @@ -1921,6 +1921,7 @@ impl SimpleEngine { context: QueryExecutionContext, enable_topk: bool, ) -> Option<(KeyByLabelNames, QueryResult)> { + let agg_id = context.agg_info.aggregation_id_for_value; let results = self .execute_query_pipeline(&context, enable_topk) .map_err(|e| { @@ -1928,9 +1929,27 @@ impl SimpleEngine { e }) .ok()?; - Some(( - context.metadata.query_output_labels, - QueryResult::vector(results, context.query_time), + let qr = QueryResult::vector(results, context.query_time); + let qr = match self.accuracy_envelope_for(agg_id) { + Some(env) => qr.with_accuracy(env), + None => qr, + }; + Some((context.metadata.query_output_labels, qr)) + } + + /// Build an [`AccuracyEnvelope`] for a single resolved + /// `agg_id` by looking up the matching `AggregationConfig` in + /// the current streaming-config snapshot and deriving its + /// [`AccuracyProfile`]. Returns `None` when the agg isn't in + /// config (e.g. post-retire / test harness with empty config). + pub(crate) fn accuracy_envelope_for( + &self, + agg_id: u64, + ) -> Option { + let snap = self.streaming_config_snapshot(); + let cfg = snap.get_aggregation_config(agg_id)?; + Some(crate::stores::sketch_db::AccuracyEnvelope::single( + crate::stores::sketch_db::AccuracyProfile::derive(cfg), )) } @@ -3241,10 +3260,30 @@ impl SimpleEngine { Vec::new() }; - Some(( - probe_context.metadata.query_output_labels, - QueryResult::vector_with_warnings(output, probe_context.query_time, warnings), - )) + // §6.4: build a per-segment accuracy envelope from each + // segment's resolved agg_id. Segments that don't resolve + // to an in-config agg drop out — their partial-ness is + // already reflected in `warnings` above. + let snap = self.streaming_config_snapshot(); + let per_segment: Vec = segments + .iter() + .filter_map(|seg| { + let cfg = snap.get_aggregation_config(seg.agg_id)?; + Some(crate::stores::sketch_db::PerSegmentAccuracy { + agg_id: seg.agg_id, + range_ms: [seg.start_ms as i64, seg.end_ms as i64], + profile: crate::stores::sketch_db::AccuracyProfile::derive(cfg), + }) + }) + .collect(); + let envelope = crate::stores::sketch_db::AccuracyEnvelope::from_segments(per_segment); + + let qr = QueryResult::vector_with_warnings(output, probe_context.query_time, warnings); + let qr = match envelope { + Some(e) => qr.with_accuracy(e), + None => qr, + }; + Some((probe_context.metadata.query_output_labels, qr)) } /// Merge precomputed outputs (extracts buckets from timestamped data) diff --git a/asap-query-engine/src/stores/sketch_db/accuracy.rs b/asap-query-engine/src/stores/sketch_db/accuracy.rs index 32f93ca4..23f232e0 100644 --- a/asap-query-engine/src/stores/sketch_db/accuracy.rs +++ b/asap-query-engine/src/stores/sketch_db/accuracy.rs @@ -91,6 +91,24 @@ impl AccuracyProfile { } } + /// Human-readable one-liner. Surfaced in Prometheus-style + /// `infos` arrays so Grafana 11+ shows it inline without a + /// custom panel. + pub fn summary(&self) -> String { + format!( + "accuracy: ε={}, δ={}, kind={}", + self.epsilon, + self.delta, + match self.kind { + AccuracyKind::Exact => "exact", + AccuracyKind::AdditiveFrequency => "additive_frequency", + AccuracyKind::RelativeCardinality => "relative_cardinality", + AccuracyKind::RankQuantile => "rank_quantile", + AccuracyKind::RelativeQuantile => "relative_quantile", + } + ) + } + /// Derive an [`AccuracyProfile`] from a pinned /// [`AggregationConfig`]. Reads `aggregation_type` and any /// necessary entries in `parameters`; falls back to exact for @@ -254,6 +272,93 @@ fn ddsketch_alpha(config: &AggregationConfig) -> f64 { .unwrap_or(0.01) } +/// Per-segment accuracy record. Attached to a multi-segment +/// [`AccuracyEnvelope`] so clients can see the error bound for +/// each piece of the schema-timeline-crossing query. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct PerSegmentAccuracy { + pub agg_id: u64, + /// Half-open millisecond range `[start_ms, end_ms)` this + /// segment covered. + pub range_ms: [i64; 2], + #[serde(flatten)] + pub profile: AccuracyProfile, +} + +/// Wire-side envelope emitted on PromQL responses as the +/// top-level `accuracy` field. Single-schema queries fill +/// `profile`; queries that span a schema-timeline boundary also +/// populate `per_segment` so the caller can see each piece's +/// bound. The top-level `profile` is the worst-case (max ε, +/// max δ) across segments — a conservative upper envelope. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct AccuracyEnvelope { + #[serde(flatten)] + pub profile: AccuracyProfile, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub per_segment: Vec, +} + +impl AccuracyEnvelope { + /// Envelope for a single resolved aggregation. + pub fn single(profile: AccuracyProfile) -> Self { + Self { + profile, + per_segment: Vec::new(), + } + } + + /// Build an envelope from a slice of per-segment tuples. + /// Top-level `profile.epsilon` is `max(segment.epsilon)` and + /// same for δ — the conservative envelope across segments. + /// Returns `None` when the slice is empty. + pub fn from_segments(segs: Vec) -> Option { + if segs.is_empty() { + return None; + } + let mut epsilon = 0.0_f64; + let mut delta = 0.0_f64; + // Pick the "most lossy" kind: any non-Exact wins over + // Exact; if mixed non-Exact kinds span segments we pick + // the first non-Exact and trust the per-segment data for + // the caller's finer needs. + let mut kind = AccuracyKind::Exact; + for s in &segs { + if s.profile.epsilon > epsilon { + epsilon = s.profile.epsilon; + } + if s.profile.delta > delta { + delta = s.profile.delta; + } + if matches!(kind, AccuracyKind::Exact) && !matches!(s.profile.kind, AccuracyKind::Exact) + { + kind = s.profile.kind; + } + } + Some(Self { + profile: AccuracyProfile { + epsilon, + delta, + kind, + }, + per_segment: segs, + }) + } + + /// Summary line suitable for Prometheus `infos`. + pub fn summary(&self) -> String { + if self.per_segment.is_empty() { + self.profile.summary() + } else { + format!( + "{} (worst-case over {} schema-timeline segments)", + self.profile.summary(), + self.per_segment.len() + ) + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/asap-query-engine/src/stores/sketch_db/mod.rs b/asap-query-engine/src/stores/sketch_db/mod.rs index 0adc2e79..4e6db7ca 100644 --- a/asap-query-engine/src/stores/sketch_db/mod.rs +++ b/asap-query-engine/src/stores/sketch_db/mod.rs @@ -41,7 +41,7 @@ pub mod schema; pub mod schema_eviction; pub mod simple_map_store; -pub use accuracy::{AccuracyKind, AccuracyProfile}; +pub use accuracy::{AccuracyEnvelope, AccuracyKind, AccuracyProfile, PerSegmentAccuracy}; pub use backfill::{ BackfillJob, BackfillRegistry, BackfillSource, BackfillStatus, Coverage, CreateError, }; diff --git a/asap-query-engine/src/tests/accuracy_in_promql_response_tests.rs b/asap-query-engine/src/tests/accuracy_in_promql_response_tests.rs new file mode 100644 index 00000000..b0e3f728 --- /dev/null +++ b/asap-query-engine/src/tests/accuracy_in_promql_response_tests.rs @@ -0,0 +1,180 @@ +//! Verify the §6.4 accuracy envelope lands on the Prometheus +//! HTTP response as an ASAP-extension `accuracy` field plus a +//! human-readable `infos` mirror, in the A+C schema agreed in +//! the design discussion. +//! +//! * **A** — top-level `accuracy: { epsilon, delta, kind, per_segment }` +//! * **C** — `infos: ["accuracy: ε=..., δ=..., kind=..."]` +//! +//! Standard PromQL clients ignore both fields; Grafana 11+ +//! renders `infos` inline; custom panels read `accuracy`. + +#[cfg(test)] +use crate::drivers::query::adapters::PrometheusResponse; +use crate::stores::sketch_db::{ + AccuracyEnvelope, AccuracyKind, AccuracyProfile, PerSegmentAccuracy, +}; +use serde_json::Value; + +fn hll_profile() -> AccuracyProfile { + AccuracyProfile { + epsilon: 0.008125, + delta: 0.0, + kind: AccuracyKind::RelativeCardinality, + } +} + +#[test] +fn prometheus_response_carries_accuracy_top_level_and_infos_mirror() { + let envelope = AccuracyEnvelope::single(hll_profile()); + let resp = PrometheusResponse::success(serde_json::json!({ + "resultType": "vector", + "result": [], + })) + .with_accuracy(envelope.clone()); + + let json: Value = serde_json::to_value(&resp).unwrap(); + + // A: structured accuracy top-level. + let accuracy = &json["accuracy"]; + assert!(!accuracy.is_null(), "accuracy field should be present"); + assert_eq!(accuracy["epsilon"], 0.008125); + assert_eq!(accuracy["delta"], 0.0); + assert_eq!(accuracy["kind"], "relative_cardinality"); + assert!( + accuracy + .get("per_segment") + .map(|v| v.as_array().unwrap().is_empty()) + .unwrap_or(true), + "per_segment must be absent / empty for single-schema queries" + ); + + // C: infos mirror. + let infos = json["infos"].as_array().expect("infos should be an array"); + assert_eq!(infos.len(), 1); + let line = infos[0].as_str().unwrap(); + assert!( + line.contains("ε=0.008125") + && line.contains("δ=0") + && line.contains("relative_cardinality"), + "infos summary must include ε, δ, kind: got {line}" + ); +} + +#[test] +fn prometheus_response_without_accuracy_skips_both_fields() { + let resp = PrometheusResponse::success(serde_json::json!({ + "resultType": "vector", + "result": [], + })); + let json = serde_json::to_string(&resp).unwrap(); + // Both top-level extensions must be absent so the wire shape + // stays byte-identical to standard Prometheus when accuracy + // isn't populated (e.g. fallback-only responses). + assert!( + !json.contains("\"accuracy\""), + "accuracy must be omitted: {json}" + ); + assert!(!json.contains("\"infos\""), "infos must be omitted: {json}"); +} + +#[test] +fn prometheus_response_per_segment_contains_all_segments_with_worst_case_top() { + let segments = vec![ + PerSegmentAccuracy { + agg_id: 1, + range_ms: [1_000, 2_000], + profile: AccuracyProfile { + epsilon: 0.01, + delta: 0.0, + kind: AccuracyKind::RelativeQuantile, + }, + }, + PerSegmentAccuracy { + agg_id: 2, + range_ms: [2_000, 3_000], + profile: AccuracyProfile { + epsilon: 0.05, + delta: 0.01, + kind: AccuracyKind::RankQuantile, + }, + }, + ]; + let envelope = AccuracyEnvelope::from_segments(segments).unwrap(); + // Worst-case envelope = (max ε=0.05, max δ=0.01, first + // non-Exact kind wins for `kind`). + assert_eq!(envelope.profile.epsilon, 0.05); + assert_eq!(envelope.profile.delta, 0.01); + assert_eq!(envelope.profile.kind, AccuracyKind::RelativeQuantile); + + let resp = PrometheusResponse::success(serde_json::json!({ + "resultType": "vector", + "result": [], + })) + .with_accuracy(envelope); + let json: Value = serde_json::to_value(&resp).unwrap(); + + assert_eq!(json["accuracy"]["epsilon"], 0.05); + let per_seg = json["accuracy"]["per_segment"].as_array().unwrap(); + assert_eq!(per_seg.len(), 2); + assert_eq!(per_seg[0]["agg_id"], 1); + assert_eq!(per_seg[0]["range_ms"], serde_json::json!([1_000, 2_000])); + assert_eq!(per_seg[1]["agg_id"], 2); + + // `infos` summary flags the multi-segment shape. + let infos = json["infos"].as_array().unwrap(); + assert!(infos[0] + .as_str() + .unwrap() + .contains("schema-timeline segments")); +} + +#[test] +fn accuracy_coexists_with_warnings_without_interference() { + // The wire contract: `warnings` stays for partial-result + // advisories, `accuracy` is orthogonal. Both must coexist. + let envelope = AccuracyEnvelope::single(hll_profile()); + let resp = PrometheusResponse::success_with_warnings( + serde_json::json!({ + "resultType": "vector", + "result": [], + }), + vec!["partial: 2 schemas".to_string()], + ) + .with_accuracy(envelope); + let json: Value = serde_json::to_value(&resp).unwrap(); + assert_eq!( + json["warnings"].as_array().unwrap()[0].as_str().unwrap(), + "partial: 2 schemas" + ); + assert_eq!(json["accuracy"]["kind"], "relative_cardinality"); + assert_eq!( + json["infos"].as_array().unwrap().len(), + 1, + "infos should carry only the accuracy summary, not the warning" + ); +} + +#[test] +fn promql_standard_client_can_decode_response_ignoring_extensions() { + // A minimal "standard Prometheus" response decoder — just + // the fields the upstream API defines. It must parse our + // extended response without error (unknown fields + // ignored). This locks down the "zero-risk extension" claim. + let envelope = AccuracyEnvelope::single(hll_profile()); + let resp = PrometheusResponse::success(serde_json::json!({ + "resultType": "vector", + "result": [], + })) + .with_accuracy(envelope); + let wire = serde_json::to_string(&resp).unwrap(); + + #[derive(serde::Deserialize)] + struct StandardPromResponse { + status: String, + #[serde(default)] + _data: Option, + } + let standard: StandardPromResponse = serde_json::from_str(&wire).unwrap(); + assert_eq!(standard.status, "success"); +} diff --git a/asap-query-engine/src/tests/mod.rs b/asap-query-engine/src/tests/mod.rs index c2bf7b84..960fd4bb 100644 --- a/asap-query-engine/src/tests/mod.rs +++ b/asap-query-engine/src/tests/mod.rs @@ -1,3 +1,4 @@ +pub mod accuracy_in_promql_response_tests; pub mod capability_matching_tests; pub mod capability_miss_http_e2e_tests; pub mod clickhouse_forwarding_tests;