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..31819026 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,113 @@ 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. Neither is a nonzero series `limit`. + #[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()); + // 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. + #[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]); +}