From e3905a8efeaf4a11fad77b2697268807746f2dae Mon Sep 17 00:00:00 2001 From: zzylol Date: Sun, 18 Oct 2026 17:44:06 +0000 Subject: [PATCH 1/2] feat(query): answer planned range queries one evaluation per step A planned /api/v1/query_range runs one evaluation per step, start + k*step through end, and returns a Prometheus matrix. Each query-time raw range is fetched from Prometheus once for all steps and sliced per step; each kept definition's missing panes are fetched once, so a kept pane built at one step is reused by the later ones. Stored panes are read per step. If any step cannot be answered, the whole request is forwarded with the reason. start/end accept Unix seconds or RFC 3339 (now also for instant queries), step float seconds or a Prometheus duration; requests Prometheus would reject, and requests with a nonzero limit, are forwarded unchanged. --- Cargo.lock | 1 + README.md | 12 +- data_plane/Cargo.toml | 1 + data_plane/src/engine.rs | 68 ++++++ data_plane/src/lib.rs | 98 ++++++--- data_plane/src/prometheus.rs | 255 ++++++++++++++++++++++- data_plane/src/query.rs | 148 +++++++++++++ data_plane/tests/answers.rs | 155 +++++++++++++- data_plane/tests/e2e_examples.rs | 218 +++++++++++++++++++- data_plane/tests/range.rs | 344 +++++++++++++++++++++++++++++++ 10 files changed, 1250 insertions(+), 50 deletions(-) create mode 100644 data_plane/tests/range.rs diff --git a/Cargo.lock b/Cargo.lock index 7365b834..83ad041b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -474,6 +474,7 @@ dependencies = [ "asap-executor", "asap-types", "axum", + "chrono", "clap", "control_plane", "form_urlencoded", diff --git a/README.md b/README.md index 9ebd9ae3..186be56a 100644 --- a/README.md +++ b/README.md @@ -53,8 +53,16 @@ sample minus `--allowed-lateness-ms`), and later evaluations reuse it; a pane read before it closes is computed again by later evaluations. Kept panes are evicted like stored ones, measured from the newest evaluation. Anything else is forwarded to Prometheus (`x-asap-execution: forwarded`); a planned query that could not -be answered also says why in `x-asap-forward-reason`. Range queries, SQL and -ClickHouse are not supported yet. +be answered also says why in `x-asap-forward-reason`. + +A range query (`/api/v1/query_range`) whose text is one of the plan's queries +is answered as one evaluation per step, `start + k * step` through `end`, +and returns a Prometheus matrix. Each query-time raw range is read from +Prometheus once for all steps and sliced per step; stored and kept panes are +read per step, so a kept pane built at one step is reused by the later ones. +If any step cannot be answered, for example because a step is off the pane +grid or reads an evicted pane, the whole range query is forwarded with the +reason. SQL and ClickHouse are not supported yet. ## Running diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 1d7eca3f..74630e2f 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -22,6 +22,7 @@ prost = "0.13" snap = "1.1" futures = "0.3" form_urlencoded = "1" +chrono = { version = "0.4", default-features = false, features = ["alloc"] } [dev-dependencies] tower = { version = "0.5", features = ["util"] } diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index e1e173c3..f0f140e6 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -32,6 +32,17 @@ pub enum Command { raw: BTreeMap>, reply: Reply>, }, + RawRangesForSteps { + query: String, + steps: Vec, + reply: Reply<(usize, Vec)>, + }, + EvaluateSteps { + root: usize, + steps: Vec, + raw: BTreeMap>, + reply: Reply>>, + }, } #[derive(Debug, thiserror::Error)] @@ -104,6 +115,37 @@ impl EngineHandle { .await } + /// The plan root answering `query` at every one of `steps` and the raw + /// samples they need; see [`QueryEngine::raw_ranges_for_steps`]. + pub async fn raw_ranges_for_steps( + &self, + query: String, + steps: Vec, + ) -> Result<(usize, Vec), Unanswered> { + self.ask(|reply| Command::RawRangesForSteps { + query, + steps, + reply, + }) + .await + } + + /// Runs a root at every one of `steps`; see [`QueryEngine::evaluate_steps`]. + pub async fn evaluate_steps( + &self, + root: usize, + steps: Vec, + raw: BTreeMap>, + ) -> Result>, Unanswered> { + self.ask(|reply| Command::EvaluateSteps { + root, + steps, + raw, + reply, + }) + .await + } + async fn ask(&self, command: impl FnOnce(Reply) -> Command) -> Result { let busy = || Unanswered::Execution("the engine is busy".into()); let (reply, answer) = oneshot::channel(); @@ -178,6 +220,32 @@ impl Engine { &raw, )); } + Command::RawRangesForSteps { + query, + steps, + reply, + } => { + let _ = reply.send(self.query.raw_ranges_for_steps( + &query, + &steps, + &self.precompute.store, + )); + } + Command::EvaluateSteps { + root, + steps, + raw, + reply, + } => { + // One watermark for every step of the request. + let _ = reply.send(self.query.evaluate_steps( + root, + &steps, + &self.precompute.store, + self.precompute.watermark_ms(), + &raw, + )); + } } } } diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 0223c40a..baaad848 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -3,8 +3,8 @@ //! //! The installed plan is ASAPPlanner's selected `PhysicalASAPDAG`. The //! engine thread builds its ingestion-time outputs pane by pane and answers -//! the plan's instant queries from them; every other query, and any planned -//! query it cannot answer, is forwarded to Prometheus. +//! the plan's instant and range queries from them; every other query, and +//! any planned query it cannot answer, is forwarded to Prometheus. pub mod engine; pub mod precompute; @@ -26,7 +26,7 @@ use axum::Router; use control_plane::InstalledPlan; use engine::EngineHandle; use precompute::RawSample; -use prometheus::InstantQuery; +use prometheus::{InstantQuery, RangeQuery}; use query::Unanswered; /// Prometheus' staleness marker: a NaN that ends a series, not a sample. @@ -125,24 +125,57 @@ async fn query( headers: HeaderMap, body: Bytes, ) -> Response { - let Some(instant) = InstantQuery::parse(query.0.as_deref(), &body) else { - return forward(state, "query", method, query, headers, body).await; + let answered = match InstantQuery::parse(query.0.as_deref(), &body) { + Some(instant) => Some(answer(&state, instant).await), + None => None, }; - let reason = match answer(&state, instant).await { - Ok(response) => { + respond(answered, state, "query", method, query, headers, body).await +} + +/// Answers a planned range query from the plan, one evaluation per step; +/// forwards anything else, and the whole range query if any step cannot +/// be answered. +async fn query_range( + state: State>, + method: Method, + query: RawQuery, + headers: HeaderMap, + body: Bytes, +) -> Response { + let answered = match RangeQuery::parse(query.0.as_deref(), &body) { + Some(range) => Some(answer_range(&state, range).await), + None => None, + }; + respond(answered, state, "query_range", method, query, headers, body).await +} + +/// The plan's answer with `x-asap-execution: plan`, or else the request +/// forwarded to `endpoint`, with the reason if the query is planned. +/// `answered` is `None` for a request the data plane does not parse. +async fn respond( + answered: Option>, + state: State>, + endpoint: &str, + method: Method, + query: RawQuery, + headers: HeaderMap, + body: Bytes, +) -> Response { + let reason = match answered { + Some(Ok(response)) => { let mut response = axum::Json(response).into_response(); response .headers_mut() .insert(EXECUTION_HEADER, HeaderValue::from_static("plan")); return response; } - Err(Unanswered::NotPlanned(_)) => None, - Err(reason) => { - tracing::debug!(%reason, "forwarding a planned query"); + None | Some(Err(Unanswered::NotPlanned(_))) => None, + Some(Err(reason)) => { + tracing::debug!(%reason, endpoint, "forwarding a planned query"); HeaderValue::from_str(&reason.to_string()).ok() } }; - let mut response = forward(state, "query", method, query, headers, body).await; + let mut response = forward(state, endpoint, method, query, headers, body).await; if let Some(reason) = reason { response.headers_mut().insert(FORWARD_REASON_HEADER, reason); } @@ -156,25 +189,42 @@ async fn answer(state: &DataPlane, query: InstantQuery) -> Result Result { + let steps = query.steps(); + let (root, ranges) = state + .engine + .raw_ranges_for_steps(query.query, steps.clone()) + .await?; + let raw = fetch(state, &ranges).await?; + let results = state + .engine + .evaluate_steps(root, steps.clone(), raw) + .await?; + Ok(prometheus::matrix_response(&steps, &results)) +} + +/// The samples of each of `ranges` from Prometheus, by node. +async fn fetch( + state: &DataPlane, + ranges: &[query::RawRange], +) -> Result>, Unanswered> { let mut raw = BTreeMap::new(); - for range in &ranges { + for range in ranges { let samples = prometheus::fetch(&state.client, &state.prometheus_url, range) .await .map_err(|e| Unanswered::Execution(format!("fetching raw samples: {e}")))?; raw.insert(range.node, samples); } - let samples = state.engine.evaluate(root, time_ms, raw).await?; - Ok(prometheus::vector_response(time_ms, &samples)) -} - -async fn query_range( - state: State>, - method: Method, - query: RawQuery, - headers: HeaderMap, - body: Bytes, -) -> Response { - forward(state, "query_range", method, query, headers, body).await + Ok(raw) } /// Sends the request to Prometheus unchanged and returns its response. diff --git a/data_plane/src/prometheus.rs b/data_plane/src/prometheus.rs index 08e753d8..f666ddb1 100644 --- a/data_plane/src/prometheus.rs +++ b/data_plane/src/prometheus.rs @@ -1,5 +1,6 @@ -//! The Prometheus HTTP API as the data plane speaks it: instant-query -//! parameters and results, and fetching raw samples for query-time ranges. +//! The Prometheus HTTP API as the data plane speaks it: instant- and +//! range-query parameters and results, and fetching raw samples for +//! query-time ranges. use std::collections::BTreeMap; use std::sync::Arc; @@ -17,17 +18,13 @@ pub struct InstantQuery { } impl InstantQuery { - /// Reads `query` and `time` (Unix seconds); `None` if either is missing - /// or `time` is in another format, such as RFC 3339. + /// Reads `query` and `time` (Unix seconds or RFC 3339); `None` if + /// `query` is missing or `time` is invalid. pub fn parse(url_query: Option<&str>, form: &[u8]) -> Option { - let params: BTreeMap = - form_urlencoded::parse(url_query.unwrap_or_default().as_bytes()) - .chain(form_urlencoded::parse(form)) - .map(|(k, v)| (k.into_owned(), v.into_owned())) - .collect(); + let params = params(url_query, form)?; let time_ms = match params.get("time") { None => None, - Some(time) => Some(seconds_to_ms(time.parse().ok()?)), + Some(time) => Some(parse_time(time)?), }; Some(Self { query: params.get("query")?.clone(), @@ -36,6 +33,111 @@ impl InstantQuery { } } +/// Parameters of a range query (`/api/v1/query_range`), from its URL query +/// and form body. +pub struct RangeQuery { + pub query: String, + /// First evaluation time, inclusive. + pub start_ms: i64, + /// Last possible evaluation time, inclusive. + pub end_ms: i64, + /// Time between consecutive evaluations, at least 1 ms. + pub step_ms: i64, +} + +impl RangeQuery { + /// Reads `query`, `start`, `end` and `step` as Prometheus does; `None` if + /// one is missing or invalid, which Prometheus then reports. + pub fn parse(url_query: Option<&str>, form: &[u8]) -> Option { + let params = params(url_query, form)?; + let start_ms = parse_time(params.get("start")?)?; + let end_ms = parse_time(params.get("end")?)?; + let step_ms = parse_duration_ms(params.get("step")?)?; + if end_ms < start_ms || step_ms <= 0 || (end_ms - start_ms) / step_ms > MAX_STEPS { + return None; + } + Some(Self { + query: params.get("query")?.clone(), + start_ms, + end_ms, + step_ms, + }) + } + + /// The evaluation times `start + k * step` up to `end`, ascending. + pub fn steps(&self) -> Vec { + (self.start_ms..=self.end_ms) + .step_by(self.step_ms as usize) + .collect() + } +} + +/// Prometheus' limit on `(end - start) / step`: 11,000 points per series. +const MAX_STEPS: i64 = 11_000; + +/// The request's parameters, from its URL query and form body; `None` if it +/// sets a nonzero `limit`, which the data plane leaves to Prometheus. +fn params(url_query: Option<&str>, form: &[u8]) -> Option> { + let params: BTreeMap = + form_urlencoded::parse(url_query.unwrap_or_default().as_bytes()) + .chain(form_urlencoded::parse(form)) + .map(|(k, v)| (k.into_owned(), v.into_owned())) + .collect(); + match params.get("limit").map(String::as_str) { + None | Some("0") => Some(params), + Some(_) => None, + } +} + +/// A Prometheus timestamp, Unix seconds or RFC 3339, in ms. +fn parse_time(time: &str) -> Option { + if let Ok(seconds) = time.parse::() { + return seconds.is_finite().then(|| seconds_to_ms(seconds)); + } + chrono::DateTime::parse_from_rfc3339(time) + .ok() + .map(|t| t.timestamp_millis()) +} + +/// A Prometheus ``: float seconds, or units `y w d h m s +/// ms` in that order, each at most once (`1h30m`); in ms, truncated as +/// Prometheus does. +fn parse_duration_ms(duration: &str) -> Option { + if let Ok(seconds) = duration.parse::() { + let ns = seconds * 1e9; + return (ns.is_finite() && ns.abs() < i64::MAX as f64).then_some(ns as i64 / 1_000_000); + } + // In the order a duration lists them. + const UNITS: [(&str, i64); 7] = [ + ("y", 365 * 86_400_000), + ("w", 7 * 86_400_000), + ("d", 86_400_000), + ("h", 3_600_000), + ("m", 60_000), + ("s", 1000), + ("ms", 1), + ]; + let (mut rest, mut total, mut last) = (duration, 0i64, None); + while !rest.is_empty() { + let digits = rest.len() - rest.trim_start_matches(|c: char| c.is_ascii_digit()).len(); + let count: i64 = rest[..digits].parse().ok()?; + rest = &rest[digits..]; + // The longest unit that matches: "ms" rather than "m". + let (index, (unit, ms)) = UNITS + .iter() + .enumerate() + .filter(|(_, (unit, _))| rest.starts_with(unit)) + .max_by_key(|(_, (unit, _))| unit.len())?; + if last.is_some_and(|last| last >= index) { + return None; + } + last = Some(index); + rest = &rest[unit.len()..]; + total = total.checked_add(count.checked_mul(*ms)?)?; + } + last.map(|_| total) +} + fn seconds_to_ms(seconds: f64) -> i64 { (seconds * 1000.0).round() as i64 } @@ -54,6 +156,26 @@ pub fn vector_response(time_ms: i64, samples: &[ResultSample]) -> Value { json!({"status": "success", "data": {"resultType": "vector", "result": result}}) } +/// A Prometheus range-query response: each series' value at each of +/// `steps` (`results[k]` at `steps[k]`), series sorted by labels. +pub fn matrix_response(steps: &[i64], results: &[Vec]) -> Value { + let mut series: BTreeMap<&BTreeMap, Vec> = BTreeMap::new(); + for (time_ms, samples) in steps.iter().zip(results) { + let time = *time_ms as f64 / 1000.0; + for s in samples { + series + .entry(&s.labels) + .or_default() + .push(json!([time, format_value(s.value)])); + } + } + let result: Vec = series + .into_iter() + .map(|(labels, values)| json!({"metric": labels, "values": values})) + .collect(); + json!({"status": "success", "data": {"resultType": "matrix", "result": result}}) +} + /// Prometheus' sample value format. fn format_value(value: f64) -> String { match value { @@ -134,7 +256,13 @@ mod tests { let q = InstantQuery::parse(None, b"query=sum%28x%29").unwrap(); assert_eq!((q.query.as_str(), q.time_ms), ("sum(x)", None)); assert!(InstantQuery::parse(Some("time=1"), b"").is_none()); - assert!(InstantQuery::parse(Some("query=up&time=2026-01-01T00:00:00Z"), b"").is_none()); + let q = InstantQuery::parse(Some("query=up&time=2026-01-01T00:00:00Z"), b"").unwrap(); + assert_eq!( + q.time_ms, + Some(1_767_225_600_000), + "RFC 3339, as Prometheus accepts" + ); + assert!(InstantQuery::parse(Some("query=up&time=now"), b"").is_none()); } /// Results use Prometheus' vector shape and value formatting. @@ -159,4 +287,109 @@ mod tests { ); assert_eq!(ms_to_seconds(4_200_001), "4200.001"); } + + fn range(params: &str) -> Option<(i64, i64, i64)> { + RangeQuery::parse(Some(params), b"").map(|q| (q.start_ms, q.end_ms, q.step_ms)) + } + + /// `start` and `end` are Unix seconds or RFC 3339; `step` is float + /// seconds or a Prometheus duration; parameters may come in the body. + #[test] + fn parses_range_query_parameters() { + assert_eq!( + range("query=up&start=1.5&end=61.5&step=15"), + Some((1500, 61_500, 15_000)) + ); + assert_eq!( + range("query=up&start=1&end=2&step=0.25"), + Some((1000, 2000, 250)) + ); + assert_eq!( + range("query=up&start=2026-01-01T00:00:00Z&end=2026-01-01T00:01:00.5%2B00:00&step=1m"), + Some((1_767_225_600_000, 1_767_225_660_500, 60_000)) + ); + assert_eq!( + range("query=up&start=0&end=0&step=1h30m"), + Some((0, 0, 90 * 60_000)) + ); + for (step, ms) in [ + ("500ms", 500), + ("1d", 86_400_000), + ("1w", 7 * 86_400_000), + ("1y", 365 * 86_400_000), + ("2m30s", 150_000), + ] { + assert_eq!( + range(&format!("query=up&start=0&end=0&step={step}")), + Some((0, 0, ms)), + "{step}" + ); + } + let q = RangeQuery::parse(None, b"query=sum%28x%29&start=0&end=60&step=30").unwrap(); + assert_eq!(q.query, "sum(x)"); + } + + /// What Prometheus rejects is not parsed: missing parameters, `end` + /// before `start`, a step that is not positive or not a duration, and + /// more than 11,000 steps. + #[test] + fn rejects_invalid_range_queries() { + for params in [ + "start=0&end=60&step=15", + "query=up&end=60&step=15", + "query=up&start=0&step=15", + "query=up&start=0&end=60", + "query=up&start=60&end=0&step=15", + "query=up&start=0&end=60&step=0", + "query=up&start=0&end=60&step=-1", + "query=up&start=0&end=60&step=1m1h", + "query=up&start=0&end=60&step=15x", + "query=up&start=yesterday&end=60&step=15", + "query=up&start=0&end=11001&step=1", + ] { + assert_eq!(range(params), None, "{params}"); + } + assert!(range("query=up&start=0&end=11000&step=1").is_some()); + } + + /// Steps run from `start` by `step` and include `end` when on a step. + #[test] + fn steps_include_the_end() { + let q = |end_ms| RangeQuery { + query: "up".into(), + start_ms: 1000, + end_ms, + step_ms: 500, + }; + assert_eq!(q(2000).steps(), vec![1000, 1500, 2000]); + assert_eq!(q(1999).steps(), vec![1000, 1500]); + assert_eq!(q(1000).steps(), vec![1000]); + } + + /// Range results use Prometheus' matrix shape: one entry per series + /// with its values in time order; a series absent at a step has no + /// value there, and series absent at every step are left out. + #[test] + fn formats_matrix_results() { + let sample = |job: &str, value| ResultSample { + labels: BTreeMap::from([("job".into(), job.into())]), + value, + }; + let results = vec![ + vec![sample("b", 1.0), sample("a", 2.0)], + vec![sample("b", 3.0)], + vec![], + ]; + assert_eq!( + matrix_response(&[1500, 2000, 2500], &results), + json!({"status": "success", "data": {"resultType": "matrix", "result": [ + {"metric": {"job": "a"}, "values": [[1.5, "2"]]}, + {"metric": {"job": "b"}, "values": [[1.5, "1"], [2.0, "3"]]}, + ]}}) + ); + assert_eq!( + matrix_response(&[1500], &[vec![]]), + json!({"status": "success", "data": {"resultType": "matrix", "result": []}}) + ); + } } diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index abfb539c..7786e2f9 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -14,6 +14,11 @@ //! bound to the samples in `(T - offset - range, T - offset]`, which the //! caller fetches, filtered by the plan's `Scan`. //! +//! A range query is one evaluation per step, in time order. Each raw range +//! is fetched once for all steps and sliced per step, and each kept +//! definition's missing panes once, so a kept pane built at one step is +//! reused by the later ones. +//! //! A query the engine cannot answer returns [`Unanswered`]; the caller then //! forwards it to Prometheus. @@ -277,6 +282,127 @@ impl QueryEngine { result_samples(&program.output, rows) } + /// The root answering `query` at each of `steps` (ascending) and the + /// raw samples they need, each fetched once for all steps: per + /// query-time raw range, `(first step - offset - range, last step - + /// offset]`; per kept definition, the span of its panes no evaluation + /// has kept. Fails if the query is not planned or a stored pane any step + /// reads is missing. + pub fn raw_ranges_for_steps( + &self, + query: &str, + steps: &[i64], + store: &SummaryStore, + ) -> Result<(usize, Vec), Unanswered> { + let (root, program) = self.program(query)?; + let (Some(&first), Some(&last)) = (steps.first(), steps.last()) else { + return Ok((root, Vec::new())); + }; + for &time_ms in steps { + for input in &program.stored { + pane(input, time_ms, store)?; + } + } + let mut ranges: Vec = program + .raw + .iter() + .map(|input| RawRange { + node: input.node, + metric: input.scan.metric.clone(), + start_ms: first - input.offset_ms - input.range_ms, + end_ms: last - input.offset_ms, + }) + .collect(); + // The first and last end of each kept definition's missing panes. + let mut missing: BTreeMap<&DefinitionId, (i64, i64)> = BTreeMap::new(); + for &time_ms in steps { + for input in &program.kept { + let end_ms = time_ms - input.offset_ms; + if self.kept.get(&input.definition, end_ms).is_none() { + let span = missing.entry(&input.definition).or_insert((end_ms, end_ms)); + *span = (span.0.min(end_ms), span.1.max(end_ms)); + } + } + } + for (definition, (first_end_ms, last_end_ms)) in missing { + let build = self.kept_build(definition); + ranges.push(RawRange { + node: build.definition.raw_input, + metric: build.scan.metric.clone(), + start_ms: first_end_ms - build.definition.pane_ms, + end_ms: last_end_ms, + }); + } + Ok((root, ranges)) + } + + /// Runs `root` at each of `steps`, in order, as [`Self::evaluate`] does, + /// each over its slice of `raw` (the samples of each + /// [`Self::raw_ranges_for_steps`] range, by node). A kept pane built at + /// one step is reused by the later ones. `results[k]` answers + /// `steps[k]`; any step that fails fails the whole call. Every step + /// keeps a pane only if it is closed at `watermark_ms`, so a pane still + /// open is recomputed at each step that reads it. + pub fn evaluate_steps( + &mut self, + root: usize, + steps: &[i64], + store: &SummaryStore, + watermark_ms: Option, + raw: &BTreeMap>, + ) -> Result>, Unanswered> { + let program = self.roots[root] + .1 + .as_ref() + .map_err(|reason| Unanswered::Unsupported(reason.clone()))?; + // Each query-time raw range: its node, offset and range. + let raw_inputs: Vec<(u64, i64, i64)> = program + .raw + .iter() + .map(|input| (input.node, input.offset_ms, input.range_ms)) + .collect(); + // Each kept output: its definition and offset, the node its pane's + // samples feed, and the node its definition's samples were fetched for. + let kept_inputs: Vec<(DefinitionId, i64, u64, u64, i64)> = program + .kept + .iter() + .map(|input| { + let definition = &self.kept_build(&input.definition).definition; + ( + input.definition.clone(), + input.offset_ms, + input.raw_input, + definition.raw_input, + definition.pane_ms, + ) + }) + .collect(); + let series: BTreeMap> = raw + .iter() + .map(|(node, samples)| (*node, series_runs(samples))) + .collect(); + let mut results = Vec::with_capacity(steps.len()); + for &time_ms in steps { + let mut step_raw = BTreeMap::new(); + for &(node, offset_ms, range_ms) in &raw_inputs { + if let Some(runs) = series.get(&node) { + let end_ms = time_ms - offset_ms; + step_raw.insert(node, slice(runs, end_ms - range_ms, end_ms)); + } + } + for (definition, offset_ms, node, fetched, pane_ms) in &kept_inputs { + let end_ms = time_ms - offset_ms; + if self.kept.get(definition, end_ms).is_none() { + if let Some(runs) = series.get(fetched) { + step_raw.insert(*node, slice(runs, end_ms - pane_ms, end_ms)); + } + } + } + results.push(self.evaluate(root, time_ms, store, watermark_ms, &step_raw)?); + } + Ok(results) + } + /// The build of a kept definition a compiled root reads; roots whose /// kept builds failed are not compiled. fn kept_build(&self, definition: &DefinitionId) -> &PaneBuild { @@ -299,6 +425,28 @@ impl QueryEngine { } } +/// `samples` split into runs of one series in time order, as Prometheus +/// returns them. +fn series_runs(samples: &[RawSample]) -> Vec<&[RawSample]> { + samples + .chunk_by(|a, b| { + (Arc::ptr_eq(&a.labels, &b.labels) || a.labels == b.labels) + && a.timestamp_ms <= b.timestamp_ms + }) + .collect() +} + +/// The samples of `runs` in `(start_ms, end_ms]`, in their order. +fn slice(runs: &[&[RawSample]], start_ms: i64, end_ms: i64) -> Vec { + let mut samples = Vec::new(); + for run in runs { + let from = run.partition_point(|s| s.timestamp_ms <= start_ms); + let to = run.partition_point(|s| s.timestamp_ms <= end_ms); + samples.extend_from_slice(&run[from..to]); + } + samples +} + /// The stored pane `input` reads at `time_ms`. fn pane<'a>( input: &StoredInput, diff --git a/data_plane/tests/answers.rs b/data_plane/tests/answers.rs index c08ec7ba..d8d30fff 100644 --- a/data_plane/tests/answers.rs +++ b/data_plane/tests/answers.rs @@ -31,7 +31,8 @@ fn series(metric: &str, labels: &[(&str, &str)]) -> BTreeMap { /// A stand-in Prometheus holding `points`. It answers /// `{__name__="m"}[Nms]` at `time` with the samples in `[time - N, time]`, /// both ends included as Prometheus 2 does, and anything else with -/// `{"forwarded": query}`. +/// `{"forwarded": query}`. Range queries are answered with +/// `{"forwarded_range": parameters}`. async fn fake_prometheus(points: Vec) -> String { let points = Arc::new(points); let handler = move |Query(params): Query>| { @@ -64,7 +65,12 @@ async fn fake_prometheus(points: Vec) -> String { Json(json!({"status": "success", "data": {"resultType": "matrix", "result": result}})) } }; - let app = Router::new().route("/api/v1/query", get(handler)); + let range = |Query(params): Query>| async move { + Json(json!({ "forwarded_range": params })) + }; + let app = Router::new() + .route("/api/v1/query", get(handler)) + .route("/api/v1/query_range", get(range)); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = format!("http://{}", listener.local_addr().unwrap()); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); @@ -78,11 +84,20 @@ struct Response { } async fn instant(app: &Router, query: &str, time_ms: i64) -> Response { + let time = (time_ms as f64 / 1000.0).to_string(); + get_json(app, "query", &[("query", query), ("time", &time)]).await +} + +/// `GET /api/v1/query_range` with `params`, as given. +async fn range(app: &Router, params: &[(&str, &str)]) -> Response { + get_json(app, "query_range", params).await +} + +async fn get_json(app: &Router, endpoint: &str, params: &[(&str, &str)]) -> Response { let uri = format!( - "/api/v1/query?{}", + "/api/v1/{endpoint}?{}", form_urlencoded::Serializer::new(String::new()) - .append_pair("query", query) - .append_pair("time", &(time_ms as f64 / 1000.0).to_string()) + .extend_pairs(params) .finish() ); let response = app @@ -193,10 +208,9 @@ async fn answers_query_time_plans_from_raw_samples() { assert_eq!(response.body, json!({"forwarded": "up"})); } -/// Example 4b reads six stored 10-minute KLL panes built from Remote Write; -/// at a time off the pane grid it forwards and says why. -#[tokio::test] -async fn answers_from_panes_built_from_remote_write() { +/// Example 4b's data plane, with Remote Write samples every 15 s in +/// `[1 min, end)` for jobs a and b. +async fn example4b(end_ms: i64) -> (Arc, Router) { let state = Arc::new( DataPlane::new( common::plan_json("example4b", false), @@ -220,7 +234,7 @@ async fn answers_from_panes_built_from_remote_write() { value: job.into(), }, ], - samples: (MINUTE..75 * MINUTE) + samples: (MINUTE..end_ms) .step_by(15_000) .map(|t| Sample { timestamp: t, @@ -239,6 +253,16 @@ async fn answers_from_panes_built_from_remote_write() { .await .unwrap(); assert_eq!(write.status(), StatusCode::NO_CONTENT); + // Wait until the engine has ingested the samples. + state.engine.stats().await; + (state, app) +} + +/// Example 4b reads six stored 10-minute KLL panes built from Remote Write; +/// at a time off the pane grid it forwards and says why. +#[tokio::test] +async fn answers_from_panes_built_from_remote_write() { + let (state, app) = example4b(75 * MINUTE).await; // Panes ending at 20..70 min; the first, (0, 10 min], is skipped. assert_eq!(state.engine.stats().await.panes_built, 6); @@ -263,3 +287,114 @@ async fn answers_from_panes_built_from_remote_write() { assert_eq!(response.execution.as_deref(), Some("forwarded")); assert!(response.reason.unwrap().contains("pane grid")); } + +const P99: &str = "quantile_over_time(0.99, latency_ms[60m])"; + +/// A planned range query is answered with a Prometheus matrix, from `start` +/// to `end` inclusive by `step` (a duration or float seconds); each value +/// equals the instant query at its step. +#[tokio::test] +async fn answers_planned_range_queries() { + let (_, app) = example4b(101 * MINUTE).await; + for step in ["10m", "600", "600.0"] { + let response = range( + &app, + &[ + ("query", P99), + ("start", "5400"), + ("end", "6000"), + ("step", step), + ], + ) + .await; + assert_eq!( + response.execution.as_deref(), + Some("plan"), + "{step}: {:?}", + response.reason + ); + assert_eq!(response.body["status"], "success"); + assert_eq!(response.body["data"]["resultType"], "matrix"); + let result = response.body["data"]["result"].as_array().unwrap(); + assert_eq!(result.len(), 2, "{result:?}"); + for series in result { + let job = series["metric"]["job"].as_str().unwrap(); + let values = series["values"].as_array().unwrap(); + let times: Vec = values.iter().map(|v| v[0].as_f64().unwrap()).collect(); + assert_eq!(times, vec![5400.0, 6000.0], "the end is included"); + for value in values { + let t = (value[0].as_f64().unwrap() * 1000.0) as i64; + let at_step = instant(&app, P99, t).await; + assert_eq!(at_step.execution.as_deref(), Some("plan")); + let expected = by_job(&at_step.body) + .into_iter() + .find(|(j, _)| j == job) + .unwrap() + .1; + assert_eq!(value[1].as_str().unwrap().parse::().unwrap(), expected); + } + } + } +} + +/// A planned range query any of whose steps the plan cannot answer is +/// forwarded whole, with the reason: a step off the 10-minute pane grid, or +/// a step reading an evicted pane while the later steps could be answered. +#[tokio::test] +async fn forwards_whole_range_queries_a_step_cannot_answer() { + let (_, app) = example4b(101 * MINUTE).await; + for (start, step, reason) in [("5400", "5m", "pane grid"), ("4800", "10m", "not stored")] { + let params = [ + ("query", P99), + ("start", start), + ("end", "6000"), + ("step", step), + ]; + let response = range(&app, ¶ms).await; + assert_eq!(response.execution.as_deref(), Some("forwarded")); + let header = response.reason.unwrap_or_default(); + assert!(header.contains(reason), "{start} {step}: {header}"); + assert_eq!( + response.body, + json!({"forwarded_range": {"query": P99, "start": start, "end": "6000", "step": step}}), + "forwarded unchanged" + ); + } +} + +/// Unplanned range queries, and range queries Prometheus would reject, are +/// forwarded unchanged with no reason: Prometheus answers them. +#[tokio::test] +async fn forwards_unplanned_and_invalid_range_queries_unchanged() { + let (_, app) = example4b(101 * MINUTE).await; + for params in [ + [ + ("query", "up"), + ("start", "5400"), + ("end", "6000"), + ("step", "15s"), + ], + [ + ("query", P99), + ("start", "5400"), + ("end", "6000"), + ("step", "0"), + ], + [ + ("query", P99), + ("start", "6000"), + ("end", "5400"), + ("step", "10m"), + ], + ] { + let response = range(&app, ¶ms).await; + assert_eq!( + response.execution.as_deref(), + Some("forwarded"), + "{params:?}" + ); + assert_eq!(response.reason, None, "{params:?}"); + let expected: BTreeMap<&str, &str> = params.into_iter().collect(); + assert_eq!(response.body, json!({ "forwarded_range": expected })); + } +} diff --git a/data_plane/tests/e2e_examples.rs b/data_plane/tests/e2e_examples.rs index c5397868..2c2d1efd 100644 --- a/data_plane/tests/e2e_examples.rs +++ b/data_plane/tests/e2e_examples.rs @@ -108,8 +108,9 @@ fn sorted(mut samples: Vec) -> Vec { /// A Prometheus holding `all`: range selectors `{__name__="m"}[Nms]` at /// `time` return the samples in `[time - N, time]`, as Prometheus 2 does; -/// every other query returns an empty vector. Returns its URL and how many -/// range selectors it has answered. +/// every other query returns an empty vector, and every range query an +/// empty matrix. Returns its URL and how many range selectors it has +/// answered. async fn prometheus(all: Arc>) -> (String, Arc) { let range_reads = Arc::new(AtomicUsize::new(0)); let reads = range_reads.clone(); @@ -153,7 +154,12 @@ async fn prometheus(all: Arc>) -> (String, Arc) { }; let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let url = format!("http://{}", listener.local_addr().unwrap()); - let app = Router::new().route("/api/v1/query", get(handler)); + let empty_matrix = || async { + Json(json!({"status": "success", "data": {"resultType": "matrix", "result": []}})) + }; + let app = Router::new() + .route("/api/v1/query", get(handler)) + .route("/api/v1/query_range", get(empty_matrix)); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); (url, range_reads) } @@ -295,6 +301,89 @@ async fn planned_answer(url: &str, query: &str, time_ms: i64) -> Vec BTreeMap> { + let client = reqwest::Client::new(); + let deadline = Instant::now() + Duration::from_secs(30); + loop { + let response = client + .get(format!("{url}/api/v1/query_range")) + .query(&[ + ("query", query), + ("start", &(start_ms as f64 / 1000.0).to_string()), + ("end", &(end_ms as f64 / 1000.0).to_string()), + ("step", step), + ]) + .send() + .await + .unwrap(); + let execution = response.headers().get("x-asap-execution").cloned(); + let reason = response.headers().get("x-asap-forward-reason").cloned(); + let body: Value = response.json().await.unwrap(); + if execution.as_ref().is_some_and(|v| v == "plan") { + assert_eq!(body["data"]["resultType"], "matrix"); + let mut steps: BTreeMap> = BTreeMap::new(); + for series in body["data"]["result"].as_array().unwrap() { + let labels: BTreeMap = + serde_json::from_value(series["metric"].clone()).unwrap(); + for point in series["values"].as_array().unwrap() { + let t = (point[0].as_f64().unwrap() * 1000.0).round() as i64; + steps.entry(t).or_default().push(ResultSample { + labels: labels.clone(), + value: point[1].as_str().unwrap().parse().unwrap(), + }); + } + } + return steps.into_iter().map(|(t, s)| (t, sorted(s))).collect(); + } + assert!( + Instant::now() < deadline, + "range {query} was not answered by the plan: {reason:?}" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + +/// Answers `query` over `steps` (from `start_ms` by `step`, through +/// `end_ms`) as one range query and checks each step against the Planner +/// executor's reference and against the same instant query. Returns how +/// many raw ranges the range query read from Prometheus. +#[allow(clippy::too_many_arguments)] +async fn check_range( + url: &str, + reads: &AtomicUsize, + at_query_time: &InstalledPlan, + query: &str, + (start_ms, end_ms, step): (i64, i64, &str), + step_ms: i64, + all: &[RawSample], +) -> usize { + let before = reads.load(Ordering::SeqCst); + let answer = planned_range_answer(url, query, start_ms, end_ms, step).await; + let range_reads = reads.load(Ordering::SeqCst) - before; + let steps: Vec = (start_ms..=end_ms).step_by(step_ms as usize).collect(); + assert_eq!(answer.keys().copied().collect::>(), steps, "{query}"); + for (t, actual) in &answer { + let expected = reference(at_query_time, query, *t, all); + assert_same(&format!("{query} @ {t}"), actual, &expected, 0.0); + let instant = planned_answer(url, query, *t).await; + assert_eq!(actual, &instant, "{query} @ {t}: range vs instant"); + } + eprintln!( + "{query}: range over {} steps read {range_reads} raw ranges", + steps.len() + ); + range_reads +} + /// `actual` and `expected` hold the same series with values within /// `tolerance` (plus float rounding). fn assert_same(query: &str, actual: &[ResultSample], expected: &[ResultSample], tolerance: f64) { @@ -485,3 +574,126 @@ async fn example4_b3() { eprintln!("{query}: raw ranges read per evaluation {reads:?}"); assert_eq!(reads, vec![5, 1, 1]); } + +/// Example 1 as range queries, a step every 30 s over [3 min, 6 min]: each +/// query reads its one raw input from Prometheus once for all 7 steps, and +/// every step equals the instant query and the Planner executor. +#[tokio::test] +async fn example1_range() { + let series = [ + series("http_requests_total", &[("job", "api"), ("instance", "1")]), + series("http_requests_total", &[("job", "api"), ("instance", "2")]), + series("http_requests_total", &[("job", "web"), ("instance", "1")]), + ]; + // Quadratic counters: the rate changes at every step. + let all = samples(&series, 0, 10 * MINUTE, 15_000, |i, t| { + (i as f64 + 1.0) * (t as f64 / 1000.0).powi(2) + }); + let workload = workload("example1"); + let inputs = DeploymentInputs::default(); + let plan = plan_workload(&workload, &inputs).unwrap(); + let at_query_time = plan_selected_at_query_time(&workload, &inputs).unwrap(); + let all = Arc::new(all); + let (prometheus_url, reads) = prometheus(all.clone()).await; + let process = start(&plan, &prometheus_url).await; + remote_write(&process.url, &all).await; + for query in &plan.queries { + let range = (3 * MINUTE, 6 * MINUTE, "30s"); + let read = check_range( + &process.url, + &reads, + &at_query_time, + query, + range, + 30_000, + &all, + ) + .await; + assert_eq!(read, 1, "{query}"); + } +} + +/// Example 4b as a range query on a deployment without raw data: every +/// step reads six stored panes and nothing from Prometheus. +#[tokio::test] +async fn example4b_range() { + let series = [ + series("latency_ms", &[("service", "a")]), + series("latency_ms", &[("service", "b")]), + ]; + // Values grow with time, so each hour's p99 differs. + let all = samples(&series, MINUTE, 101 * MINUTE, 15_000, |i, t| { + (i * 1000) as f64 + (t / 15_000) as f64 + }); + let workload = workload("example4b"); + let inputs = without_raw_data(); + let plan = plan_workload(&workload, &inputs).unwrap(); + let at_query_time = plan_selected_at_query_time(&workload, &inputs).unwrap(); + let all = Arc::new(all); + let (prometheus_url, reads) = prometheus(all.clone()).await; + let process = start(&plan, &prometheus_url).await; + remote_write(&process.url, &all).await; + let query = &plan.queries[0]; + let range = (90 * MINUTE, 100 * MINUTE, "10m"); + let read = check_range( + &process.url, + &reads, + &at_query_time, + query, + range, + 10 * MINUTE, + &all, + ) + .await; + assert_eq!(read, 0); +} + +/// Example 4's B3 as a range query, a step a minute over [10 min, 14 min]: +/// one Prometheus read covers the raw samples of all 9 kept panes, each +/// pane is built once and reused by later steps, and every step equals the +/// instant query and the Planner executor. Samples run to 15 min, so every +/// pane read is closed and kept. The same range query again reads one raw +/// range, for the panes ending at 6..8 min: they were evicted, measured from +/// the newest step. +#[tokio::test] +async fn example4_b3_range() { + let series = [ + series("latency_ms", &[("service", "a")]), + series("latency_ms", &[("service", "b")]), + ]; + let all = samples(&series, 0, 15 * MINUTE, 15_000, |i, t| { + (i * 100) as f64 + (t / 15_000 * 25 % 101) as f64 + }); + let workload = workload("example4-b3"); + let inputs = DeploymentInputs::default(); + let plan = plan_workload(&workload, &inputs).unwrap(); + let at_query_time = plan_selected_at_query_time(&workload, &inputs).unwrap(); + let all = Arc::new(all); + let (prometheus_url, reads) = prometheus(all.clone()).await; + let process = start(&plan, &prometheus_url).await; + remote_write(&process.url, &all).await; + let query = &plan.queries[0]; + let range = (10 * MINUTE, 14 * MINUTE, "60"); + let read = check_range( + &process.url, + &reads, + &at_query_time, + query, + range, + MINUTE, + &all, + ) + .await; + assert_eq!(read, 1); + let read = check_range( + &process.url, + &reads, + &at_query_time, + query, + range, + MINUTE, + &all, + ) + .await; + assert_eq!(read, 1, "the evicted panes"); +} diff --git a/data_plane/tests/range.rs b/data_plane/tests/range.rs new file mode 100644 index 00000000..10e55dda --- /dev/null +++ b/data_plane/tests/range.rs @@ -0,0 +1,344 @@ +//! Range queries in the query engine: one evaluation per step, raw samples +//! fetched once for all steps, and stored and kept panes shared across +//! steps as they are across the evaluations of a recurring query. + +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use asap_executor::physical_planner::frontier_from_timing; +use asap_executor::values::Batch; +use control_plane::{plan_selected_at_query_time, InstalledPlan}; +use data_plane::precompute::{Precompute, RawSample}; +use data_plane::query::{KeptResultStats, QueryEngine, RawRange, ResultSample, Unanswered}; +use data_plane::sds::{Sds, SummaryStore}; + +const MINUTE: i64 = 60_000; + +fn sds(plan: &InstalledPlan) -> Sds { + let dag = &plan.dag.dag; + Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()).unwrap() +} + +/// Samples every 15 s in `[from, to)` of `metric` for each `service`, with +/// value `f(series index, t)`. +fn samples( + metric: &str, + services: &[&str], + from: i64, + to: i64, + f: impl Fn(usize, i64) -> f64, +) -> Vec { + let series: Vec<_> = services + .iter() + .map(|service| { + Arc::new(BTreeMap::from([ + ("__name__".to_string(), metric.to_string()), + ("service".to_string(), service.to_string()), + ])) + }) + .collect(); + let f = &f; + (from..to) + .step_by(15_000) + .flat_map(|t| { + series.iter().enumerate().map(move |(i, labels)| RawSample { + labels: labels.clone(), + timestamp_ms: t, + value: f(i, t), + }) + }) + .collect() +} + +/// What a deployment fetches for each raw range: its metric's samples in +/// `(start, end]`. +fn fetch(ranges: &[RawRange], all: &[RawSample]) -> BTreeMap> { + ranges + .iter() + .map(|range| { + let samples = all + .iter() + .filter(|s| { + s.labels["__name__"] == range.metric + && range.start_ms < s.timestamp_ms + && s.timestamp_ms <= range.end_ms + }) + .cloned() + .collect(); + (range.node, samples) + }) + .collect() +} + +fn sorted(mut samples: Vec) -> Vec { + samples.sort_by(|a, b| a.labels.cmp(&b.labels)); + samples +} + +/// `query` at `time_ms` as an instant query. +fn instant( + engine: &mut QueryEngine, + query: &str, + time_ms: i64, + store: &SummaryStore, + all: &[RawSample], +) -> Vec { + let (root, ranges) = engine.raw_ranges(query, time_ms, store).unwrap(); + sorted( + engine + .evaluate(root, time_ms, store, None, &fetch(&ranges, all)) + .unwrap(), + ) +} + +/// `query` at every step as one range query once every sample in `all` +/// has arrived; also returns its raw ranges. +fn range( + engine: &mut QueryEngine, + query: &str, + steps: &[i64], + store: &SummaryStore, + all: &[RawSample], +) -> (Vec>, Vec) { + let watermark_ms = all.iter().map(|s| s.timestamp_ms).max(); + range_at_watermark(engine, query, steps, store, watermark_ms, all) +} + +/// `query` at every step as one range query when the watermark is +/// `watermark_ms`; also returns its raw ranges. +fn range_at_watermark( + engine: &mut QueryEngine, + query: &str, + steps: &[i64], + store: &SummaryStore, + watermark_ms: Option, + all: &[RawSample], +) -> (Vec>, Vec) { + let (root, ranges) = engine.raw_ranges_for_steps(query, steps, store).unwrap(); + let results = engine + .evaluate_steps(root, steps, store, watermark_ms, &fetch(&ranges, all)) + .unwrap(); + (results.into_iter().map(sorted).collect(), ranges) +} + +fn minutes(range: std::ops::RangeInclusive) -> Vec { + range.map(|m| m * MINUTE).collect() +} + +/// Example 1 runs at query time from raw samples. A range query fetches +/// each raw input once, `(start - 1m, end]`, and every step answers like +/// the same instant query. +#[test] +fn raw_samples_are_fetched_once_and_sliced_per_step() { + let plan = common::plan("example1", true); + let query = "sum by (job) (rate(http_requests_total[1m]))"; + let series: Vec<_> = [("a", "1"), ("a", "2"), ("b", "1")] + .into_iter() + .map(|(job, instance)| { + Arc::new(BTreeMap::from([ + ("__name__".to_string(), "http_requests_total".to_string()), + ("job".to_string(), job.to_string()), + ("instance".to_string(), instance.to_string()), + ])) + }) + .collect(); + let all: Vec = (0..10 * MINUTE) + .step_by(15_000) + .flat_map(|t| { + series.iter().enumerate().map(move |(i, labels)| RawSample { + labels: labels.clone(), + timestamp_ms: t, + // Rates change over time, so each step's answer differs. + value: (i as f64 + 1.0) * (t as f64 / 1000.0).powi(2), + }) + }) + .collect(); + let store = SummaryStore::new(); + let mut engine = QueryEngine::new(&plan, &sds(&plan)); + // Steps 30 s apart in [3 min, 6 min], inclusive. + let steps: Vec = (6..=12).map(|k| k * 30_000).collect(); + let (results, ranges) = range(&mut engine, query, &steps, &store, &all); + assert_eq!( + ranges + .iter() + .map(|r| (r.metric.as_str(), r.start_ms, r.end_ms)) + .collect::>(), + vec![("http_requests_total", 2 * MINUTE, 6 * MINUTE)], + "one fetch covers every step" + ); + assert_eq!(results.len(), steps.len()); + for (t, result) in steps.iter().zip(&results) { + assert_eq!(result.len(), 2, "two jobs at {t}"); + assert_eq!( + result, + &instant(&mut engine, query, *t, &store, &all), + "{t}" + ); + } + assert_ne!(results[0], results[1]); +} + +/// Example 4b keeps no raw data: every step reads six stored 10-minute +/// panes, no raw samples, and answers like the same instant query. Steps +/// off the pane grid, or reading an evicted pane, forward the whole query. +#[test] +fn every_step_reads_stored_panes() { + let plan = common::plan("example4b", false); + let query = "quantile_over_time(0.99, latency_ms[60m])"; + let sds = sds(&plan); + let mut precompute = Precompute::new(plan.dag.dag.clone(), &sds, 0).unwrap(); + // Values grow with time, so each hour's p99 differs. + let all = samples("latency_ms", &["a", "b"], MINUTE, 101 * MINUTE, |i, t| { + (i * 1000) as f64 + (t / 15_000) as f64 + }); + precompute.ingest(&all); + let store = &precompute.store; + let mut engine = QueryEngine::new(&plan, &sds); + let steps = minutes(9..=10) + .into_iter() + .map(|t| t * 10) + .collect::>(); + let (results, ranges) = range(&mut engine, query, &steps, store, &[]); + assert!( + ranges.is_empty(), + "every input is a stored pane: {ranges:?}" + ); + for (t, result) in steps.iter().zip(&results) { + assert_eq!(result.len(), 2); + assert_eq!(result, &instant(&mut engine, query, *t, store, &[]), "{t}"); + } + assert_ne!(results[0], results[1]); + + let missing = |steps: &[i64]| match engine.raw_ranges_for_steps(query, steps, store) { + Err(Unanswered::MissingPane(reason)) => reason, + other => panic!("{steps:?}: {:?}", other.map(|(_, r)| r)), + }; + // A step 5 minutes apart lands off the 10-minute grid. + assert!(missing(&[90 * MINUTE, 95 * MINUTE]).contains("pane grid")); + // 80 min reads the pane ending at 30 min, already evicted: the whole + // range is forwarded even though its later steps could be answered. + assert!(missing( + &minutes(8..=10) + .into_iter() + .map(|t| t * 10) + .collect::>() + ) + .contains("not stored")); +} + +/// #509 Example 4's B3: five 1-minute KLL panes built at query time and +/// kept. A range query over 10..14 min, a step a minute, fetches the raw +/// samples of all its panes once, builds each of the 9 panes once and reuses +/// it at the later steps; every step answers like the Planner executor. +/// Kept panes are evicted as for instant evaluations, measured from the +/// newest step: the same range query again rebuilds the panes ending at +/// 6..8 min, which are older than 14 min minus 6. +#[test] +fn kept_panes_are_shared_across_steps() { + let workload = common::workload("example4-b3"); + let inputs = common::inputs(true); + let plan = control_plane::plan_workload(&workload, &inputs).unwrap(); + let reference_plan = plan_selected_at_query_time(&workload, &inputs).unwrap(); + let query = "quantile_over_time(0.99, latency_ms[5m])"; + let all = samples("latency_ms", &["a", "b"], 0, 30 * MINUTE + 1, |i, t| { + (i * 100) as f64 + (t / 15_000 * 25 % 101) as f64 + }); + let store = SummaryStore::new(); + let mut engine = QueryEngine::new(&plan, &sds(&plan)); + let mut reference = QueryEngine::new(&reference_plan, &sds(&reference_plan)); + let steps = minutes(10..=14); + + let (results, ranges) = range(&mut engine, query, &steps, &store, &all); + assert_eq!( + ranges + .iter() + .map(|r| (r.start_ms, r.end_ms)) + .collect::>(), + vec![(5 * MINUTE, 14 * MINUTE)], + "one fetch for the kept panes of every step" + ); + for (t, result) in steps.iter().zip(&results) { + assert_eq!(result.len(), 2); + assert_eq!( + result, + &instant(&mut reference, query, *t, &store, &all), + "{t}" + ); + } + assert_ne!(results[0], results[1]); + // Panes ending at 6..14 min are built once; each step after the first + // reuses four of its five. + assert_eq!( + engine.kept_stats, + KeptResultStats { + built: 9, + reused: 16 + } + ); + + let (again, ranges) = range(&mut engine, query, &steps, &store, &all); + assert_eq!( + ranges + .iter() + .map(|r| (r.start_ms, r.end_ms)) + .collect::>(), + vec![(5 * MINUTE, 8 * MINUTE)], + "only the evicted panes are fetched" + ); + assert_eq!(again, results); + // Panes 6..8 are older than the newest evaluation's retention, so each + // step reading one rebuilds it: 3 at 10 min, 2 at 11, 1 at 12. + assert_eq!( + engine.kept_stats, + KeptResultStats { + built: 15, + reused: 35 + } + ); +} + +/// A pane still open at the request's watermark is computed for each step +/// that reads it and not kept: with the watermark at 12 min, the panes +/// ending at 13 and 14 min are recomputed at every step instead of reused, +/// and the answers still match the Planner executor. +#[test] +fn open_panes_are_recomputed_at_each_step() { + let workload = common::workload("example4-b3"); + let inputs = common::inputs(true); + let plan = control_plane::plan_workload(&workload, &inputs).unwrap(); + let reference_plan = plan_selected_at_query_time(&workload, &inputs).unwrap(); + let query = "quantile_over_time(0.99, latency_ms[5m])"; + let all = samples("latency_ms", &["a", "b"], 0, 30 * MINUTE + 1, |i, t| { + (i * 100) as f64 + (t / 15_000 * 25 % 101) as f64 + }); + let store = SummaryStore::new(); + let mut engine = QueryEngine::new(&plan, &sds(&plan)); + let mut reference = QueryEngine::new(&reference_plan, &sds(&reference_plan)); + let steps = minutes(10..=14); + let (results, _) = + range_at_watermark(&mut engine, query, &steps, &store, Some(12 * MINUTE), &all); + for (t, result) in steps.iter().zip(&results) { + assert_eq!( + result, + &instant(&mut reference, query, *t, &store, &all), + "{t}" + ); + } + // Panes 6..12 are built once and kept; 13 is built at 13 and 14 min, 14 + // at 14 min. Reused: four panes at 11, 12 and 13 min, three at 14. + assert_eq!( + engine.kept_stats, + KeptResultStats { + built: 10, + reused: 15 + } + ); + let definition = sds(&plan).kept_definitions.keys().next().unwrap().clone(); + let kept_ends: Vec = (0..=30) + .filter(|m| engine.kept.get(&definition, m * MINUTE).is_some()) + .collect(); + // 6..8 are evicted, measured from the newest step (14 min). + assert_eq!(kept_ends, vec![9, 10, 11, 12]); +} From 8d26bc772fce73a9c1d0b4a35807d7f44ade5803 Mon Sep 17 00:00:00 2001 From: zzylol Date: Sun, 18 Oct 2026 17:44:27 +0000 Subject: [PATCH 2/2] test(query): a nonzero series limit is forwarded to Prometheus --- data_plane/src/prometheus.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/data_plane/src/prometheus.rs b/data_plane/src/prometheus.rs index f666ddb1..31819026 100644 --- a/data_plane/src/prometheus.rs +++ b/data_plane/src/prometheus.rs @@ -331,7 +331,7 @@ mod tests { /// What Prometheus rejects is not parsed: missing parameters, `end` /// before `start`, a step that is not positive or not a duration, and - /// more than 11,000 steps. + /// more than 11,000 steps. Neither is a nonzero series `limit`. #[test] fn rejects_invalid_range_queries() { for params in [ @@ -350,6 +350,10 @@ mod tests { assert_eq!(range(params), None, "{params}"); } assert!(range("query=up&start=0&end=11000&step=1").is_some()); + // A series limit is left to Prometheus; `limit=0` means no limit. + assert_eq!(range("query=up&start=0&end=60&step=15&limit=5"), None); + assert!(range("query=up&start=0&end=60&step=15&limit=0").is_some()); + assert!(InstantQuery::parse(Some("query=up&limit=5"), b"").is_none()); } /// Steps run from `start` by `step` and include `end` when on a step.