From b23b37e3136d42834350a966424795ec579bdb9b Mon Sep 17 00:00:00 2001 From: zzylol Date: Sun, 18 Oct 2026 15:36:25 +0000 Subject: [PATCH 1/4] 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 (end - width, end], like PromQL ranges. When the watermark (newest sample minus --allowed-lateness-ms) passes a pane's end, it selects the pane's rows with the executor's RawSampleSelector (the Scan's label matchers) and runs 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 and counted. A full engine queue answers Remote Write with 503 so senders retry. The SDS finds the plan's materialization frontier itself from each node's timing (Sds::ingestion_frontier), since the executor exports frontier_from_timing only for tests; a test checks both agree on every example. SummaryStore keeps panes per definition, so a lookup borrows the definition id and eviction splits off the panes still needed. Forwarded queries keep main's timeout (--forward-timeout-seconds, through DataPlane::with_forward_timeout), header forwarding and Remote Write body limit. --- Cargo.lock | 1 + README.md | 24 +-- data_plane/Cargo.toml | 3 +- data_plane/src/engine.rs | 99 +++++++++++ data_plane/src/lib.rs | 94 ++++++++--- data_plane/src/main.rs | 23 +-- data_plane/src/precompute.rs | 291 +++++++++++++++++++++++++++++++++ data_plane/src/sds.rs | 80 +++++++-- data_plane/tests/common/mod.rs | 24 +++ data_plane/tests/precompute.rs | 136 +++++++++++++++ data_plane/tests/sds.rs | 33 ++-- data_plane/tests/server.rs | 43 ++--- 12 files changed, 752 insertions(+), 99 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 e3aa5f0c..5d652394 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -679,6 +679,7 @@ dependencies = [ "axum", "clap", "control_plane", + "futures", "http-body-util", "prost", "reqwest", diff --git a/README.md b/README.md index 9dcb4f79..e4d68e5a 100644 --- a/README.md +++ b/README.md @@ -31,18 +31,19 @@ The backend is being rebuilt on the Planner's physical DAG: | Part | State | |---|---| | Control plane: plan a PromQL workload, write `plan.json` | done | -| Data plane: load and validate `plan.json`; accept, count and drop Remote Write samples | done | -| Stored summaries (SDS) | next | -| Precompute engine: run ingestion-time nodes per pane | planned | +| Data plane: load and validate `plan.json`, accept Remote Write | done | +| 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 | -Until the precompute engine lands, the data plane validates and counts Remote -Write samples, then drops them: Prometheus must receive Remote Write itself -(for example, configure the sender with both endpoints). Until the query engine -lands, the data plane forwards every query to Prometheus, with its -end-to-end headers such as `Authorization` and `X-Scope-OrgID`, and marks the -response `x-asap-execution: forwarded`. SQL and ClickHouse are not supported -yet. +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, with its end-to-end headers such as `Authorization` and +`X-Scope-OrgID`, and marks the response `x-asap-execution: forwarded`. SQL and +ClickHouse are not supported yet. ## Running @@ -51,8 +52,7 @@ yet. cargo run -p control_plane -- \ --workload examples/planner-layering-example1.json --out plan.json -# Serve Remote Write on /api/v1/write (samples are counted and dropped for now) -# and the query API on /api/v1/query[_range], forwarded to Prometheus. +# Serve Remote Write on /api/v1/write and the query API on /api/v1/query[_range]. cargo run -p data_plane -- --plan plan.json --prometheus-url http://localhost:9090 ``` diff --git a/data_plane/Cargo.toml b/data_plane/Cargo.toml index 0a5351e4..cc64affc 100644 --- a/data_plane/Cargo.toml +++ b/data_plane/Cargo.toml @@ -20,9 +20,10 @@ axum.workspace = true reqwest.workspace = true prost = "0.13" snap = "1.1" +futures = "0.3" [dev-dependencies] tower = { version = "0.5", features = ["util"] } http-body-util = "0.1" -# The tests find the frontier the way the Planner does; the runtime does not. +# The tests check the SDS's frontier against the executor's own. asap-executor = { workspace = true, features = ["test-support"] } diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs new file mode 100644 index 00000000..4505b786 --- /dev/null +++ b/data_plane/src/engine.rs @@ -0,0 +1,99 @@ +//! 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 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 = Sds::ingestion_frontier(&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 e770ff91..71904dd6 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -1,14 +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: Remote Write samples are validated, counted and -//! dropped (Prometheus must receive Remote Write itself), 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; use std::time::Duration; @@ -19,6 +21,12 @@ 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"; @@ -28,35 +36,52 @@ pub const EXECUTION_HEADER: &str = "x-asap-execution"; pub const DEFAULT_FORWARD_TIMEOUT: Duration = Duration::from_secs(120); 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, - /// Samples accepted (and, until the precompute engine lands, dropped) - /// since startup. + /// Samples accepted since startup. pub samples: AtomicU64, client: reqwest::Client, } impl DataPlane { - /// `forward_timeout` bounds each forwarded query, from sending the - /// request to reading the whole response. - pub fn new(queries: Vec, prometheus_url: String, forward_timeout: Duration) -> 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::builder() - .timeout(forward_timeout) - .build() - .expect("reqwest client with only a timeout set"), - } + client: forward_client(DEFAULT_FORWARD_TIMEOUT), + }) + } + + /// Bounds each forwarded query by `timeout`, from sending the request to + /// reading the whole response (default [`DEFAULT_FORWARD_TIMEOUT`]). + pub fn with_forward_timeout(mut self, timeout: Duration) -> Self { + self.client = forward_client(timeout); + self } } +fn forward_client(timeout: Duration) -> reqwest::Client { + reqwest::Client::builder() + .timeout(timeout) + .build() + .expect("reqwest client with only a timeout set") +} + pub fn router(state: Arc) -> Router { // axum refuses bodies over 2 MB by default; Remote Write allows up to // `max_compressed_bytes` and answers larger bodies with 413. @@ -72,9 +97,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 c22bb7b0..ff511085 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -3,7 +3,6 @@ use std::sync::Arc; use std::time::Duration; use clap::Parser; -use control_plane::InstalledPlan; use data_plane::{router, DataPlane, DEFAULT_FORWARD_TIMEOUT}; /// Serve Remote Write ingestion and the Prometheus query API for a plan. @@ -12,9 +11,7 @@ struct Args { /// Plan written by the control plane. #[arg(long)] plan: PathBuf, - /// Prometheus base URL that answers forwarded queries. Until the - /// precompute engine lands, Remote Write samples sent to this server are - /// counted and dropped, so Prometheus must receive Remote Write itself. + /// Prometheus base URL that answers forwarded queries. #[arg(long, default_value = "http://localhost:9090")] prometheus_url: String, /// Seconds a forwarded query may take before it fails with 502. @@ -22,6 +19,9 @@ struct Args { forward_timeout_seconds: u64, #[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] @@ -30,17 +30,12 @@ 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 plan = std::fs::read_to_string(&args.plan)?; + let state = Arc::new( + DataPlane::new(plan, args.prometheus_url, args.allowed_lateness_ms)? + .with_forward_timeout(Duration::from_secs(args.forward_timeout_seconds)), ); - let state = Arc::new(DataPlane::new( - plan.queries, - args.prometheus_url, - Duration::from_secs(args.forward_timeout_seconds), - )); + 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..03571c65 --- /dev/null +++ b/data_plane/src/precompute.rs @@ -0,0 +1,291 @@ +//! 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 selects the pane's rows from the buffered samples with +//! the executor's `RawSampleSelector` (the `Scan`'s label matchers), then runs +//! 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::{compile, CompiledPhysicalDAG, InputContract, Source}; +use asap_executor::runtime::{Limits, RunContext, Scope}; +use asap_executor::values::{Batch, SchemaRef}; +use asap_executor::{RawSampleSelector, Sample}; +use futures::executor::block_on; +use futures::StreamExt; +use planner_types::ir::physical_export::{PhysicalASAPDAG, RawSampleWindow}; +use planner_types::ir::{NonASAPOp, Operator as IrOperator}; + +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, + /// Turns a pane's samples into the build's input: the `Scan`'s label + /// matchers, then the pane's `TimeRange` rows. + selector: RawSampleSelector, + raw_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 { + 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 = 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()); + let selector = RawSampleSelector::new( + &dag, + &RawSampleWindow { + time_range_node_id: definition.raw_input as usize, + width_ms: definition.pane_ms, + end_offset_ms: sds.outputs[&definition.builder].1.offset_ms, + scan_node_id: scan as usize, + }, + )?; + let build = compile( + &dag, + BTreeMap::from([( + definition.raw_input, + InputContract::bounded(raw_schema.clone()), + )]), + &[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(), + selector, + raw_schema, + build, + retention_ms: max_offset + 2 * definition.pane_ms, + open: BTreeMap::new(), + built_through: None, + }); + } + Ok(Self { + 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.selector.metric()) { + 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(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`. +/// 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) * 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 start = end - builder.definition.pane_ms; + let scope = Scope::Ingestion { + window_start_ms: start, + window_end_ms: end, + revision: 1, + }; + let samples: Vec = samples + .iter() + .map(|s| Sample { + labels: (*s.labels).clone(), + timestamp_ms: s.timestamp_ms, + value: s.value, + }) + .collect(); + let selected = builder.selector.window_batch(end, &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(&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?.to_rows()); + } + Ok(rows) + }) +} + +fn output_schema(dag: &PhysicalASAPDAG, id: u64) -> &planner_types::ir::schema::Schema { + &dag.nodes + .iter() + .find(|n| n.id as u64 == id) + .expect("SDS names DAG nodes") + .output_schema +} + +/// The time-series `Scan` that `id` reads. +fn scan_below(dag: &PhysicalASAPDAG, id: u64) -> Option { + let inputs: HashMap = dag + .edges + .iter() + .map(|e| (e.consumer as u64, e.producer as u64)) + .collect(); + let mut current = id; + loop { + let node = dag.nodes.iter().find(|n| n.id as u64 == current)?; + if let IrOperator::NonASAP(NonASAPOp::Scan { .. }) = &node.payload { + return Some(current); + } + current = *inputs.get(¤t)?; + } +} diff --git a/data_plane/src/sds.rs b/data_plane/src/sds.rs index 05ca7478..771488af 100644 --- a/data_plane/src/sds.rs +++ b/data_plane/src/sds.rs @@ -11,14 +11,16 @@ //! 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. -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, BTreeSet, HashMap}; use planner_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPDAGNode}; +use planner_types::ir::properties::ExecutionTiming::IngestionTime; use planner_types::ir::schema::Schema; use planner_types::ir::{NonASAPOp, Operator}; use planner_types::physical::Materialization; @@ -33,7 +35,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. @@ -69,9 +71,50 @@ pub enum SdsError { /// not select such a plan. #[error("node {0} is kept across evaluations (MatAtQueryKeptAcrossEvaluations), which the data plane does not support")] KeptAcrossEvaluations(u64), + #[error("ingestion-time node {0} reads a query-time node")] + QueryTimeFeedsIngestion(u64), } impl Sds { + /// The plan's materialization frontier: every node computed at ingestion + /// time whose output is read at query time, by a query-time node or as a + /// query's answer (a root, or a node a scalar root reads). In node order. + pub fn ingestion_frontier(dag: &PhysicalASAPDAG) -> Result, SdsError> { + let at_ingestion: HashMap = dag + .nodes + .iter() + .map(|n| (n.id as u64, n.output_state.timing == IngestionTime)) + .collect(); + let timing = |id: u64| { + at_ingestion + .get(&id) + .copied() + .ok_or(SdsError::UnknownNode(id)) + }; + let mut frontier = BTreeSet::new(); + let roots = dag.roots.iter().copied().chain( + dag.scalar_roots + .iter() + .flat_map(|(_, expr)| expr.operator_refs().into_iter().copied()), + ); + for root in roots { + if timing(root as u64)? { + frontier.insert(root as u64); + } + } + for edge in &dag.edges { + let (producer, consumer) = (edge.producer as u64, edge.consumer as u64); + match (timing(producer)?, timing(consumer)?) { + (true, false) => { + frontier.insert(producer); + } + (false, true) => return Err(SdsError::QueryTimeFeedsIngestion(consumer)), + _ => {} + } + } + Ok(frontier.into_iter().collect()) + } + /// The definitions behind `frontier`, the DAG outputs computed at /// ingestion time and read at query time. pub fn from_dag(dag: &PhysicalASAPDAG, frontier: &[u64]) -> Result { @@ -216,9 +259,15 @@ fn hex(bytes: &[u8]) -> String { /// Built panes, by definition and pane end. A pane is stored once, when it /// is complete, and never changes. -#[derive(Debug, Default)] +#[derive(Debug)] pub struct SummaryStore { - panes: BTreeMap<(DefinitionId, i64), S>, + panes: BTreeMap>, +} + +impl Default for SummaryStore { + fn default() -> Self { + Self::new() + } } impl SummaryStore { @@ -229,24 +278,33 @@ impl SummaryStore { } pub fn insert(&mut self, definition: &DefinitionId, end_ms: i64, state: S) { - self.panes.insert((definition.clone(), end_ms), state); + match self.panes.get_mut(definition) { + Some(panes) => { + panes.insert(end_ms, state); + } + None => { + self.panes + .insert(definition.clone(), BTreeMap::from([(end_ms, state)])); + } + } } pub fn get(&self, definition: &DefinitionId, end_ms: i64) -> Option<&S> { - self.panes.get(&(definition.clone(), end_ms)) + self.panes.get(definition)?.get(&end_ms) } /// Drops the panes of `definition` that end at or before `end_ms`. pub fn evict_through(&mut self, definition: &DefinitionId, end_ms: i64) { - self.panes - .retain(|(id, end), _| id != definition || *end > end_ms); + if let Some(panes) = self.panes.get_mut(definition) { + *panes = panes.split_off(&end_ms.saturating_add(1)); + } } pub fn len(&self) -> usize { - self.panes.len() + self.panes.values().map(BTreeMap::len).sum() } pub fn is_empty(&self) -> bool { - self.panes.is_empty() + self.panes.values().all(BTreeMap::is_empty) } } diff --git a/data_plane/tests/common/mod.rs b/data_plane/tests/common/mod.rs new file mode 100644 index 00000000..6029a045 --- /dev/null +++ b/data_plane/tests/common/mod.rs @@ -0,0 +1,24 @@ +use std::path::Path; + +use control_plane::{plan_workload, DeploymentInputs, InstalledPlan}; + +/// The deployment inputs of a backend that keeps raw data or not. +pub fn inputs(raw_data_retained: bool) -> DeploymentInputs { + let mut inputs = DeploymentInputs::default(); + inputs.deployment_capabilities.raw_data_retained = raw_data_retained; + inputs +} + +/// 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(); + plan_workload(&workload, &inputs(raw_data_retained)).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..3fdea416 --- /dev/null +++ b/data_plane/tests/precompute.rs @@ -0,0 +1,136 @@ +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use control_plane::{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, &Sds::ingestion_frontier(&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, at `from`, `from + 15 s`, ... before `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(); + // 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"); + assert_eq!(pane.num_rows(), 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 * 37); + assert_eq!(stats.accepted, 2 * 99); +} + +/// 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(&[sample("latency_ms", "a", 45 * MINUTE, 1.0)]); + assert_eq!(engine.store.get(&id, 2 * PANE).unwrap().num_rows(), 1); + assert_eq!(engine.store.get(&id, 3 * PANE).unwrap().num_rows(), 0); + assert_eq!(engine.store.get(&id, 4 * PANE).unwrap().num_rows(), 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 (mut engine, sds) = precompute(plan_workload(&workload, &common::inputs(false)).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().num_rows(), 1); +} + +/// Panes are `(end - width, end]`, like PromQL ranges: a sample at a +/// boundary belongs to the pane it ends. +#[test] +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); +} + +/// The engine puts a sample at a pane boundary into the pane it ends, and a +/// sample one millisecond later into the next pane. +#[test] +fn engine_assigns_boundary_samples_to_the_pane_they_end() { + let (mut engine, sds) = precompute(common::plan("example4b", false)); + let id = sds.definitions.keys().next().unwrap().clone(); + engine.ingest(&[ + // Opens the skipped first pane (0, 10 min]. + sample("latency_ms", "a", MINUTE, 1.0), + sample("latency_ms", "a", 15 * MINUTE, 1.0), + sample("latency_ms", "b", 2 * PANE, 1.0), + sample("latency_ms", "c", 2 * PANE + 1, 1.0), + // Moves the watermark past (20 min, 30 min]. + sample("latency_ms", "a", 45 * MINUTE, 1.0), + ]); + // (10 min, 20 min]: series a and b, the one at its end included. + assert_eq!(engine.store.get(&id, 2 * PANE).unwrap().num_rows(), 2); + // (20 min, 30 min]: only c, one millisecond after the boundary. + assert_eq!(engine.store.get(&id, 3 * PANE).unwrap().num_rows(), 1); +} diff --git a/data_plane/tests/sds.rs b/data_plane/tests/sds.rs index a25a2d07..0f5361cd 100644 --- a/data_plane/tests/sds.rs +++ b/data_plane/tests/sds.rs @@ -1,25 +1,30 @@ -use std::path::Path; +mod common; -use asap_executor::physical_planner::frontier_from_timing; -use control_plane::{plan_workload, DeploymentInputs, InstalledPlan}; +use common::plan; use data_plane::sds::{DefinitionId, Sds, SdsError, SummaryStore}; use planner_types::ir::physical_export::PhysicalASAPDAG; use planner_types::ir::{NonASAPOp, Operator}; use planner_types::physical::{Materialization, Medium}; -/// The selected plan of a #509 example workload in `examples/`, for a -/// deployment that keeps raw data or not. -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 mut inputs = DeploymentInputs::default(); - inputs.deployment_capabilities.raw_data_retained = raw_data_retained; - plan_workload(&workload, &inputs).unwrap() +fn sds_of(dag: &PhysicalASAPDAG) -> Result { + Sds::from_dag(dag, &Sds::ingestion_frontier(dag).unwrap()) } -fn sds_of(dag: &PhysicalASAPDAG) -> Result { - Sds::from_dag(dag, &frontier_from_timing(dag).unwrap()) +/// The SDS's frontier is the executor's: every ingestion-time output read at +/// query time, on every example with and without raw data. +#[test] +fn ingestion_frontier_matches_the_executor() { + for example in ["example1", "example3b", "example4b"] { + for raw_data_retained in [true, false] { + let dag = plan(example, raw_data_retained).dag.dag; + let executor = asap_executor::physical_planner::frontier_from_timing(&dag).unwrap(); + assert_eq!( + Sds::ingestion_frontier(&dag).unwrap(), + executor, + "{example} raw_data_retained={raw_data_retained}" + ); + } + } } /// Example 4b's six ingestion-time KLL panes are one definition with a diff --git a/data_plane/tests/server.rs b/data_plane/tests/server.rs index 969c7625..eba0b69c 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; use std::time::Duration; @@ -8,7 +10,7 @@ use axum::http::{header, HeaderMap, Method, Request, StatusCode}; use axum::routing::get; use axum::Router; use data_plane::remote_write::{encode, Label, Sample, TimeSeries, WriteRequest}; -use data_plane::{router, DataPlane, DEFAULT_FORWARD_TIMEOUT, EXECUTION_HEADER}; +use data_plane::{router, DataPlane, EXECUTION_HEADER}; use http_body_util::BodyExt; use tower::ServiceExt; @@ -38,6 +40,11 @@ async fn serve(app: Router) -> String { url } +/// A data plane for #509 Example 1 that forwards to `prometheus_url`. +fn data_plane(prometheus_url: String) -> Arc { + Arc::new(DataPlane::new(common::plan_json("example1", true), prometheus_url, 0).unwrap()) +} + async fn send(app: &Router, request: Request) -> (StatusCode, Option, String) { let response = app.clone().oneshot(request).await.unwrap(); let status = response.status(); @@ -53,11 +60,7 @@ async fn send(app: &Router, request: Request) -> (StatusCode, Option Date: Tue, 20 Oct 2026 19:12:03 +0000 Subject: [PATCH 2/4] fix(precompute): bound sample times, queue by samples, report counters - One future timestamp no longer closes every pane up to it: only samples of a metric a stored summary reads move the watermark, samples more than --max-future-ms (default 5 min) ahead of the wall clock are dropped and counted, and when the watermark jumps the panes that would already be evicted are skipped instead of built. Pane arithmetic is checked, so timestamps near i64::MIN/MAX neither overflow nor loop. - Pin ASAPPlanner main ecea897f, whose RawSampleSelector drops stale markers (a range window keeps the samples around one; an instant selector whose latest sample is one has no row). The HTTP-layer stale filter is removed. - A stopped engine thread makes Remote Write answer 500 and /-/ready 503, instead of 503 forever. - The engine queue is bounded by queued samples (--max-queued-samples, default 2,000,000); beyond it Remote Write gets 503. - /api/v1/status/asap reports the engine's counters. Sample counters count each sample once: accepted, dropped late, skipped with the first pane, dropped as future, unrouted; pane counters count built, failed and skipped-as-expired panes. - --allowed-lateness-ms is unsigned, and a zero-width pane is refused. - The plan is parsed once, on the engine thread. run_root executes one root; the selector's unused end offset is 0. --- Cargo.lock | 14 +-- Cargo.toml | 10 +- README.md | 9 +- data_plane/src/engine.rs | 183 ++++++++++++++++++++++------- data_plane/src/lib.rs | 94 ++++++++++----- data_plane/src/main.rs | 26 ++++- data_plane/src/precompute.rs | 207 +++++++++++++++++++++++++-------- data_plane/tests/precompute.rs | 178 ++++++++++++++++++++++++++-- data_plane/tests/server.rs | 110 ++++++++++++++++++ 9 files changed, 685 insertions(+), 146 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5d652394..18448423 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -252,7 +252,7 @@ dependencies = [ [[package]] name = "asap-executor" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f9dd41e095997fe2c34c1a1278f9f20c50a814ff#f9dd41e095997fe2c34c1a1278f9f20c50a814ff" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ecea897f78640c377ce1d80846156f330f62dee2#ecea897f78640c377ce1d80846156f330f62dee2" dependencies = [ "ahash", "arrow", @@ -270,7 +270,7 @@ dependencies = [ [[package]] name = "asap-frontend-common" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f9dd41e095997fe2c34c1a1278f9f20c50a814ff#f9dd41e095997fe2c34c1a1278f9f20c50a814ff" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ecea897f78640c377ce1d80846156f330f62dee2#ecea897f78640c377ce1d80846156f330f62dee2" dependencies = [ "asap-types", "serde", @@ -280,7 +280,7 @@ dependencies = [ [[package]] name = "asap-frontend-promql" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f9dd41e095997fe2c34c1a1278f9f20c50a814ff#f9dd41e095997fe2c34c1a1278f9f20c50a814ff" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ecea897f78640c377ce1d80846156f330f62dee2#ecea897f78640c377ce1d80846156f330f62dee2" dependencies = [ "asap-frontend-common", "asap-types", @@ -290,7 +290,7 @@ dependencies = [ [[package]] name = "asap-logical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f9dd41e095997fe2c34c1a1278f9f20c50a814ff#f9dd41e095997fe2c34c1a1278f9f20c50a814ff" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ecea897f78640c377ce1d80846156f330f62dee2#ecea897f78640c377ce1d80846156f330f62dee2" dependencies = [ "asap-types", "asap_sketchlib", @@ -302,7 +302,7 @@ dependencies = [ [[package]] name = "asap-physical-asap-optimization" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f9dd41e095997fe2c34c1a1278f9f20c50a814ff#f9dd41e095997fe2c34c1a1278f9f20c50a814ff" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ecea897f78640c377ce1d80846156f330f62dee2#ecea897f78640c377ce1d80846156f330f62dee2" dependencies = [ "asap-logical-asap-optimization", "asap-types", @@ -312,7 +312,7 @@ dependencies = [ [[package]] name = "asap-plan-selection" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f9dd41e095997fe2c34c1a1278f9f20c50a814ff#f9dd41e095997fe2c34c1a1278f9f20c50a814ff" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ecea897f78640c377ce1d80846156f330f62dee2#ecea897f78640c377ce1d80846156f330f62dee2" dependencies = [ "asap-logical-asap-optimization", "asap-physical-asap-optimization", @@ -326,7 +326,7 @@ dependencies = [ [[package]] name = "asap-types" version = "0.1.0" -source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=f9dd41e095997fe2c34c1a1278f9f20c50a814ff#f9dd41e095997fe2c34c1a1278f9f20c50a814ff" +source = "git+https://github.com/ProjectASAP/ASAPPlanner?rev=ecea897f78640c377ce1d80846156f330f62dee2#ecea897f78640c377ce1d80846156f330f62dee2" dependencies = [ "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 3469c8f0..bf6734c4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,11 +9,11 @@ version = "0.1.0" [workspace.dependencies] # Keep every Planner crate on the same immutable revision. Alias upstream # asap-types as planner-types to name its origin at use sites. -planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f9dd41e095997fe2c34c1a1278f9f20c50a814ff" } -asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f9dd41e095997fe2c34c1a1278f9f20c50a814ff" } -asap-logical-asap-optimization = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f9dd41e095997fe2c34c1a1278f9f20c50a814ff" } -asap-plan-selection = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f9dd41e095997fe2c34c1a1278f9f20c50a814ff" } -asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "f9dd41e095997fe2c34c1a1278f9f20c50a814ff" } +planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" } +asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" } +asap-logical-asap-optimization = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" } +asap-plan-selection = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" } +asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/README.md b/README.md index e4d68e5a..e642564d 100644 --- a/README.md +++ b/README.md @@ -37,8 +37,13 @@ The backend is being rebuilt on the Planner's physical DAG: | 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. +of a stored summary's metric passes their end; later samples for them are +dropped. The first pane after startup is skipped because it may be partial. +Samples more than `--max-future-ms` (default 5 min) ahead of the wall clock +are dropped. Remote Write gets 503 while more than `--max-queued-samples` +(default 2,000,000) samples wait for the engine, and 500 if the engine has +stopped, which also makes `/-/ready` fail. `/api/v1/status/asap` reports the +engine's sample and pane counters as JSON. Until the query engine lands, the data plane forwards every query to Prometheus, with its end-to-end headers such as `Authorization` and diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index 4505b786..fa5bf42a 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -2,75 +2,170 @@ //! 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 std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::mpsc::{channel, Receiver, Sender}; +use std::sync::{Arc, Mutex}; use control_plane::InstalledPlan; -use tokio::sync::oneshot; -use crate::precompute::{Precompute, PrecomputeStats, RawSample}; +use crate::precompute::{Precompute, PrecomputeOptions, PrecomputeStats, RawSample}; use crate::sds::Sds; -/// Commands queued for the engine before ingestion pushes back. -const QUEUE: usize = 1024; +/// How the engine thread ingests samples. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct EngineOptions { + /// Lateness, future bound and pane checks of the precompute engine. + pub precompute: PrecomputeOptions, + /// Samples queued for the engine, not yet ingested, beyond which Remote + /// Write is refused with 503 so senders retry. + pub max_queued_samples: u64, +} + +impl Default for EngineOptions { + fn default() -> Self { + Self { + precompute: PrecomputeOptions::default(), + // Two Remote Write requests of the default sample limit. + max_queued_samples: 2_000_000, + } + } +} -pub enum Command { +enum Command { Ingest(Vec), - Stats(oneshot::Sender), + Stop, } -#[derive(Debug, thiserror::Error)] -#[error("the engine is busy")] -pub struct Busy; +/// Why the engine did not take samples. +#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] +pub enum IngestError { + /// More samples are queued than `max_queued_samples`; retry later. + #[error("the engine is busy: too many samples queued")] + Busy, + /// The engine thread has stopped; no sample will be ingested again. + #[error("the engine has stopped")] + Stopped, +} /// The HTTP side of the engine thread. #[derive(Clone)] pub struct EngineHandle { - commands: SyncSender, + commands: Sender, + /// Samples sent and not yet ingested. + queued_samples: Arc, + max_queued_samples: u64, + /// False once the engine thread has exited, for any reason. + running: Arc, + /// The counters as of the last command the engine ran. + stats: Arc>, +} + +/// Marks the engine stopped when its thread exits, panics included. +struct RunningGuard(Arc); + +impl Drop for RunningGuard { + fn drop(&mut self) { + self.0.store(false, Ordering::SeqCst); + } } impl EngineHandle { - /// Starts the engine for the plan serialized in `plan_json`. Fails if the + /// Starts the engine for the plan serialized in `plan_json`, parsed once + /// on the engine thread, and returns the plan's queries. 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); + pub fn spawn(plan_json: String, options: EngineOptions) -> anyhow::Result<(Self, Vec)> { + let (commands, receiver) = channel(); let (ready, started) = std::sync::mpsc::channel(); + let running = Arc::new(AtomicBool::new(true)); + let queued_samples = Arc::new(AtomicU64::new(0)); + let stats = Arc::new(Mutex::new(PrecomputeStats::default())); + let shared = Shared { + queued_samples: queued_samples.clone(), + stats: stats.clone(), + }; + let guard = RunningGuard(running.clone()); 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)); + .spawn(move || { + let _guard = guard; + match Engine::new(&plan_json, options.precompute) { + Ok((engine, queries)) => { + let _ = ready.send(Ok(queries)); + engine.serve(receiver, shared); + } + Err(error) => { + let _ = ready.send(Err(error)); + } } })?; - started.recv()??; - Ok(Self { commands }) + let queries = started.recv()??; + let handle = Self { + commands, + queued_samples, + max_queued_samples: options.max_queued_samples, + running, + stats, + }; + Ok((handle, queries)) } - /// 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), + /// Queues `samples` for precompute. `Busy` when the queue would hold + /// more than `max_queued_samples`; `Stopped` when the engine has exited. + pub fn try_ingest(&self, samples: Vec) -> Result<(), IngestError> { + if !self.is_running() { + return Err(IngestError::Stopped); } + let count = samples.len() as u64; + let admitted = + self.queued_samples + .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |queued| { + queued + .checked_add(count) + .filter(|&total| total <= self.max_queued_samples) + }); + if admitted.is_err() { + return Err(IngestError::Busy); + } + self.commands.send(Command::Ingest(samples)).map_err(|_| { + self.queued_samples.fetch_sub(count, Ordering::SeqCst); + IngestError::Stopped + }) + } + + /// Whether the engine thread is still running. + pub fn is_running(&self) -> bool { + self.running.load(Ordering::SeqCst) + } + + /// Samples queued and not yet ingested. + pub fn queued_samples(&self) -> u64 { + self.queued_samples.load(Ordering::SeqCst) } - /// 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() + /// The precompute counters as of the last batch the engine ingested. + pub fn stats(&self) -> PrecomputeStats { + *self.stats.lock().unwrap_or_else(|e| e.into_inner()) + } + + /// Stops the engine thread after the commands queued before; later + /// ingestion fails with [`IngestError::Stopped`]. + pub fn stop(&self) { + let _ = self.commands.send(Command::Stop); } } +/// What the engine thread publishes to the HTTP side. +struct Shared { + queued_samples: Arc, + stats: Arc>, +} + struct Engine { precompute: Precompute, } impl Engine { - fn new(plan_json: &str, allowed_lateness_ms: i64) -> anyhow::Result { + fn new(plan_json: &str, options: PrecomputeOptions) -> anyhow::Result<(Self, Vec)> { let plan: InstalledPlan = serde_json::from_str(plan_json)?; plan.validate()?; let dag = plan.dag.dag; @@ -81,19 +176,25 @@ impl Engine { outputs = sds.outputs.len(), "stored summaries" ); - Ok(Self { - precompute: Precompute::new(dag, &sds, allowed_lateness_ms)?, - }) + let engine = Self { + precompute: Precompute::new(dag, &sds, options)?, + }; + Ok((engine, plan.queries)) } - fn serve(mut self, commands: Receiver) { + fn serve(mut self, commands: Receiver, shared: Shared) { for command in commands { match command { - Command::Ingest(samples) => self.precompute.ingest(&samples), - Command::Stats(reply) => { - let _ = reply.send(self.precompute.stats); + Command::Ingest(samples) => { + self.precompute.ingest(&samples); + *shared.stats.lock().unwrap_or_else(|e| e.into_inner()) = self.precompute.stats; + shared + .queued_samples + .fetch_sub(samples.len() as u64, Ordering::SeqCst); } + Command::Stop => break, } } + tracing::warn!("engine stopped"); } } diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index 71904dd6..d501c70e 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -21,13 +21,9 @@ 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 engine::{EngineHandle, EngineOptions, IngestError}; 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"; @@ -48,18 +44,30 @@ pub struct DataPlane { } impl DataPlane { - /// Installs the plan serialized in `plan_json` (an [`InstalledPlan`]). - /// Panes close `allowed_lateness_ms` after the newest sample passes them. + /// Installs the plan serialized in `plan_json` (a + /// `control_plane::InstalledPlan`) with [`EngineOptions::default`], except + /// that panes close `allowed_lateness_ms` after the newest sample passes + /// them. pub fn new( plan_json: String, prometheus_url: String, - allowed_lateness_ms: i64, + allowed_lateness_ms: u64, + ) -> anyhow::Result { + let mut options = EngineOptions::default(); + options.precompute.allowed_lateness_ms = allowed_lateness_ms; + Self::with_options(plan_json, prometheus_url, options) + } + + /// Installs the plan serialized in `plan_json` with `options`. + pub fn with_options( + plan_json: String, + prometheus_url: String, + options: EngineOptions, ) -> anyhow::Result { - let plan: InstalledPlan = serde_json::from_str(&plan_json)?; - plan.validate()?; + let (engine, queries) = EngineHandle::spawn(plan_json, options)?; Ok(Self { - queries: plan.queries, - engine: EngineHandle::spawn(plan_json, allowed_lateness_ms)?, + queries, + engine, prometheus_url: prometheus_url.trim_end_matches('/').to_string(), limits: remote_write::Limits::default(), samples: AtomicU64::new(0), @@ -90,7 +98,8 @@ pub fn router(state: Arc) -> Router { .route("/api/v1/write", post(write).layer(write_body_limit)) .route("/api/v1/query", get(query).post(query)) .route("/api/v1/query_range", get(query_range).post(query_range)) - .route("/-/ready", get(|| async { StatusCode::OK })) + .route("/api/v1/status/asap", get(status)) + .route("/-/ready", get(ready)) .with_state(state) } @@ -106,17 +115,13 @@ async fn write(State(state): State>, body: Bytes) -> Response { .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, - }), - ); + // Stale markers are kept: the executor's sample selector + // drops them from range windows and ends instant selectors. + samples.extend(series.samples.into_iter().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) { @@ -124,14 +129,51 @@ async fn write(State(state): State>, body: Bytes) -> Response { 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(), + // Remote Write senders retry on 503. + Err(IngestError::Busy) => ( + StatusCode::SERVICE_UNAVAILABLE, + IngestError::Busy.to_string(), + ) + .into_response(), + Err(IngestError::Stopped) => { + tracing::error!("Remote Write refused: the engine has stopped"); + ( + StatusCode::INTERNAL_SERVER_ERROR, + IngestError::Stopped.to_string(), + ) + .into_response() + } } } Err(error) => (StatusCode::BAD_REQUEST, error.to_string()).into_response(), } } +/// Ready while the engine thread runs; a stopped engine never ingests +/// again, so the data plane must be restarted. +async fn ready(State(state): State>) -> StatusCode { + if state.engine.is_running() { + StatusCode::OK + } else { + StatusCode::SERVICE_UNAVAILABLE + } +} + +/// The data plane's counters, as JSON: the precompute engine's sample and +/// pane counters, the samples queued for it, and whether it runs. +async fn status(State(state): State>) -> Response { + let body = serde_json::json!({ + "engine_running": state.engine.is_running(), + "samples_queued": state.engine.queued_samples(), + "precompute": state.engine.stats(), + }); + ( + [(header::CONTENT_TYPE, "application/json")], + body.to_string(), + ) + .into_response() +} + async fn query( state: State>, method: Method, diff --git a/data_plane/src/main.rs b/data_plane/src/main.rs index ff511085..88186e41 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -3,6 +3,8 @@ use std::sync::Arc; use std::time::Duration; use clap::Parser; +use data_plane::engine::EngineOptions; +use data_plane::precompute::PrecomputeOptions; use data_plane::{router, DataPlane, DEFAULT_FORWARD_TIMEOUT}; /// Serve Remote Write ingestion and the Prometheus query API for a plan. @@ -20,8 +22,14 @@ struct Args { #[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, + #[arg(long, default_value_t = PrecomputeOptions::default().allowed_lateness_ms)] + allowed_lateness_ms: u64, + /// Samples further ahead of the wall clock than this are dropped. + #[arg(long, default_value_t = PrecomputeOptions::default().max_future_ms)] + max_future_ms: u64, + /// Samples queued for the engine beyond which Remote Write gets 503. + #[arg(long, default_value_t = EngineOptions::default().max_queued_samples)] + max_queued_samples: u64, } #[tokio::main] @@ -32,8 +40,18 @@ async fn main() -> anyhow::Result<()> { let args = Args::parse(); let plan = std::fs::read_to_string(&args.plan)?; let state = Arc::new( - DataPlane::new(plan, args.prometheus_url, args.allowed_lateness_ms)? - .with_forward_timeout(Duration::from_secs(args.forward_timeout_seconds)), + DataPlane::with_options( + plan, + args.prometheus_url, + EngineOptions { + precompute: PrecomputeOptions { + allowed_lateness_ms: args.allowed_lateness_ms, + max_future_ms: args.max_future_ms, + }, + max_queued_samples: args.max_queued_samples, + }, + )? + .with_forward_timeout(Duration::from_secs(args.forward_timeout_seconds)), ); tracing::info!(queries = state.queries.len(), "installed plan"); let listener = tokio::net::TcpListener::bind(&args.listen).await?; diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index 03571c65..e4708602 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -7,11 +7,15 @@ //! `end`. Closing it selects the pane's rows from the buffered samples with //! the executor's `RawSampleSelector` (the `Scan`'s label matchers), then runs //! 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. +//! 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. +//! +//! Only samples of a metric some definition reads move the watermark, and a +//! sample further ahead of the wall clock than `max_future_ms` is dropped, so +//! one bad timestamp cannot close every pane up to it. When the watermark +//! jumps, panes that would already be evicted are skipped, not built. use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; @@ -45,17 +49,47 @@ pub enum PrecomputeError { Executor(#[from] asap_executor::Error), } -/// Counters for what happened to incoming samples. -#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +/// Counters for what happened to incoming samples, each sample counted once +/// (`samples_*`), and to panes, each pane of each definition counted once +/// (`panes_*`). +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, serde::Serialize)] pub struct PrecomputeStats { /// Samples buffered for at least one pane. - pub accepted: u64, - /// Samples for a pane already built or skipped. - pub dropped: u64, + pub samples_accepted: u64, + /// Samples of a pane already built or skipped, other than the first + /// pane after startup. + pub samples_dropped_late: u64, + /// Samples of the first pane after startup, which is never built. + pub samples_skipped_first_pane: u64, + /// Samples more than `max_future_ms` ahead of the wall clock. + pub samples_dropped_future: u64, + /// Samples of a metric no stored summary reads. + pub samples_unrouted: u64, /// Panes built and stored. pub panes_built: u64, /// Panes whose build failed; they are skipped, not retried. pub panes_failed: u64, + /// Panes not built because they would already be evicted when the + /// watermark reached them. + pub panes_skipped_expired: u64, +} + +/// How the precompute engine treats sample times. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct PrecomputeOptions { + /// A pane closes when the newest routed sample is this far past its end. + pub allowed_lateness_ms: u64, + /// Samples further ahead of the wall clock than this are dropped. + pub max_future_ms: u64, +} + +impl Default for PrecomputeOptions { + fn default() -> Self { + Self { + allowed_lateness_ms: 15_000, + max_future_ms: 5 * 60_000, + } + } } struct Builder { @@ -65,19 +99,30 @@ struct Builder { selector: RawSampleSelector, raw_schema: SchemaRef, build: CompiledPhysicalDAG, - /// Oldest pane offset any output reads, plus one pane: panes ending + /// Largest offset any output reads, plus two panes: panes ending at or /// before `watermark - retention_ms` are no longer read. retention_ms: i64, /// Samples of open panes, by pane end. open: BTreeMap>, + /// End of the first pane seen, which is skipped. `None` until then. + first_pane_end: Option, /// End of the newest pane built or skipped; samples in it or before it /// are dropped. `None` until the first sample. built_through: Option, } +/// What one definition did with one sample. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +enum Outcome { + Late, + FirstPane, + Buffered, +} + pub struct Precompute { builders: Vec, allowed_lateness_ms: i64, + max_future_ms: i64, newest_sample_ms: Option, pub store: SummaryStore, pub stats: PrecomputeStats, @@ -88,21 +133,26 @@ impl Precompute { pub fn new( dag: PhysicalASAPDAG, sds: &Sds, - allowed_lateness_ms: i64, + options: PrecomputeOptions, ) -> Result { let mut builders = Vec::new(); for definition in sds.definitions.values() { let error = |message: String| PrecomputeError::Definition(definition.id.clone(), message); + if definition.pane_ms <= 0 { + return Err(error(format!("pane width {} ms", definition.pane_ms))); + } let scan = 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()); + // `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, &RawSampleWindow { time_range_node_id: definition.raw_input as usize, width_ms: definition.pane_ms, - end_offset_ms: sds.outputs[&definition.builder].1.offset_ms, + end_offset_ms: 0, scan_node_id: scan as usize, }, )?; @@ -121,19 +171,25 @@ impl Precompute { .map(|(_, binding)| binding.offset_ms) .max() .unwrap_or(0); + let retention_ms = max_offset + .checked_add(2 * definition.pane_ms) + .ok_or_else(|| error("retention overflows".into()))?; builders.push(Builder { definition: definition.clone(), selector, raw_schema, build, - retention_ms: max_offset + 2 * definition.pane_ms, + retention_ms, open: BTreeMap::new(), + first_pane_end: None, built_through: None, }); } + let millis = |value: u64| i64::try_from(value).unwrap_or(i64::MAX); Ok(Self { builders, - allowed_lateness_ms, + allowed_lateness_ms: millis(options.allowed_lateness_ms), + max_future_ms: millis(options.max_future_ms), newest_sample_ms: None, store: SummaryStore::new(), stats: PrecomputeStats::default(), @@ -143,33 +199,61 @@ impl Precompute { /// 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) + .map(|newest| newest.saturating_sub(self.allowed_lateness_ms)) } - /// Buffers `samples` into their panes, then builds every pane the - /// watermark has passed. + /// [`Self::ingest_at`] the current wall-clock time. pub fn ingest(&mut self, samples: &[RawSample]) { + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| i64::try_from(d.as_millis()).unwrap_or(i64::MAX)); + self.ingest_at(samples, now_ms); + } + + /// Buffers `samples` into their panes, then builds every pane the + /// watermark has passed. `now_ms` is the wall-clock time that bounds + /// sample times from above. + pub fn ingest_at(&mut self, samples: &[RawSample], now_ms: i64) { + let latest_allowed_ms = now_ms.saturating_add(self.max_future_ms); for sample in samples { + if sample.timestamp_ms > latest_allowed_ms { + self.stats.samples_dropped_future += 1; + continue; + } let metric = sample.labels.get("__name__").map(String::as_str); - let mut accepted = false; + let mut outcome: Option = None; for builder in &mut self.builders { if metric != Some(builder.selector.metric()) { 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; + let this = match pane_end(sample.timestamp_ms, builder.definition.pane_ms) { + None => Outcome::Late, + Some(end) => { + let first = *builder.first_pane_end.get_or_insert(end); + let built = *builder.built_through.get_or_insert(end); + if end == first { + Outcome::FirstPane + } else if end <= built { + Outcome::Late + } else { + builder.open.entry(end).or_default().push(sample.clone()); + Outcome::Buffered + } + } + }; + outcome = outcome.max(Some(this)); } - if accepted { - self.stats.accepted += 1; + 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)); + } } - self.newest_sample_ms = self.newest_sample_ms.max(Some(sample.timestamp_ms)); } self.close_panes(); } @@ -183,35 +267,59 @@ impl Precompute { continue; }; let pane_ms = builder.definition.pane_ms; + let id = &builder.definition.id; + // Panes ending at or before this are evicted once built: skip them. + let expired_through = watermark.saturating_sub(builder.retention_ms); + let Some(mut end) = built.checked_add(pane_ms) else { + continue; + }; + if end <= expired_through { + let Some(first_kept) = expired_through + .checked_add(1) + .and_then(|t| pane_end(t, pane_ms)) + else { + continue; + }; + self.stats.panes_skipped_expired += + (first_kept.saturating_sub(end) / pane_ms) as u64; + builder.open = builder.open.split_off(&first_kept); + end = first_kept; + } // 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(builder, end, &samples) { Ok(state) => { - self.store.insert(&builder.definition.id, end, state); + self.store.insert(id, end, state); self.stats.panes_built += 1; } Err(error) => { - tracing::error!(definition = ?builder.definition.id, end, %error, "pane build failed"); + tracing::error!(definition = ?id, end, %error, "pane build failed"); self.stats.panes_failed += 1; } } builder.built_through = Some(end); - end += pane_ms; + match end.checked_add(pane_ms) { + Some(next) => end = next, + None => break, + } } - self.store - .evict_through(&builder.definition.id, watermark - builder.retention_ms); + self.store.evict_through(id, expired_through); } } } -/// 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) * pane_ms +/// End of the tumbling pane of width `pane_ms` that contains `timestamp_ms`, +/// or `None` when it does not fit in an `i64`. 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) -> Option { + timestamp_ms + .checked_neg()? + .checked_div_euclid(pane_ms)? + .checked_neg()? + .checked_mul(pane_ms) } /// Selects the pane's rows from `samples` with the plan's `Scan` matchers, @@ -221,12 +329,13 @@ fn build_pane( end: i64, samples: &[RawSample], ) -> Result { - let start = end - builder.definition.pane_ms; let scope = Scope::Ingestion { - window_start_ms: start, + window_start_ms: end - builder.definition.pane_ms, window_end_ms: end, revision: 1, }; + // The selector takes owned label maps, so each sample's labels are + // copied once per pane build. let samples: Vec = samples .iter() .map(|s| Sample { @@ -243,20 +352,20 @@ fn build_pane( let build = builder .build .instantiate(BTreeMap::from([(builder.definition.raw_input, input)]))?; - let rows = run(&build, builder.build.roots(), scope)?; + let rows = run_root(&build, builder.definition.builder, 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( +/// Executes `root` of `dag` to completion and returns its rows. +pub fn run_root( dag: &asap_executor::plan::PhysicalDAG<'_, Batch, SchemaRef>, - roots: &[u64], + root: 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 output = dag.execute(&[root], context)?.remove(0); let mut rows = Vec::new(); while let Some(batch) = output.next().await { rows.extend(batch?.to_rows()); diff --git a/data_plane/tests/precompute.rs b/data_plane/tests/precompute.rs index 3fdea416..58d4a4b4 100644 --- a/data_plane/tests/precompute.rs +++ b/data_plane/tests/precompute.rs @@ -4,17 +4,29 @@ use std::collections::BTreeMap; use std::sync::Arc; use control_plane::{plan_workload, InstalledPlan}; -use data_plane::precompute::{pane_end, Precompute, RawSample}; +use data_plane::precompute::{pane_end, Precompute, PrecomputeOptions, RawSample}; use data_plane::sds::Sds; use planner_types::workload::{PlanningWorkload, Query}; const MINUTE: i64 = 60_000; const PANE: i64 = 10 * MINUTE; +const DAY: i64 = 24 * 60 * MINUTE; +/// The Prometheus stale marker's bits. +const STALE_MARKER: u64 = 0x7ff0_0000_0000_0002; + fn precompute(plan: InstalledPlan) -> (Precompute, Sds) { + precompute_with(plan, 0) +} + +fn precompute_with(plan: InstalledPlan, allowed_lateness_ms: u64) -> (Precompute, Sds) { let dag = plan.dag.dag; let sds = Sds::from_dag(&dag, &Sds::ingestion_frontier(&dag).unwrap()).unwrap(); - (Precompute::new(dag, &sds, 0).unwrap(), sds) + let options = PrecomputeOptions { + allowed_lateness_ms, + ..PrecomputeOptions::default() + }; + (Precompute::new(dag, &sds, options).unwrap(), sds) } fn sample(metric: &str, job: &str, timestamp_ms: i64, value: f64) -> RawSample { @@ -58,8 +70,9 @@ fn builds_panes_after_the_first_as_the_watermark_passes() { 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 * 37); - assert_eq!(stats.accepted, 2 * 99); + assert_eq!(stats.samples_skipped_first_pane, 2 * 37); + assert_eq!(stats.samples_dropped_late, 0); + assert_eq!(stats.samples_accepted, 2 * 99); } /// Samples of other metrics are ignored; samples for a built pane are dropped. @@ -69,9 +82,17 @@ fn ignores_other_metrics_and_drops_late_samples() { 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); + assert_eq!(engine.stats.samples_unrouted, before.samples_unrouted + 1); + assert_eq!(engine.stats.samples_accepted, before.samples_accepted); engine.ingest(&[sample("latency_ms", "a", 15 * MINUTE, 1.0)]); - assert_eq!(engine.stats.dropped, before.dropped + 1); + assert_eq!( + engine.stats.samples_dropped_late, + before.samples_dropped_late + 1 + ); + assert_eq!( + engine.stats.samples_skipped_first_pane, + before.samples_skipped_first_pane + ); } /// A pane without samples is stored empty, so a query can tell "no data" @@ -106,12 +127,12 @@ fn applies_label_matchers() { /// boundary belongs to the pane it ends. #[test] 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); + assert_eq!(pane_end(1, PANE), Some(PANE)); + assert_eq!(pane_end(PANE, PANE), Some(PANE)); + assert_eq!(pane_end(PANE + 1, PANE), Some(2 * PANE)); + assert_eq!(pane_end(0, PANE), Some(0)); + assert_eq!(pane_end(-1, PANE), Some(0)); + assert_eq!(pane_end(-PANE, PANE), Some(-PANE)); } /// The engine puts a sample at a pane boundary into the pane it ends, and a @@ -134,3 +155,136 @@ fn engine_assigns_boundary_samples_to_the_pane_they_end() { // (20 min, 30 min]: only c, one millisecond after the boundary. assert_eq!(engine.store.get(&id, 3 * PANE).unwrap().num_rows(), 1); } + +/// Samples of a metric no definition reads do not move the watermark, even +/// far in the future: the next real sample is still accepted. +#[test] +fn unrouted_future_sample_does_not_close_panes() { + let (mut engine, _) = precompute(common::plan("example4b", false)); + let now = 30 * MINUTE; + engine.ingest_at(&samples("latency_ms", &["a"], 0, 25 * MINUTE), now); + let built = engine.stats.panes_built; + engine.ingest_at(&[sample("other_metric", "a", 30 * DAY, 1.0)], 30 * DAY); + assert_eq!(engine.stats.panes_built, built, "no pane closed"); + assert_eq!(engine.stats.samples_unrouted, 1); + engine.ingest_at(&[sample("latency_ms", "a", 26 * MINUTE, 1.0)], now); + assert_eq!(engine.stats.samples_dropped_late, 0); +} + +/// A routed sample further ahead of the wall clock than `max_future_ms` is +/// dropped and counted, and does not move the watermark. +#[test] +fn drops_samples_beyond_the_future_bound() { + let (mut engine, _) = precompute(common::plan("example4b", false)); + let now = 30 * MINUTE; + let bound = PrecomputeOptions::default().max_future_ms as i64; + engine.ingest_at(&samples("latency_ms", &["a"], 0, 25 * MINUTE), now); + let built = engine.stats.panes_built; + engine.ingest_at(&[sample("latency_ms", "a", now + bound + 1, 1.0)], now); + assert_eq!(engine.stats.samples_dropped_future, 1); + assert_eq!(engine.stats.panes_built, built); + // Exactly at the bound is accepted. + engine.ingest_at(&[sample("latency_ms", "a", now + bound, 1.0)], now); + assert_eq!(engine.stats.samples_dropped_future, 1); +} + +/// When the watermark jumps by 30 days, only the panes still retained are +/// built; the rest are skipped and counted, not built and evicted. +#[test] +fn watermark_jump_builds_only_retained_panes() { + let (mut engine, sds) = precompute(common::plan("example4b", false)); + let id = sds.definitions.keys().next().unwrap().clone(); + engine.ingest_at(&samples("latency_ms", &["a"], 0, 25 * MINUTE), 30 * DAY); + let built = engine.stats.panes_built; + engine.ingest_at(&[sample("latency_ms", "a", 30 * DAY, 1.0)], 30 * DAY); + // Example 4b reads panes up to 50 min back, plus two panes: 70 min. + assert_eq!(engine.stats.panes_built - built, 7); + assert_eq!( + engine.stats.panes_skipped_expired, + (30 * DAY - 70 * MINUTE) as u64 / PANE as u64 - 2 + ); + assert!(engine.store.get(&id, 30 * DAY).is_some()); + assert!(engine.store.get(&id, 2 * PANE).is_none()); + assert_eq!(engine.store.len(), 7); +} + +/// Timestamps at the ends of `i64` neither overflow nor loop. +#[test] +fn extreme_timestamps_do_not_overflow() { + assert_eq!(pane_end(i64::MIN, PANE), None); + assert_eq!(pane_end(i64::MAX, PANE), None); + let (mut engine, _) = precompute(common::plan("example4b", false)); + engine.ingest_at( + &[ + sample("latency_ms", "a", i64::MIN, 1.0), + sample("latency_ms", "a", i64::MIN + 1, 1.0), + sample("latency_ms", "a", 0, 1.0), + sample("latency_ms", "a", i64::MAX - 1, 1.0), + sample("latency_ms", "a", i64::MAX, 1.0), + ], + i64::MAX, + ); + assert!(engine.stats.panes_built <= 7); +} + +/// A pane of zero width is refused when the engine is built. +#[test] +fn rejects_zero_width_panes() { + let plan = common::plan("example4b", false); + let dag = plan.dag.dag; + let mut sds = Sds::from_dag(&dag, &Sds::ingestion_frontier(&dag).unwrap()).unwrap(); + for definition in sds.definitions.values_mut() { + definition.pane_ms = 0; + } + assert!(Precompute::new(dag, &sds, PrecomputeOptions::default()).is_err()); +} + +/// Panes older than the retention (largest offset plus two panes) are +/// evicted as the watermark advances. +#[test] +fn evicts_panes_past_retention() { + let (mut engine, sds) = precompute(common::plan("example4b", false)); + let id = sds.definitions.keys().next().unwrap().clone(); + engine.ingest(&samples("latency_ms", &["a"], 0, 25 * MINUTE)); + assert!(engine.store.get(&id, 2 * PANE).is_some()); + // Watermark 100 min: panes ending at or before 30 min are evicted. + engine.ingest(&[sample("latency_ms", "a", 100 * MINUTE, 1.0)]); + assert!(engine.store.get(&id, 2 * PANE).is_none()); + assert!(engine.store.get(&id, 3 * PANE).is_none()); + assert!(engine.store.get(&id, 4 * PANE).is_some()); + assert!(engine.store.get(&id, 10 * PANE).is_some()); +} + +/// Out-of-order samples within the allowed lateness reach their pane. +#[test] +fn accepts_out_of_order_samples_within_lateness() { + let (mut engine, sds) = precompute_with(common::plan("example4b", false), 5 * MINUTE as u64); + let id = sds.definitions.keys().next().unwrap().clone(); + engine.ingest(&[sample("latency_ms", "a", MINUTE, 1.0)]); + engine.ingest(&[sample("latency_ms", "a", 24 * MINUTE, 1.0)]); + // Watermark 19 min: (10, 20 min] is still open. + engine.ingest(&[sample("latency_ms", "b", 15 * MINUTE, 1.0)]); + engine.ingest(&[sample("latency_ms", "a", 26 * MINUTE, 1.0)]); + // Watermark 21 min: (10, 20 min] holds b's late sample. + assert_eq!(engine.store.get(&id, 2 * PANE).unwrap().num_rows(), 1); + assert_eq!(engine.stats.samples_dropped_late, 0); + engine.ingest(&[sample("latency_ms", "c", 18 * MINUTE, 1.0)]); + assert_eq!(engine.stats.samples_dropped_late, 1); +} + +/// Stale markers are not samples: a series whose only sample in a pane is +/// a stale marker has no state, and one with a value and a marker keeps it. +#[test] +fn stale_markers_are_not_samples() { + let (mut engine, sds) = precompute(common::plan("example4b", false)); + let id = sds.definitions.keys().next().unwrap().clone(); + let stale = f64::from_bits(STALE_MARKER); + engine.ingest(&[ + sample("latency_ms", "a", MINUTE, 1.0), + sample("latency_ms", "a", 12 * MINUTE, 1.0), + sample("latency_ms", "a", 13 * MINUTE, stale), + sample("latency_ms", "b", 14 * MINUTE, stale), + sample("latency_ms", "a", 25 * MINUTE, 1.0), + ]); + assert_eq!(engine.store.get(&id, 2 * PANE).unwrap().num_rows(), 1); +} diff --git a/data_plane/tests/server.rs b/data_plane/tests/server.rs index eba0b69c..695d22aa 100644 --- a/data_plane/tests/server.rs +++ b/data_plane/tests/server.rs @@ -9,6 +9,7 @@ use axum::extract::RawQuery; use axum::http::{header, HeaderMap, Method, Request, StatusCode}; use axum::routing::get; use axum::Router; +use data_plane::engine::EngineOptions; use data_plane::remote_write::{encode, Label, Sample, TimeSeries, WriteRequest}; use data_plane::{router, DataPlane, EXECUTION_HEADER}; use http_body_util::BodyExt; @@ -252,3 +253,112 @@ async fn forwarded_query_times_out() { .await; assert_eq!(status, StatusCode::BAD_GATEWAY, "{body}"); } + +/// A Remote Write body of `count` samples of one Example 1 series. +fn write_body(count: usize) -> Vec { + encode(&WriteRequest { + timeseries: vec![TimeSeries { + labels: vec![Label { + name: "__name__".into(), + value: "http_requests_total".into(), + }], + samples: (0..count) + .map(|i| Sample { + value: i as f64, + timestamp: 1_000 * (i as i64 + 1), + }) + .collect(), + }], + }) +} + +async fn write(app: &Router, count: usize) -> StatusCode { + let request = Request::post("/api/v1/write") + .body(Body::from(write_body(count))) + .unwrap(); + send(app, request).await.0 +} + +/// GET `path` and parse the JSON body. +async fn get_json(app: &Router, path: &str) -> (StatusCode, serde_json::Value) { + let (status, _, body) = send(app, Request::get(path).body(Body::empty()).unwrap()).await; + (status, serde_json::from_str(&body).unwrap_or_default()) +} + +/// Remote Write that would queue more samples than `max_queued_samples` gets +/// 503 and is not counted; a request that fits is accepted. +#[tokio::test] +async fn refuses_remote_write_beyond_the_sample_queue() { + let options = EngineOptions { + max_queued_samples: 2, + ..EngineOptions::default() + }; + let state = Arc::new( + DataPlane::with_options( + common::plan_json("example1", true), + "http://unused".into(), + options, + ) + .unwrap(), + ); + let app = router(state.clone()); + assert_eq!(write(&app, 3).await, StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(state.samples.load(Ordering::Relaxed), 0); + assert_eq!(write(&app, 2).await, StatusCode::NO_CONTENT); + assert_eq!(state.samples.load(Ordering::Relaxed), 2); +} + +/// Once the engine thread has stopped, Remote Write gets 500 instead of a +/// 503 that senders would retry forever, and the data plane is not ready. +#[tokio::test] +async fn stopped_engine_fails_writes_and_readiness() { + let state = data_plane("http://unused".into()); + let app = router(state.clone()); + let (status, _, _) = send(&app, Request::get("/-/ready").body(Body::empty()).unwrap()).await; + assert_eq!(status, StatusCode::OK); + state.engine.stop(); + for _ in 0..200 { + if !state.engine.is_running() { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(!state.engine.is_running()); + assert_eq!(write(&app, 1).await, StatusCode::INTERNAL_SERVER_ERROR); + let (status, _, _) = send(&app, Request::get("/-/ready").body(Body::empty()).unwrap()).await; + assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE); + let (_, body) = get_json(&app, "/api/v1/status/asap").await; + assert_eq!(body["engine_running"], false); +} + +/// The status endpoint reports what the engine did with every sample. +#[tokio::test] +async fn reports_sample_counters() { + let state = data_plane("http://unused".into()); + let app = router(state.clone()); + assert_eq!(write(&app, 3).await, StatusCode::NO_CONTENT); + let mut body = serde_json::Value::Null; + for _ in 0..200 { + let (status, json) = get_json(&app, "/api/v1/status/asap").await; + assert_eq!(status, StatusCode::OK); + body = json; + if body["samples_queued"] == 0 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert_eq!(body["engine_running"], true); + assert_eq!(body["samples_queued"], 0); + let counters = &body["precompute"]; + let total: u64 = [ + "samples_accepted", + "samples_dropped_late", + "samples_skipped_first_pane", + "samples_dropped_future", + "samples_unrouted", + ] + .iter() + .map(|name| counters[name].as_u64().unwrap()) + .sum(); + assert_eq!(total, 3, "{body}"); +} From ac6cec1bc1dddf6aeea14ecc13e24e8a92c1e6b6 Mon Sep 17 00:00:00 2001 From: zzylol Date: Tue, 20 Oct 2026 19:16:13 +0000 Subject: [PATCH 3/4] fix(engine): admit queued samples with compare_exchange AtomicU64::fetch_update is deprecated on current stable (renamed try_update), which fails clippy -D warnings in CI. --- data_plane/src/engine.rs | 24 +++++++++++++++--------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index fa5bf42a..865f41a2 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -116,15 +116,21 @@ impl EngineHandle { return Err(IngestError::Stopped); } let count = samples.len() as u64; - let admitted = - self.queued_samples - .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |queued| { - queued - .checked_add(count) - .filter(|&total| total <= self.max_queued_samples) - }); - if admitted.is_err() { - return Err(IngestError::Busy); + let mut queued = self.queued_samples.load(Ordering::SeqCst); + loop { + let total = queued + .checked_add(count) + .filter(|&total| total <= self.max_queued_samples) + .ok_or(IngestError::Busy)?; + match self.queued_samples.compare_exchange( + queued, + total, + Ordering::SeqCst, + Ordering::SeqCst, + ) { + Ok(_) => break, + Err(current) => queued = current, + } } self.commands.send(Command::Ingest(samples)).map_err(|_| { self.queued_samples.fetch_sub(count, Ordering::SeqCst); From 9b2df0f0cb02eae454a2820d816ddf3cd3caa41e Mon Sep 17 00:00:00 2001 From: zzylol Date: Tue, 20 Oct 2026 19:34:51 +0000 Subject: [PATCH 4/4] fix(precompute): cap the watermark at the wall clock; admit into an empty queue - The watermark is the newest routed sample capped at the wall clock, minus the allowed lateness: one series stamped ahead of the others (but within --max-future-ms) no longer closes panes early and drops on-time samples as late. - A request is always admitted into an empty queue, so a --max-queued-samples smaller than one request cannot refuse it forever. - A sample of a metric no definition reads counts as unrouted even when it is beyond the future bound. - Remote Write requests without samples are not queued. - EngineHandle::stop is test-only (doc(hidden)); a test covers the running flag clearing when the engine thread panics. - The default retention's doubling of the pane width is overflow-checked. --- README.md | 7 +++-- data_plane/src/engine.rs | 53 ++++++++++++++++++++++++++++++---- data_plane/src/lib.rs | 3 ++ data_plane/src/precompute.rs | 29 +++++++++++++++---- data_plane/tests/precompute.rs | 31 ++++++++++++++++++++ data_plane/tests/server.rs | 28 +++++++++++++----- 6 files changed, 130 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index e642564d..4747bf47 100644 --- a/README.md +++ b/README.md @@ -37,11 +37,12 @@ The backend is being rebuilt on the Planner's physical DAG: | 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 -of a stored summary's metric passes their end; later samples for them are +of a stored summary's metric, capped at the wall clock, passes their end; later samples for them are dropped. The first pane after startup is skipped because it may be partial. Samples more than `--max-future-ms` (default 5 min) ahead of the wall clock -are dropped. Remote Write gets 503 while more than `--max-queued-samples` -(default 2,000,000) samples wait for the engine, and 500 if the engine has +are dropped. Remote Write gets 503 when admitting it would leave more than +`--max-queued-samples` (default 2,000,000) samples waiting for the engine +(a request into an empty queue is always admitted), and 500 if the engine has stopped, which also makes `/-/ready` fail. `/api/v1/status/asap` reports the engine's sample and pane counters as JSON. diff --git a/data_plane/src/engine.rs b/data_plane/src/engine.rs index 865f41a2..b10d6b05 100644 --- a/data_plane/src/engine.rs +++ b/data_plane/src/engine.rs @@ -118,10 +118,7 @@ impl EngineHandle { let count = samples.len() as u64; let mut queued = self.queued_samples.load(Ordering::SeqCst); loop { - let total = queued - .checked_add(count) - .filter(|&total| total <= self.max_queued_samples) - .ok_or(IngestError::Busy)?; + let total = admit(queued, count, self.max_queued_samples)?; match self.queued_samples.compare_exchange( queued, total, @@ -153,13 +150,29 @@ impl EngineHandle { *self.stats.lock().unwrap_or_else(|e| e.into_inner()) } - /// Stops the engine thread after the commands queued before; later - /// ingestion fails with [`IngestError::Stopped`]. + /// For tests: stops the engine thread after the commands queued before. + /// Samples queued between this call and the stop are acknowledged and + /// then lost; ingestion after the stop fails with + /// [`IngestError::Stopped`]. + #[doc(hidden)] pub fn stop(&self) { let _ = self.commands.send(Command::Stop); } } +/// The queue length after admitting `count` samples to a queue of +/// `queued`: admitted while the total stays within `max_queued`, and always +/// into an empty queue, so a bound smaller than one request cannot refuse +/// that request forever. +fn admit(queued: u64, count: u64, max_queued: u64) -> Result { + let total = queued.checked_add(count).ok_or(IngestError::Busy)?; + if queued == 0 || total <= max_queued { + Ok(total) + } else { + Err(IngestError::Busy) + } +} + /// What the engine thread publishes to the HTTP side. struct Shared { queued_samples: Arc, @@ -204,3 +217,31 @@ impl Engine { tracing::warn!("engine stopped"); } } + +#[cfg(test)] +mod tests { + use super::*; + + /// A request is admitted while the queue stays within its bound, and + /// always into an empty queue; otherwise it is `Busy`. + #[test] + fn admission() { + assert_eq!(admit(0, 3, 2), Ok(3)); + assert_eq!(admit(1, 1, 2), Ok(2)); + assert_eq!(admit(1, 2, 2), Err(IngestError::Busy)); + assert_eq!(admit(u64::MAX, 1, u64::MAX), Err(IngestError::Busy)); + } + + /// The running flag clears when the engine thread panics. + #[test] + fn running_guard_clears_on_panic() { + let running = Arc::new(AtomicBool::new(true)); + let guard = RunningGuard(running.clone()); + let thread = std::thread::spawn(move || { + let _guard = guard; + panic!("engine bug"); + }); + assert!(thread.join().is_err()); + assert!(!running.load(Ordering::SeqCst)); + } +} diff --git a/data_plane/src/lib.rs b/data_plane/src/lib.rs index d501c70e..34b49bf2 100644 --- a/data_plane/src/lib.rs +++ b/data_plane/src/lib.rs @@ -124,6 +124,9 @@ async fn write(State(state): State>, body: Bytes) -> Response { })); } let count = samples.len() as u64; + if count == 0 { + return StatusCode::NO_CONTENT.into_response(); + } match state.engine.try_ingest(samples) { Ok(()) => { state.samples.fetch_add(count, Ordering::Relaxed); diff --git a/data_plane/src/precompute.rs b/data_plane/src/precompute.rs index e4708602..307947f9 100644 --- a/data_plane/src/precompute.rs +++ b/data_plane/src/precompute.rs @@ -124,6 +124,10 @@ pub struct Precompute { allowed_lateness_ms: i64, max_future_ms: i64, newest_sample_ms: Option, + /// The newest wall-clock time an ingest ran at: the watermark never + /// passes it, so a series stamped ahead of the others cannot close + /// panes early. + newest_clock_ms: Option, pub store: SummaryStore, pub stats: PrecomputeStats, } @@ -172,7 +176,7 @@ impl Precompute { .max() .unwrap_or(0); let retention_ms = max_offset - .checked_add(2 * definition.pane_ms) + .checked_add(definition.pane_ms.saturating_mul(2)) .ok_or_else(|| error("retention overflows".into()))?; builders.push(Builder { definition: definition.clone(), @@ -191,15 +195,21 @@ impl Precompute { allowed_lateness_ms: millis(options.allowed_lateness_ms), max_future_ms: millis(options.max_future_ms), newest_sample_ms: None, + newest_clock_ms: None, store: SummaryStore::new(), stats: PrecomputeStats::default(), }) } - /// The watermark: no sample older than this is expected any more. + /// 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. pub fn watermark_ms(&self) -> Option { - self.newest_sample_ms - .map(|newest| newest.saturating_sub(self.allowed_lateness_ms)) + let newest = match self.newest_clock_ms { + Some(clock) => self.newest_sample_ms?.min(clock), + None => self.newest_sample_ms?, + }; + Some(newest.saturating_sub(self.allowed_lateness_ms)) } /// [`Self::ingest_at`] the current wall-clock time. @@ -214,13 +224,22 @@ impl Precompute { /// watermark has passed. `now_ms` is the wall-clock time that bounds /// sample times from above. pub fn ingest_at(&mut self, samples: &[RawSample], now_ms: i64) { + self.newest_clock_ms = self.newest_clock_ms.max(Some(now_ms)); 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())) + { + self.stats.samples_unrouted += 1; + continue; + } if sample.timestamp_ms > latest_allowed_ms { self.stats.samples_dropped_future += 1; continue; } - let metric = sample.labels.get("__name__").map(String::as_str); let mut outcome: Option = None; for builder in &mut self.builders { if metric != Some(builder.selector.metric()) { diff --git a/data_plane/tests/precompute.rs b/data_plane/tests/precompute.rs index 58d4a4b4..553c8911 100644 --- a/data_plane/tests/precompute.rs +++ b/data_plane/tests/precompute.rs @@ -288,3 +288,34 @@ fn stale_markers_are_not_samples() { ]); assert_eq!(engine.store.get(&id, 2 * PANE).unwrap().num_rows(), 1); } + +/// One series stamped ahead of the wall clock, though within +/// `max_future_ms`, does not close panes early: the watermark is bounded by +/// the wall clock, so an on-time sample of the open pane is still accepted. +#[test] +fn series_ahead_of_the_clock_does_not_close_panes_early() { + let (mut engine, sds) = precompute_with(common::plan("example4b", false), 15_000); + let id = sds.definitions.keys().next().unwrap().clone(); + let base = PANE; + // Opens the skipped first pane (0, 10 min]. + engine.ingest_at(&[sample("latency_ms", "a", MINUTE, 1.0)], MINUTE); + let now = base + 9 * MINUTE; + engine.ingest_at(&[sample("latency_ms", "b", base + 13 * MINUTE, 1.0)], now); + assert!( + engine.store.get(&id, base + PANE).is_none(), + "(10, 20 min] still open" + ); + engine.ingest_at(&[sample("latency_ms", "a", base + 9 * MINUTE, 1.0)], now); + assert_eq!(engine.stats.samples_dropped_late, 0); + assert_eq!(engine.stats.samples_accepted, 2); +} + +/// A far-future sample of a metric no definition reads is unrouted, not +/// dropped as future. +#[test] +fn far_future_unrouted_sample_counts_as_unrouted() { + let (mut engine, _) = precompute(common::plan("example4b", false)); + engine.ingest_at(&[sample("other_metric", "a", 30 * DAY, 1.0)], 0); + assert_eq!(engine.stats.samples_unrouted, 1); + assert_eq!(engine.stats.samples_dropped_future, 0); +} diff --git a/data_plane/tests/server.rs b/data_plane/tests/server.rs index 695d22aa..1d472557 100644 --- a/data_plane/tests/server.rs +++ b/data_plane/tests/server.rs @@ -285,10 +285,10 @@ async fn get_json(app: &Router, path: &str) -> (StatusCode, serde_json::Value) { (status, serde_json::from_str(&body).unwrap_or_default()) } -/// Remote Write that would queue more samples than `max_queued_samples` gets -/// 503 and is not counted; a request that fits is accepted. +/// A request larger than `max_queued_samples` is still admitted when nothing +/// is queued, so a small queue bound cannot refuse a large request forever. #[tokio::test] -async fn refuses_remote_write_beyond_the_sample_queue() { +async fn admits_a_large_request_into_an_empty_queue() { let options = EngineOptions { max_queued_samples: 2, ..EngineOptions::default() @@ -302,10 +302,24 @@ async fn refuses_remote_write_beyond_the_sample_queue() { .unwrap(), ); let app = router(state.clone()); - assert_eq!(write(&app, 3).await, StatusCode::SERVICE_UNAVAILABLE); - assert_eq!(state.samples.load(Ordering::Relaxed), 0); - assert_eq!(write(&app, 2).await, StatusCode::NO_CONTENT); - assert_eq!(state.samples.load(Ordering::Relaxed), 2); + assert_eq!(write(&app, 3).await, StatusCode::NO_CONTENT); + assert_eq!(state.samples.load(Ordering::Relaxed), 3); +} + +/// A Remote Write request without samples is accepted and not queued. +#[tokio::test] +async fn does_not_queue_empty_requests() { + let state = data_plane("http://unused".into()); + let app = router(state.clone()); + state.engine.stop(); + for _ in 0..200 { + if !state.engine.is_running() { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + // A stopped engine refuses samples, but there is nothing to refuse. + assert_eq!(write(&app, 0).await, StatusCode::NO_CONTENT); } /// Once the engine thread has stopped, Remote Write gets 500 instead of a