From a62e7de5c3cbf53a63eb5e57863209115583c90b Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 7 Oct 2026 03:11:52 +0000 Subject: [PATCH 1/2] feat(precompute): build stored summary panes from Remote Write samples An engine thread owns the installed plan (Planner IR holds Rcs) and receives Remote Write samples from the HTTP server. For each stored summary definition it buffers its Scan metric's samples by tumbling pane. When the watermark (newest sample minus --allowed-lateness-ms) passes a pane's end, it runs the plan's Scan over them (applying label matchers) and the definition's build from its TimeRange input, then stores the output. Panes without samples are stored empty; the first pane seen is skipped because it may be partial; samples for built panes are dropped. A full engine queue answers Remote Write with 503 so senders retry. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + README.md | 8 +- data_plane/Cargo.toml | 1 + data_plane/src/engine.rs | 100 +++++++++++ data_plane/src/lib.rs | 69 ++++++-- data_plane/src/main.rs | 18 +- data_plane/src/precompute.rs | 304 +++++++++++++++++++++++++++++++++ data_plane/tests/common/mod.rs | 21 +++ data_plane/tests/precompute.rs | 116 +++++++++++++ data_plane/tests/sds.rs | 16 +- data_plane/tests/server.rs | 20 ++- 11 files changed, 636 insertions(+), 38 deletions(-) create mode 100644 data_plane/src/engine.rs create mode 100644 data_plane/src/precompute.rs create mode 100644 data_plane/tests/common/mod.rs create mode 100644 data_plane/tests/precompute.rs diff --git a/Cargo.lock b/Cargo.lock index 8ca158eb..33d20ff4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -490,6 +490,7 @@ dependencies = [ "axum", "clap", "control_plane", + "futures", "http-body-util", "prost", "reqwest", diff --git a/README.md b/README.md index d1ec5393..744af865 100644 --- a/README.md +++ b/README.md @@ -31,10 +31,14 @@ The backend is being rebuilt on the Planner's physical DAG: |---|---| | Control plane: plan a PromQL workload, write `plan.json` | done | | Data plane: load and validate `plan.json`, accept Remote Write | done | -| Stored summaries (SDS) | next | -| Precompute engine: run ingestion-time nodes per pane | planned | +| Stored summaries (SDS): one definition per computation, panes by end time | done | +| Precompute engine: build ingestion-time outputs once per tumbling pane | done | | Query engine: run query-time nodes over stored panes and raw data | planned | +Panes close `--allowed-lateness-ms` (default 15 s) after the newest sample +passes their end; later samples for them are dropped. The first pane after +startup is skipped because it may be partial. + Until the query engine lands, the data plane forwards every query to Prometheus and marks the response `x-asap-execution: forwarded`. SQL and ClickHouse are not supported yet. diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index cf8e346d..3c6a4655 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -20,6 +20,7 @@ axum.workspace = true reqwest.workspace = true prost = "0.13" snap = "1.1" +futures = "0.3" [dev-dependencies] tower = { version = "0.5", features = ["util"] } diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs new file mode 100644 index 00000000..b5b643a9 --- /dev/null +++ b/data_plane/src/engine.rs @@ -0,0 +1,100 @@ +//! The engine thread: owns the installed plan and everything that executes +//! it. Planner IR holds `Rc`s, so the plan cannot be shared across the HTTP +//! server's threads; handlers send the engine commands instead. + +use std::sync::mpsc::{sync_channel, Receiver, SyncSender, TrySendError}; + +use asap_executor::physical_planner::frontier_from_timing; +use control_plane::InstalledPlan; +use tokio::sync::oneshot; + +use crate::precompute::{Precompute, PrecomputeStats, RawSample}; +use crate::sds::Sds; + +/// Commands queued for the engine before ingestion pushes back. +const QUEUE: usize = 1024; + +pub enum Command { + Ingest(Vec), + Stats(oneshot::Sender), +} + +#[derive(Debug, thiserror::Error)] +#[error("the engine is busy")] +pub struct Busy; + +/// The HTTP side of the engine thread. +#[derive(Clone)] +pub struct EngineHandle { + commands: SyncSender, +} + +impl EngineHandle { + /// Starts the engine for the plan serialized in `plan_json`. Fails if the + /// plan is invalid or the data plane cannot execute it. + pub fn spawn(plan_json: String, allowed_lateness_ms: i64) -> anyhow::Result { + let (commands, receiver) = sync_channel(QUEUE); + let (ready, started) = std::sync::mpsc::channel(); + std::thread::Builder::new() + .name("engine".into()) + .spawn(move || match Engine::new(&plan_json, allowed_lateness_ms) { + Ok(engine) => { + let _ = ready.send(Ok(())); + engine.serve(receiver); + } + Err(error) => { + let _ = ready.send(Err(error)); + } + })?; + started.recv()??; + Ok(Self { commands }) + } + + /// Queues `samples` for precompute, or fails if the engine is behind. + pub fn try_ingest(&self, samples: Vec) -> Result<(), Busy> { + match self.commands.try_send(Command::Ingest(samples)) { + Ok(()) => Ok(()), + Err(TrySendError::Full(_) | TrySendError::Disconnected(_)) => Err(Busy), + } + } + + /// Precompute counters, once every command queued before has run. + pub async fn stats(&self) -> PrecomputeStats { + let (reply, stats) = oneshot::channel(); + let _ = self.commands.send(Command::Stats(reply)); + stats.await.unwrap_or_default() + } +} + +struct Engine { + precompute: Precompute, +} + +impl Engine { + fn new(plan_json: &str, allowed_lateness_ms: i64) -> anyhow::Result { + let plan: InstalledPlan = serde_json::from_str(plan_json)?; + plan.validate()?; + let dag = plan.dag.dag; + let frontier = frontier_from_timing(&dag)?; + let sds = Sds::from_dag(&dag, &frontier)?; + tracing::info!( + definitions = sds.definitions.len(), + outputs = sds.outputs.len(), + "stored summaries" + ); + Ok(Self { + precompute: Precompute::new(dag, &sds, allowed_lateness_ms)?, + }) + } + + fn serve(mut self, commands: Receiver) { + for command in commands { + match command { + Command::Ingest(samples) => self.precompute.ingest(&samples), + Command::Stats(reply) => { + let _ = reply.send(self.precompute.stats); + } + } + } + } +} diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 7180dacf..9de90c76 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -1,13 +1,16 @@ //! Backend data plane: receives Prometheus Remote Write samples and serves //! the Prometheus query API for an installed plan. //! -//! The installed plan is ASAPPlanner's selected `PhysicalASAPDAG`. Executing -//! it is not implemented yet: samples are validated and counted, and every -//! query is forwarded to Prometheus. +//! The installed plan is ASAPPlanner's selected `PhysicalASAPDAG`. The +//! engine thread builds its ingestion-time outputs pane by pane; queries are +//! still forwarded to Prometheus. +pub mod engine; +pub mod precompute; pub mod remote_write; pub mod sds; +use std::collections::BTreeMap; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; @@ -17,14 +20,20 @@ use axum::http::{header, HeaderMap, HeaderValue, Method, StatusCode}; use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; use axum::Router; +use control_plane::InstalledPlan; +use engine::EngineHandle; +use precompute::RawSample; + +/// Prometheus' staleness marker: a NaN that ends a series, not a sample. +const STALE_NAN_BITS: u64 = 0x7ff0_0000_0000_0002; /// Response header naming what answered a query. pub const EXECUTION_HEADER: &str = "x-asap-execution"; pub struct DataPlane { - /// The installed plan's queries, in root order. The plan itself is not - /// `Send` (its operator payloads hold `Rc`s), so it stays with its loader. + /// The installed plan's queries, in root order. pub queries: Vec, + pub engine: EngineHandle, /// Base URL of the Prometheus that answers forwarded queries. pub prometheus_url: String, pub limits: remote_write::Limits, @@ -34,14 +43,23 @@ pub struct DataPlane { } impl DataPlane { - pub fn new(queries: Vec, prometheus_url: String) -> Self { - Self { - queries, + /// Installs the plan serialized in `plan_json` (an [`InstalledPlan`]). + /// Panes close `allowed_lateness_ms` after the newest sample passes them. + pub fn new( + plan_json: String, + prometheus_url: String, + allowed_lateness_ms: i64, + ) -> anyhow::Result { + let plan: InstalledPlan = serde_json::from_str(&plan_json)?; + plan.validate()?; + Ok(Self { + queries: plan.queries, + engine: EngineHandle::spawn(plan_json, allowed_lateness_ms)?, prometheus_url: prometheus_url.trim_end_matches('/').to_string(), limits: remote_write::Limits::default(), samples: AtomicU64::new(0), client: reqwest::Client::new(), - } + }) } } @@ -57,9 +75,36 @@ pub fn router(state: Arc) -> Router { async fn write(State(state): State>, body: Bytes) -> Response { match remote_write::decode(&body, &state.limits) { Ok(request) => { - let samples: usize = request.timeseries.iter().map(|s| s.samples.len()).sum(); - state.samples.fetch_add(samples as u64, Ordering::Relaxed); - StatusCode::NO_CONTENT.into_response() + let mut samples = Vec::new(); + for series in request.timeseries { + let labels: Arc> = Arc::new( + series + .labels + .into_iter() + .map(|label| (label.name, label.value)) + .collect(), + ); + samples.extend( + series + .samples + .into_iter() + .filter(|s| s.value.to_bits() != STALE_NAN_BITS) + .map(|s| RawSample { + labels: labels.clone(), + timestamp_ms: s.timestamp, + value: s.value, + }), + ); + } + let count = samples.len() as u64; + match state.engine.try_ingest(samples) { + Ok(()) => { + state.samples.fetch_add(count, Ordering::Relaxed); + StatusCode::NO_CONTENT.into_response() + } + // Remote Write senders retry on 5xx. + Err(busy) => (StatusCode::SERVICE_UNAVAILABLE, busy.to_string()).into_response(), + } } Err(error) => (StatusCode::BAD_REQUEST, error.to_string()).into_response(), } diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index e82015a4..c6ff5484 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -2,7 +2,6 @@ use std::path::PathBuf; use std::sync::Arc; use clap::Parser; -use control_plane::InstalledPlan; use data_plane::{router, DataPlane}; /// Serve Remote Write ingestion and the Prometheus query API for a plan. @@ -16,6 +15,9 @@ struct Args { prometheus_url: String, #[arg(long, default_value = "0.0.0.0:8088")] listen: String, + /// How long after the newest sample passes a pane's end it is built. + #[arg(long, default_value_t = 15_000)] + allowed_lateness_ms: i64, } #[tokio::main] @@ -24,13 +26,13 @@ async fn main() -> anyhow::Result<()> { .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) .init(); let args = Args::parse(); - let plan = InstalledPlan::load(&args.plan)?; - tracing::info!( - queries = plan.queries.len(), - nodes = plan.dag.dag.nodes.len(), - "installed plan" - ); - let state = Arc::new(DataPlane::new(plan.queries, args.prometheus_url)); + let plan = std::fs::read_to_string(&args.plan)?; + let state = Arc::new(DataPlane::new( + plan, + args.prometheus_url, + args.allowed_lateness_ms, + )?); + tracing::info!(queries = state.queries.len(), "installed plan"); let listener = tokio::net::TcpListener::bind(&args.listen).await?; tracing::info!(address = %listener.local_addr()?, "listening"); axum::serve(listener, router(state)).await?; diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs new file mode 100644 index 00000000..1854e5c9 --- /dev/null +++ b/data_plane/src/precompute.rs @@ -0,0 +1,304 @@ +//! Precompute engine: builds each stored summary pane once, when the +//! watermark passes its end. +//! +//! For every [`SummaryDefinition`] the engine buffers the samples of its +//! `Scan`'s metric by tumbling pane. A pane `[end - pane, end)` closes when +//! the watermark (the newest sample time minus the allowed lateness) reaches +//! `end`. Closing it runs the plan's `Scan` over the buffered samples, which +//! applies the label matchers, then the definition's build from its +//! `TimeRange` input up to its output, under ingestion scope. The output is +//! stored and never changes; samples older than a closed pane are dropped. +//! The first pane a definition sees may have started before the data plane +//! did, so it is never built: queries that need it are not answered from +//! stored panes. + +use std::collections::{BTreeMap, HashMap}; +use std::sync::Arc; + +use asap_executor::operators::Operator; +use asap_executor::physical_planner::promql_rows::series_row; +use asap_executor::physical_planner::{ + bind_with_data_sources, compile, CompiledPhysicalDAG, InputContract, Source, +}; +use asap_executor::runtime::{Limits, RunContext, Scope}; +use asap_executor::sources::{DataSources, MemorySource}; +use asap_executor::values::{Batch, SchemaRef}; +use futures::executor::block_on; +use futures::StreamExt; +use planner_types::ir::export::{ + LogicalASAPOperatorPayload as Payload, NonASAPOpKind, PhysicalASAPDAG, +}; +use planner_types::ir::operator::Source as ScanSource; + +use crate::sds::{DefinitionId, Sds, SummaryDefinition, SummaryStore}; + +/// One raw sample: its series' full label set (including `__name__`), time +/// and value. +#[derive(Debug, Clone, PartialEq)] +pub struct RawSample { + pub labels: Arc>, + pub timestamp_ms: i64, + pub value: f64, +} + +#[derive(Debug, thiserror::Error)] +pub enum PrecomputeError { + #[error("stored summary {0:?}: {1}")] + Definition(DefinitionId, String), + #[error(transparent)] + Executor(#[from] asap_executor::Error), +} + +/// Counters for what happened to incoming samples. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub struct PrecomputeStats { + /// Samples buffered for at least one pane. + pub accepted: u64, + /// Samples for a pane already built or skipped. + pub dropped: u64, + /// Panes built and stored. + pub panes_built: u64, + /// Panes whose build failed; they are skipped, not retried. + pub panes_failed: u64, +} + +struct Builder { + definition: SummaryDefinition, + metric: String, + scan: u64, + scan_schema: SchemaRef, + build: CompiledPhysicalDAG, + /// Oldest pane offset any output reads, plus one pane: panes ending + /// before `watermark - retention_ms` are no longer read. + retention_ms: i64, + /// Samples of open panes, by pane end. + open: BTreeMap>, + /// End of the newest pane built or skipped; samples in it or before it + /// are dropped. `None` until the first sample. + built_through: Option, +} + +pub struct Precompute { + dag: PhysicalASAPDAG, + builders: Vec, + allowed_lateness_ms: i64, + newest_sample_ms: Option, + pub store: SummaryStore, + pub stats: PrecomputeStats, +} + +impl Precompute { + /// Compiles one build per definition of `sds`, a description of `dag`. + pub fn new( + dag: PhysicalASAPDAG, + sds: &Sds, + allowed_lateness_ms: i64, + ) -> Result { + let mut builders = Vec::new(); + for definition in sds.definitions.values() { + let error = + |message: String| PrecomputeError::Definition(definition.id.clone(), message); + let (scan, metric, scan_schema) = scan_below(&dag, definition.raw_input) + .ok_or_else(|| error("no time-series Scan below its TimeRange".into()))?; + let raw_schema = Arc::new(output_schema(&dag, definition.raw_input).clone()); + if raw_schema != scan_schema { + return Err(error("TimeRange rows differ from its Scan's".into())); + } + let build = compile( + &dag, + BTreeMap::from([(definition.raw_input, InputContract::bounded(raw_schema))]), + &[definition.builder], + )?; + let max_offset = sds + .outputs + .values() + .filter(|(id, _)| id == &definition.id) + .map(|(_, binding)| binding.offset_ms) + .max() + .unwrap_or(0); + builders.push(Builder { + definition: definition.clone(), + metric, + scan, + scan_schema, + build, + retention_ms: max_offset + 2 * definition.pane_ms, + open: BTreeMap::new(), + built_through: None, + }); + } + Ok(Self { + dag, + builders, + allowed_lateness_ms, + newest_sample_ms: None, + store: SummaryStore::new(), + stats: PrecomputeStats::default(), + }) + } + + /// The watermark: no sample older than this is expected any more. + pub fn watermark_ms(&self) -> Option { + self.newest_sample_ms + .map(|newest| newest - self.allowed_lateness_ms) + } + + /// Buffers `samples` into their panes, then builds every pane the + /// watermark has passed. + pub fn ingest(&mut self, samples: &[RawSample]) { + for sample in samples { + let metric = sample.labels.get("__name__").map(String::as_str); + let mut accepted = false; + for builder in &mut self.builders { + if metric != Some(builder.metric.as_str()) { + continue; + } + let end = pane_end(sample.timestamp_ms, builder.definition.pane_ms); + // The first pane seen may be partial: skip it. + let built = *builder.built_through.get_or_insert(end); + if end <= built { + self.stats.dropped += 1; + continue; + } + builder.open.entry(end).or_default().push(sample.clone()); + accepted = true; + } + if accepted { + self.stats.accepted += 1; + } + self.newest_sample_ms = self.newest_sample_ms.max(Some(sample.timestamp_ms)); + } + self.close_panes(); + } + + fn close_panes(&mut self) { + let Some(watermark) = self.watermark_ms() else { + return; + }; + for builder in &mut self.builders { + let Some(built) = builder.built_through else { + continue; + }; + let pane_ms = builder.definition.pane_ms; + // Every pane after the first through the watermark is built, so + // a pane without samples is stored as empty, not missing. + let mut end = built + pane_ms; + while end <= watermark { + let samples = builder.open.remove(&end).unwrap_or_default(); + match build_pane(&self.dag, builder, end, &samples) { + Ok(state) => { + self.store.insert(&builder.definition.id, end, state); + self.stats.panes_built += 1; + } + Err(error) => { + tracing::error!(definition = ?builder.definition.id, end, %error, "pane build failed"); + self.stats.panes_failed += 1; + } + } + builder.built_through = Some(end); + end += pane_ms; + } + self.store + .evict_through(&builder.definition.id, watermark - builder.retention_ms); + } + } +} + +/// End of the tumbling pane of width `pane_ms` that contains `timestamp_ms`. +pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> i64 { + (timestamp_ms.div_euclid(pane_ms) + 1) * pane_ms +} + +/// Runs the plan's `Scan` over `samples`, then the definition's build over +/// the rows it selects, for the pane ending at `end`. +fn build_pane( + dag: &PhysicalASAPDAG, + builder: &Builder, + end: i64, + samples: &[RawSample], +) -> Result { + let start = end - builder.definition.pane_ms; + let scope = Scope::Ingestion { + window_start_ms: start, + window_end_ms: end, + revision: 1, + }; + let rows = samples + .iter() + .map(|s| series_row(&builder.scan_schema, &s.labels, s.timestamp_ms, s.value)) + .collect::, _>>()?; + let raw = Batch::try_new(builder.scan_schema.clone(), rows)?; + let mut sources = DataSources::default(); + sources.register( + ScanSource::TimeSeries { + metric: builder.metric.clone(), + }, + Arc::new(MemorySource::new(builder.scan_schema.clone(), vec![raw])?), + )?; + let scan = bind_with_data_sources(dag, BTreeMap::new(), &[builder.scan], &sources)?; + let selected = run(&scan, &[builder.scan], scope.clone())?; + let selected = Batch::try_new(builder.scan_schema.clone(), selected)?; + let input: Source<'_> = Box::new(Operator::source( + builder.scan_schema.clone(), + vec![selected], + )?); + let build = builder + .build + .instantiate(BTreeMap::from([(builder.definition.raw_input, input)]))?; + let rows = run(&build, builder.build.roots(), scope)?; + let schema = Arc::new(builder.definition.output_schema.clone()); + Ok(Batch::try_new(schema, rows)?) +} + +/// Executes `dag`'s single root to completion and returns its rows. +pub fn run( + dag: &asap_executor::plan::PhysicalDAG<'_, Batch, SchemaRef>, + roots: &[u64], + scope: Scope, +) -> Result>, asap_executor::Error> { + block_on(async { + let context = RunContext::new(scope, Limits::default())?; + let mut output = dag.execute(roots, context)?.remove(0); + let mut rows = Vec::new(); + while let Some(batch) = output.next().await { + rows.extend(batch?.rows().iter().cloned()); + } + Ok(rows) + }) +} + +fn output_schema(dag: &PhysicalASAPDAG, id: u64) -> &planner_types::ir::schema::Schema { + &dag.nodes + .iter() + .find(|n| u64::from(n.id.0) == id) + .expect("SDS names DAG nodes") + .output_schema +} + +/// The time-series `Scan` that `id` reads: its node id, metric and row schema. +fn scan_below(dag: &PhysicalASAPDAG, id: u64) -> Option<(u64, String, SchemaRef)> { + let inputs: HashMap = dag + .edges + .iter() + .map(|e| (u64::from(e.consumer.0), u64::from(e.producer.0))) + .collect(); + let mut current = id; + loop { + let node = dag.nodes.iter().find(|n| u64::from(n.id.0) == current)?; + if let Payload::Relational { + operator: + NonASAPOpKind::Scan { + source: ScanSource::TimeSeries { metric }, + .. + }, + } = &node.payload + { + return Some(( + current, + metric.clone(), + Arc::new(node.output_schema.clone()), + )); + } + current = *inputs.get(¤t)?; + } +} diff --git a/data_plane/tests/common/mod.rs b/data_plane/tests/common/mod.rs new file mode 100644 index 00000000..cde1de88 --- /dev/null +++ b/data_plane/tests/common/mod.rs @@ -0,0 +1,21 @@ +use std::path::Path; + +use control_plane::{backend_capabilities, plan_workload, InstalledPlan}; + +/// The selected plan of a #509 example workload in `examples/`, for a +/// deployment that keeps raw data or not. +pub fn plan(example: &str, raw_data_retained: bool) -> InstalledPlan { + let path = Path::new(env!("CARGO_MANIFEST_DIR")) + .join(format!("../examples/planner-layering-{example}.json")); + let workload = serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); + let capabilities = planner_types::deployment::DeploymentCapabilities { + raw_data_retained, + ..backend_capabilities() + }; + plan_workload(&workload, &capabilities).unwrap() +} + +#[allow(dead_code)] +pub fn plan_json(example: &str, raw_data_retained: bool) -> String { + serde_json::to_string(&plan(example, raw_data_retained)).unwrap() +} diff --git a/data_plane/tests/precompute.rs b/data_plane/tests/precompute.rs new file mode 100644 index 00000000..e7b9511e --- /dev/null +++ b/data_plane/tests/precompute.rs @@ -0,0 +1,116 @@ +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use asap_executor::physical_planner::frontier_from_timing; +use control_plane::{backend_capabilities, plan_workload, InstalledPlan}; +use data_plane::precompute::{pane_end, Precompute, RawSample}; +use data_plane::sds::Sds; +use planner_types::workload::{PlanningWorkload, Query}; + +const MINUTE: i64 = 60_000; +const PANE: i64 = 10 * MINUTE; + +fn precompute(plan: InstalledPlan) -> (Precompute, Sds) { + let dag = plan.dag.dag; + let sds = Sds::from_dag(&dag, &frontier_from_timing(&dag).unwrap()).unwrap(); + (Precompute::new(dag, &sds, 0).unwrap(), sds) +} + +fn sample(metric: &str, job: &str, timestamp_ms: i64, value: f64) -> RawSample { + RawSample { + labels: Arc::new(BTreeMap::from([ + ("__name__".to_string(), metric.to_string()), + ("job".to_string(), job.to_string()), + ])), + timestamp_ms, + value, + } +} + +/// One sample every 15 s for each job in `[from, to)`. +fn samples(metric: &str, jobs: &[&str], from: i64, to: i64) -> Vec { + (from..to) + .step_by(15_000) + .flat_map(|t| jobs.iter().map(move |job| sample(metric, job, t, t as f64))) + .collect() +} + +/// Panes are built once the watermark passes them, one state per series. +/// The first pane seen may be partial, so it is skipped. +#[test] +fn builds_panes_after_the_first_as_the_watermark_passes() { + let (mut engine, sds) = precompute(common::plan("example4b", false)); + let id = sds.definitions.keys().next().unwrap().clone(); + engine.ingest(&samples("latency_ms", &["a", "b"], 0, 35 * MINUTE)); + assert_eq!(engine.store.get(&id, PANE).map(|_| ()), None, "first pane"); + for end in [2 * PANE, 3 * PANE] { + let pane = engine.store.get(&id, end).expect("built pane"); + assert_eq!(pane.rows().len(), 2, "one state per series"); + } + assert_eq!( + engine.store.get(&id, 4 * PANE).map(|_| ()), + None, + "open pane" + ); + let stats = engine.stats; + assert_eq!(stats.panes_built, 2); + assert_eq!(stats.panes_failed, 0); + // Samples in [0, 10 min) are dropped with the first pane. + assert_eq!(stats.dropped, 2 * 40); + assert_eq!(stats.accepted, 2 * 100); +} + +/// Samples of other metrics are ignored; samples for a built pane are dropped. +#[test] +fn ignores_other_metrics_and_drops_late_samples() { + let (mut engine, _) = precompute(common::plan("example4b", false)); + engine.ingest(&samples("latency_ms", &["a"], 0, 25 * MINUTE)); + let before = engine.stats; + engine.ingest(&[sample("other", "a", 25 * MINUTE, 1.0)]); + assert_eq!(engine.stats, before); + engine.ingest(&[sample("latency_ms", "a", 15 * MINUTE, 1.0)]); + assert_eq!(engine.stats.dropped, before.dropped + 1); +} + +/// A pane without samples is stored empty, so a query can tell "no data" +/// from "not built". +#[test] +fn stores_empty_panes() { + let (mut engine, sds) = precompute(common::plan("example4b", false)); + let id = sds.definitions.keys().next().unwrap().clone(); + engine.ingest(&samples("latency_ms", &["a"], 0, 20 * MINUTE)); + engine.ingest(&samples("latency_ms", &["a"], 40 * MINUTE, 41 * MINUTE)); + assert_eq!(engine.store.get(&id, 2 * PANE).unwrap().rows().len(), 1); + assert_eq!(engine.store.get(&id, 3 * PANE).unwrap().rows().len(), 0); + assert_eq!(engine.store.get(&id, 4 * PANE).unwrap().rows().len(), 0); +} + +/// The plan's label matchers select which series a pane summarizes. +#[test] +fn applies_label_matchers() { + let path = std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("../examples/planner-layering-example4b.json"); + let mut workload: PlanningWorkload = + serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); + workload.query_workload.repeating_queries.as_mut().unwrap()[0].query = + Query("quantile_over_time(0.99, latency_ms{job=\"a\"}[60m])".into()); + let capabilities = planner_types::deployment::DeploymentCapabilities { + raw_data_retained: false, + ..backend_capabilities() + }; + let (mut engine, sds) = precompute(plan_workload(&workload, &capabilities).unwrap()); + let id = sds.definitions.keys().next().unwrap().clone(); + engine.ingest(&samples("latency_ms", &["a", "b"], 0, 25 * MINUTE)); + assert_eq!(engine.store.get(&id, 2 * PANE).unwrap().rows().len(), 1); +} + +/// A sample belongs to the pane whose end follows it, including at negative times. +#[test] +fn pane_end_is_the_next_boundary() { + assert_eq!(pane_end(0, PANE), PANE); + assert_eq!(pane_end(PANE - 1, PANE), PANE); + assert_eq!(pane_end(PANE, PANE), 2 * PANE); + assert_eq!(pane_end(-1, PANE), 0); +} diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs index 7dffbba1..9eaec559 100644 --- a/data_plane/tests/sds.rs +++ b/data_plane/tests/sds.rs @@ -1,23 +1,11 @@ -use std::path::Path; +mod common; use asap_executor::physical_planner::frontier_from_timing; -use control_plane::{backend_capabilities, plan_workload, InstalledPlan}; +use common::plan; use data_plane::sds::{DefinitionId, Sds, SdsError, SummaryStore}; use planner_types::ir::export::{LogicalASAPOperatorPayload as Payload, NonASAPOpKind}; use planner_types::ir::physical_export::PhysicalASAPDAG; -/// The selected plan of a #509 example workload in `examples/`. -fn plan(example: &str, raw_data_retained: bool) -> InstalledPlan { - let path = Path::new(env!("CARGO_MANIFEST_DIR")) - .join(format!("../examples/planner-layering-{example}.json")); - let workload = serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap(); - let capabilities = planner_types::deployment::DeploymentCapabilities { - raw_data_retained, - ..backend_capabilities() - }; - plan_workload(&workload, &capabilities).unwrap() -} - fn sds_of(dag: &PhysicalASAPDAG) -> Result { Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()) } diff --git a/data_plane/tests/server.rs b/data_plane/tests/server.rs index 60823445..bf116c17 100644 --- a/data_plane/tests/server.rs +++ b/data_plane/tests/server.rs @@ -1,3 +1,5 @@ +mod common; + use std::sync::atomic::Ordering; use std::sync::Arc; @@ -45,7 +47,14 @@ async fn send(app: &Router, request: Request) -> (StatusCode, Option Date: Wed, 7 Oct 2026 03:12:57 +0000 Subject: [PATCH 2/2] fix(precompute): make panes left-open like PromQL ranges A PromQL range at T covers (T - range, T], and the executor's series windows follow it. Panes were [start, end), so a sample at a pane boundary was summarized in the next pane. Panes are now (end - width, end]. Co-Authored-By: Claude Opus 5.5 --- data_plane/src/precompute.rs | 6 ++++-- data_plane/src/sds.rs | 5 +++-- data_plane/tests/precompute.rs | 26 +++++++++++++++----------- 3 files changed, 22 insertions(+), 15 deletions(-) diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 1854e5c9..a0aabe86 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -2,7 +2,7 @@ //! watermark passes its end. //! //! For every [`SummaryDefinition`] the engine buffers the samples of its -//! `Scan`'s metric by tumbling pane. A pane `[end - pane, end)` closes when +//! `Scan`'s metric by tumbling pane. A pane `(end - pane, end]` closes when //! the watermark (the newest sample time minus the allowed lateness) reaches //! `end`. Closing it runs the plan's `Scan` over the buffered samples, which //! applies the label matchers, then the definition's build from its @@ -205,8 +205,10 @@ impl Precompute { } /// End of the tumbling pane of width `pane_ms` that contains `timestamp_ms`. +/// Panes are `(end - pane_ms, end]`, like PromQL ranges: a sample at a pane +/// boundary belongs to the pane it ends. pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> i64 { - (timestamp_ms.div_euclid(pane_ms) + 1) * pane_ms + -(-timestamp_ms).div_euclid(pane_ms) * pane_ms } /// Runs the plan's `Scan` over `samples`, then the definition's build over diff --git a/data_plane/src/sds.rs b/data_plane/src/sds.rs index 4dc662b0..d3873a83 100644 --- a/data_plane/src/sds.rs +++ b/data_plane/src/sds.rs @@ -9,7 +9,8 @@ //! frontier node <- ... <- TimeRange(width) <- TimeShift(offset)* <- Scan //! ``` //! -//! At evaluation time `T` it covers `[T - offset - width, T - offset)`. +//! At evaluation time `T` it covers `(T - offset - width, T - offset]`, the +//! PromQL range convention. //! Outputs that differ only in `offset` compute the same thing over different //! panes, so they share one [`SummaryDefinition`]: the data plane builds each //! pane of a definition once, and every output reads the pane it needs. @@ -30,7 +31,7 @@ pub struct DefinitionId(pub String); #[derive(Debug, Clone, PartialEq)] pub struct SummaryDefinition { pub id: DefinitionId, - /// Pane width: each stored summary covers `[end - pane_ms, end)`. + /// Pane width: each stored summary covers `(end - pane_ms, end]`. pub pane_ms: i64, /// The frontier output that builds a pane; any output of the definition /// computes the same, and this is the first in node order. diff --git a/data_plane/tests/precompute.rs b/data_plane/tests/precompute.rs index e7b9511e..3823f75d 100644 --- a/data_plane/tests/precompute.rs +++ b/data_plane/tests/precompute.rs @@ -29,7 +29,7 @@ fn sample(metric: &str, job: &str, timestamp_ms: i64, value: f64) -> RawSample { } } -/// One sample every 15 s for each job in `[from, to)`. +/// One sample every 15 s for each job, at `from`, `from + 15 s`, ... before `to`. fn samples(metric: &str, jobs: &[&str], from: i64, to: i64) -> Vec { (from..to) .step_by(15_000) @@ -43,7 +43,8 @@ fn samples(metric: &str, jobs: &[&str], from: i64, to: i64) -> Vec { fn builds_panes_after_the_first_as_the_watermark_passes() { let (mut engine, sds) = precompute(common::plan("example4b", false)); let id = sds.definitions.keys().next().unwrap().clone(); - engine.ingest(&samples("latency_ms", &["a", "b"], 0, 35 * MINUTE)); + // The first sample, at 1 min, opens pane (0, 10 min], which is skipped. + engine.ingest(&samples("latency_ms", &["a", "b"], MINUTE, 35 * MINUTE)); assert_eq!(engine.store.get(&id, PANE).map(|_| ()), None, "first pane"); for end in [2 * PANE, 3 * PANE] { let pane = engine.store.get(&id, end).expect("built pane"); @@ -57,9 +58,9 @@ fn builds_panes_after_the_first_as_the_watermark_passes() { let stats = engine.stats; assert_eq!(stats.panes_built, 2); assert_eq!(stats.panes_failed, 0); - // Samples in [0, 10 min) are dropped with the first pane. - assert_eq!(stats.dropped, 2 * 40); - assert_eq!(stats.accepted, 2 * 100); + // Samples in (0, 10 min] are dropped with the first pane. + assert_eq!(stats.dropped, 2 * 37); + assert_eq!(stats.accepted, 2 * 99); } /// Samples of other metrics are ignored; samples for a built pane are dropped. @@ -81,7 +82,7 @@ fn stores_empty_panes() { let (mut engine, sds) = precompute(common::plan("example4b", false)); let id = sds.definitions.keys().next().unwrap().clone(); engine.ingest(&samples("latency_ms", &["a"], 0, 20 * MINUTE)); - engine.ingest(&samples("latency_ms", &["a"], 40 * MINUTE, 41 * MINUTE)); + engine.ingest(&[sample("latency_ms", "a", 45 * MINUTE, 1.0)]); assert_eq!(engine.store.get(&id, 2 * PANE).unwrap().rows().len(), 1); assert_eq!(engine.store.get(&id, 3 * PANE).unwrap().rows().len(), 0); assert_eq!(engine.store.get(&id, 4 * PANE).unwrap().rows().len(), 0); @@ -106,11 +107,14 @@ fn applies_label_matchers() { assert_eq!(engine.store.get(&id, 2 * PANE).unwrap().rows().len(), 1); } -/// A sample belongs to the pane whose end follows it, including at negative times. +/// Panes are `(end - width, end]`, like PromQL ranges: a sample at a +/// boundary belongs to the pane it ends. #[test] -fn pane_end_is_the_next_boundary() { - assert_eq!(pane_end(0, PANE), PANE); - assert_eq!(pane_end(PANE - 1, PANE), PANE); - assert_eq!(pane_end(PANE, PANE), 2 * PANE); +fn panes_are_left_open() { + assert_eq!(pane_end(1, PANE), PANE); + assert_eq!(pane_end(PANE, PANE), PANE); + assert_eq!(pane_end(PANE + 1, PANE), 2 * PANE); + assert_eq!(pane_end(0, PANE), 0); assert_eq!(pane_end(-1, PANE), 0); + assert_eq!(pane_end(-PANE, PANE), -PANE); }