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