diff --git a/README.md b/README.md index 88909a1d..c1df9747 100644 --- a/README.md +++ b/README.md @@ -54,9 +54,19 @@ stopped, which also makes `/-/ready` fail. `/api/v1/status/asap` reports the engine's sample and pane counters as JSON. An instant query whose text is one of the plan's queries is answered by the -plan (`x-asap-execution: plan`). Its stored inputs read the panes ending at -`T - offset`, so `T` must be on the pane grid and those panes stored; its -query-time bound raw-sample windows are read from Prometheus. Anything else is forwarded +plan (`x-asap-execution: plan`) at its evaluation time `T`. Stored outputs +built in an `IngestionRun` are read from the SDS as the panes ending at +`T - offset`, so `T` must be on the pane grid and those panes stored. The +plan's query-time `RawSampleWindow`s are read from Prometheus. A stored output +built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`) is built from its +pane's raw samples the first time an evaluation needs it. It is stored in the +SDS under the same key once its pane is closed, that is, once its end is at or +before the watermark at the time the evaluation chose its fetch. Later +evaluations reuse it. It is evicted like an `IngestionRun` output: the plan's +`retention_ms` plus one pane, measured from the newest closed pane at or +before the watermark. `/api/v1/status/asap` reports `query_run` counters, +including summary states stored while their own metric lagged behind the +global watermark. Anything else is forwarded to Prometheus with its end-to-end headers, such as `Authorization` and `X-Scope-OrgID` (`x-asap-execution: forwarded`); a planned query that could not be answered also says why in `x-asap-forward-reason`. Range queries, SQL and @@ -75,8 +85,7 @@ cargo run -p data_plane -- --plan plan.json --prometheus-url http://localhost:90 `--deployment-capabilities caps.json` replaces the default `DeploymentCapabilities`: every summary the Planner can build, raw data kept -in Prometheus, and no stored outputs built in a `QueryRun` -(`MatAtQueryKeptAcrossEvaluations`). For example +in Prometheus, and stored outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`). For example `{"raw_data_retained": false}` makes the Planner build every input at ingestion time. `--measured-operation-costs costs.json` gives Stage 3 the deployment's measured sketch operation costs; otherwise it prices every node diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs index ba1ee7a5..5e575d53 100644 --- a/control_plane/src/lib.rs +++ b/control_plane/src/lib.rs @@ -82,14 +82,15 @@ pub fn load_plan(path: &Path) -> Result { } /// What this backend's data plane can execute and keep: every summary the -/// Planner executor builds, raw samples kept in Prometheus, and no -/// stored outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`): the SDS describes such -/// outputs, but nothing builds them until the query engine does. +/// Planner executor builds, raw samples kept in Prometheus, and stored +/// outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`): the +/// query engine builds their summary states on first use and stores them in +/// the SDS once their coverage is closed. pub fn backend_capabilities() -> DeploymentCapabilities { DeploymentCapabilities { supported_summaries: None, raw_data_retained: true, - query_time_results_retained: false, + query_time_results_retained: true, } } diff --git a/control_plane/tests/plan_workload.rs b/control_plane/tests/plan_workload.rs index f4855511..1d592a63 100644 --- a/control_plane/tests/plan_workload.rs +++ b/control_plane/tests/plan_workload.rs @@ -88,3 +88,21 @@ fn load_rejects_a_plan_without_a_query_per_root() { )); std::fs::remove_file(path).unwrap(); } + +/// The backend retains query-time results (`query_time_results_retained`), +/// so for a few series sampled every millisecond the Planner selects #509 +/// Example 4's B3: five 1-minute KLL panes, stored outputs built in a +/// `QueryRun` (`MatAtQueryKeptAcrossEvaluations`). +#[test] +fn selects_stored_outputs_built_in_a_query_run() { + let inputs = DeploymentInputs::default(); + assert!(inputs.deployment_capabilities.query_time_results_retained); + let plan = plan_workload(&example("example4-b3"), &inputs).unwrap(); + let query_run: Vec<_> = plan + .stored_outputs + .iter() + .filter(|o| o.built_at == asap_planner::StoredOutputBuild::QueryRun) + .collect(); + assert_eq!(query_run.len(), 5, "five one-minute summaries"); + assert!(query_run.iter().all(|o| o.pane.width_ms == 60_000)); +} diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index df882c5a..c9df45c0 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -12,7 +12,9 @@ use asap_planner::InstallablePlan; use tokio::sync::oneshot; use crate::precompute::{Precompute, PrecomputeOptions, PrecomputeStats, RawSample}; -use crate::query::{BoundRawSampleWindow, QueryEngine, ResultSample, Unanswered}; +use crate::query::{ + IngestionProgress, Prepared, QueryEngine, QueryRunStats, ResultSample, Unanswered, +}; use crate::sds::Sds; /// How the engine thread ingests samples. @@ -42,11 +44,12 @@ enum Command { BoundRawSampleWindows { query: String, time_ms: i64, - reply: Reply<(usize, Vec)>, + reply: Reply, }, Evaluate { root: usize, time_ms: i64, + progress: IngestionProgress, raw: BTreeMap>, reply: Reply>, }, @@ -75,6 +78,8 @@ pub struct EngineHandle { running: Arc, /// The counters as of the last command the engine ran. stats: Arc>, + /// The stored output built in a `QueryRun` counters as of the last evaluation. + query_run_stats: Arc>, } /// Marks the engine stopped when its thread exits, panics included. @@ -96,9 +101,11 @@ impl EngineHandle { let running = Arc::new(AtomicBool::new(true)); let queued_samples = Arc::new(AtomicU64::new(0)); let stats = Arc::new(Mutex::new(PrecomputeStats::default())); + let query_run_stats = Arc::new(Mutex::new(QueryRunStats::default())); let shared = Shared { queued_samples: queued_samples.clone(), stats: stats.clone(), + query_run_stats: query_run_stats.clone(), }; let guard = RunningGuard(running.clone()); std::thread::Builder::new() @@ -122,6 +129,7 @@ impl EngineHandle { max_queued_samples: options.max_queued_samples, running, stats, + query_run_stats, }; Ok((handle, queries)) } @@ -158,7 +166,7 @@ impl EngineHandle { &self, query: String, time_ms: i64, - ) -> Result<(usize, Vec), Unanswered> { + ) -> Result { self.ask(|reply| Command::BoundRawSampleWindows { query, time_ms, @@ -167,22 +175,32 @@ impl EngineHandle { .await } - /// Runs a root over stored panes and `raw`; see [`QueryEngine::evaluate`]. + /// Runs a prepared root over stored summaries and `raw`; see + /// [`QueryEngine::evaluate`]. pub async fn evaluate( &self, - root: usize, + prepared: Prepared, time_ms: i64, raw: BTreeMap>, ) -> Result, Unanswered> { self.ask(|reply| Command::Evaluate { - root, + root: prepared.root, time_ms, + progress: prepared.progress, raw, reply, }) .await } + /// The stored output built in a `QueryRun` counters as of the last evaluation. + pub fn query_run_stats(&self) -> QueryRunStats { + *self + .query_run_stats + .lock() + .unwrap_or_else(|e| e.into_inner()) + } + async fn ask(&self, command: impl FnOnce(Reply) -> Command) -> Result { let stopped = || Unanswered::Execution("the engine has stopped".into()); let (reply, answer) = oneshot::channel(); @@ -232,6 +250,7 @@ fn admit(queued: u64, count: u64, max_queued: u64) -> Result { struct Shared { queued_samples: Arc, stats: Arc>, + query_run_stats: Arc>, } struct Engine { @@ -249,18 +268,6 @@ impl Engine { outputs = sds.outputs.len(), "stored summaries" ); - let query_run = sds - .definitions - .values() - .filter(|d| d.built_at == asap_planner::StoredOutputBuild::QueryRun) - .count(); - if query_run > 0 { - tracing::warn!( - definitions = query_run, - "the plan has stored outputs built in a QueryRun, which this data plane does not build yet; \ - queries that read them are forwarded" - ); - } let queries = plan.queries.iter().map(|q| q.query.clone()).collect(); let query = QueryEngine::new(&plan, &sds); let engine = Self { @@ -289,20 +296,30 @@ impl Engine { &query, time_ms, &self.precompute.store, + self.precompute.progress(), )); } Command::Evaluate { root, time_ms, + progress, raw, reply, } => { - let _ = reply.send(self.query.evaluate( + let answer = self.query.evaluate( root, time_ms, - &self.precompute.store, + &mut self.precompute.store, + &progress, &raw, - )); + ); + // Published before the reply, so a caller that awaited + // the evaluation reads its counters. + *shared + .query_run_stats + .lock() + .unwrap_or_else(|e| e.into_inner()) = self.query.query_run_stats; + let _ = reply.send(answer); } Command::Stop => break, } diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index b65ac699..455fd9f0 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -176,6 +176,7 @@ async fn status(State(state): State>) -> Response { "engine_running": state.engine.is_running(), "samples_queued": state.engine.queued_samples(), "precompute": state.engine.stats(), + "query_run": state.engine.query_run_stats(), }); ( [(header::CONTENT_TYPE, "application/json")], @@ -249,10 +250,11 @@ async fn answer( .duration_since(std::time::UNIX_EPOCH) .map_or(0, |d| d.as_millis() as i64) }); - let (root, ranges) = state + let prepared = state .engine .bound_raw_sample_windows(query.query, time_ms) .await?; + let ranges = &prepared.windows; let mut fetch_headers = forwarded_headers(headers); // The raw read is a GET without the caller's body. fetch_headers.remove(header::CONTENT_TYPE); @@ -267,7 +269,7 @@ async fn answer( .map_err(|e| Unanswered::Execution(format!("fetching raw samples: {e}")))? .into_iter() .collect(); - let samples = state.engine.evaluate(root, time_ms, raw).await?; + let samples = state.engine.evaluate(prepared, time_ms, raw).await?; Ok(prometheus::vector_response(time_ms, &samples)) } diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 3cdedce2..19d6c475 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -29,6 +29,8 @@ use asap_executor::{RawSampleSelector, Sample}; use futures::executor::block_on; use futures::StreamExt; +use asap_planner::StoredOutputBuild; + use crate::sds::{ DefinitionId, Sds, Selection, SummaryDefinition, SummaryInstanceKey, SummaryStore, }; @@ -68,6 +70,10 @@ pub struct PrecomputeStats { pub samples_dropped_future: u64, /// Samples of a metric no stored summary reads. pub samples_unrouted: u64, + /// Samples of a metric only stored outputs built in a `QueryRun` read: they move the + /// watermark, which says when such a summary is closed, but are not + /// buffered (the query engine reads them from Prometheus). + pub samples_query_run_only: u64, /// Panes built and stored. pub panes_built: u64, /// Panes whose build failed; they are skipped, not retried. @@ -95,13 +101,84 @@ impl Default for PrecomputeOptions { } } -struct Builder { - definition: SummaryDefinition, - /// Turns a pane's samples into the build's input: the `Scan`'s label - /// matchers, then the pane's `TimeRange` rows. +/// How one definition's summaries are built from raw samples. The +/// precompute engine builds stored outputs built in an `IngestionRun` with it, and the query +/// engine `QueryRun` ones. +pub struct SummaryBuild { + pub definition: SummaryDefinition, + /// Turns a coverage's samples into the build's input: the `Scan`'s label + /// matchers, then the coverage's `TimeRange` rows. selector: RawSampleSelector, raw_schema: SchemaRef, build: CompiledPhysicalDAG, +} + +impl SummaryBuild { + /// Compiles `definition`'s build from its `TimeRange` up to its builder. + /// A definition built from other stored outputs is refused until the + /// engine can build it once its inputs are sealed. + pub fn new( + dag: &PhysicalASAPDAG, + definition: &SummaryDefinition, + ) -> Result { + let error = |message: String| PrecomputeError::Definition(definition.id.clone(), message); + if !definition.stored_inputs.is_empty() { + return Err(error( + "built from other stored outputs, which the data plane cannot build yet".into(), + )); + } + if definition.pane_ms <= 0 { + return Err(error(format!("pane width {} ms", definition.pane_ms))); + } + let raw_schema = Arc::new(output_schema(dag, definition.raw_input()).clone()); + // `window_batch` is called with each coverage's end, so the window's + // offset from evaluation time is not used. + let selector = RawSampleSelector::new(dag, &definition.window)?; + let build = compile( + dag, + BTreeMap::from([( + definition.raw_input(), + InputContract::bounded(raw_schema.clone()), + )]), + &[definition.builder], + )?; + Ok(Self { + definition: definition.clone(), + selector, + raw_schema, + build, + }) + } + + /// The metric whose samples the build reads. + pub fn metric(&self) -> &str { + self.selector.metric() + } + + /// Selects the coverage ending at `end_ms` from `samples` with the plan's + /// `Scan` matchers, then runs the build over those rows under `scope`. + pub fn run( + &self, + end_ms: i64, + samples: &[RawSample], + scope: Scope, + ) -> Result { + let selected = self + .selector + .window_batch(end_ms, &executor_samples(samples))?; + let input: Source<'_> = + Box::new(Operator::source(self.raw_schema.clone(), vec![selected])?); + let build = self + .build + .instantiate(BTreeMap::from([(self.definition.raw_input(), input)]))?; + let rows = run_root(&build, self.definition.builder, scope)?; + let schema = Arc::new(self.definition.output_schema.clone()); + Ok(Batch::try_new(schema, rows)?) + } +} + +struct Builder { + summary: SummaryBuild, /// Samples of open panes, by pane end. open: BTreeMap>, /// End of the first pane seen, which is skipped. `None` until then. @@ -121,6 +198,9 @@ enum Outcome { pub struct Precompute { builders: Vec, + /// Metrics read by stored outputs built in a `QueryRun`, with each one's newest routed + /// sample time. + query_run_metrics: BTreeMap>, allowed_lateness_ms: i64, max_future_ms: i64, newest_sample_ms: Option, @@ -147,43 +227,35 @@ impl Precompute { .ingestion_build_order() .map_err(|e| PrecomputeError::Plan(e.to_string()))?; for id in &order { - let definition = &sds.definitions[id]; - let error = - |message: String| PrecomputeError::Definition(definition.id.clone(), message); - if !definition.stored_inputs.is_empty() { - return Err(error( - "built from other stored outputs, which the precompute engine cannot build yet" - .into(), - )); - } - if definition.pane_ms <= 0 { - return Err(error(format!("pane width {} ms", definition.pane_ms))); - } - let raw_schema = Arc::new(output_schema(&dag, definition.raw_input()).clone()); - // `window_batch` is called with each pane's end, so the window's - // offset from evaluation time is not used. - let selector = RawSampleSelector::new(&dag, &definition.window)?; - let build = compile( - &dag, - BTreeMap::from([( - definition.raw_input(), - InputContract::bounded(raw_schema.clone()), - )]), - &[definition.builder], - )?; builders.push(Builder { - definition: definition.clone(), - selector, - raw_schema, - build, + summary: SummaryBuild::new(&dag, &sds.definitions[id])?, open: BTreeMap::new(), first_pane_end: None, built_through: None, }); } + // Samples of a metric only stored outputs built in a `QueryRun` read move the + // watermark, which says when such a summary's coverage is closed. + // A `QueryRun` definition whose selector does not build is not + // tracked here; the query engine forwards the queries that read it. + let mut query_run_metrics = BTreeMap::new(); + for definition in sds.definitions.values() { + if definition.built_at == StoredOutputBuild::QueryRun { + match RawSampleSelector::new(&dag, &definition.window) { + Ok(selector) => { + query_run_metrics.insert(selector.metric().to_string(), None); + } + Err(error) => tracing::warn!( + definition = ?definition.id, %error, + "stored output built in a `QueryRun` cannot be built; its queries are forwarded" + ), + } + } + } let millis = |value: u64| i64::try_from(value).unwrap_or(i64::MAX); Ok(Self { builders, + query_run_metrics, allowed_lateness_ms: millis(options.allowed_lateness_ms), max_future_ms: millis(options.max_future_ms), newest_sample_ms: None, @@ -193,6 +265,20 @@ impl Precompute { }) } + /// What ingestion has seen, for the query engine's store decisions: the + /// watermark and the newest sample of each metric a `QueryRun` stored + /// output reads. + pub fn progress(&self) -> crate::query::IngestionProgress { + crate::query::IngestionProgress { + watermark_ms: self.watermark_ms(), + newest_sample_ms: self + .query_run_metrics + .iter() + .filter_map(|(metric, newest)| Some((metric.clone(), (*newest)?))) + .collect(), + } + } + /// The watermark: no sample older than this is expected any more. It is /// the newest routed sample, capped at the wall clock, minus the allowed /// lateness. @@ -220,10 +306,12 @@ impl Precompute { let latest_allowed_ms = now_ms.saturating_add(self.max_future_ms); for sample in samples { let metric = sample.labels.get("__name__").map(String::as_str); - if !self - .builders - .iter() - .any(|b| metric == Some(b.selector.metric())) + let query_run = metric.is_some_and(|m| self.query_run_metrics.contains_key(m)); + if !query_run + && !self + .builders + .iter() + .any(|b| metric == Some(b.summary.metric())) { self.stats.samples_unrouted += 1; continue; @@ -234,10 +322,10 @@ impl Precompute { } let mut outcome: Option = None; for builder in &mut self.builders { - if metric != Some(builder.selector.metric()) { + if metric != Some(builder.summary.metric()) { continue; } - let this = match pane_end(sample.timestamp_ms, builder.definition.pane_ms) { + let this = match pane_end(sample.timestamp_ms, builder.summary.definition.pane_ms) { None => Outcome::Late, Some(end) => { let first = *builder.first_pane_end.get_or_insert(end); @@ -255,15 +343,14 @@ impl Precompute { outcome = outcome.max(Some(this)); } match outcome { - None => self.stats.samples_unrouted += 1, - Some(outcome) => { - match outcome { - Outcome::Buffered => self.stats.samples_accepted += 1, - Outcome::FirstPane => self.stats.samples_skipped_first_pane += 1, - Outcome::Late => self.stats.samples_dropped_late += 1, - } - self.newest_sample_ms = self.newest_sample_ms.max(Some(sample.timestamp_ms)); - } + None => self.stats.samples_query_run_only += 1, + Some(Outcome::Buffered) => self.stats.samples_accepted += 1, + Some(Outcome::FirstPane) => self.stats.samples_skipped_first_pane += 1, + Some(Outcome::Late) => self.stats.samples_dropped_late += 1, + } + self.newest_sample_ms = self.newest_sample_ms.max(Some(sample.timestamp_ms)); + if let Some(newest) = metric.and_then(|m| self.query_run_metrics.get_mut(m)) { + *newest = (*newest).max(Some(sample.timestamp_ms)); } } self.close_panes(); @@ -277,8 +364,8 @@ impl Precompute { let Some(built) = builder.built_through else { continue; }; - let pane_ms = builder.definition.pane_ms; - let id = &builder.definition.id; + let pane_ms = builder.summary.definition.pane_ms; + let id = &builder.summary.definition.id; // The newest evaluation stored panes can serve is at the end of // the newest closed pane. Panes are kept for one extra pane // width beyond the plan's retention, so an evaluation one pane @@ -286,6 +373,7 @@ impl Precompute { // skip them. let newest_closed_end = watermark.div_euclid(pane_ms).saturating_mul(pane_ms); let oldest_kept_end = builder + .summary .definition .oldest_kept_end_ms(newest_closed_end.saturating_sub(pane_ms)); let Some(mut end) = built.checked_add(pane_ms) else { @@ -304,7 +392,12 @@ impl Precompute { // a pane without samples is stored as empty, not missing. while end <= watermark { let samples = builder.open.remove(&end).unwrap_or_default(); - match build_pane(builder, end, &samples) { + let scope = Scope::Ingestion { + window_start_ms: end - pane_ms, + window_end_ms: end, + revision: 1, + }; + match builder.summary.run(end, &samples, scope) { Ok(state) => { self.store.insert( SummaryInstanceKey::new(id.clone(), Selection::ending_at(end)), @@ -341,33 +434,6 @@ pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> Option { .checked_mul(pane_ms) } -/// Selects the pane's rows from `samples` with the plan's `Scan` matchers, -/// then runs the definition's build over them, for the pane ending at `end`. -fn build_pane( - builder: &Builder, - end: i64, - samples: &[RawSample], -) -> Result { - let scope = Scope::Ingestion { - window_start_ms: end - builder.definition.pane_ms, - window_end_ms: end, - revision: 1, - }; - let selected = builder - .selector - .window_batch(end, &executor_samples(samples))?; - let input: Source<'_> = Box::new(Operator::source( - builder.raw_schema.clone(), - vec![selected], - )?); - let build = builder - .build - .instantiate(BTreeMap::from([(builder.definition.raw_input(), input)]))?; - let rows = run_root(&build, builder.definition.builder, scope)?; - let schema = Arc::new(builder.definition.output_schema.clone()); - Ok(Batch::try_new(schema, rows)?) -} - /// `samples` as the executor's [`Sample`]s. The selector takes owned label /// maps, so each sample's labels are copied. pub fn executor_samples(samples: &[RawSample]) -> Vec { diff --git a/data_plane/src/query.rs b/data_plane/src/query.rs index d73012a8..3416a5d7 100644 --- a/data_plane/src/query.rs +++ b/data_plane/src/query.rs @@ -1,12 +1,20 @@ //! Query engine: answers a planned query at evaluation time `T` by running //! its root of the plan's DAG over stored panes and raw samples. //! -//! Each root is compiled once with two kinds of inputs: -//! - stored outputs (the SDS frontier): bound to the pane of their definition -//! ending at `T - offset`, which must be stored; -//! - query-time bound raw-sample windows, `TimeRange(range) <- TimeShift(offset)* <- Scan`: -//! bound to the samples in `(T - offset - range, T - offset]`, which the -//! caller fetches, filtered by the plan's `Scan`. +//! Each root is compiled once with three kinds of inputs: +//! - stored outputs built in an ingestion run: bound to their definition's +//! summary ending at `T - offset`, which must be stored; +//! - stored outputs built in a query run (`StoredOutputBuild::QueryRun`): +//! bound to their definition's summary ending at `T - offset`. A missing +//! one is built from its coverage's raw samples, which the caller fetches, +//! and stored in the SDS for later evaluations once its pane is closed +//! (its end at or before the watermark when the evaluation chose its +//! fetch); these summary states are evicted by the plan's `retention_ms` +//! plus one pane, measured from the newest closed pane at or before the +//! watermark, as for stored outputs built in an `IngestionRun`; +//! - the plan's query-time `RawSampleWindow`s: bound to the samples in +//! `(T - offset - width, T - offset]`, which the caller fetches, selected +//! by the window's `RawSampleSelector`. //! //! A query the engine cannot answer returns [`Unanswered`]; the caller then //! forwards it to Prometheus. @@ -23,10 +31,10 @@ use asap_executor::planner::ir::{NonASAPOp, Operator as IrOperator, ScalarExpr}; use asap_executor::runtime::Scope; use asap_executor::values::{Batch, SchemaRef, Value}; use asap_executor::{series_labels_of_output_row, RawSampleSelector}; -use asap_planner::InstallablePlan; +use asap_planner::{InstallablePlan, StoredOutputBuild}; -use crate::precompute::{executor_samples, run_root, RawSample}; -use crate::sds::{DefinitionId, Sds, Selection, SummaryStore}; +use crate::precompute::{executor_samples, run_root, RawSample, SummaryBuild}; +use crate::sds::{DefinitionId, Sds, Selection, SummaryInstanceKey, SummaryStore}; /// A bound raw-sample window: a `RawSampleWindow` of the plan bound to one /// evaluation time, i.e. `metric`'s raw samples in `(start_ms, end_ms]` @@ -112,14 +120,74 @@ struct RawInput { offset_ms: i64, } +/// A stored output built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`). +struct QueryRunInput { + node: u64, + definition: DefinitionId, + offset_ms: i64, + /// The output's own `TimeRange`, which its coverage's samples feed. + raw_input: u64, +} + struct Program { compiled: CompiledPhysicalDAG, stored: Vec, + query_run: Vec, + /// For each `QueryRun` definition the root reads, the node its one + /// fetch is keyed by: the lowest `TimeRange` among its outputs. + query_run_fetch_node: BTreeMap, raw: Vec, output: Schema, } +/// What happened to the stored outputs built in a `QueryRun` evaluations read. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, serde::Serialize)] +pub struct QueryRunStats { + /// Summary states built from raw samples because the SDS had none. + pub built: u64, + /// Of those, the ones stored in the SDS: their coverage was closed. + pub stored: u64, + /// Of those stored, the ones whose own metric's newest sample had not + /// passed the coverage's end: the watermark is global, so a metric + /// lagging behind others can have a summary state stored before all its samples + /// arrived. A per-metric watermark is the follow-up. + pub stored_while_metric_behind: u64, + /// Summaries read from an earlier evaluation instead of rebuilt. + pub reused: u64, +} + +/// What ingestion had seen when an evaluation decided what to fetch. The +/// store decision uses this snapshot, not the state when the evaluation +/// runs: samples ingested in between were not in the fetched data. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct IngestionProgress { + /// The precompute engine's watermark; `None` before any sample. + pub watermark_ms: Option, + /// The newest sample time of each metric stored outputs built in a `QueryRun` read. + pub newest_sample_ms: BTreeMap, +} + +/// One raw read an evaluation needs, prepared by [`QueryEngine::bound_raw_sample_windows`]. +#[derive(Debug, Clone)] +pub struct Prepared { + /// The root answering the query. + pub root: usize, + /// The bound raw-sample windows to fetch. + pub windows: Vec, + /// What ingestion had seen when `ranges` were chosen. + pub progress: IngestionProgress, +} + pub struct QueryEngine { + /// How many summary states of stored outputs built in a `QueryRun` + /// evaluations built, stored and reused. + pub query_run_stats: QueryRunStats, + /// How each `QueryRun` definition's summaries are built, or why they + /// cannot be. + query_run_builds: BTreeMap>, + /// The label matchers of each `QueryRun` definition's `Scan`, pushed into + /// its fetch. + query_run_matchers: BTreeMap>, /// Each query, with its compiled root or why it cannot run. roots: Vec<(String, Result)>, } @@ -132,30 +200,56 @@ impl QueryEngine { .iter() .map(|w| (w.time_range_node_id as u64, w)) .collect(); + let query_run_builds: BTreeMap<_, _> = sds + .definitions + .values() + .filter(|d| d.built_at == StoredOutputBuild::QueryRun) + .map(|d| { + let build = SummaryBuild::new(dag, d).map_err(|e| e.to_string()); + (d.id.clone(), build) + }) + .collect(); let roots = plan .queries .iter() .map(|q| { - let program = program(dag, sds, &windows, q.root_node_id as u64); + let program = program(dag, sds, &query_run_builds, &windows, q.root_node_id as u64); (q.query.trim().to_string(), program) }) .collect(); - Self { roots } + let query_run_matchers = query_run_builds + .keys() + .map(|id| { + let scan = sds.definitions[id].window.scan_node_id as u64; + (id.clone(), label_matchers(dag, scan)) + }) + .collect(); + Self { + query_run_stats: QueryRunStats::default(), + query_run_builds, + query_run_matchers, + roots, + } } - /// The root answering `query` at `time_ms`, and the raw samples it needs. - /// Fails if the query is not planned or a stored pane it reads is missing. + /// The root answering `query` at `time_ms`, and the raw samples it needs: + /// its query-time windows, and for each `QueryRun` definition one range + /// covering every summary it reads that is not stored. Fails if the + /// query is not planned, a time does not fit, or an ingestion-built + /// summary it reads is missing. `progress` is what ingestion has seen + /// now; pass the returned copy to [`Self::evaluate`]. pub fn bound_raw_sample_windows( &self, query: &str, time_ms: i64, store: &SummaryStore, - ) -> Result<(usize, Vec), Unanswered> { + progress: IngestionProgress, + ) -> Result { let (root, program) = self.program(query)?; for input in &program.stored { pane(input, time_ms, store)?; } - let ranges = program + let mut ranges: Vec = program .raw .iter() .map(|input| { @@ -169,16 +263,53 @@ impl QueryEngine { }) }) .collect::>()?; - Ok((root, ranges)) + // One fetch per `QueryRun` definition, covering all its missing + // summaries. + let mut missing: BTreeMap<&DefinitionId, (i64, i64)> = BTreeMap::new(); + for input in &program.query_run { + let build = self.query_run_build(&input.definition); + let (start_ms, end_ms) = window(time_ms, input.offset_ms, build.definition.pane_ms)?; + if store + .get(&input.definition, Selection::ending_at(end_ms)) + .is_none() + { + let span = missing + .entry(&input.definition) + .or_insert((start_ms, end_ms)); + *span = (span.0.min(start_ms), span.1.max(end_ms)); + } + } + for (definition, (start_ms, end_ms)) in missing { + let build = self.query_run_build(definition); + ranges.push(BoundRawSampleWindow { + node: program.query_run_fetch_node[definition], + metric: build.metric().to_string(), + matchers: self.query_run_matchers[definition].clone(), + start_ms, + end_ms, + }); + } + Ok(Prepared { + root, + windows: ranges, + progress, + }) } - /// Runs `root` at `time_ms` over stored panes and `raw`, the samples of - /// each of its [`BoundRawSampleWindow`]s by node. + /// Runs `root` at `time_ms` over stored summaries and `raw`, the samples + /// of each of its [`BoundRawSampleWindow`]s by node. A stored output built in a `QueryRun` not stored + /// is built from its raw samples and stored in `store`, under the same + /// key as those built in an `IngestionRun`, if its coverage was closed in + /// `progress` (the snapshot [`Self::bound_raw_sample_windows`] chose the fetch by). + /// Then stored outputs built in a `QueryRun` older than their definition's retention, + /// measured from the newest closed pane at or before the watermark, are + /// evicted. pub fn evaluate( - &self, + &mut self, root: usize, time_ms: i64, - store: &SummaryStore, + store: &mut SummaryStore, + progress: &IngestionProgress, raw: &BTreeMap>, ) -> Result, Unanswered> { let program = self.roots[root] @@ -197,6 +328,70 @@ impl QueryEngine { Operator::source(pane.schema().clone(), vec![pane.clone()]).map_err(failed)?; sources.insert(input.node, Box::new(source)); } + for input in &program.query_run { + let build = self.query_run_builds[&input.definition] + .as_ref() + .expect("compiled roots read only `QueryRun` definitions that build"); + let (_, end_ms) = window(time_ms, input.offset_ms, build.definition.pane_ms)?; + let selection = Selection::ending_at(end_ms); + let summary = match store.get(&input.definition, selection) { + Some(summary) => { + self.query_run_stats.reused += 1; + summary.clone() + } + None => { + // `bound_raw_sample_windows` asked for these samples unless the + // summary was stored then and evicted since. + let samples = raw + .get(&program.query_run_fetch_node[&input.definition]) + .ok_or_else(|| { + Unanswered::MissingPane(format!( + "stored output built in a `QueryRun` ending at {end_ms} of {:?} was evicted", + input.definition + )) + })?; + // Built as at its coverage's end, so every evaluation + // that reads it reads the same summary. + let build_scope = Scope::Query { + evaluation_time_ms: end_ms, + revision: 1, + }; + let summary = build + .run(end_ms, samples, build_scope) + .map_err(|e| Unanswered::Execution(e.to_string()))?; + self.query_run_stats.built += 1; + // Stored only once closed, as the precompute engine closes + // panes: a sample may still arrive for an open one. + if progress.watermark_ms.is_some_and(|w| end_ms <= w) { + self.query_run_stats.stored += 1; + let metric_newest = progress.newest_sample_ms.get(build.metric()); + if metric_newest.is_none_or(|newest| *newest < end_ms) { + self.query_run_stats.stored_while_metric_behind += 1; + } + store.insert( + SummaryInstanceKey::new(input.definition.clone(), selection), + summary.clone(), + ); + } + summary + } + }; + let source = + Operator::source(summary.schema().clone(), vec![summary]).map_err(failed)?; + sources.insert(input.node, Box::new(source)); + } + if let Some(watermark) = progress.watermark_ms { + for build in self.query_run_builds.values().flatten() { + let definition = &build.definition; + let pane_ms = definition.pane_ms; + // As for stored outputs built in an `IngestionRun`: measured from the newest + // closed coverage, one pane of margin. + let newest_closed_end = watermark.div_euclid(pane_ms).saturating_mul(pane_ms); + let oldest = + definition.oldest_kept_end_ms(newest_closed_end.saturating_sub(pane_ms)); + store.evict_before(&definition.id, Selection::ending_at(oldest)); + } + } for input in &program.raw { let samples = raw.get(&input.node).map(Vec::as_slice).unwrap_or_default(); let rows = input @@ -214,6 +409,13 @@ impl QueryEngine { result_samples(&program.output, rows) } + /// The build of a `QueryRun` definition a compiled root reads. + fn query_run_build(&self, definition: &DefinitionId) -> &SummaryBuild { + self.query_run_builds[definition] + .as_ref() + .expect("compiled roots read only `QueryRun` definitions that build") + } + fn program(&self, query: &str) -> Result<(usize, &Program), Unanswered> { let query = query.trim(); let root = self @@ -265,6 +467,7 @@ fn pane<'a>( fn program( dag: &PhysicalASAPDAG, sds: &Sds, + query_run_builds: &BTreeMap>, windows: &HashMap, root: u64, ) -> Result { @@ -276,7 +479,7 @@ fn program( .or_default() .push(edge.producer as u64); } - let (mut stored, mut raw) = (Vec::new(), Vec::new()); + let (mut stored, mut query_run, mut raw) = (Vec::new(), Vec::new(), Vec::new()); let mut pending = vec![root]; let mut seen = BTreeSet::new(); while let Some(id) = pending.pop() { @@ -284,12 +487,25 @@ fn program( continue; } if let Some((definition, binding)) = sds.outputs.get(&id) { - stored.push(StoredInput { - node: id, - definition: definition.clone(), - offset_ms: binding.offset_ms, - pane_ms: sds.definitions[definition].pane_ms, - }); + match sds.definitions[definition].built_at { + StoredOutputBuild::IngestionRun => stored.push(StoredInput { + node: id, + definition: definition.clone(), + offset_ms: binding.offset_ms, + pane_ms: sds.definitions[definition].pane_ms, + }), + StoredOutputBuild::QueryRun => { + query_run_builds[definition] + .as_ref() + .map_err(Clone::clone)?; + query_run.push(QueryRunInput { + node: id, + definition: definition.clone(), + offset_ms: binding.offset_ms, + raw_input: binding.raw_input, + }); + } + } continue; } match &nodes[&id].payload { @@ -316,17 +532,30 @@ fn program( } } let mut inputs = BTreeMap::new(); - for input in &stored { - let schema = Arc::new(nodes[&input.node].output_schema.clone()); - inputs.insert(input.node, InputContract::bounded(schema)); + for node in stored + .iter() + .map(|i| i.node) + .chain(query_run.iter().map(|i| i.node)) + { + let schema = Arc::new(nodes[&node].output_schema.clone()); + inputs.insert(node, InputContract::bounded(schema)); } for input in &raw { inputs.insert(input.node, InputContract::bounded(input.schema.clone())); } let compiled = compile(dag, inputs, &[root]).map_err(|e| e.to_string())?; + let mut query_run_fetch_node: BTreeMap = BTreeMap::new(); + for input in &query_run { + let node = query_run_fetch_node + .entry(input.definition.clone()) + .or_insert(input.raw_input); + *node = (*node).min(input.raw_input); + } Ok(Program { compiled, stored, + query_run, + query_run_fetch_node, raw, output: nodes[&root].output_schema.clone(), }) diff --git a/data_plane/src/sds.rs b/data_plane/src/sds.rs index 184f5f2c..299c7528 100644 --- a/data_plane/src/sds.rs +++ b/data_plane/src/sds.rs @@ -86,6 +86,9 @@ impl SummaryDefinition { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct PaneBinding { pub offset_ms: i64, + /// The output's own `TimeRange` node: where its coverage's raw samples + /// feed when a stored output built in a `QueryRun` is built for it. + pub raw_input: u64, } /// The stored summaries of one plan. @@ -154,6 +157,7 @@ impl Sds { definition.retention_ms = definition.retention_ms.max(retention_ms); let binding = PaneBinding { offset_ms: output.pane.end_offset_ms, + raw_input: output.pane.time_range_node_id as u64, }; outputs.insert(node, (id, binding)); } diff --git a/data_plane/tests/precompute.rs b/data_plane/tests/precompute.rs index 72cb9269..004655fc 100644 --- a/data_plane/tests/precompute.rs +++ b/data_plane/tests/precompute.rs @@ -432,7 +432,7 @@ fn does_not_build_query_run_outputs() { let (mut engine, _) = precompute(plan); engine.ingest(&samples("latency_ms", &["a"], 0, 45 * MINUTE)); assert_eq!(engine.stats.panes_built, 0); - assert_eq!(engine.stats.samples_unrouted, 180); + assert_eq!(engine.stats.samples_query_run_only, 180); } /// Retention is measured from the newest evaluation the stored panes can @@ -473,3 +473,31 @@ fn keeps_one_pane_of_margin_beyond_the_retention() { // the one more the evaluation at 280 reads. assert_eq!(engine.store.len(), 7); } + +/// Samples of a metric only stored outputs built in a `QueryRun` read (B3) are not buffered +/// but move the watermark, which says when those summaries are closed. +#[test] +fn query_run_metrics_move_the_watermark() { + let (mut engine, _) = precompute_with(common::plan("example4-b3", true), 0); + assert_eq!(engine.watermark_ms(), None); + engine.ingest(&samples("latency_ms", &["a"], 0, 3 * MINUTE)); + assert_eq!(engine.stats.samples_query_run_only, 12); + assert_eq!(engine.stats.samples_accepted, 0); + assert_eq!(engine.watermark_ms(), Some(3 * MINUTE - 15_000)); +} + +/// Samples of a metric only `QueryRun` stored outputs read get the same future bound and +/// wall-clock cap as routed samples: one far ahead is dropped, and one +/// ahead of the clock but within the bound moves the watermark only to the +/// clock. +#[test] +fn query_run_metrics_are_bounded_like_routed_samples() { + let (mut engine, _) = precompute_with(common::plan("example4-b3", true), 0); + let now = 10 * MINUTE; + engine.ingest_at(&[sample("latency_ms", "a", 30 * DAY, 1.0)], now); + assert_eq!(engine.stats.samples_dropped_future, 1); + assert_eq!(engine.watermark_ms(), None); + engine.ingest_at(&[sample("latency_ms", "a", now + 2 * MINUTE, 1.0)], now); + assert_eq!(engine.stats.samples_query_run_only, 1); + assert_eq!(engine.watermark_ms(), Some(now)); +} diff --git a/data_plane/tests/query.rs b/data_plane/tests/query.rs index ae9b698e..2e12f204 100644 --- a/data_plane/tests/query.rs +++ b/data_plane/tests/query.rs @@ -81,14 +81,21 @@ fn fetch(ranges: &[BoundRawSampleWindow], all: &[RawSample]) -> BTreeMap Vec { - let (root, ranges) = f +fn answer(f: &mut Fixture, query: &str, time_ms: i64, all: &[RawSample]) -> Vec { + let prepared = f .query - .bound_raw_sample_windows(query, time_ms, &f.precompute.store) + .bound_raw_sample_windows(query, time_ms, &f.precompute.store, f.precompute.progress()) .unwrap(); + let raw = fetch(&prepared.windows, all); let mut result = f .query - .evaluate(root, time_ms, &f.precompute.store, &fetch(&ranges, all)) + .evaluate( + prepared.root, + time_ms, + &mut f.precompute.store, + &prepared.progress, + &raw, + ) .unwrap(); result.sort_by(|a, b| a.labels.cmp(&b.labels)); result @@ -112,23 +119,40 @@ fn stored_panes_answer_like_a_single_build() { assert!( panes .query - .bound_raw_sample_windows(P99, 70 * MINUTE, &panes.precompute.store) + .bound_raw_sample_windows( + P99, + 70 * MINUTE, + &panes.precompute.store, + Default::default() + ) .is_err(), "no panes before ingestion" ); panes.precompute.ingest(&all); - let (_, ranges) = panes + let ranges = panes .query - .bound_raw_sample_windows(P99, 70 * MINUTE, &panes.precompute.store) - .unwrap(); + .bound_raw_sample_windows( + P99, + 70 * MINUTE, + &panes.precompute.store, + Default::default(), + ) + .unwrap() + .windows; assert!(ranges.is_empty(), "every input is a stored pane"); - let from_panes = answer(&panes, P99, 70 * MINUTE, &all); + let from_panes = answer(&mut panes, P99, 70 * MINUTE, &all); - let single = fixture(common::plan("example4b", true)); - let (_, ranges) = single + let mut single = fixture(common::plan("example4b", true)); + let ranges = single .query - .bound_raw_sample_windows(P99, 70 * MINUTE, &single.precompute.store) - .unwrap(); + .bound_raw_sample_windows( + P99, + 70 * MINUTE, + &single.precompute.store, + Default::default(), + ) + .unwrap() + .windows; assert_eq!( ranges .iter() @@ -137,7 +161,7 @@ fn stored_panes_answer_like_a_single_build() { vec![(10 * MINUTE, 70 * MINUTE)], "one raw hour, left-open" ); - let from_raw = answer(&single, P99, 70 * MINUTE, &all); + let from_raw = answer(&mut single, P99, 70 * MINUTE, &all); assert_eq!(from_panes, from_raw); assert_eq!(from_panes.len(), 2); @@ -159,9 +183,9 @@ fn query_time_plans_read_raw_ranges() { ("http_requests_total", &[("job", "b"), ("instance", "1")]), ]; let all = samples(series, 0, 10 * MINUTE, |_, t| t as f64 / 1000.0); - let f = fixture(common::plan("example1", true)); + let mut f = fixture(common::plan("example1", true)); let result = answer( - &f, + &mut f, "sum by (job) (rate(http_requests_total[1m]))", 5 * MINUTE, &all, @@ -177,7 +201,7 @@ fn query_time_plans_read_raw_ranges() { assert!((by_job[1].1 - 1.0).abs() < 1e-9, "{by_job:?}"); let topk = answer( - &f, + &mut f, "topk by (job) (10, sum_over_time(http_requests_total[1m]))", 5 * MINUTE, &all, @@ -196,11 +220,13 @@ fn reports_what_it_cannot_answer() { let mut f = fixture(common::plan("example4b", false)); let store = &f.precompute.store; assert!(matches!( - f.query.bound_raw_sample_windows("up", 70 * MINUTE, store), + f.query + .bound_raw_sample_windows("up", 70 * MINUTE, store, Default::default()), Err(Unanswered::NotPlanned(_)) )); assert!(matches!( - f.query.bound_raw_sample_windows(P99, 70 * MINUTE, store), + f.query + .bound_raw_sample_windows(P99, 70 * MINUTE, store, Default::default()), Err(Unanswered::MissingPane(_)) )); let all = samples(&[("latency_ms", &[])], MINUTE, 75 * MINUTE, |_, t| t as f64); @@ -208,16 +234,17 @@ fn reports_what_it_cannot_answer() { let store = &f.precompute.store; assert!(f .query - .bound_raw_sample_windows(P99, 70 * MINUTE, store) + .bound_raw_sample_windows(P99, 70 * MINUTE, store, Default::default()) .is_ok()); assert!(matches!( f.query - .bound_raw_sample_windows(P99, 70 * MINUTE + 1, store), + .bound_raw_sample_windows(P99, 70 * MINUTE + 1, store, Default::default()), Err(Unanswered::MissingPane(_)) )); // The pane ending at 10 min was skipped as possibly partial. assert!(matches!( - f.query.bound_raw_sample_windows(P99, 60 * MINUTE, store), + f.query + .bound_raw_sample_windows(P99, 60 * MINUTE, store, Default::default()), Err(Unanswered::MissingPane(_)) )); } @@ -230,20 +257,23 @@ fn extreme_evaluation_times_are_not_answered() { for time_ms in [i64::MIN, i64::MIN + 1] { assert!(matches!( f.query - .bound_raw_sample_windows(P99, time_ms, &f.precompute.store), + .bound_raw_sample_windows(P99, time_ms, &f.precompute.store, Default::default()), Err(Unanswered::Unsupported(_)) )); } // The window ending at `i64::MAX` fits; binding it does not panic. assert!(f .query - .bound_raw_sample_windows(P99, i64::MAX, &f.precompute.store) + .bound_raw_sample_windows(P99, i64::MAX, &f.precompute.store, Default::default()) .is_ok()); let panes = fixture(common::plan("example4b", false)); assert!(matches!( - panes - .query - .bound_raw_sample_windows(P99, i64::MIN, &panes.precompute.store), + panes.query.bound_raw_sample_windows( + P99, + i64::MIN, + &panes.precompute.store, + Default::default() + ), Err(Unanswered::Unsupported(_)) )); } @@ -276,12 +306,12 @@ fn examples_answer_like_the_query_time_reference() { let reference = plan.query_time_reference().unwrap(); let mut installed = fixture(plan.clone()); installed.precompute.ingest(&all); - let by_reference = fixture(reference); + let mut by_reference = fixture(reference); for query in plan.queries.iter().map(|q| q.query.as_str()) { - let expected = answer(&by_reference, query, 70 * MINUTE, &all); + let expected = answer(&mut by_reference, query, 70 * MINUTE, &all); assert!(!expected.is_empty(), "{example} raw={raw} {query}"); assert_eq!( - answer(&installed, query, 70 * MINUTE, &all), + answer(&mut installed, query, 70 * MINUTE, &all), expected, "{example} raw={raw} {query}" ); @@ -300,14 +330,16 @@ fn bound_raw_sample_windows_carry_the_scans_label_matchers() { ); let plan = control_plane::plan_workload(&workload, &common::inputs(true)).unwrap(); let f = fixture(plan); - let (_, ranges) = f + let ranges = f .query .bound_raw_sample_windows( "quantile_over_time(0.99, latency_ms{job=\"a\"}[60m])", 70 * MINUTE, &f.precompute.store, + Default::default(), ) - .unwrap(); + .unwrap() + .windows; assert_eq!(ranges.len(), 1); assert_eq!( ranges[0].matchers, diff --git a/data_plane/tests/query_run.rs b/data_plane/tests/query_run.rs new file mode 100644 index 00000000..85ececb0 --- /dev/null +++ b/data_plane/tests/query_run.rs @@ -0,0 +1,381 @@ +//! Stored outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`): +//! #509 Example 4's B3, five 1-minute KLL panes built at query time and +//! stored in the SDS for the next evaluations of +//! `quantile_over_time(0.99, latency_ms[5m])` every minute. + +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use asap_executor::values::Batch; +use asap_planner::{InstallablePlan, StoredOutputBuild}; +use data_plane::engine::{EngineHandle, EngineOptions}; +use data_plane::precompute::RawSample; +use data_plane::query::{ + BoundRawSampleWindow, IngestionProgress, QueryEngine, QueryRunStats, ResultSample, +}; +use data_plane::sds::{DefinitionId, Sds, Selection, SummaryStore}; + +const MINUTE: i64 = 60_000; +const P99: &str = "quantile_over_time(0.99, latency_ms[5m])"; + +/// A query engine and the SDS store its stored outputs built in a `QueryRun` go to. +struct Fixture { + engine: QueryEngine, + store: SummaryStore, + definition: DefinitionId, +} + +fn fixture(plan: &InstallablePlan) -> Fixture { + let sds = Sds::from_plan(plan).unwrap(); + assert!( + sds.ingestion_build_order().unwrap().is_empty(), + "nothing is built at ingestion" + ); + let definition = sds + .definitions + .values() + .find(|d| d.built_at == StoredOutputBuild::QueryRun) + .map(|d| d.id.clone()) + .unwrap_or_else(|| DefinitionId(String::new())); + Fixture { + engine: QueryEngine::new(plan, &sds), + store: SummaryStore::new(), + definition, + } +} + +impl Fixture { + /// The ends, in minutes, of the stored stored outputs built in a `QueryRun`. + fn ends(&self) -> Vec { + (0..=30) + .map(|m| m * MINUTE) + .filter(|end| { + self.store + .get(&self.definition, Selection::ending_at(*end)) + .is_some() + }) + .map(|end| end / MINUTE) + .collect() + } +} + +/// Two series every 15 s in `[0, 30 min]`, the `n`th sample `25n mod 101` +/// (plus 100 for series b): each 5-minute window's p99 falls in an older +/// pane and changes every minute. KLL is exact below k. +fn samples() -> Vec { + let series: Vec<_> = ["a", "b"] + .iter() + .map(|job| { + Arc::new(BTreeMap::from([ + ("__name__".to_string(), "latency_ms".to_string()), + ("job".to_string(), job.to_string()), + ])) + }) + .collect(); + (0..=30 * MINUTE) + .step_by(15_000) + .flat_map(|t| { + series.iter().enumerate().map(move |(i, labels)| RawSample { + labels: labels.clone(), + timestamp_ms: t, + value: (i * 100) as f64 + (t / 15_000 * 25 % 101) as f64, + }) + }) + .collect() +} + +fn fetch(ranges: &[BoundRawSampleWindow], 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() +} + +/// Answers `P99` at `time_ms` with ingestion's watermark at `watermark_ms` +/// (every metric's newest sample there too), returning the answer and the +/// bound raw-sample windows it read. +fn answer_at_watermark( + k: &mut Fixture, + time_ms: i64, + watermark_ms: Option, + all: &[RawSample], +) -> (Vec, Vec) { + let progress = IngestionProgress { + watermark_ms, + newest_sample_ms: watermark_ms + .map(|w| BTreeMap::from([("latency_ms".to_string(), w)])) + .unwrap_or_default(), + }; + let prepared = k + .engine + .bound_raw_sample_windows(P99, time_ms, &k.store, progress) + .unwrap(); + let ranges = prepared.windows.clone(); + let mut result = k + .engine + .evaluate( + prepared.root, + time_ms, + &mut k.store, + &prepared.progress, + &fetch(&ranges, all), + ) + .unwrap(); + result.sort_by(|a, b| a.labels.cmp(&b.labels)); + (result, ranges) +} + +/// Answers `P99` at `time_ms` once every sample through `time_ms` has +/// arrived: the watermark is `time_ms`. +fn answer( + k: &mut Fixture, + time_ms: i64, + all: &[RawSample], +) -> (Vec, Vec) { + answer_at_watermark(k, time_ms, Some(time_ms), all) +} + +fn spans(ranges: &[BoundRawSampleWindow]) -> Vec<(i64, i64)> { + ranges + .iter() + .map(|r| (r.start_ms / MINUTE, r.end_ms / MINUTE)) + .collect() +} + +/// The installed B3 plan and the Planner's reference for it: the same +/// selection with nothing materialized. +fn plans() -> (InstallablePlan, InstallablePlan) { + let plan = common::plan("example4-b3", true); + let reference = plan.query_time_reference().unwrap(); + (plan, reference) +} + +/// The first evaluation builds all five summaries from one fetch and +/// answers like the reference; the next one, a minute later, fetches only +/// the newest minute, builds one summary, reuses four, and still answers +/// like the reference. +#[test] +fn query_run_outputs_answer_like_the_reference_and_are_reused() { + let (plan, reference_plan) = plans(); + let all = samples(); + let mut k = fixture(&plan); + let mut reference = fixture(&reference_plan); + + let (first, ranges) = answer(&mut k, 10 * MINUTE, &all); + assert_eq!( + spans(&ranges), + vec![(5, 10)], + "one fetch for five summaries" + ); + assert_eq!(first, answer(&mut reference, 10 * MINUTE, &all).0); + assert_eq!(first.len(), 2); + assert_eq!( + k.engine.query_run_stats, + QueryRunStats { + built: 5, + stored: 5, + stored_while_metric_behind: 0, + reused: 0 + } + ); + + let (second, ranges) = answer(&mut k, 11 * MINUTE, &all); + assert_eq!(spans(&ranges), vec![(10, 11)], "only the newest summary"); + assert_eq!(second, answer(&mut reference, 11 * MINUTE, &all).0); + assert_ne!(second, first, "the answer moves with the window"); + assert_eq!( + ( + k.engine.query_run_stats.built, + k.engine.query_run_stats.reused + ), + (6, 4) + ); + + // The same evaluation again builds nothing. + let (again, ranges) = answer(&mut k, 11 * MINUTE, &all); + assert!(ranges.is_empty()); + assert_eq!(again, second); + assert_eq!( + ( + k.engine.query_run_stats.built, + k.engine.query_run_stats.reused + ), + (6, 9) + ); + assert_eq!( + reference.engine.query_run_stats, + QueryRunStats::default(), + "the reference stores nothing" + ); +} + +/// Stored stored outputs built in a `QueryRun` are evicted by the plan's retention (5 min, +/// the oldest summary a 5-minute window reads), plus one pane, measured +/// from the newest closed coverage at or before the watermark, as for +/// stored outputs built in an `IngestionRun`. A far-future evaluation does not move that +/// point: its summary states are built but not stored, and storing resumes. +#[test] +fn query_run_outputs_are_evicted_by_the_plans_retention() { + let (plan, _) = plans(); + let retention = plan.stored_outputs.iter().map(|o| o.retention_ms).max(); + assert_eq!(retention, Some(5 * MINUTE as u64)); + let all = samples(); + let mut k = fixture(&plan); + answer(&mut k, 10 * MINUTE, &all); + assert_eq!(k.ends(), vec![6, 7, 8, 9, 10]); + answer(&mut k, 11 * MINUTE, &all); + assert_eq!(k.ends(), vec![6, 7, 8, 9, 10, 11]); + answer(&mut k, 12 * MINUTE, &all); + assert_eq!(k.ends(), vec![7, 8, 9, 10, 11, 12]); + // An evaluation far ahead of the watermark is answered, but its + // summaries' coverages are not closed, so nothing is stored or evicted. + let (far, ranges) = answer_at_watermark(&mut k, 20 * MINUTE, Some(12 * MINUTE), &all); + assert_eq!(far.len(), 2); + assert_eq!(spans(&ranges), vec![(15, 20)]); + assert_eq!(k.ends(), vec![7, 8, 9, 10, 11, 12]); + // Keeping resumes with the watermark. + answer(&mut k, 13 * MINUTE, &all); + assert_eq!(k.ends(), vec![8, 9, 10, 11, 12, 13]); +} + +/// A summary state is stored only once its coverage is closed (its end at or +/// before the watermark). One read before then is built for that +/// evaluation but not stored, so a later evaluation builds it again and +/// includes a sample that arrived late. +#[test] +fn query_run_outputs_are_stored_only_once_closed() { + let (plan, reference_plan) = plans(); + let mut all = samples(); + let mut k = fixture(&plan); + let mut reference = fixture(&reference_plan); + + // Watermark at 9.5 min: the summary ending at 10 min is still open. + let (first, ranges) = answer_at_watermark(&mut k, 10 * MINUTE, Some(570_000), &all); + assert_eq!(spans(&ranges), vec![(5, 10)]); + assert_eq!(first, answer(&mut reference, 10 * MINUTE, &all).0); + assert_eq!( + k.ends(), + vec![6, 7, 8, 9], + "the open summary state is not stored" + ); + + // A late sample for the open coverage arrives before it closes. + let late = RawSample { + timestamp_ms: 10 * MINUTE - 1_000, + value: 1_000.0, + ..all.iter().find(|s| s.labels["job"] == "a").unwrap().clone() + }; + all.push(late); + let (again, ranges) = answer_at_watermark(&mut k, 10 * MINUTE, Some(585_000), &all); + assert_eq!( + spans(&ranges), + vec![(9, 10)], + "the open summary is built again" + ); + assert_eq!(again, answer(&mut reference, 10 * MINUTE, &all).0); + assert_ne!(again, first, "the late sample is included"); + + // Once the watermark reaches 10 min, the summary state is stored. + answer_at_watermark(&mut k, 10 * MINUTE, Some(10 * MINUTE), &all); + assert_eq!(k.ends(), vec![6, 7, 8, 9, 10]); + // Without any sample there is no watermark, and nothing is stored. + let mut fresh = fixture(&plan); + answer_at_watermark(&mut fresh, 10 * MINUTE, None, &all); + assert!(fresh.store.is_empty()); +} + +/// The store decision uses the watermark when the evaluation chose what to +/// fetch, not when it runs: ingestion that closes a coverage after the +/// fetch was chosen does not make the partial summary built from that +/// fetch permanent. +#[tokio::test] +async fn stores_by_the_watermark_the_fetch_was_chosen_at() { + let (plan, _) = plans(); + let options = EngineOptions { + precompute: data_plane::precompute::PrecomputeOptions { + allowed_lateness_ms: 0, + ..Default::default() + }, + ..EngineOptions::default() + }; + let (engine, _) = EngineHandle::spawn(serde_json::to_string(&plan).unwrap(), options).unwrap(); + let all = samples(); + let through = |t: i64| -> Vec { + all.iter() + .filter(|s| s.timestamp_ms <= t) + .cloned() + .collect() + }; + let settle = |engine: &EngineHandle| { + let engine = engine.clone(); + async move { + while engine.queued_samples() > 0 { + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + } + }; + engine.try_ingest(through(570_000)).unwrap(); + settle(&engine).await; + let prepared = engine + .bound_raw_sample_windows(P99.into(), 10 * MINUTE) + .await + .unwrap(); + assert_eq!(prepared.progress.watermark_ms, Some(570_000)); + // The fetch sees only what had arrived; then ingestion closes 10 min. + let raw = fetch(&prepared.windows, &through(570_000)); + engine.try_ingest(through(11 * MINUTE)).unwrap(); + settle(&engine).await; + engine.evaluate(prepared, 10 * MINUTE, raw).await.unwrap(); + assert_eq!(engine.query_run_stats().stored, 4, "6..9 stored, 10 not"); + // So the next evaluation builds the summary ending at 10 min again. + let prepared = engine + .bound_raw_sample_windows(P99.into(), 10 * MINUTE) + .await + .unwrap(); + assert_eq!(spans(&prepared.windows), vec![(9, 10)]); +} + +/// A summary state stored while its own metric's newest sample has not reached its +/// end (the global watermark moved on another metric) is counted. +#[test] +fn counts_outputs_stored_while_their_metric_is_behind() { + let (plan, _) = plans(); + let all = samples(); + let mut k = fixture(&plan); + let progress = IngestionProgress { + watermark_ms: Some(10 * MINUTE), + newest_sample_ms: BTreeMap::from([("latency_ms".to_string(), 8 * MINUTE)]), + }; + let prepared = k + .engine + .bound_raw_sample_windows(P99, 10 * MINUTE, &k.store, progress) + .unwrap(); + let raw = fetch(&prepared.windows, &all); + k.engine + .evaluate( + prepared.root, + 10 * MINUTE, + &mut k.store, + &prepared.progress, + &raw, + ) + .unwrap(); + assert_eq!(k.engine.query_run_stats.stored, 5); + assert_eq!( + k.engine.query_run_stats.stored_while_metric_behind, 2, + "9 and 10" + ); +} diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs index 53019fd4..f4663089 100644 --- a/data_plane/tests/sds.rs +++ b/data_plane/tests/sds.rs @@ -116,7 +116,7 @@ fn identity_depends_on_the_source() { ); } -/// Outputs built in a query run and kept are stored summaries too: their +/// Stored outputs built in a `QueryRun` are stored summaries too: their /// definition says so, and no ingestion run builds them. #[test] fn query_run_outputs_have_query_run_definitions() { @@ -251,3 +251,27 @@ fn build_order_refuses_ingestion_reading_query_run() { }) ); } + +/// B3's five 1-minute KLL panes, stored outputs built in a `QueryRun` +/// (`MatAtQueryKeptAcrossEvaluations`), are one `QueryRun` definition, each output reading the +/// summary `i * 1min` before evaluation. Nothing is built at ingestion. +#[test] +fn query_run_outputs_share_one_definition() { + let sds = Sds::from_plan(&plan("example4-b3", true)).unwrap(); + assert!(sds.ingestion_build_order().unwrap().is_empty()); + let mut query_run = sds + .definitions + .values() + .filter(|d| d.built_at == StoredOutputBuild::QueryRun); + let definition = query_run.next().unwrap(); + assert!(query_run.next().is_none()); + assert_eq!(definition.pane_ms, 60_000); + let mut offsets: Vec = sds + .outputs + .values() + .filter(|(id, _)| id == &definition.id) + .map(|(_, binding)| binding.offset_ms) + .collect(); + offsets.sort(); + assert_eq!(offsets, (0..5).map(|i| i * 60_000).collect::>()); +} diff --git a/examples/planner-layering-example4-b3.json b/examples/planner-layering-example4-b3.json new file mode 100644 index 00000000..9ad9481d --- /dev/null +++ b/examples/planner-layering-example4-b3.json @@ -0,0 +1,44 @@ +{ + "query_workload": { + "language": "promql", + "query_batch": null, + "repeating_queries": [ + { + "query": "quantile_over_time(0.99, latency_ms[5m])", + "demand": { + "fixed_interval": 60000 + }, + "requirements": { + "accuracy": { + "accuracy_metric": "rank", + "max_error": 0.01, + "failure_probability": 0.01, + "accuracy_scope": "whole_result" + }, + "max_latency_ms": 200.0 + }, + "predictability": { + "predictable": { + "known_at": null + } + } + } + ] + }, + "data_workload": { + "arrival": { + "ingesting": { + "rows_per_second": 10000.0 + } + }, + "sources": { + "latency_ms": { + "series_count": 10, + "metric_type": null, + "sample_interval": 1, + "stream_norms": null, + "max_distinct_per_evaluation": null + } + } + } +}