diff --git a/Cargo.lock b/Cargo.lock index e3aa5f0c..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", @@ -679,6 +679,7 @@ dependencies = [ "axum", "clap", "control_plane", + "futures", "http-body-util", "prost", "reqwest", 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 9dcb4f79..4747bf47 100644 --- a/README.md +++ b/README.md @@ -31,18 +31,25 @@ 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 +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 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. + +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 +58,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..b10d6b05 --- /dev/null +++ b/data_plane/src/engine.rs @@ -0,0 +1,247 @@ +//! 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::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::mpsc::{channel, Receiver, Sender}; +use std::sync::{Arc, Mutex}; + +use control_plane::InstalledPlan; + +use crate::precompute::{Precompute, PrecomputeOptions, PrecomputeStats, RawSample}; +use crate::sds::Sds; + +/// 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, + } + } +} + +enum Command { + Ingest(Vec), + Stop, +} + +/// 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: 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`, 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, 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 || { + 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)); + } + } + })?; + 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. `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 mut queued = self.queued_samples.load(Ordering::SeqCst); + loop { + let total = admit(queued, count, self.max_queued_samples)?; + 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); + 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) + } + + /// 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()) + } + + /// 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, + stats: Arc>, +} + +struct Engine { + precompute: Precompute, +} + +impl Engine { + 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; + 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" + ); + let engine = Self { + precompute: Precompute::new(dag, &sds, options)?, + }; + Ok((engine, plan.queries)) + } + + fn serve(mut self, commands: Receiver, shared: Shared) { + for command in commands { + match command { + 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"); + } +} + +#[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 e770ff91..34b49bf2 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,8 @@ use axum::http::{header, HeaderMap, HeaderValue, Method, StatusCode}; use axum::response::{IntoResponse, Response}; use axum::routing::{get, post}; use axum::Router; +use engine::{EngineHandle, EngineOptions, IngestError}; +use precompute::RawSample; /// Response header naming what answered a query. pub const EXECUTION_HEADER: &str = "x-asap-execution"; @@ -28,33 +32,62 @@ 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 { + /// 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: 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 (engine, queries) = EngineHandle::spawn(plan_json, options)?; + Ok(Self { queries, + engine, 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 { @@ -65,21 +98,85 @@ 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) } 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(), + ); + // 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; + if count == 0 { + return StatusCode::NO_CONTENT.into_response(); + } + match state.engine.try_ingest(samples) { + Ok(()) => { + state.samples.fetch_add(count, Ordering::Relaxed); + StatusCode::NO_CONTENT.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 c22bb7b0..88186e41 100644 --- a/data_plane/src/main.rs +++ b/data_plane/src/main.rs @@ -3,7 +3,8 @@ use std::sync::Arc; use std::time::Duration; use clap::Parser; -use control_plane::InstalledPlan; +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. @@ -12,9 +13,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 +21,15 @@ 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 = 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] @@ -30,17 +38,22 @@ 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::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)), ); - 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..307947f9 --- /dev/null +++ b/data_plane/src/precompute.rs @@ -0,0 +1,419 @@ +//! 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. +//! +//! 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; + +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, 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 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 { + 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, + /// 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, + /// 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, +} + +impl Precompute { + /// Compiles one build per definition of `sds`, a description of `dag`. + pub fn new( + dag: PhysicalASAPDAG, + sds: &Sds, + 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: 0, + 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); + let retention_ms = max_offset + .checked_add(definition.pane_ms.saturating_mul(2)) + .ok_or_else(|| error("retention overflows".into()))?; + builders.push(Builder { + definition: definition.clone(), + selector, + raw_schema, + build, + 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: 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. It is + /// the newest routed sample, capped at the wall clock, minus the allowed + /// lateness. + pub fn watermark_ms(&self) -> Option { + 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. + 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) { + 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 mut outcome: Option = None; + for builder in &mut self.builders { + if metric != Some(builder.selector.metric()) { + continue; + } + 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)); + } + 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.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; + 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. + while end <= watermark { + let samples = builder.open.remove(&end).unwrap_or_default(); + match build_pane(builder, end, &samples) { + Ok(state) => { + self.store.insert(id, end, state); + self.stats.panes_built += 1; + } + Err(error) => { + tracing::error!(definition = ?id, end, %error, "pane build failed"); + self.stats.panes_failed += 1; + } + } + builder.built_through = Some(end); + match end.checked_add(pane_ms) { + Some(next) => end = next, + None => break, + } + } + self.store.evict_through(id, expired_through); + } + } +} + +/// 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, +/// 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 scope = Scope::Ingestion { + 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 { + 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_root(&build, builder.definition.builder, scope)?; + let schema = Arc::new(builder.definition.output_schema.clone()); + Ok(Batch::try_new(schema, rows)?) +} + +/// Executes `root` of `dag` to completion and returns its rows. +pub fn run_root( + dag: &asap_executor::plan::PhysicalDAG<'_, Batch, SchemaRef>, + root: u64, + scope: Scope, +) -> Result>, asap_executor::Error> { + block_on(async { + let context = RunContext::new(scope, Limits::default())?; + 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()); + } + 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..553c8911 --- /dev/null +++ b/data_plane/tests/precompute.rs @@ -0,0 +1,321 @@ +mod common; + +use std::collections::BTreeMap; +use std::sync::Arc; + +use control_plane::{plan_workload, InstalledPlan}; +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(); + 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 { + 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.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. +#[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.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.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" +/// 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), 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 +/// 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); +} + +/// 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); +} + +/// 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/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..1d472557 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; @@ -7,8 +9,9 @@ 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, DEFAULT_FORWARD_TIMEOUT, EXECUTION_HEADER}; +use data_plane::{router, DataPlane, EXECUTION_HEADER}; use http_body_util::BodyExt; use tower::ServiceExt; @@ -38,6 +41,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 +61,7 @@ async fn send(app: &Router, request: Request) -> (StatusCode, Option 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()) +} + +/// 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 admits_a_large_request_into_an_empty_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::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 +/// 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}"); +}