diff --git a/README.md b/README.md index afbd541f..8c72d1a3 100644 --- a/README.md +++ b/README.md @@ -46,8 +46,11 @@ across the ingestion workers by a hash of their label set; a pane closes passes its end, and later samples for it are dropped. The first pane after startup is skipped because it may be partial. Each query, including every step of a range query, runs on one query thread; queries read the stored -panes concurrently, by shared reference. The data plane uses jemalloc as its -allocator. +panes concurrently, by shared reference. Ingestion updates each open pane's summary +state per sample, with the updater the Planner executor's batch build uses; +a definition whose build has another shape (a row filter, a keyed family, +operators between its range and its summary) is built from the pane's +buffered samples instead. The data plane uses jemalloc as its allocator. 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 diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index 3a657b81..7353efcf 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -148,8 +148,9 @@ impl EngineHandle { for (index, mut commands) in receivers.into_iter().enumerate() { let (plan_json, shared, peers) = (plan_json.clone(), precompute.clone(), peers.clone()); spawn(format!("ingest-{index}"), move |ready| { - let started = install(&plan_json) - .and_then(|(plan, sds)| Ok(ShardWorker::new(&plan, &sds, index, shared)?)); + let started = install(&plan_json).and_then(|(plan, sds)| { + Ok(ShardWorker::new(&plan, &sds, index, shared, true)?) + }); let mut worker = match started { Ok(worker) => { let _ = ready.send(Ok(())); diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 7e6954f2..4eb8ba32 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -9,6 +9,7 @@ //! answer, is forwarded to Prometheus. pub mod engine; +mod per_sample; pub mod precompute; pub mod prometheus; pub mod query; diff --git a/data_plane/src/per_sample.rs b/data_plane/src/per_sample.rs new file mode 100644 index 00000000..655bb676 --- /dev/null +++ b/data_plane/src/per_sample.rs @@ -0,0 +1,293 @@ +//! Per-sample ingestion: a definition's build, `Scan -> TimeShift* -> +//! TimeRange(pane) -> SummaryAgg`, run one sample at a time instead of +//! over a pane's buffered rows. +//! +//! - **Per series, once:** the stateless prefix. The executor's +//! `RawSampleSelector` turns the series' first sample into its input row, +//! which applies the `Scan`'s label matchers; the row's group columns give +//! its group. The result is cached by label set. +//! - **Per sample:** the value updates its open pane's state for its group +//! in place, through the updater the executor's batch build uses for the +//! family (`asap_executor::factory::create_planner_accumulator`), with the +//! same value and timestamp, in the same order. +//! - **Per group, once:** the output row's layout. The definition's compiled +//! build runs over the group's first row, and its output row is kept as +//! the group's template; at pane close each group's row is its template +//! with the pane's timestamp and the group's state. +//! +//! A build of another shape (a row filter, a keyed or shared-grid family, +//! an operator between the `TimeRange` and the `SummaryAgg`, matchers on the +//! sample's value or time) keeps the batch build. + +use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; + +use asap_executor::factory::{create_planner_accumulator, AccumulatorUpdater}; +use asap_executor::planner::ir::operator::Reduction; +use asap_executor::planner::ir::scalar::ColumnRef; +use asap_executor::planner::ir::schema::{ + ExactKind, FieldDataType, GroupingStrategy, SummaryInputExpr, SummaryUpdate, +}; +use asap_executor::planner::ir::{ASAPOp, NonASAPOp, Operator}; +use asap_executor::runtime::Scope; +use asap_executor::values::{Batch, Value}; +use asap_planner::PhysicalASAPDAG; + +use crate::precompute::{PaneBuild, RawSample}; + +type Labels = Arc>; + +/// One group of a definition: its key and what its output rows repeat. +struct Group { + /// The group's first input row, for its template. + row: Batch, + /// Its output row, with the timestamp and state to be replaced; built + /// when the group's first pane closes. + template: Option>, +} + +/// One group's state in one open pane. +struct OpenState { + updater: Box, + /// The previous sample's time, for counters, which need increasing times. + previous_ms: Option, +} + +/// One open pane: its groups' states, by group key, or why it failed. +type OpenPane = Result>, OpenState>, String>; + +/// A definition's build, run per sample. +pub(crate) struct PerSampleBuild { + family: FieldDataType, + update: SummaryUpdate, + grouping: GroupingStrategy, + /// Input columns: the value, the time, the group key. + value: usize, + time: usize, + groups: Vec, + /// Counters (rate, increase) need strictly increasing times in a group. + ordered_time: bool, + /// Output columns of the state and the timestamp. + state_column: usize, + time_column: usize, + /// Each series' group key, or `None` if the `Scan` does not select it. + series: HashMap>>>, + group_rows: HashMap>, Group>, + /// How much the output timestamp is after the pane's end. + time_offset_ms: Option, + open: BTreeMap, +} + +impl PerSampleBuild { + /// The per-sample build of `pane`'s definition, or `None` when its shape + /// keeps the batch build. + pub(crate) fn new(dag: &PhysicalASAPDAG, pane: &PaneBuild) -> Option { + let definition = &pane.definition; + let node = |id: u64| dag.nodes.iter().find(|n| n.id as u64 == id); + let builder = node(definition.builder)?; + let inputs: Vec = dag + .edges + .iter() + .filter(|e| e.consumer as u64 == definition.builder) + .map(|e| e.producer as u64) + .collect(); + if inputs != [definition.raw_input()] { + return None; + } + let Operator::ASAP(ASAPOp::SummaryAgg { + family, + input: update, + reduction, + grouping, + filter: None, + .. + }) = &builder.payload + else { + return None; + }; + if update.item.is_some() || *grouping != GroupingStrategy::PerSubpopulationInstance { + return None; + } + let input = &node(definition.raw_input())?.output_schema; + let SummaryInputExpr::Column(column) = &update.weight else { + return None; + }; + let name = match column { + ColumnRef::SampleValue => "value", + ColumnRef::Named(name) => name.as_str(), + _ => return None, + }; + let value = input.fields.iter().position(|f| f.name == name)?; + let time = input.time_index?; + let groups: Vec = match reduction { + Reduction::PerEntity => (0..input.fields.len()) + .filter(|&c| c != value && c != time) + .collect(), + Reduction::Reduce(keys) if !keys.is_without() && !keys.keys().is_empty() => { + keys.keys().to_vec() + } + _ => return None, + }; + // The cached prefix holds only if the matchers read labels alone. + let scan = node(definition.window.scan_node_id as u64)?; + let Operator::NonASAP(NonASAPOp::Scan { predicates, .. }) = &scan.payload else { + return None; + }; + let read = serde_json::to_value(predicates).ok()?; + if columns_read(&read).any(|c| c == value || c == time) { + return None; + } + create_planner_accumulator(family, update, grouping).ok()?; + let output = &definition.output_schema; + let state_column = output + .fields + .iter() + .position(|f| !matches!(f.dtype, FieldDataType::Plain(_)))?; + let time_column = output.time_index?; + let ordered_time = matches!( + family, + FieldDataType::ExactAggregate(ExactKind::Rate | ExactKind::Increase, _) + ); + Some(Self { + family: family.clone(), + update: update.clone(), + grouping: grouping.clone(), + value, + time, + groups, + ordered_time, + state_column, + time_column, + series: HashMap::new(), + group_rows: HashMap::new(), + time_offset_ms: None, + open: BTreeMap::new(), + }) + } + + /// Updates the pane ending at `end_ms` with `sample`. + pub(crate) fn update(&mut self, pane: &PaneBuild, sample: &RawSample, end_ms: i64) { + let key = match self.series.get(&sample.labels) { + Some(key) => key.clone(), + None => { + let key = self.series_group(pane, sample); + self.series.insert(sample.labels.clone(), key.clone()); + key + } + }; + let Some(key) = key else { return }; + let Ok(states) = self + .open + .entry(end_ms) + .or_insert_with(|| Ok(BTreeMap::new())) + else { + return; + }; + let state = match states.get_mut(&key) { + Some(state) => state, + None => { + let updater = + create_planner_accumulator(&self.family, &self.update, &self.grouping) + .expect("checked at install"); + states.entry(key).or_insert(OpenState { + updater, + previous_ms: None, + }) + } + }; + let time = sample.timestamp_ms; + let result = if self.ordered_time && state.previous_ms.is_some_and(|p| time <= p) { + Err("counter samples must have strictly increasing timestamps within each group".into()) + } else { + state.previous_ms = Some(time); + state + .updater + .update_value(&Value::Float64(sample.value), time) + }; + if let Err(error) = result { + self.open.insert(end_ms, Err(error)); + } + } + + /// The group of `sample`'s series: its input row's group columns, if + /// the `Scan` selects it. + fn series_group(&mut self, pane: &PaneBuild, sample: &RawSample) -> Option>> { + let row = pane + .selector + .rows(sample.timestamp_ms, &[sample.to_sample()]) + .ok()?; + let cells = row.rows().first()?; + debug_assert!(matches!(cells[self.value], Value::Float64(_))); + debug_assert!(matches!(cells[self.time], Value::Timestamp(_))); + let key = self + .groups + .iter() + .map(|&c| cells[c].key()) + .collect::, _>>() + .ok()?; + self.group_rows.entry(key.clone()).or_insert(Group { + row, + template: None, + }); + Some(key) + } + + /// The pane ending at `end_ms`, which closes: each group's row, in group + /// key order as the batch build emits them. + pub(crate) fn close( + &mut self, + pane: &PaneBuild, + end_ms: i64, + scope: Scope, + ) -> Result { + let states = self + .open + .remove(&end_ms) + .unwrap_or_else(|| Ok(BTreeMap::new()))?; + let mut rows = Vec::with_capacity(states.len()); + for (key, state) in states { + let group = self.group_rows.get_mut(&key).expect("grouped series"); + if group.template.is_none() { + let built = pane + .run_rows(group.row.clone(), scope.clone()) + .map_err(|e| e.to_string())?; + let [row] = built.rows() else { + return Err("a group's build did not give one row".into()); + }; + let Value::Timestamp(time) = row[self.time_column] else { + return Err("output time column is not a timestamp".into()); + }; + let offset = time - end_ms; + if *self.time_offset_ms.get_or_insert(offset) != offset { + return Err("output timestamps differ between groups".into()); + } + group.template = Some(row.clone()); + } + let mut row = group.template.clone().expect("built above"); + row[self.time_column] = + Value::Timestamp(end_ms + self.time_offset_ms.expect("set with the template")); + row[self.state_column] = Value::Summary { + family: self.family.clone(), + state: Arc::from(state.updater.into_accumulator()), + }; + rows.push(row); + } + let schema = Arc::new(pane.definition.output_schema.clone()); + Batch::try_new(schema, rows).map_err(|e| e.to_string()) + } +} + +/// The column indices an expression's JSON form reads (`{"Column": i}`). +fn columns_read(expression: &serde_json::Value) -> Box + '_> { + match expression { + serde_json::Value::Object(fields) => Box::new(fields.iter().flat_map(|(name, value)| { + let column = match (name.as_str(), value) { + ("Column", serde_json::Value::Number(n)) => n.as_u64().map(|c| c as usize), + _ => None, + }; + column.into_iter().chain(columns_read(value)) + })), + serde_json::Value::Array(items) => Box::new(items.iter().flat_map(columns_read)), + _ => Box::new(std::iter::empty()), + } +} diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 656df256..fe171eb3 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -29,6 +29,7 @@ use asap_planner::{InstallablePlan, PhysicalASAPDAG}; use futures::executor::block_on; use futures::StreamExt; +use crate::per_sample::PerSampleBuild; use crate::sds::{DefinitionId, Sds, SummaryDefinition, SummaryStore}; /// One raw sample: its series' full label set (including `__name__`), time @@ -112,7 +113,11 @@ impl PaneBuild { end_ms: i64, scope: Scope, ) -> Result { - let rows = self.selector.rows(end_ms, samples)?; + self.run_rows(self.selector.rows(end_ms, samples)?, scope) + } + + /// The build over `rows`, input rows already selected, under `scope`. + pub(crate) fn run_rows(&self, rows: Batch, scope: Scope) -> Result { let input: Source<'_> = Box::new(Operator::source(rows.schema().clone(), vec![rows])?); let build = self .build @@ -287,7 +292,9 @@ pub fn shard_of(labels: &BTreeMap, workers: usize) -> usize { struct Builder { pane: PaneBuild, - /// This worker's samples of open panes, by pane end. + /// The build run per sample, unless its shape keeps the batch build. + per_sample: Option, + /// For the batch build, this worker's samples of open panes, by pane end. open: BTreeMap>, /// End of the newest pane this worker built its part of, or skipped; /// samples in it or before it are dropped. `None` before the first @@ -309,19 +316,30 @@ pub struct ShardWorker { impl ShardWorker { /// Worker `index` of `shared`'s workers, with one build per - /// ingestion-time definition of `sds`, the stored summaries of `plan`. + /// ingestion-time definition of `sds`, the stored summaries of `plan`: + /// per sample where its shape allows ([`crate::per_sample`]) and + /// `per_sample` is set, else over each pane's buffered samples. pub fn new( plan: &InstallablePlan, sds: &Sds, index: usize, shared: Arc, + per_sample: bool, ) -> Result { let builders = sds .definitions .values() .map(|definition| { + let pane = PaneBuild::new(&plan.dag.dag, definition)?; + let per_sample = per_sample + .then(|| PerSampleBuild::new(&plan.dag.dag, &pane)) + .flatten(); + if index == 0 { + tracing::info!(definition = ?definition.id, per_sample = per_sample.is_some(), "pane build"); + } Ok(Builder { - pane: PaneBuild::new(&plan.dag.dag, definition)?, + pane, + per_sample, open: BTreeMap::new(), built_through: None, }) @@ -367,11 +385,14 @@ impl ShardWorker { dropped += 1; continue; } - builder - .open - .entry(end) - .or_default() - .push(sample.to_sample()); + match &mut builder.per_sample { + Some(per_sample) => per_sample.update(&builder.pane, sample, end), + None => builder + .open + .entry(end) + .or_default() + .push(sample.to_sample()), + } buffered = true; } accepted += u64::from(buffered); @@ -419,16 +440,21 @@ impl ShardWorker { .get_or_insert(pane_end(start_ms, pane_ms)); let mut end = built + pane_ms; while end <= watermark { - let samples = builder.open.remove(&end).unwrap_or_default(); let scope = Scope::Ingestion { window_start_ms: end - pane_ms, window_end_ms: end, revision: 1, }; - let part = builder - .pane - .run(&samples, end, scope) - .map_err(|e| e.to_string()); + let part = match &mut builder.per_sample { + Some(per_sample) => per_sample.close(&builder.pane, end, scope), + None => { + let samples = builder.open.remove(&end).unwrap_or_default(); + builder + .pane + .run(&samples, end, scope) + .map_err(|e| e.to_string()) + } + }; self.shared.add_part(&builder.pane.definition, end, part); builder.built_through = Some(end); end += pane_ms; @@ -460,14 +486,43 @@ impl Precompute { sds: &Sds, allowed_lateness_ms: i64, shards: usize, + ) -> Result { + Self::workers(plan, sds, allowed_lateness_ms, shards, true) + } + + /// One worker that builds every pane over its buffered samples with + /// the executor's batch build: the reference per-sample builds equal. + pub fn batch( + plan: &InstallablePlan, + sds: &Sds, + allowed_lateness_ms: i64, + ) -> Result { + Self::workers(plan, sds, allowed_lateness_ms, 1, false) + } + + fn workers( + plan: &InstallablePlan, + sds: &Sds, + allowed_lateness_ms: i64, + shards: usize, + per_sample: bool, ) -> Result { let shared = Arc::new(Shared::new(shards, allowed_lateness_ms)); let workers = (0..shared.workers()) - .map(|index| ShardWorker::new(plan, sds, index, shared.clone())) + .map(|index| ShardWorker::new(plan, sds, index, shared.clone(), per_sample)) .collect::>()?; Ok(Self { shared, workers }) } + /// How many definitions are built per sample. + pub fn per_sample_definitions(&self) -> usize { + self.workers[0] + .builders + .iter() + .filter(|b| b.per_sample.is_some()) + .count() + } + /// Routes `samples` to their workers, then lets every worker build the /// panes the global watermark has passed. pub fn ingest(&mut self, samples: &[RawSample]) { diff --git a/data_plane/tests/per_sample.rs b/data_plane/tests/per_sample.rs new file mode 100644 index 00000000..d15a3098 --- /dev/null +++ b/data_plane/tests/per_sample.rs @@ -0,0 +1,149 @@ +//! Ingestion updates each open pane's summary state per sample, through the +//! executor's own updater; the panes it stores equal the executor's batch +//! build of the same samples. + +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use asap_executor::planner::ir::schema::{FieldDataType, SketchStatistic}; +use asap_executor::summary_kernels::exact::ExactAccumulator; +use asap_executor::values::{Batch, Value}; +use data_plane::precompute::{Precompute, RawSample}; +use data_plane::sds::Sds; + +const MINUTE: i64 = 60_000; + +/// Samples every 15 s in `[from, to)` of `metric` for each `job`; values +/// grow with time and differ by job. +fn samples(metric: &str, jobs: usize, from: i64, to: i64) -> Vec { + let series: Vec<_> = (0..jobs) + .map(|j| { + Arc::new(BTreeMap::from([ + ("__name__".to_string(), metric.to_string()), + ("job".to_string(), format!("job{j}")), + ("instance".to_string(), format!("{}", j % 3)), + ])) + }) + .collect(); + (from..to) + .step_by(15_000) + .flat_map(|t| { + series.iter().enumerate().map(move |(j, labels)| RawSample { + labels: labels.clone(), + timestamp_ms: t, + value: (j * 1000) as f64 + (t / 15_000 % 97) as f64 + t as f64 / 1e6, + }) + }) + .collect() +} + +/// The state in `value`, as comparable data: an exact accumulator's +/// serialized bytes, or a quantile sketch's quantiles at every percent. +/// KLL seeds its compaction randomly per sketch, so two builds of the same +/// samples agree on content, not on bytes; below k they hold every sample. +fn state_of(value: &Value) -> Vec { + let Value::Summary { family, state } = value else { + panic!("not a state: {value:?}"); + }; + if let Some(exact) = state.as_any().downcast_ref::() { + return serde_json::to_vec(exact).unwrap(); + } + assert!(matches!(family, FieldDataType::Sketch(..)), "{family:?}"); + (1..100) + .flat_map(|p| { + let quantile = SketchStatistic::Quantile { + q: p as f64 / 100.0, + }; + state.estimate(&quantile).unwrap().to_le_bytes() + }) + .collect() +} + +/// Every cell of `pane` as comparable data, rows in order. +fn cells(pane: &Batch) -> Vec>> { + pane.rows() + .iter() + .map(|row| { + row.iter() + .map(|value| match value { + Value::Summary { .. } => state_of(value), + other => other.key().unwrap(), + }) + .collect() + }) + .collect() +} + +/// For Example 1's counter rates and sums over 10 s panes, Example 3b's +/// 1-minute KLL panes and Example 4b's 10-minute KLL panes, on a deployment +/// without raw data: every pane built per sample, on one shard or four, +/// equals the executor's batch build of the same samples, row by row. +#[test] +fn per_sample_panes_equal_the_batch_build() { + for (example, metric, to) in [ + ("example1", "http_requests_total", 12 * MINUTE), + ("example3b", "latency_ms", 12 * MINUTE), + ("example4b", "latency_ms", 75 * MINUTE), + ] { + let plan = common::plan(example, false); + let sds = Sds::from_plan(&plan).unwrap(); + let all = samples(metric, 7, MINUTE, to + 1); + let mut batch = Precompute::batch(&plan, &sds, 0).unwrap(); + batch.ingest(&all); + assert_eq!(batch.per_sample_definitions(), 0); + for shards in [1, 4] { + let mut per_sample = Precompute::with_shards(&plan, &sds, 0, shards).unwrap(); + assert_eq!( + per_sample.per_sample_definitions(), + sds.definitions.len(), + "{example}: every definition is built per sample" + ); + per_sample.ingest(&all); + assert_eq!(per_sample.stats(), batch.stats(), "{example}"); + let mut compared = 0; + for definition in sds.definitions.values() { + let mut end = 0; + while end <= to { + let expected = batch.store().get(&definition.id, end); + let actual = per_sample.store().get(&definition.id, end); + assert_eq!(actual.is_some(), expected.is_some(), "{example} pane {end}"); + if let (Some(actual), Some(expected)) = (actual, expected) { + assert_eq!(actual.schema(), expected.schema()); + assert_eq!(cells(&actual), cells(&expected), "{example} pane {end}"); + compared += usize::from(!expected.rows().is_empty()); + } + end += definition.pane_ms; + } + } + assert!(compared > 0, "{example}: no pane compared"); + } + } +} + +/// The plan's label matchers apply per sample as in the batch build: only +/// the selected series get a state. +#[test] +fn per_sample_panes_apply_label_matchers() { + let mut workload = common::workload("example4b"); + workload.query_workload.repeating_queries.as_mut().unwrap()[0].query = + asap_executor::planner::workload::Query( + "quantile_over_time(0.99, latency_ms{instance=\"1\"}[60m])".into(), + ); + let plan = control_plane::plan_workload(&workload, &common::inputs(false)).unwrap(); + let sds = Sds::from_plan(&plan).unwrap(); + let id = sds.definitions.keys().next().unwrap().clone(); + let all = samples("latency_ms", 7, MINUTE, 25 * MINUTE); + let mut per_sample = Precompute::new(&plan, &sds, 0).unwrap(); + assert_eq!(per_sample.per_sample_definitions(), 1); + per_sample.ingest(&all); + let mut batch = Precompute::batch(&plan, &sds, 0).unwrap(); + batch.ingest(&all); + let (actual, expected) = ( + per_sample.store().get(&id, 20 * MINUTE).unwrap(), + batch.store().get(&id, 20 * MINUTE).unwrap(), + ); + assert_eq!(expected.rows().len(), 2, "jobs 1 and 4 are instance 1"); + assert_eq!(cells(&actual), cells(&expected)); +}