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