diff --git a/README.md b/README.md index 2713b32..0f8431b 100644 --- a/README.md +++ b/README.md @@ -84,7 +84,12 @@ All of this sits on the host-neutral runtime in the same crate ## Build & test For a presentation-ready walkthrough of the real YAML pipeline integration, -see the [live OTAP pipeline demo guide](./docs/demo-guide.md). +see the [live OTAP pipeline demo guide](./docs/demo-guide.md). The guide also +includes a [Prometheus + Grafana setup](./docs/demo-guide.md#write-kll-estimates-to-prometheus-and-view-them-in-grafana) +that writes the KLL p50/p99 estimates to Prometheus and displays them in a +provisioned Grafana dashboard. For a continuous KLL vs exact comparison with +every processor in its own pinned process, see the +[multi-process streaming KLL demo](./docs/kll-windows.md). ```sh cd asap-precompute-rs diff --git a/asap-precompute-rs/Cargo.lock b/asap-precompute-rs/Cargo.lock index 089542e..aec0f19 100644 --- a/asap-precompute-rs/Cargo.lock +++ b/asap-precompute-rs/Cargo.lock @@ -372,6 +372,7 @@ dependencies = [ "otel-arrow-dfe-telemetry", "proptest", "prost 0.13.5", + "prost 0.14.4", "serde", "serde_json", "thiserror 1.0.69", diff --git a/asap-precompute-rs/Cargo.toml b/asap-precompute-rs/Cargo.toml index ea28761..71e7512 100644 --- a/asap-precompute-rs/Cargo.toml +++ b/asap-precompute-rs/Cargo.toml @@ -10,6 +10,8 @@ license = "MIT" asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib", rev = "010457de46ba9ec5574a53476bbe8b4bdb02b965" } serde = { version = "1", features = ["derive"] } prost = "0.13" +# OTAP's generated protobuf types implement the 0.14 Message trait. +prost-otap = { package = "prost", version = "0.14", optional = true } thiserror = "1" # Apache Arrow is only pulled in by the OTAP codec. Consumers on the @@ -73,6 +75,7 @@ otap = ["dep:arrow-array", "dep:arrow-schema"] # folded into `otap`. otap-engine = [ "otap", + "dep:prost-otap", "dep:otel-arrow-dfe-engine", "dep:otel-arrow-dfe-otap", "dep:otel-arrow-dfe-config", diff --git a/asap-precompute-rs/plugins/asap_sketches/sample.toml b/asap-precompute-rs/plugins/asap_sketches/sample.toml index 2d8d40c..03fc2d9 100644 --- a/asap-precompute-rs/plugins/asap_sketches/sample.toml +++ b/asap-precompute-rs/plugins/asap_sketches/sample.toml @@ -23,6 +23,9 @@ plugin = "asap_sketches" ## Window size. Rust duration string, parsed via humantime; ## passed verbatim into `PrecomputeConfig::window`. window_size = "10s" + # window_slide = "5s" # KLL rolling lookback of 10s, emit every 5s + # Set transmit_sketch=false for overlapping sliding windows. Equal size + # and slide can transmit full disjoint panes for downstream merging. ## Output metric name stamped onto every emitted envelope. output_metric_name = "http_request_duration_ms" diff --git a/asap-precompute-rs/src/bin/asap_otap_demo.rs b/asap-precompute-rs/src/bin/asap_otap_demo.rs index f996c8d..fed0671 100644 --- a/asap-precompute-rs/src/bin/asap_otap_demo.rs +++ b/asap-precompute-rs/src/bin/asap_otap_demo.rs @@ -1,6 +1,9 @@ //! Multi-process OTAP sketch create -> merge -> estimate demonstration. //! Each ASAP processor runs in its own OS process and OTAP RuntimePipeline. +#[path = "demo/prometheus_export.rs"] +mod prometheus_export; + use asap_precompute_rs::envelope::{Encoding, SketchEnvelope, SketchType}; use asap_precompute_rs::observation::KeyValue; use asap_precompute_rs::otap::codec::{decode_pdata_to_observations, encode_envelopes_to_pdata}; @@ -756,6 +759,7 @@ struct DemoManifest { } struct ParentOptions { + prometheus_otlp_endpoint: Option, output_dir: Option, result_manifest: Option, points_per_source: u64, @@ -766,6 +770,7 @@ struct ParentOptions { fn parent_options(args: &[std::ffi::OsString]) -> Result { let mut options = ParentOptions { + prometheus_otlp_endpoint: None, output_dir: None, result_manifest: None, points_per_source: 100, @@ -780,6 +785,9 @@ fn parent_options(args: &[std::ffi::OsString]) -> Result .get(index + 1) .ok_or_else(|| format!("missing value for {flag}"))?; match flag.as_ref() { + "--prometheus-otlp-endpoint" => { + options.prometheus_otlp_endpoint = Some(value.to_string_lossy().into_owned()); + } "--output-dir" => options.output_dir = Some(PathBuf::from(value)), "--result-manifest" => options.result_manifest = Some(PathBuf::from(value)), "--points-per-source" => { @@ -812,6 +820,9 @@ fn parent_options(args: &[std::ffi::OsString]) -> Result } index += 2; } + if options.prometheus_otlp_endpoint.is_some() && options.scenario != "kll" { + return Err("--prometheus-otlp-endpoint requires --scenario kll".into()); + } Ok(options) } @@ -954,6 +965,10 @@ fn run_parent(options: ParentOptions) -> Result<(), String> { } } let validation_elapsed_nanoseconds = validation_start.elapsed().as_nanos(); + if let Some(endpoint) = options.prometheus_otlp_endpoint { + prometheus_export::export(&out, &endpoint)?; + println!("KLL estimates exported to Prometheus"); + } if let Some(path) = options.result_manifest { let manifest = DemoManifest { parent_pid: std::process::id(), diff --git a/asap-precompute-rs/src/bin/asap_stream_bench.rs b/asap-precompute-rs/src/bin/asap_stream_bench.rs index ad25e9f..892eda4 100644 --- a/asap-precompute-rs/src/bin/asap_stream_bench.rs +++ b/asap-precompute-rs/src/bin/asap_stream_bench.rs @@ -66,6 +66,7 @@ struct Options { endpoint: String, generator_workers: usize, signals_per_second: usize, + rolling_windows: usize, } impl Options { fn read() -> Result { @@ -86,6 +87,7 @@ impl Options { endpoint: get("ASAP_ENDPOINT", "http://merge:4318"), generator_workers: number("ASAP_GENERATOR_WORKERS", "1")?, signals_per_second: number("ASAP_SIGNALS_PER_SECOND", "100000")?, + rolling_windows: number("ASAP_ROLLING_WINDOWS", "1")?, }; if !["raw", "exact", "kll"].contains(&o.scenario.as_str()) || !["generator", "branch", "merge", "estimate", "backend"].contains(&o.role.as_str()) @@ -95,6 +97,7 @@ impl Options { || o.source > 1 || o.generator_workers == 0 || o.signals_per_second == 0 + || o.rolling_windows == 0 { return Err("invalid role/scenario/window/batch/source configuration".into()); } @@ -113,6 +116,7 @@ struct Stats { output_send_wall_ns: u64, output_messages: u64, window_latency_ms: BTreeMap, + latest_results: Vec, } struct Shared { stats: Mutex, @@ -280,7 +284,13 @@ fn create_processor( let o = OPTIONS.get().unwrap(); Ok(ProcessorWrapper::local( StreamProcessor { - processor: model::Processor::new(&o.scenario, &o.role, o.points, o.batch), + processor: model::Processor::new( + &o.scenario, + &o.role, + o.points, + o.batch, + o.rolling_windows, + ), normalizer: (o.role == "branch") .then(|| model::InputNormalizer::new(o.source, o.points, o.batch)), }, @@ -317,6 +327,7 @@ impl exporter::Exporter for ValidatingBackend { stats.received_signals += count; if let Some((window, start)) = completed { assert_eq!(window, stats.completed_windows, "backend window gap"); + stats.latest_results.clone_from(&self.0.latest); stats.completed_windows += 1; let elapsed_ms = now_ns().saturating_sub(start) / 1_000_000; *stats.window_latency_ms.entry(elapsed_ms).or_default() += 1; @@ -340,7 +351,11 @@ fn create_backend( ) -> Result, otel_arrow_dfe_config::error::Error> { let o = OPTIONS.get().unwrap(); Ok(ExporterWrapper::local( - ValidatingBackend(model::Backend::new(&o.scenario, o.points)), + ValidatingBackend(model::Backend::new( + &o.scenario, + o.points, + o.rolling_windows, + )), node, config, runtime, diff --git a/asap-precompute-rs/src/bin/demo/prometheus_export.rs b/asap-precompute-rs/src/bin/demo/prometheus_export.rs new file mode 100644 index 0000000..567b99d --- /dev/null +++ b/asap-precompute-rs/src/bin/demo/prometheus_export.rs @@ -0,0 +1,202 @@ +//! Optional demo-only OTLP/HTTP export. Curl supplies HTTP/TLS without adding +//! an HTTP client dependency to the collector runtime. +use otel_arrow_dfe_pdata::proto::opentelemetry::collector::metrics::v1::{ + ExportMetricsServiceRequest, ExportMetricsServiceResponse, +}; +use otel_arrow_dfe_pdata::proto::opentelemetry::metrics::v1::metric::Data; +use prost_otap::Message; +use std::io::Write; +use std::path::Path; +use std::process::{Command, Stdio}; +use std::time::{SystemTime, UNIX_EPOCH}; + +fn prepare_gauges(bytes: &[u8], now: Option) -> Result, String> { + let mut request = ExportMetricsServiceRequest::decode(bytes).map_err(|e| e.to_string())?; + let mut points = 0; + for resource in &mut request.resource_metrics { + for scope in &mut resource.scope_metrics { + let scope_attributes = scope + .scope + .as_ref() + .map(|s| s.attributes.as_slice()) + .unwrap_or_default(); + for metric in &mut scope.metrics { + let Some(Data::Gauge(gauge)) = &mut metric.data else { + return Err("Prometheus demo export expects only estimated gauges".into()); + }; + for point in &mut gauge.data_points { + // The OTAP codec stores series labels on the scope. Prometheus + // identifies series by point attributes, so materialize them here. + for attribute in scope_attributes { + if !point.attributes.iter().any(|a| a.key == attribute.key) { + point.attributes.push(attribute.clone()); + } + } + // Synthetic windows stay in the trace files. Export is a current + // snapshot of the completed estimate, not historical telemetry. + if let Some(now) = now { + point.start_time_unix_nano = 0; + point.time_unix_nano = now; + } + // Window boundaries are timestamps/metadata, not series + // dimensions: retaining them as labels creates a new + // Prometheus series for every emitted rolling window. + point.attributes.retain(|attribute| { + !matches!( + attribute.key.as_str(), + "sketch.window_start_ms" | "sketch.window_end_ms" + ) + }); + points += 1; + } + } + } + } + if points == 0 { + return Err("no estimated gauge points to export".into()); + } + Ok(request.encode_to_vec()) +} + +pub fn export(path: &Path, endpoint: &str) -> Result<(), String> { + if !(endpoint.starts_with("http://") || endpoint.starts_with("https://")) { + return Err("Prometheus OTLP endpoint must be an http:// or https:// URL".into()); + } + let now = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|e| e.to_string())? + .as_nanos() as u64; + export_bytes( + &std::fs::read(path).map_err(|e| e.to_string())?, + endpoint, + Some(now), + ) +} + +/// Export estimated OTLP gauges. `None` preserves window-end timestamps for +/// the streaming demo; `Some` publishes a current snapshot for the finite demo. +pub fn export_bytes(bytes: &[u8], endpoint: &str, now: Option) -> Result<(), String> { + if !(endpoint.starts_with("http://") || endpoint.starts_with("https://")) { + return Err("Prometheus OTLP endpoint must be an http:// or https:// URL".into()); + } + let body = prepare_gauges(bytes, now)?; + let mut child = Command::new("curl") + .args([ + "--silent", + "--show-error", + "--fail-with-body", + "--max-time", + "30", + "--header", + "Content-Type: application/x-protobuf", + "--header", + "Accept: application/x-protobuf", + "--data-binary", + "@-", + "--url", + endpoint, + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .map_err(|e| format!("start curl for Prometheus OTLP export: {e}"))?; + let write_result = child.stdin.take().expect("piped stdin").write_all(&body); + let output = child.wait_with_output().map_err(|e| e.to_string())?; + if !output.status.success() { + return Err(format!( + "Prometheus OTLP export failed: {} {}", + String::from_utf8_lossy(&output.stderr), + String::from_utf8_lossy(&output.stdout) + )); + } + write_result.map_err(|e| format!("write OTLP request: {e}"))?; + let response = ExportMetricsServiceResponse::decode(output.stdout.as_slice()) + .map_err(|e| format!("invalid Prometheus OTLP response: {e}"))?; + if let Some(partial) = response.partial_success { + if partial.rejected_data_points != 0 || !partial.error_message.is_empty() { + return Err(format!( + "Prometheus OTLP partial success: {} rejected: {}", + partial.rejected_data_points, partial.error_message + )); + } + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use otel_arrow_dfe_pdata::proto::opentelemetry::common::v1::{ + any_value::Value, AnyValue, InstrumentationScope, KeyValue, + }; + use otel_arrow_dfe_pdata::proto::opentelemetry::metrics::v1::{ + Gauge, Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, + }; + + #[test] + fn export_uses_current_timestamp_and_preserves_estimate() { + let request = ExportMetricsServiceRequest { + resource_metrics: vec![ResourceMetrics { + scope_metrics: vec![ScopeMetrics { + scope: Some(InstrumentationScope { attributes: vec![KeyValue { key: "quantile".into(), value: Some(AnyValue { value: Some(Value::StringValue("0.99".into())) }) }], ..Default::default() }), + metrics: vec![Metric { + name: "request.duration.estimate".into(), + data: Some(Data::Gauge(Gauge { data_points: vec![NumberDataPoint { + attributes: vec![KeyValue { key: "sketch.window_end_ms".into(), value: Some(AnyValue { value: Some(Value::StringValue("2000".into())) }) }], + time_unix_nano: 2_000_000_000, + value: Some(otel_arrow_dfe_pdata::proto::opentelemetry::metrics::v1::number_data_point::Value::AsDouble(198.0)), + ..Default::default() + }] })), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }], + }; + let updated = ExportMetricsServiceRequest::decode( + prepare_gauges(&request.encode_to_vec(), Some(123456)) + .unwrap() + .as_slice(), + ) + .unwrap(); + let metric = &updated.resource_metrics[0].scope_metrics[0].metrics[0]; + let streaming = ExportMetricsServiceRequest::decode( + prepare_gauges(&request.encode_to_vec(), None) + .unwrap() + .as_slice(), + ) + .unwrap(); + let Some(Data::Gauge(streaming_gauge)) = + &streaming.resource_metrics[0].scope_metrics[0].metrics[0].data + else { + panic!("expected streaming gauge") + }; + assert_eq!(streaming_gauge.data_points[0].time_unix_nano, 2_000_000_000); + assert_eq!(metric.name, "request.duration.estimate"); + let Some(Data::Gauge(gauge)) = &metric.data else { + panic!("expected gauge") + }; + assert_eq!(gauge.data_points[0].time_unix_nano, 123456); + let Some(Data::Gauge(original)) = + &request.resource_metrics[0].scope_metrics[0].metrics[0].data + else { + panic!("expected gauge") + }; + assert_eq!(gauge.data_points[0].value, original.data_points[0].value); + assert_eq!( + gauge.data_points[0].attributes, + request.resource_metrics[0].scope_metrics[0] + .scope + .as_ref() + .unwrap() + .attributes + ); + assert!(prepare_gauges( + &ExportMetricsServiceRequest::default().encode_to_vec(), + Some(1) + ) + .is_err()); + } +} diff --git a/asap-precompute-rs/src/bin/stream_bench/model.rs b/asap-precompute-rs/src/bin/stream_bench/model.rs index 4219619..0e03090 100644 --- a/asap-precompute-rs/src/bin/stream_bench/model.rs +++ b/asap-precompute-rs/src/bin/stream_bench/model.rs @@ -8,7 +8,8 @@ use asap_precompute_rs::{ sketches::KLLWrapper, }; use otel_arrow_dfe_otap::pdata::OtapPdata; -use std::collections::BTreeMap; +use serde::Serialize; +use std::collections::{BTreeMap, VecDeque}; // Independent official generators can drift under high load. Keep the skew // explicitly bounded while allowing short scheduler/exporter bursts. @@ -111,16 +112,25 @@ fn decode_sketch(o: &Observation, count: usize) -> Result { .map_err(|e| e.to_string())?; Ok(sketch) } -fn quantiles_pdata(values: [f64; 2], meta: Meta) -> Result { +const QUANTILES: [&str; 2] = ["0.5", "0.99"]; +/// Tumbling quantiles, plus rolling quantiles when the estimate retains windows. +fn quantiles_pdata( + tumbling: [f64; 2], + rolling: Option<[f64; 2]>, + meta: Meta, +) -> Result { let _cpu = profile::scope(Category::PdataCodec); - let envelopes = ["0.5", "0.99"] + let envelopes = [("tumbling", Some(tumbling)), ("rolling", rolling)] .into_iter() - .zip(values) - .map(|(q, value)| { - let mut env = meta.envelope(value); - env.metric_name = "http.server.request.duration.estimate".into(); - env.labels.push(KeyValue::new("quantile", q)); - env + .filter_map(|(kind, values)| Some((kind, values?))) + .flat_map(|(kind, values)| { + QUANTILES.into_iter().zip(values).map(move |(q, value)| { + let mut env = meta.envelope(value); + env.metric_name = "http.server.request.duration.estimate".into(); + env.labels.push(KeyValue::new("quantile", q)); + env.labels.push(KeyValue::new("window_kind", kind)); + env + }) }) .collect::>(); encode_envelopes_to_pdata(&envelopes).map_err(|e| e.to_string()) @@ -128,10 +138,9 @@ fn quantiles_pdata(values: [f64; 2], meta: Meta) -> Result { pub fn exact(values: &[f64]) -> [f64; 2] { [0.5, 0.99].map(|q| values[((values.len() - 1) as f64 * q).round() as usize]) } -pub fn reference(points: usize) -> [f64; 2] { - let mut values = (1..=points as u64 * 2).map(value).collect::>(); - values.sort_by(f64::total_cmp); - exact(&values) +/// Exact quantiles of `copies` concatenated copies of one sorted window. +fn exact_repeated(sorted: &[f64], copies: usize) -> [f64; 2] { + [0.5, 0.99].map(|q| sorted[((sorted.len() * copies - 1) as f64 * q).round() as usize / copies]) } pub struct InputNormalizer { @@ -209,9 +218,13 @@ pub struct Processor { pub batch: usize, windows: BTreeMap, next: [u64; 3], + /// Completed windows merged into each rolling estimate; 1 disables rolling. + rolling: usize, + kll_history: VecDeque, + exact_history: VecDeque>, } impl Processor { - pub fn new(scenario: &str, role: &str, points: usize, batch: usize) -> Self { + pub fn new(scenario: &str, role: &str, points: usize, batch: usize, rolling: usize) -> Self { Self { scenario: scenario.into(), role: role.into(), @@ -219,7 +232,43 @@ impl Processor { batch, windows: BTreeMap::new(), next: [0; 3], + rolling, + kll_history: VecDeque::new(), + exact_history: VecDeque::new(), + } + } + /// Merge the last `rolling` window sketches; quantiles of their union. + fn rolling_kll(&mut self, sketch: KLLWrapper) -> Result, String> { + if self.rolling <= 1 { + return Ok(None); + } + self.kll_history.push_back(sketch); + if self.kll_history.len() > self.rolling { + self.kll_history.pop_front(); + } + let mut merged = new_sketch(5); + for sketch in &self.kll_history { + merged.merge(sketch).map_err(|e| e.to_string())?; + } + Ok(Some([merged.quantile(0.5), merged.quantile(0.99)])) + } + /// Exact rolling quantiles must retain and re-sort every value in the range. + fn rolling_exact(&mut self, values: Vec) -> Option<[f64; 2]> { + if self.rolling <= 1 { + return None; + } + self.exact_history.push_back(values); + if self.exact_history.len() > self.rolling { + self.exact_history.pop_front(); } + let mut all = self + .exact_history + .iter() + .flatten() + .copied() + .collect::>(); + all.sort_unstable_by(f64::total_cmp); + Some(exact(&all)) } pub fn process(&mut self, pdata: OtapPdata) -> Result, String> { let _cpu = profile::scope(Category::ProcessorBookkeeping); @@ -308,13 +357,12 @@ impl Processor { }; if self.scenario == "kll" { return Ok(vec![if estimate { - quantiles_pdata( - { - let _cpu = profile::scope(Category::Computation); - [w.sketch.quantile(0.5), w.sketch.quantile(0.99)] - }, - out_meta, - )? + let (tumbling, rolling) = { + let _cpu = profile::scope(Category::Computation); + let tumbling = [w.sketch.quantile(0.5), w.sketch.quantile(0.99)]; + (tumbling, self.rolling_kll(w.sketch)?) + }; + quantiles_pdata(tumbling, rolling, out_meta)? } else { sketch_pdata( &w.sketch, @@ -350,11 +398,12 @@ impl Processor { } }; if estimate { - let quantiles = { + let (tumbling, rolling) = { let _cpu = profile::scope(Category::Computation); - exact(&values) + let tumbling = exact(&values); + (tumbling, self.rolling_exact(values)) }; - return Ok(vec![quantiles_pdata(quantiles, out_meta)?]); + return Ok(vec![quantiles_pdata(tumbling, rolling, out_meta)?]); } values .chunks(self.batch) @@ -445,23 +494,39 @@ impl Processor { } } +/// One validated result and its exact reference, reported for dashboards. +#[derive(Serialize, Clone, Debug, Default)] +pub struct ResultPoint { + pub window_kind: String, + pub quantile: String, + pub value: f64, + pub reference: f64, +} pub struct Backend { scenario: String, points: usize, - expected: [f64; 2], + rolling: usize, + /// The sorted two-source corpus; every window replays it. + sorted: Vec, windows: BTreeMap, next: [u64; 2], next_quantile: u64, + /// Results of the most recently completed window. + pub latest: Vec, } impl Backend { - pub fn new(scenario: &str, points: usize) -> Self { + pub fn new(scenario: &str, points: usize, rolling: usize) -> Self { + let mut sorted = (1..=points as u64 * 2).map(value).collect::>(); + sorted.sort_by(f64::total_cmp); Self { scenario: scenario.into(), points, - expected: reference(points), + rolling, + sorted, windows: BTreeMap::new(), next: [0; 2], next_quantile: 0, + latest: Vec::new(), } } /// Return completed input signals and optional (window id, start timestamp). @@ -509,36 +574,56 @@ impl Backend { } return Ok((obs.len() as u64, completed)); } - if meta.window != self.next_quantile || meta.source != 2 || obs.len() != 2 { + let mut kinds = vec![("tumbling", exact(&self.sorted))]; + if self.rolling > 1 { + let copies = (meta.window as usize + 1).min(self.rolling); + kinds.push(("rolling", exact_repeated(&self.sorted, copies))); + } + if meta.window != self.next_quantile || meta.source != 2 || obs.len() != kinds.len() * 2 { return Err("quantile window missing/duplicated".into()); } - for (q, expected) in ["0.5", "0.99"].into_iter().zip(self.expected) { - let matching = obs + let label = |o: &Observation, key: &str| { + o.labels .iter() - .filter(|o| { - o.labels - .iter() - .any(|kv| kv.key == "quantile" && kv.value == q) - }) - .collect::>(); - if matching.len() != 1 { - return Err("missing/duplicate quantile".into()); - } - let o = matching[0]; - let other = Meta::read(o)?; - let tolerance = if self.scenario == "exact" { - 0.0 - } else { - (expected.abs() * 0.05).max(0.001) - }; - if other.window != meta.window - || other.source != 2 - || !o.value.float.is_finite() - || (o.value.float - expected).abs() > tolerance - { - return Err(format!("{q}: {} != {expected}", o.value.float)); + .find(|kv| kv.key == key) + .map(|kv| kv.value.clone()) + }; + let mut latest = Vec::with_capacity(obs.len()); + for (kind, expected) in kinds { + for (q, expected) in QUANTILES.into_iter().zip(expected) { + let matching = obs + .iter() + .filter(|o| { + label(o, "quantile").as_deref() == Some(q) + && label(o, "window_kind").as_deref() == Some(kind) + }) + .collect::>(); + if matching.len() != 1 { + return Err("missing/duplicate quantile".into()); + } + let o = matching[0]; + let other = Meta::read(o)?; + let tolerance = if self.scenario == "exact" { + 0.0 + } else { + (expected.abs() * 0.05).max(0.001) + }; + if other.window != meta.window + || other.source != 2 + || !o.value.float.is_finite() + || (o.value.float - expected).abs() > tolerance + { + return Err(format!("{kind} {q}: {} != {expected}", o.value.float)); + } + latest.push(ResultPoint { + window_kind: kind.into(), + quantile: q.into(), + value: o.value.float, + reference: expected, + }); } } + self.latest = latest; self.next_quantile += 1; Ok(( (self.points * 2) as u64, @@ -550,15 +635,15 @@ impl Backend { #[cfg(test)] mod tests { use super::*; - fn exercise(scenario: &str) { + fn exercise(scenario: &str, rolling: usize) { let (n, batch) = (1024, 127); let mut branches = [ - Processor::new(scenario, "branch", n, batch), - Processor::new(scenario, "branch", n, batch), + Processor::new(scenario, "branch", n, batch, rolling), + Processor::new(scenario, "branch", n, batch, rolling), ]; - let mut merge = Processor::new(scenario, "merge", n, batch); - let mut estimate = Processor::new(scenario, "estimate", n, batch); - let mut backend = Backend::new(scenario, n); + let mut merge = Processor::new(scenario, "merge", n, batch, rolling); + let mut estimate = Processor::new(scenario, "estimate", n, batch, rolling); + let mut backend = Backend::new(scenario, n, rolling); let (mut received, mut completed) = (0, 0); for window in 0..3 { for offset in (0..n).step_by(batch) { @@ -588,11 +673,25 @@ mod tests { } assert_eq!(received, 6 * n as u64); assert_eq!(completed, 3); + if scenario != "raw" { + let rolling_points = backend + .latest + .iter() + .filter(|p| p.window_kind == "rolling") + .count(); + assert_eq!(rolling_points, if rolling > 1 { 2 } else { 0 }); + } } #[test] fn streaming_windows_survive_batch_boundaries() { for scenario in ["raw", "exact", "kll"] { - exercise(scenario); + exercise(scenario, 1); + } + } + #[test] + fn rolling_estimates_merge_retained_windows() { + for scenario in ["exact", "kll"] { + exercise(scenario, 2); } } #[test] @@ -604,10 +703,10 @@ mod tests { started_ns: 1, }; let input = || values_pdata(&[value(1)], meta).unwrap(); - let mut backend = Backend::new("raw", 10); + let mut backend = Backend::new("raw", 10, 1); backend.receive(input()).unwrap(); assert!(backend.receive(input()).is_err()); - let mut processor = Processor::new("kll", "branch", 10, 2); + let mut processor = Processor::new("kll", "branch", 10, 2, 1); assert!(processor .process(values_pdata(&[value(1)], Meta { offset: 1, ..meta }).unwrap()) .is_err()); diff --git a/asap-precompute-rs/src/config.rs b/asap-precompute-rs/src/config.rs index 3041825..ba8e032 100644 --- a/asap-precompute-rs/src/config.rs +++ b/asap-precompute-rs/src/config.rs @@ -20,8 +20,8 @@ pub type AggId = u64; /// Picks the windowing strategy. /// -/// Only [`AggregationMode::Tumbling`] is implemented; -/// [`AggregationMode::Sliding`] is deferred (see [`crate::window`]). +/// Tumbling uses the general runtime; sliding KLL uses +/// [`crate::kll_windows::KllRollingPrecompute`]. #[derive( Copy, Clone, Debug, Default, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize, )] @@ -33,7 +33,7 @@ pub enum AggregationMode { Tumbling, /// Rotates every [`WindowSpec::slide`]; observations land in /// every window whose `[start, end)` range covers their - /// timestamp. Only Tumbling is implemented today. + /// timestamp. KLL implements this by merging disjoint retained panes. Sliding, /// Processes one batch end-to-end with no windowing (`mode: batch`). Batch, diff --git a/asap-precompute-rs/src/kll_windows.rs b/asap-precompute-rs/src/kll_windows.rs new file mode 100644 index 0000000..6047f8f --- /dev/null +++ b/asap-precompute-rs/src/kll_windows.rs @@ -0,0 +1,512 @@ +//! Retained, mergeable KLL panes and aligned sliding-window estimates. +//! +//! A pane is a half-open tumbling interval. Range queries merge complete +//! sketches rather than quantiles or raw observations. Exact time selection +//! requires pane-aligned bounds; the quantile itself has KLL rank error. +use std::collections::{BTreeMap, HashMap, HashSet}; +use std::sync::Mutex; +use std::time::Duration; + +use crate::config::{AggregationMode, PrecomputeConfig, PrecomputeConfigSet}; +use crate::envelope::{Encoding, SketchEnvelope, SketchType}; +use crate::matchers::series_attrs; +use crate::observation::{KeyValue, Observation, ObservationValueKind}; +use crate::precompute::{Precompute, PrecomputeError, Sketch, StatsSnapshot}; +use crate::sketches::KLLWrapper; + +fn invalid(message: &str) -> PrecomputeError { + PrecomputeError::Other(format!("KLL windows: {message}")) +} + +struct PaneSeries { + sketch: KLLWrapper, + resource_labels: Vec, + labels: Vec, + count: u64, +} + +/// In-memory KLL tumbling panes, retained for aligned time-range queries. +/// +/// Each observation is inserted once. Closed panes can also be merged from +/// full ASAPv1 envelopes produced upstream. Delivery must be exactly once: +/// sending the same full pane twice would count its observations twice. +/// Empty panes consume no storage; advancing time expires old panes. +pub struct KllWindowStore { + config: PrecomputeConfig, + pane_ms: u64, + retention_ms: u64, + closed_until: u64, + panes: BTreeMap>, + k: i32, + seed: Option, +} + +impl KllWindowStore { + /// Create a store. Retention must be a positive multiple of the pane size. + /// Only KLL/ASAPv1 full sketches are accepted; no delta replay or partial + /// pane splitting is supported. + pub fn new( + config: PrecomputeConfig, + pane: Duration, + retention: Duration, + ) -> Result { + let pane_ms = + u64::try_from(pane.as_millis()).map_err(|_| invalid("pane duration overflow"))?; + let retention_ms = u64::try_from(retention.as_millis()) + .map_err(|_| invalid("retention duration overflow"))?; + if !pane.as_nanos().is_multiple_of(1_000_000) + || !retention.as_nanos().is_multiple_of(1_000_000) + || pane_ms == 0 + || retention_ms < pane_ms + || !retention_ms.is_multiple_of(pane_ms) + { + return Err(invalid( + "retention must be a positive multiple of a nonzero millisecond pane", + )); + } + if config.sketch_type != SketchType::KLLSketch + || config.encoding != Encoding::Msgpack + || config.delta_transmission + { + return Err(invalid("requires KLL with full ASAPv1/Msgpack encoding")); + } + let k = config.sketch_params.get("k").copied().unwrap_or(200.0); + if !k.is_finite() || k < 8.0 || k > i32::MAX as f64 || k.fract() != 0.0 { + return Err(invalid("k must be an integer between 8 and i32::MAX")); + } + if config + .quantiles + .iter() + .any(|q| !q.is_finite() || !(0.0..=1.0).contains(q)) + { + return Err(invalid("quantiles must be finite values in [0, 1]")); + } + let seed = config + .sketch_params + .get("seed") + .copied() + .filter(|value| *value != 0.0) + .map(f64::to_bits); + Ok(Self { + config, + pane_ms, + retention_ms, + closed_until: 0, + panes: BTreeMap::new(), + k: k as i32, + seed, + }) + } + + /// Mark complete panes through `now_ms` and expire panes older than retention. + /// The watermark never moves backward, including after a clock adjustment. + pub fn advance(&mut self, now_ms: u64) { + let boundary = now_ms / self.pane_ms * self.pane_ms; + if boundary <= self.closed_until { + return; + } + self.closed_until = boundary; + let cutoff = self.closed_until.saturating_sub(self.retention_ms); + self.panes = self.panes.split_off(&cutoff); + } + + /// Number of nonempty panes currently retained (including the active pane). + pub fn retained_panes(&self) -> usize { + self.panes.len() + } + + /// Number of distinct series across all retained panes. + pub fn retained_series(&self) -> usize { + self.panes + .values() + .flat_map(|pane| pane.keys()) + .collect::>() + .len() + } + + /// Insert a finite scalar into its event-time pane. Once the watermark has + /// closed a pane, scalar observations for it are rejected as late. + pub fn observe(&mut self, observation: &Observation) -> Result<(), PrecomputeError> { + if !self.config.matches(observation) { + return Ok(()); + } + if observation.value.kind != ObservationValueKind::Float + || !observation.value.float.is_finite() + { + return Err(invalid("expected a finite scalar observation")); + } + if observation.timestamp_ms < self.closed_until { + return Err(PrecomputeError::LateData); + } + let start = observation.timestamp_ms / self.pane_ms * self.pane_ms; + start + .checked_add(self.pane_ms) + .ok_or_else(|| invalid("pane end overflow"))?; + self.advance(observation.timestamp_ms); + let (resource, labels) = self.labels(&observation.resource_labels, &observation.labels); + let key = self.config.series_key_for_entry(&resource, &labels); + self.admit(&key)?; + let k = self.k; + let seed = self.seed; + let entry = self + .panes + .entry(start) + .or_default() + .entry(key) + .or_insert_with(|| PaneSeries { + sketch: KLLWrapper::new(k, seed).with_wire_encoding(Encoding::Msgpack), + resource_labels: resource, + labels, + count: 0, + }); + entry.sketch.update(observation.value.float); + entry.count = entry.count.saturating_add(1); + Ok(()) + } + + /// Merge one full upstream tumbling pane. Multiple sources may contribute + /// to the same pane. Overlapping wide sketches and unaligned panes are + /// rejected because their observations cannot be assigned to one pane. + pub fn merge_pane(&mut self, envelope: &SketchEnvelope) -> Result<(), PrecomputeError> { + if envelope.sketch_type != SketchType::KLLSketch + || envelope.encoding != Encoding::Msgpack + || envelope.payload.is_empty() + { + return Err(invalid( + "inbound pane must contain a full ASAPv1 KLL sketch", + )); + } + if self.config.agg_id != 0 && envelope.agg_id != self.config.agg_id { + return Err(PrecomputeError::AggIdMismatch { + envelope: envelope.agg_id, + config: self.config.agg_id, + }); + } + if !envelope.window_start_ms.is_multiple_of(self.pane_ms) + || envelope.window_start_ms.checked_add(self.pane_ms) != Some(envelope.window_end_ms) + { + return Err(invalid( + "inbound sketch must cover exactly one aligned pane", + )); + } + let mut sketch = KLLWrapper::new(self.k, self.seed).with_wire_encoding(Encoding::Msgpack); + // Decode before changing state: invalid payloads cannot advance retention. + sketch.apply_delta_encoded(&envelope.payload, Encoding::Msgpack)?; + let cutoff = self.closed_until.saturating_sub(self.retention_ms); + if envelope.window_start_ms < cutoff { + return Err(PrecomputeError::LateData); + } + let (resource, labels) = self.labels(&envelope.resource_labels, &envelope.labels); + let key = self.config.series_key_for_entry(&resource, &labels); + self.advance(envelope.window_end_ms); + self.admit(&key)?; + let entry = self + .panes + .entry(envelope.window_start_ms) + .or_default() + .entry(key) + .or_insert_with(|| PaneSeries { + sketch: KLLWrapper::new(self.k, self.seed).with_wire_encoding(Encoding::Msgpack), + resource_labels: resource, + labels, + count: 0, + }); + entry.sketch.merge(&sketch)?; + entry.count = entry.count.saturating_add(envelope.count); + Ok(()) + } + + fn labels(&self, resource: &[KeyValue], labels: &[KeyValue]) -> (Vec, Vec) { + if self.config.global_aggregation { + return (vec![], vec![]); + } + ( + if self.config.omit_resource_attrs { + vec![] + } else { + resource.to_vec() + }, + series_attrs(labels, &self.config.aggregate_by), + ) + } + + fn admit(&self, key: &str) -> Result<(), PrecomputeError> { + if self.config.max_series == 0 || self.panes.values().any(|pane| pane.contains_key(key)) { + return Ok(()); + } + let series: HashSet<_> = self.panes.values().flat_map(|pane| pane.keys()).collect(); + if series.len() as u64 >= self.config.max_series { + return Err(PrecomputeError::SeriesCapExceeded); + } + Ok(()) + } + + /// Merge all complete panes in `[start_ms, end_ms)` into one full KLL per + /// series. Queries never consume or mutate stored sketches. Bounds must + /// align to panes, lie within retention, and exclude the active pane. + /// Missing panes mean no observations, not interpolated samples. + pub fn query( + &self, + start_ms: u64, + end_ms: u64, + ) -> Result, PrecomputeError> { + if start_ms >= end_ms + || !start_ms.is_multiple_of(self.pane_ms) + || !end_ms.is_multiple_of(self.pane_ms) + { + return Err(invalid( + "query bounds must describe a nonempty aligned range", + )); + } + if end_ms > self.closed_until + || start_ms < self.closed_until.saturating_sub(self.retention_ms) + { + return Err(invalid("query includes an open or expired pane")); + } + let mut merged: BTreeMap = BTreeMap::new(); + for (_, pane) in self.panes.range(start_ms..end_ms) { + for (key, source) in pane { + let entry = merged.entry(key.clone()).or_insert_with(|| PaneSeries { + sketch: KLLWrapper::new(self.k, self.seed) + .with_wire_encoding(Encoding::Msgpack), + resource_labels: source.resource_labels.clone(), + labels: source.labels.clone(), + count: 0, + }); + entry.sketch.merge(&source.sketch)?; + entry.count = entry.count.saturating_add(source.count); + } + } + merged + .into_values() + .map(|entry| { + Ok(SketchEnvelope { + schema_version: 1, + sketch_type: SketchType::KLLSketch, + agg_id: self.config.agg_id, + resource_labels: entry.resource_labels, + labels: entry.labels, + window_start_ms: start_ms, + window_end_ms: end_ms, + encoding: Encoding::Msgpack, + payload: entry.sketch.snapshot()?, + hash_spec: None, + metric_name: self.config.metric_name.clone(), + count: entry.count, + aggregation_temporality: 1, + value: 0.0, + }) + }) + .collect() + } + + /// Estimate configured quantiles after merging complete panes. This is a + /// quantile of the union of their observations, never a quantile of p99s. + pub fn estimate( + &self, + start_ms: u64, + end_ms: u64, + ) -> Result, PrecomputeError> { + self.quantile_over_time(start_ms, end_ms, &self.config.quantiles) + } + + /// Answer requested quantiles over an aligned range of retained raw-observation + /// panes. Quantiles may differ on each query; sketches are merged before + /// estimating and the retained state is unchanged. + pub fn quantile_over_time( + &self, + start_ms: u64, + end_ms: u64, + quantiles: &[f64], + ) -> Result, PrecomputeError> { + if quantiles + .iter() + .any(|q| !q.is_finite() || !(0.0..=1.0).contains(q)) + { + return Err(invalid("quantiles must be finite values in [0, 1]")); + } + let mut results = Vec::new(); + for envelope in self.query(start_ms, end_ms)? { + let mut sketch = + KLLWrapper::new(self.k, self.seed).with_wire_encoding(Encoding::Msgpack); + sketch.apply_delta_encoded(&envelope.payload, Encoding::Msgpack)?; + for point in sketch.estimate(quantiles, 0) { + let mut result = envelope.clone(); + result.payload.clear(); + result.encoding = Encoding::Unspecified; + result.value = point.value; + result.labels.extend(point.labels); + results.push(result); + } + } + Ok(results) + } +} + +struct RollingState { + store: KllWindowStore, + last_emit: Option, + upstream_complete_until: Option, + stats: StatsSnapshot, + closed: bool, +} + +/// KLL sliding estimator implementing the same runtime API as tumbling +/// precompute. `size` is the lookback and `slide` is the pane/emission interval. +/// It emits the most recent complete range once per boundary, including warmup +/// ranges with fewer observations. Missed ticks coalesce to the latest range. +pub struct KllRollingPrecompute { + state: Mutex, +} + +impl KllRollingPrecompute { + /// Construct an aligned sliding KLL estimator. Overlapping output is scalar; + /// when size equals slide, disjoint panes may be transmitted as sketches. + /// full upstream inputs must be disjoint tumbling panes of `slide` width. + pub fn new(config: PrecomputeConfig) -> Result { + if config.mode != AggregationMode::Sliding + || (config.transmit_sketch && config.window.size != config.window.slide) + || !config.window.allowed_lateness.is_zero() + { + return Err(invalid( + "overlapping sliding requires scalar output and zero allowed_lateness", + )); + } + if !config.transmit_sketch && config.quantiles.is_empty() { + return Err(invalid( + "scalar sliding output requires at least one quantile", + )); + } + let store = KllWindowStore::new(config.clone(), config.window.slide, config.window.size)?; + Ok(Self { + state: Mutex::new(RollingState { + store, + last_emit: None, + upstream_complete_until: None, + stats: StatsSnapshot::default(), + closed: false, + }), + }) + } +} + +impl Precompute for KllRollingPrecompute { + fn observe(&self, observation: &Observation) -> Result<(), PrecomputeError> { + let mut state = self.state.lock().expect("rolling KLL lock"); + if state.closed { + return Err(invalid("instance is closed")); + } + state.stats.input_observations = state.stats.input_observations.saturating_add(1); + if observation.value.kind == ObservationValueKind::Envelope { + drop(state); + return self.observe_envelope( + observation + .value + .envelope + .as_deref() + .ok_or_else(|| invalid("missing envelope"))?, + ); + } + let result = state.store.observe(observation); + match &result { + Err(PrecomputeError::LateData) => state.stats.dropped_late += 1, + Err(PrecomputeError::SeriesCapExceeded) => state.stats.dropped_overflow += 1, + _ => {} + } + result + } + fn observe_envelope(&self, envelope: &SketchEnvelope) -> Result<(), PrecomputeError> { + let mut state = self.state.lock().expect("rolling KLL lock"); + if state.closed { + return Err(invalid("instance is closed")); + } + state.stats.input_envelopes += 1; + let result = state.store.merge_pane(envelope); + if result.is_ok() { + state.upstream_complete_until = Some( + state + .upstream_complete_until + .unwrap_or(0) + .max(envelope.window_end_ms), + ); + } + match &result { + Err(PrecomputeError::LateData) => state.stats.dropped_late += 1, + Err(PrecomputeError::SeriesCapExceeded) => state.stats.dropped_overflow += 1, + _ => {} + } + result + } + fn tick(&self, now_ms: u64) -> Vec { + let mut state = self.state.lock().expect("rolling KLL lock"); + if state.closed { + return vec![]; + } + let wall_boundary = now_ms / state.store.pane_ms * state.store.pane_ms; + let end = state + .upstream_complete_until + .map_or(wall_boundary, |complete| complete.min(wall_boundary)); + if state.last_emit.is_some_and(|last| end <= last) || end == 0 { + return vec![]; + } + if state.upstream_complete_until.is_none() { + state.store.advance(now_ms); + } + let start = end.saturating_sub(state.store.retention_ms); + let results = if state.store.config.transmit_sketch { + state.store.query(start, end) + } else { + state.store.estimate(start, end) + } + .unwrap_or_default(); + if !results.is_empty() { + state.last_emit = Some(end); + } + state.stats.last_tick_ms = now_ms; + state.stats.last_emitted_envelopes = results.len() as u64; + state.stats.output_envelopes += results.len() as u64; + results + } + fn drain(&self) -> Vec { + // A partial pane cannot represent a complete rolling interval. Stop + // without inventing a future boundary or repeating the previous result. + vec![] + } + fn update_config(&self, configs: &PrecomputeConfigSet) { + let mut state = self.state.lock().expect("rolling KLL lock"); + if let Some(config) = configs + .configs + .iter() + .find(|c| c.agg_id == state.store.config.agg_id) + { + if *config != state.store.config { + if let Ok(replacement) = Self::new(config.clone()) { + let mut next = replacement.state.into_inner().expect("rolling KLL lock"); + next.stats = state.stats; + next.closed = state.closed; + *state = next; + } + } + } + } + fn active_config(&self) -> Option { + Some( + self.state + .lock() + .expect("rolling KLL lock") + .store + .config + .clone(), + ) + } + fn stats(&self) -> StatsSnapshot { + let state = self.state.lock().expect("rolling KLL lock"); + let mut stats = state.stats; + stats.active_series = state.store.retained_series() as i64; + stats + } + fn shutdown(&self) -> Result<(), PrecomputeError> { + self.state.lock().expect("rolling KLL lock").closed = true; + Ok(()) + } +} diff --git a/asap-precompute-rs/src/lib.rs b/asap-precompute-rs/src/lib.rs index 8943f61..edff0a0 100644 --- a/asap-precompute-rs/src/lib.rs +++ b/asap-precompute-rs/src/lib.rs @@ -21,6 +21,7 @@ pub mod adapter; pub mod config; pub mod control_channel; pub mod envelope; +pub mod kll_windows; pub mod matchers; pub mod observation; #[cfg(feature = "otap")] diff --git a/asap-precompute-rs/src/otap/config.rs b/asap-precompute-rs/src/otap/config.rs index e0219ab..86deae0 100644 --- a/asap-precompute-rs/src/otap/config.rs +++ b/asap-precompute-rs/src/otap/config.rs @@ -43,6 +43,12 @@ pub const SKETCH_TYPE_COUNTMINSKETCH: &str = "countminsketch"; /// runtime config + sketch factory + observer triple. #[derive(Debug, thiserror::Error)] pub enum ConfigError { + /// Sliding KLL pane/lookback configuration is invalid. + #[error("asap_sketches: invalid sliding window: {reason}")] + InvalidSlidingWindow { + /// Specific validation failure. + reason: String, + }, /// User-facing `sketch_type` string didn't match any of the five /// supported sketches. #[error("asap_sketches: unsupported sketch_type {value:?} (want one of: {valid})")] @@ -88,6 +94,8 @@ pub struct PluginConfig { pub sketch_type: String, /// Window rotation period. Mandatory. pub window_size: Duration, + /// Sliding KLL emission interval/pane size; zero selects tumbling windows. + pub window_slide: Duration, /// Output metric name stamped onto every emitted envelope. pub output_metric_name: String, /// Controller-plan join key. @@ -130,6 +138,7 @@ impl Default for PluginConfig { Self { sketch_type: SKETCH_TYPE_KLL.to_string(), window_size: Duration::from_secs(10), + window_slide: Duration::ZERO, output_metric_name: "asap_sketch".to_string(), agg_id: 0, sketch_params: SketchParams::new(), @@ -187,12 +196,18 @@ pub fn resolve(config: &PluginConfig) -> Result<(PrecomputeConfig, SketchDispatc encoding: config.encoding.name(), }); } + let sliding = !config.window_slide.is_zero(); let pcfg = PrecomputeConfig { agg_id: config.agg_id, sketch_type: dispatch.sketch_type, - mode: crate::config::AggregationMode::Tumbling, + mode: if sliding { + crate::config::AggregationMode::Sliding + } else { + crate::config::AggregationMode::Tumbling + }, window: WindowSpec { size: config.window_size, + slide: config.window_slide, ..Default::default() }, matchers: Vec::new(), @@ -211,6 +226,13 @@ pub fn resolve(config: &PluginConfig) -> Result<(PrecomputeConfig, SketchDispatc global_aggregation: config.global_aggregation, emit_window_stats: config.emit_window_stats, }; + if sliding { + crate::kll_windows::KllRollingPrecompute::new(pcfg.clone()).map_err(|e| { + ConfigError::InvalidSlidingWindow { + reason: e.to_string(), + } + })?; + } Ok((pcfg, dispatch)) } diff --git a/asap-precompute-rs/src/otap/processor.rs b/asap-precompute-rs/src/otap/processor.rs index fcd3cae..65ace8e 100644 --- a/asap-precompute-rs/src/otap/processor.rs +++ b/asap-precompute-rs/src/otap/processor.rs @@ -119,6 +119,10 @@ pub struct AsapSketchesUserConfig { /// Window rotation period — humantime format (e.g. `"10s"`). #[serde(with = "humantime_serde")] pub window_size: Duration, + /// Optional KLL sliding interval/pane size (for example `"5s"`). + /// Omit it for ordinary tumbling windows. + #[serde(default, with = "humantime_serde::option")] + pub window_slide: Option, /// Stamped onto every emitted envelope. pub output_metric_name: String, /// Controller-plan join key. Defaults to 0. @@ -167,6 +171,11 @@ impl AsapSketchesUserConfig { fn into_plugin_config(self) -> Result { use crate::config::SketchParams; + if self.window_slide.is_some_and(|slide| slide.is_zero()) { + return Err(OtapConfigError::InvalidUserConfig { + error: "window_slide must be positive when specified".into(), + }); + } let mut params = SketchParams::new(); for (k, v) in self.sketch_params { // Only numeric tuning knobs survive — strings / bools are @@ -180,6 +189,7 @@ impl AsapSketchesUserConfig { let cfg = PluginConfig { sketch_type: self.sketch_type, window_size: self.window_size, + window_slide: self.window_slide.unwrap_or_default(), output_metric_name: self.output_metric_name, agg_id: self.agg_id, sketch_params: params, @@ -206,8 +216,36 @@ const fn default_transmit_sketch() -> bool { true } +fn timer_period(config: &PluginConfig) -> Duration { + if config.window_slide.is_zero() { + config.window_size + } else { + config.window_slide + } +} + +fn build_precompute( + config: crate::config::PrecomputeConfig, + dispatch: super::config::SketchDispatch, +) -> Result, crate::precompute::PrecomputeError> { + if config.mode == crate::config::AggregationMode::Sliding { + Ok(Arc::new(crate::kll_windows::KllRollingPrecompute::new( + config, + )?)) + } else { + Ok(Arc::new(PrecomputeImpl::new( + Some(config), + Some(dispatch.factory), + Some(dispatch.observer), + ))) + } +} + fn requires_precompute_rebuild(current: &PluginConfig, next: &PluginConfig) -> bool { !current.sketch_type.eq_ignore_ascii_case(&next.sketch_type) + || current.window_slide != next.window_slide + || ((!current.window_slide.is_zero() || !next.window_slide.is_zero()) + && current.window_size != next.window_size) || current.sketch_params != next.sketch_params || current.encoding != next.encoding || current.default_key != next.default_key @@ -270,11 +308,11 @@ pub fn create_asap_sketches_processor( resolve_plugin_config(&plugin_config).map_err(|e| OtapConfigError::InvalidUserConfig { error: format!("asap_sketches: configuration: {e}"), })?; - let precompute: Arc = Arc::new(PrecomputeImpl::new( - Some(precompute_config), - Some(dispatch.factory), - Some(dispatch.observer), - )); + let precompute = build_precompute(precompute_config, dispatch).map_err(|e| { + OtapConfigError::InvalidUserConfig { + error: e.to_string(), + } + })?; Ok(ProcessorWrapper::local( AsapSketchesProcessor::new(precompute, plugin_config), @@ -312,7 +350,7 @@ pub struct AsapSketchesProcessor { impl AsapSketchesProcessor { fn new(precompute: Arc, plugin_config: PluginConfig) -> Self { - let window_size = plugin_config.window_size; + let window_size = timer_period(&plugin_config); let sketch_encoder = OtapSketchEncoder::with_sketch_params(plugin_config.sketch_params.clone()); Self { @@ -432,6 +470,12 @@ impl local::Processor for AsapSketchesProcessor { } } } + if !self.plugin_config.window_slide.is_zero() { + // Completed upstream panes already have a boundary. Emit + // promptly instead of waiting another whole slide interval. + let envs = self.precompute.tick(asap_wall_clock_ms()); + self.emit_envelopes(&envs, effect_handler).await?; + } Ok(()) } Message::Control(NodeControlMsg::TimerTick { .. }) => { @@ -465,11 +509,9 @@ impl local::Processor for AsapSketchesProcessor { let factory_changed = requires_precompute_rebuild(&self.plugin_config, &plugin_config); if factory_changed { - let replacement: Arc = Arc::new(PrecomputeImpl::new( - Some(pcfg.clone()), - Some(dispatch.factory), - Some(dispatch.observer), - )); + let Ok(replacement) = build_precompute(pcfg.clone(), dispatch) else { + return Ok(()); + }; let pending = self.precompute.drain(); self.emit_envelopes(&pending, effect_handler).await?; self.precompute = replacement; @@ -483,16 +525,13 @@ impl local::Processor for AsapSketchesProcessor { }); } - if plugin_config.window_size != self.window_size { + let period = timer_period(&plugin_config); + if period != self.window_size { if let Some(timer) = self.timer.take() { let _ = timer.cancel().await; } - self.timer = Some( - effect_handler - .start_periodic_timer(plugin_config.window_size) - .await?, - ); - self.window_size = plugin_config.window_size; + self.timer = Some(effect_handler.start_periodic_timer(period).await?); + self.window_size = period; } self.plugin_config = plugin_config; Ok(()) @@ -514,8 +553,26 @@ fn asap_wall_clock_ms() -> u64 { #[cfg(test)] mod tests { - //! Unit tests confined to config-shape validation — full lifecycle - //! integration tests live in `tests/otap_pipeline_e2e.rs`. + #[test] + fn sliding_config_selects_pane_timer_and_validates_alignment() { + let parse = |size: &str, slide: &str, transmit: bool| { + serde_json::from_value::(serde_json::json!({"sketch_type":"kll","window_size":size,"window_slide":slide,"output_metric_name":"rolling","encoding":"Msgpack","transmit_sketch":transmit,"quantiles":[0.5,0.99]})).unwrap().into_plugin_config() + }; + let config = parse("60s", "5s", false).unwrap(); + assert_eq!( + super::timer_period(&config), + std::time::Duration::from_secs(5) + ); + let (runtime, _) = super::resolve_plugin_config(&config).unwrap(); + assert_eq!(runtime.mode, crate::config::AggregationMode::Sliding); + assert!(parse("60s", "7s", false).is_err()); + assert!(parse("60s", "5s", true).is_err()); + assert!(parse("5s", "5s", true).is_ok()); + assert!(parse("5s", "0s", false).is_err()); + assert!(parse("3ms", "1.5ms", false).is_err()); + } + // Unit tests confined to config-shape validation — full lifecycle + // Integration tests live in `tests/otap_pipeline_e2e.rs`. use super::*; use serde_json::json; diff --git a/asap-precompute-rs/src/sketches/kll.rs b/asap-precompute-rs/src/sketches/kll.rs index e8a142e..483bb4d 100644 --- a/asap-precompute-rs/src/sketches/kll.rs +++ b/asap-precompute-rs/src/sketches/kll.rs @@ -28,14 +28,9 @@ pub struct KLLWrapper { sk: KLL, k: i32, seed: Option, - /// Snapshot of all observations seen since the last reset. Kept - /// alongside the in-tree compactor so [`Self::merge`] / - /// [`Self::apply_delta`] can replay the peer's items into our - /// compactor without poking at private state. The compactor - /// itself drives [`Self::snapshot`] via the wire-format accessors - /// added in `asap_sketchlib` (wire_levels/wire_items/wire_coin), - /// so cross-language byte-parity now lives against the compactor's - /// actual state, not this history vec. + /// Replay history used only by the legacy proto path. The ASAPv1 path + /// retains the compacted KLL state alone, keeping pane memory independent + /// of the number of incoming observations. history: Vec, /// Outbound wire format. KLL is full-only, so msgpack means the /// self-describing ASAPv1 form; proto means the legacy `KllState` envelope. @@ -87,7 +82,9 @@ impl KLLWrapper { /// Insert a single observation. pub fn update(&mut self, value: f64) { if value.is_finite() { - self.history.push(value); + if !self.wire_encoding.is_msgpack() { + self.history.push(value); + } self.sk.update(&value); } } @@ -100,10 +97,14 @@ impl KLLWrapper { I: IntoIterator, { let values = values.into_iter(); - self.history.reserve(values.size_hint().0); + if !self.wire_encoding.is_msgpack() { + self.history.reserve(values.size_hint().0); + } for value in values { if value.is_finite() { - self.history.push(value); + if !self.wire_encoding.is_msgpack() { + self.history.push(value); + } self.sk.update(&value); } } @@ -308,6 +309,16 @@ impl SketchObserver for KLLObserver { #[cfg(test)] mod tests { + #[test] + fn asapv1_keeps_compacted_state_without_raw_replay_history() { + use super::*; + let mut sketch = KLLWrapper::new(200, Some(42)).with_wire_encoding(Encoding::Msgpack); + sketch.update_batch((0..20_000).map(|value| value as f64)); + assert!(sketch.history.is_empty()); + // The compactor reports retained weight, not the exact raw input count. + assert!(sketch.inner().count().abs_diff(20_000) < 400); + assert!(sketch.snapshot().unwrap().len() < 20_000); + } use super::*; #[test] diff --git a/asap-precompute-rs/src/window.rs b/asap-precompute-rs/src/window.rs index b572d11..0657f60 100644 --- a/asap-precompute-rs/src/window.rs +++ b/asap-precompute-rs/src/window.rs @@ -42,8 +42,8 @@ pub struct SeriesEntry { pub count: u64, } -/// Per-Precompute window manager. Tumbling-only for now; -/// Sliding lands in a follow-up. +/// General tumbling-window manager. Retained KLL panes and sliding +/// estimates use [`crate::kll_windows`] instead. /// /// Locking is owned by the enclosing [`std::sync::Mutex`] in /// [`crate::precompute::PrecomputeImpl`]; this struct itself is diff --git a/asap-precompute-rs/tests/kll_windows.rs b/asap-precompute-rs/tests/kll_windows.rs new file mode 100644 index 0000000..844049e --- /dev/null +++ b/asap-precompute-rs/tests/kll_windows.rs @@ -0,0 +1,291 @@ +//! Pane merging, expiration, and aligned rolling KLL semantics. +use asap_precompute_rs::kll_windows::{KllRollingPrecompute, KllWindowStore}; +use asap_precompute_rs::precompute::PrecomputeError; +use asap_precompute_rs::{ + AggregationMode, Encoding, KeyValue, Observation, ObservationValue, Precompute, + PrecomputeConfig, PrecomputeConfigSet, SketchType, WindowSpec, +}; +use std::time::Duration; + +fn config() -> PrecomputeConfig { + PrecomputeConfig { + sketch_type: SketchType::KLLSketch, + encoding: Encoding::Msgpack, + mode: AggregationMode::Sliding, + transmit_sketch: false, + metric_name: "request.duration.rolling_estimate".into(), + quantiles: vec![0.5, 0.99], + window: WindowSpec { + size: Duration::from_millis(30), + slide: Duration::from_millis(10), + ..Default::default() + }, + ..Default::default() + } +} +fn observation(time: u64, value: f64, source: &str) -> Observation { + Observation::new( + time, + "request.duration", + vec![KeyValue::new("service.name", source)], + vec![KeyValue::new("route", "/checkout")], + ObservationValue::float(value), + ) +} +fn store() -> KllWindowStore { + KllWindowStore::new( + config(), + Duration::from_millis(10), + Duration::from_millis(30), + ) + .unwrap() +} + +#[test] +fn aligned_ranges_merge_tumbling_panes_without_consuming_them() { + let mut s = store(); + for (time, value) in [(1, 1.), (9, 2.), (10, 3.), (20, 4.)] { + s.observe(&observation(time, value, "checkout")).unwrap(); + } + s.advance(30); + let all = s.query(0, 30).unwrap(); + assert_eq!(all.len(), 1); + assert_eq!(all[0].count, 4); + assert!(all[0].payload.starts_with(b"ASAPv1")); + assert_eq!((all[0].window_start_ms, all[0].window_end_ms), (0, 30)); + assert_eq!(s.query(10, 30).unwrap()[0].count, 2); + assert_eq!(s.query(0, 30).unwrap()[0].count, 4); + assert!(s.query(1, 30).is_err()); + assert!(s.query(0, 40).is_err()); + assert!(s.query(10, 10).is_err()); +} + +#[test] +fn upstream_sources_merge_by_pane_and_preserve_series_identity() { + let mut receiver = store(); + for (value, source) in [(1., "checkout"), (3., "checkout"), (100., "other")] { + let mut upstream = store(); + upstream.observe(&observation(1, value, source)).unwrap(); + upstream.advance(10); + receiver + .merge_pane(&upstream.query(0, 10).unwrap()[0]) + .unwrap(); + } + let merged = receiver.query(0, 10).unwrap(); + assert_eq!(merged.len(), 2); + assert_eq!( + merged + .iter() + .find(|e| e.resource_labels[0].value == "checkout") + .unwrap() + .count, + 2 + ); + assert_eq!( + merged + .iter() + .find(|e| e.resource_labels[0].value == "other") + .unwrap() + .count, + 1 + ); + let mut wide = merged[0].clone(); + wide.window_end_ms = 20; + assert!(receiver.merge_pane(&wide).is_err()); + let mut malformed = merged[0].clone(); + malformed.payload = vec![1, 2, 3]; + malformed.window_start_ms = 100; + malformed.window_end_ms = 110; + assert!(receiver.merge_pane(&malformed).is_err()); + // A corrupt future frame did not evict retained observations. + assert_eq!(receiver.query(0, 10).unwrap().len(), 2); +} + +#[test] +fn expiration_boundaries_idle_gaps_and_late_observations() { + let mut s = store(); + s.observe(&observation(1, 1., "checkout")).unwrap(); + s.observe(&observation(10, 2., "checkout")).unwrap(); + s.advance(30); + assert_eq!(s.query(0, 30).unwrap()[0].count, 2); + assert!(matches!( + s.observe(&observation(29, 3., "checkout")), + Err(PrecomputeError::LateData) + )); + s.advance(40); + assert!(s.query(0, 30).is_err()); + assert_eq!(s.query(10, 40).unwrap()[0].count, 1); + s.advance(1000); + assert_eq!(s.retained_panes(), 0); + assert!(s.query(970, 1000).unwrap().is_empty()); + // Backward wall-clock motion cannot reopen expired panes. + s.advance(20); + assert!(s.query(10, 20).is_err()); +} + +#[test] +fn sliding_outputs_use_union_of_observations_and_evict_old_panes() { + let p = KllRollingPrecompute::new(config()).unwrap(); + for (time, value) in [(1, 1.), (11, 2.), (21, 3.)] { + p.observe(&observation(time, value, "checkout")).unwrap(); + } + let first = p.tick(30); + assert_eq!(first.len(), 2); + assert!(first + .iter() + .all(|e| e.count == 3 && e.window_start_ms == 0 && e.window_end_ms == 30)); + assert_eq!(first[0].value, 2.); + assert!(p.tick(35).is_empty()); + p.observe(&observation(31, 100., "checkout")).unwrap(); + let second = p.tick(40); + assert_eq!(second.len(), 2); + assert_eq!(second[0].value, 3.); + assert_eq!( + (second[0].window_start_ms, second[0].window_end_ms), + (10, 40) + ); + assert_eq!(second[1].value, 100.); + assert!(p.tick(100).is_empty()); + assert!(p.drain().is_empty()); + p.shutdown().unwrap(); + assert!(p.observe(&observation(101, 1., "checkout")).is_err()); +} + +#[test] +fn many_panes_stay_bounded_and_unequal_compacted_panes_keep_weights() { + let mut s = store(); + // Force compaction. One large pane and one tiny high-value pane make an + // unweighted merge or averaging pane quantiles visibly incorrect. + for value in 0..20_000 { + s.observe(&observation(1, value as f64, "checkout")) + .unwrap(); + } + s.observe(&observation(11, 1_000_000., "checkout")).unwrap(); + s.advance(20); + let estimated = s.estimate(0, 20).unwrap(); + assert_eq!(estimated[0].count, 20_001); + assert!((estimated[0].value - 10_000.).abs() < 800.); + assert!((estimated[1].value - 19_800.).abs() < 800.); + assert!(s.query(0, 20).unwrap()[0].payload.len() < 20_000); + let mut downstream = store(); + downstream.merge_pane(&s.query(0, 10).unwrap()[0]).unwrap(); + assert_eq!(downstream.query(0, 10).unwrap()[0].count, 20_000); + for pane in 2..1000 { + s.observe(&observation(pane * 10 + 1, pane as f64, "checkout")) + .unwrap(); + assert!(s.retained_panes() <= 4); + } +} + +#[test] +fn rolling_can_consume_upstream_panes_and_resets_on_window_change() { + let p = KllRollingPrecompute::new(config()).unwrap(); + let mut upstream = store(); + upstream.observe(&observation(1, 42., "checkout")).unwrap(); + upstream.advance(10); + p.observe_envelope(&upstream.query(0, 10).unwrap()[0]) + .unwrap(); + assert_eq!(p.tick(30)[0].value, 42.); + let mut next = config(); + next.window.size = Duration::from_millis(20); + p.update_config(&PrecomputeConfigSet { + version: 1, + configs: vec![next], + }); + assert!(p.tick(40).is_empty()); +} + +#[test] +fn invalid_windows_quantiles_and_nonfinite_values_are_rejected() { + let mut cfg = config(); + cfg.window.size = Duration::from_millis(25); + assert!(KllRollingPrecompute::new(cfg).is_err()); + let mut cfg = config(); + cfg.window.slide = Duration::ZERO; + assert!(KllRollingPrecompute::new(cfg).is_err()); + let mut cfg = config(); + cfg.quantiles = vec![f64::NAN]; + assert!(KllRollingPrecompute::new(cfg).is_err()); + let mut cfg = config(); + cfg.sketch_type = SketchType::DDSketch; + assert!(KllRollingPrecompute::new(cfg).is_err()); + let mut cfg = config(); + cfg.transmit_sketch = true; + assert!(KllRollingPrecompute::new(cfg).is_err()); + let mut s = store(); + assert!(s + .observe(&observation(1, f64::INFINITY, "checkout")) + .is_err()); + assert_eq!(s.retained_panes(), 0); +} + +#[test] +fn series_cap_applies_across_retained_panes() { + let mut cfg = config(); + cfg.max_series = 1; + let mut s = + KllWindowStore::new(cfg, Duration::from_millis(10), Duration::from_millis(30)).unwrap(); + s.observe(&observation(1, 1., "checkout")).unwrap(); + assert!(matches!( + s.observe(&observation(11, 2., "other")), + Err(PrecomputeError::SeriesCapExceeded) + )); + s.advance(50); + s.observe(&observation(51, 2., "other")).unwrap(); +} + +#[test] +fn upstream_timer_waits_for_new_complete_panes() { + let p = KllRollingPrecompute::new(config()).unwrap(); + assert!(p.tick(30).is_empty()); + let mut upstream = store(); + upstream.observe(&observation(21, 42., "checkout")).unwrap(); + upstream.advance(30); + p.observe_envelope(&upstream.query(20, 30).unwrap()[0]) + .unwrap(); + let first = p.tick(30); + assert_eq!(first[0].window_end_ms, 30); + assert_eq!(first[0].value, 42.); + assert!(p.tick(40).is_empty()); + upstream + .observe(&observation(31, 100., "checkout")) + .unwrap(); + upstream.advance(40); + p.observe_envelope(&upstream.query(30, 40).unwrap()[0]) + .unwrap(); + let next = p.tick(40); + assert_eq!(next[0].count, 2); + assert_eq!(next[0].window_end_ms, 40); +} + +#[test] +fn disjoint_mode_transmits_full_panes_for_downstream_rolling_queries() { + let mut cfg = config(); + cfg.window.size = cfg.window.slide; + cfg.transmit_sketch = true; + let creator = KllRollingPrecompute::new(cfg).unwrap(); + creator.observe(&observation(1, 42., "checkout")).unwrap(); + let pane = creator.tick(10); + assert_eq!(pane.len(), 1); + assert_eq!(pane[0].encoding, Encoding::Msgpack); + let mut retained = store(); + retained.merge_pane(&pane[0]).unwrap(); + let answer = retained.quantile_over_time(0, 10, &[0.25, 0.75]).unwrap(); + assert_eq!(answer.len(), 2); + assert!(answer.iter().all(|e| e.value == 42.)); + assert!(retained.quantile_over_time(0, 10, &[1.5]).is_err()); +} + +#[test] +fn small_high_value_pane_cannot_outweigh_a_large_compacted_low_value_pane() { + let mut s = store(); + for _ in 0..20_000 { + s.observe(&observation(1, 1., "checkout")).unwrap(); + } + for _ in 0..10 { + s.observe(&observation(11, 1_000_000., "checkout")).unwrap(); + } + s.advance(20); + let result = s.quantile_over_time(0, 20, &[0.5, 0.99]).unwrap(); + assert!(result.iter().all(|e| e.value == 1. && e.count == 20_010)); +} diff --git a/benchmarks/STREAMING.md b/benchmarks/STREAMING.md index 0638520..059e266 100644 --- a/benchmarks/STREAMING.md +++ b/benchmarks/STREAMING.md @@ -156,6 +156,18 @@ networks it creates, including on failure; it saves logs before removal. - `snapshots.json`: raw before/interval/drained counters and latency histograms. - `result.json` and seven component logs per run. +## Live demo mode + +`--demo` runs the selected scenarios side by side, each on its own CPU slice +(generators, one CPU per processor, `--demo-backend-cores` for the backend), +at `--demo-rate` signals/s per source until Ctrl+C or `--run-seconds`. It +serves per-process throughput, CPU, egress, RSS and the backend's latest +validated answers as Prometheus metrics on `--metrics-address`:`--metrics-port` +(default `127.0.0.1:9465`). `--rolling-windows N` (`ASAP_ROLLING_WINDOWS`) +makes estimate also answer over the last N completed windows; the default of +1 disables rolling answers, so capacity runs are unaffected. See +[the multi-process KLL demo](../docs/kll-windows.md). + CI runs the full 189-run matrix, checks actual network traffic, completed windows, zero loss after drain, and resource artifacts. It uploads results even on failure. The old PR's 2.43× was measured with the file-backed batch harness and is not a diff --git a/benchmarks/run-streaming.py b/benchmarks/run-streaming.py index 8cd667b..e563a31 100755 --- a/benchmarks/run-streaming.py +++ b/benchmarks/run-streaming.py @@ -3,16 +3,20 @@ import argparse import concurrent.futures import csv +import collections import http.client +import http.server import json import math import os import platform from pathlib import Path import shlex +import signal import socket import statistics import subprocess +import threading import time import urllib.request import uuid @@ -50,8 +54,10 @@ def delta_histogram(before, after): class Run: - def __init__(self, args, scenario, rate, repeat, traffic_rate, window_points): + def __init__(self, args, scenario, rate, repeat, traffic_rate, window_points, cores=None, rolling_windows=1): self.args, self.scenario, self.rate = args, scenario, rate + # Demo mode runs several pipelines at once, each on its own CPU slice. + self.cores, self.rolling_windows = cores, rolling_windows self.generator_core_count, self.traffic_rate, self.window_points = args.generator_cores, traffic_rate, window_points self.observation_seconds = max( args.duration, @@ -92,7 +98,7 @@ def wait_for(self, predicate, timeout=120): raise TimeoutError("streaming benchmark readiness/drain timeout") def setup(self): - cores = sorted(os.sched_getaffinity(0)) + cores = self.cores or sorted(os.sched_getaffinity(0)) # The backend is benchmark infrastructure, so it receives every host CPU # left after the isolated generator and processor assignments. Keep at # least one CPU available for it. Memory is intentionally uncapped for @@ -141,6 +147,7 @@ def setup(self): "ASAP_GENERATOR_WORKERS": str(self.generator_core_count), "ASAP_SIGNALS_PER_SECOND": str(self.traffic_rate), "ASAP_PROFILE_CPU": "1" if self.args.profile_cpu else "0", + "ASAP_ROLLING_WINDOWS": str(self.rolling_windows), } cmd = ["create", "--name", name, "--network", self.networks[0], "--network-alias", role, "--cpuset-cpus", ",".join(map(str, cpu_set))] @@ -458,6 +465,103 @@ def draw_resources(rows, path): path.write_text("\n".join(svg) + "\n") +def demo_metrics(runs, history): + """Prometheus text exposition over the last ~30 s of control snapshots.""" + lines = [] + def gauge(name, labels, value): + rendered = ",".join(f'{key}="{val}"' for key, val in labels.items()) + lines.append(f"{name}{{{rendered}}} {value}") + for scenario, run in runs.items(): + if len(history[scenario]) < 2: + continue + before, after = history[scenario][0], history[scenario][-1] + seconds = (after["backend"]["timestamp_ns"] - before["backend"]["timestamp_ns"]) / 1e9 + validated = after["backend"]["stats"]["received_signals"] - before["backend"]["stats"]["received_signals"] + gauge("asap_demo_validated_signals_per_second", {"scenario": scenario}, validated / seconds) + gauge("asap_demo_completed_windows", {"scenario": scenario}, after["backend"]["stats"]["completed_windows"]) + processor_cpu = 0 + for role in ROLES: + start, end = before[role], after[role] + elapsed = (end["timestamp_ns"] - start["timestamp_ns"]) / 1e9 + cpu = (end["cpu_nanoseconds"] - start["cpu_nanoseconds"]) / 1e9 + if role in PROCESSOR_ROLES: + processor_cpu += cpu + iface = run.data_ifaces[role] + tx = end["interfaces"][iface]["tx_bytes"] - start["interfaces"][iface]["tx_bytes"] + received = end["stats"]["received_signals"] - start["stats"]["received_signals"] + labels = {"scenario": scenario, "component": role} + gauge("asap_demo_cpu_cores", labels, cpu / elapsed) + gauge("asap_demo_data_tx_mbit_per_second", labels, tx * 8 / elapsed / 1e6) + gauge("asap_demo_rss_mib", labels, end["rss_kib"] / 1024) + if role.startswith("branch"): + gauge("asap_demo_ingested_signals_per_second", labels, received / elapsed) + if validated: + gauge("asap_demo_processor_cpu_seconds_per_million_signals", {"scenario": scenario}, + processor_cpu / validated * 1e6) + for point in after["backend"]["stats"].get("latest_results", []): + labels = {"scenario": scenario, "window_kind": point["window_kind"], "quantile": point["quantile"]} + gauge("asap_demo_estimate", labels, point["value"]) + gauge("asap_demo_reference", labels, point["reference"]) + gauge("asap_demo_relative_error", labels, + abs(point["value"] - point["reference"]) / abs(point["reference"]) if point["reference"] else 0) + return "\n".join(lines) + "\n" + + +def demo(args): + """Run every scenario's pipeline concurrently at one offered rate until Ctrl+C.""" + cores = sorted(os.sched_getaffinity(0)) + per_pipeline = args.generator_cores + len(PROCESSOR_ROLES) + args.demo_backend_cores + if len(cores) < per_pipeline * len(args.scenarios): + raise SystemExit(f"demo needs {per_pipeline} CPUs per scenario ({args.generator_cores} generator, " + f"{len(PROCESSOR_ROLES)} processors, {args.demo_backend_cores} backend); " + f"only {len(cores)} are available") + def stop(*_): + raise KeyboardInterrupt + # Background jobs ignore SIGINT; SIGTERM stops the demo with the same cleanup. + signal.signal(signal.SIGTERM, stop) + runs, history, body = {}, {}, {"text": ""} + class Metrics(http.server.BaseHTTPRequestHandler): + def do_GET(self): + payload = body["text"].encode() + self.send_response(200 if self.path == "/metrics" else 404) + self.send_header("Content-Type", "text/plain; version=0.0.4") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + def log_message(self, *_): + pass + server = http.server.ThreadingHTTPServer((args.metrics_address, args.metrics_port), Metrics) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + for index, scenario in enumerate(args.scenarios): + run = Run(args, scenario, 0, "demo", args.demo_rate, args.window_points[0], + cores=cores[index * per_pipeline:(index + 1) * per_pipeline], + rolling_windows=args.rolling_windows) + runs[scenario] = run + run.setup() + # Windows complete in bursts; rates over ~30 s keep the gauges readable. + history[scenario] = collections.deque(maxlen=max(2, round(30 / args.sample_interval) + 1)) + for run in runs.values(): + for role in ("generator_a", "generator_b"): + request(run.urls[role], "/run", "POST") + print(f"demo: {', '.join(args.scenarios)} at {args.demo_rate:,} signals/s/source, " + f"window {args.window_points[0]:,}, rolling {args.rolling_windows}; " + f"metrics on http://{args.metrics_address}:{args.metrics_port}/metrics; Ctrl+C stops", flush=True) + deadline = time.monotonic() + args.run_seconds if args.run_seconds else None + while deadline is None or time.monotonic() < deadline: + time.sleep(args.sample_interval) + for scenario, run in runs.items(): + run.healthy() + history[scenario].append(run.statuses()) + body["text"] = demo_metrics(runs, history) + except KeyboardInterrupt: + print("demo: stopping", flush=True) + finally: + for run in runs.values(): + run.close() + server.shutdown() + + def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--image", default="asap-stream-benchmark:local") @@ -487,7 +591,22 @@ def main(): parser.add_argument("--sustainable-max-backlog-growth-windows", type=float, default=1.0) parser.add_argument("--sustainable-max-p99-ms", type=float, default=5000) parser.add_argument("--resume", action="store_true", help="reuse completed runs from output/runs.json") + parser.add_argument("--demo", action="store_true", + help="run the selected scenarios side by side at --demo-rate until Ctrl+C and serve Prometheus metrics") + parser.add_argument("--demo-rate", type=int, default=100000, help="demo offered signals/s per source") + parser.add_argument("--demo-backend-cores", type=int, default=1, help="demo backend CPUs per scenario") + parser.add_argument("--rolling-windows", type=int, default=1, + help="completed windows merged into each rolling estimate; 1 disables rolling results") + parser.add_argument("--run-seconds", type=float, default=0, help="demo duration; 0 runs until Ctrl+C") + parser.add_argument("--metrics-address", default="127.0.0.1") + parser.add_argument("--metrics-port", type=int, default=9465) args = parser.parse_args() + if args.demo: + if args.demo_rate < 1 or args.rolling_windows < 1 or args.demo_backend_cores < 1 or args.run_seconds < 0: + parser.error("invalid demo rate, rolling windows, backend cores, or duration") + args.output.mkdir(parents=True, exist_ok=True) + demo(args) + return if any(points < 1 for points in args.window_points) or args.batch_size < 1 or any(args.batch_size > points for points in args.window_points) or args.generator_cores < 1 or any(rate < 1 for rate in args.traffic_rates) or not all(math.isfinite(t) for t in (args.warmup, args.duration, args.sample_interval)) or args.warmup < 0 or args.duration <= 0 or args.minimum_observed_windows < 2 or args.sample_interval <= 0 or args.repetitions < 1 or any(not math.isfinite(r) or r < 0 for r in args.rates_mbit) or not 0 < args.sustainable_min_delivery <= 1 or args.sustainable_max_backlog_growth_windows < 0 or args.sustainable_max_p99_ms <= 0: parser.error("invalid benchmark sizes, times, repetitions, or rates") args.output.mkdir(parents=True, exist_ok=True) diff --git a/demos/kll-grafana/compose.yaml b/demos/kll-grafana/compose.yaml new file mode 100644 index 0000000..52246e0 --- /dev/null +++ b/demos/kll-grafana/compose.yaml @@ -0,0 +1,30 @@ +# Host networking keeps Prometheus, Grafana and the demo metrics endpoint on +# localhost only (Linux). Prometheus scrapes run-streaming.py --demo on :9465 +# and still accepts OTLP pushes from the finite asap-otap-demo export. +services: + prometheus: + image: prom/prometheus:v3.5.0 + network_mode: host + command: + - --config.file=/etc/prometheus/prometheus.yml + - --web.enable-otlp-receiver + - --web.listen-address=127.0.0.1:${PROMETHEUS_PORT:-9090} + volumes: + - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro + - prometheus-data:/prometheus + grafana: + image: grafana/grafana:12.1.0 + network_mode: host + environment: + GF_SERVER_HTTP_ADDR: 127.0.0.1 + GF_SERVER_HTTP_PORT: ${GRAFANA_PORT:-3000} + GF_AUTH_ANONYMOUS_ENABLED: "true" + GF_AUTH_ANONYMOUS_ORG_ROLE: Viewer + PROMETHEUS_PORT: ${PROMETHEUS_PORT:-9090} + volumes: + - ./grafana/provisioning:/etc/grafana/provisioning:ro + - ./grafana/dashboards:/var/lib/grafana/dashboards:ro + - grafana-data:/var/lib/grafana +volumes: + prometheus-data: + grafana-data: diff --git a/demos/kll-grafana/grafana/dashboards/kll-stream.json b/demos/kll-grafana/grafana/dashboards/kll-stream.json new file mode 100644 index 0000000..7dacd60 --- /dev/null +++ b/demos/kll-grafana/grafana/dashboards/kll-stream.json @@ -0,0 +1,260 @@ +{ + "uid": "asap-kll-stream", + "title": "ASAP KLL vs exact: multi-process streaming", + "schemaVersion": 39, + "version": 1, + "refresh": "5s", + "time": { + "from": "now-15m", + "to": "now" + }, + "panels": [ + { + "id": 1, + "type": "timeseries", + "title": "Backend-validated throughput", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_demo_validated_signals_per_second", + "legendFormat": "{{scenario}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short", + "min": 0 + }, + "overrides": [] + } + }, + { + "id": 2, + "type": "bargauge", + "title": "Processor CPU-seconds per million signals (branches + merge + estimate)", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_demo_processor_cpu_seconds_per_million_signals", + "legendFormat": "{{scenario}}", + "instant": true + } + ], + "fieldConfig": { + "defaults": { + "decimals": 2 + }, + "overrides": [] + }, + "options": { + "orientation": "horizontal", + "displayMode": "basic", + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + } + } + }, + { + "id": 3, + "type": "timeseries", + "title": "CPU cores per pinned processor", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 8 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_demo_cpu_cores{component=~\"branch_a|branch_b|merge|estimate\"}", + "legendFormat": "{{scenario}} {{component}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "short" + }, + "overrides": [] + } + }, + { + "id": 4, + "type": "timeseries", + "title": "Data-network egress per process", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 8 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_demo_data_tx_mbit_per_second{component=~\"branch_a|branch_b|merge|estimate\"}", + "legendFormat": "{{scenario}} {{component}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "Mbits" + }, + "overrides": [] + } + }, + { + "id": 5, + "type": "timeseries", + "title": "Tumbling p50/p99 vs exact reference", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 16 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_demo_estimate{window_kind=\"tumbling\"}", + "legendFormat": "{{scenario}} p{{quantile}}" + }, + { + "refId": "B", + "expr": "asap_demo_reference{window_kind=\"tumbling\",scenario=\"kll\"}", + "legendFormat": "reference p{{quantile}}" + } + ], + "fieldConfig": { + "defaults": { + "decimals": 4 + }, + "overrides": [] + } + }, + { + "id": 6, + "type": "timeseries", + "title": "Rolling p50/p99 vs exact reference", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 16 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_demo_estimate{window_kind=\"rolling\"}", + "legendFormat": "{{scenario}} p{{quantile}}" + }, + { + "refId": "B", + "expr": "asap_demo_reference{window_kind=\"rolling\",scenario=\"kll\"}", + "legendFormat": "reference p{{quantile}}" + } + ], + "fieldConfig": { + "defaults": { + "decimals": 4 + }, + "overrides": [] + } + }, + { + "id": 7, + "type": "timeseries", + "title": "Relative error vs exact", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 0, + "y": 24 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_demo_relative_error", + "legendFormat": "{{scenario}} {{window_kind}} p{{quantile}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "percentunit" + }, + "overrides": [] + } + }, + { + "id": 8, + "type": "timeseries", + "title": "Resident memory per process", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 12, + "x": 12, + "y": 24 + }, + "targets": [ + { + "refId": "A", + "expr": "asap_demo_rss_mib{component=~\"branch_a|branch_b|merge|estimate\"}", + "legendFormat": "{{scenario}} {{component}}" + } + ], + "fieldConfig": { + "defaults": { + "unit": "decmbytes" + }, + "overrides": [] + } + } + ] +} diff --git a/demos/kll-grafana/grafana/dashboards/kll.json b/demos/kll-grafana/grafana/dashboards/kll.json new file mode 100644 index 0000000..a78a6da --- /dev/null +++ b/demos/kll-grafana/grafana/dashboards/kll.json @@ -0,0 +1,81 @@ +{ + "uid": "asap-kll", + "title": "ASAP KLL estimates", + "schemaVersion": 39, + "version": 1, + "refresh": "5s", + "time": { + "from": "now-15m", + "to": "now" + }, + "panels": [ + { + "id": 1, + "type": "stat", + "title": "Latest KLL estimates", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 0 + }, + "targets": [ + { + "refId": "A", + "expr": "request_duration_estimate{quantile=~\"0.5|0.99\"}", + "legendFormat": "quantile {{quantile}}", + "instant": true + } + ], + "fieldConfig": { + "defaults": { + "decimals": 3 + }, + "overrides": [] + }, + "options": { + "reduceOptions": { + "calcs": [ + "lastNotNull" + ], + "fields": "", + "values": false + } + } + }, + { + "id": 2, + "type": "timeseries", + "title": "KLL estimates over time", + "datasource": { + "type": "prometheus", + "uid": "asap-prometheus" + }, + "gridPos": { + "h": 8, + "w": 24, + "x": 0, + "y": 8 + }, + "targets": [ + { + "refId": "A", + "expr": "request_duration_estimate{quantile=~\"0.5|0.99\"}", + "legendFormat": "quantile {{quantile}}", + "instant": false + } + ], + "fieldConfig": { + "defaults": { + "decimals": 3 + }, + "overrides": [] + }, + "options": {} + } + ] +} diff --git a/demos/kll-grafana/grafana/provisioning/dashboards/asap.yaml b/demos/kll-grafana/grafana/provisioning/dashboards/asap.yaml new file mode 100644 index 0000000..2dd466d --- /dev/null +++ b/demos/kll-grafana/grafana/provisioning/dashboards/asap.yaml @@ -0,0 +1,7 @@ +apiVersion: 1 +providers: + - name: ASAP demo + folder: ASAP + type: file + options: + path: /var/lib/grafana/dashboards diff --git a/demos/kll-grafana/grafana/provisioning/datasources/prometheus.yaml b/demos/kll-grafana/grafana/provisioning/datasources/prometheus.yaml new file mode 100644 index 0000000..1dfc28d --- /dev/null +++ b/demos/kll-grafana/grafana/provisioning/datasources/prometheus.yaml @@ -0,0 +1,9 @@ +apiVersion: 1 +datasources: + - name: Prometheus + uid: asap-prometheus + type: prometheus + access: proxy + url: http://127.0.0.1:${PROMETHEUS_PORT} + isDefault: true + editable: false diff --git a/demos/kll-grafana/prometheus.yml b/demos/kll-grafana/prometheus.yml new file mode 100644 index 0000000..760af13 --- /dev/null +++ b/demos/kll-grafana/prometheus.yml @@ -0,0 +1,9 @@ +global: + scrape_interval: 15s +scrape_configs: + # benchmarks/run-streaming.py --demo (default --metrics-port 9465). + - job_name: asap-demo + scrape_interval: 2s + static_configs: + - targets: ["127.0.0.1:9465"] +# The finite demo's estimates arrive through the OTLP HTTP receiver instead. diff --git a/demos/kll-grafana/run-demo.sh b/demos/kll-grafana/run-demo.sh new file mode 100755 index 0000000..fa2271b --- /dev/null +++ b/demos/kll-grafana/run-demo.sh @@ -0,0 +1,21 @@ +#!/usr/bin/env bash +# Multi-process streaming demo: every processor of every scenario's pipeline is +# its own container pinned to one CPU, connected by OTLP/HTTP over TCP. Builds +# the benchmark image, starts Prometheus/Grafana, then runs until Ctrl+C. +# Extra arguments go to benchmarks/run-streaming.py (e.g. --demo-rate 200000). +set -euo pipefail +repo_root=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd) +cd "$repo_root" +cargo_target_dir=${CARGO_TARGET_DIR:-"$repo_root/asap-precompute-rs/target"} +read -r -a docker_command <<< "${DOCKER_COMMAND:-docker}" +cargo build --manifest-path asap-precompute-rs/Cargo.toml \ + --release --features stream-benchmark --bin asap-stream-bench +context=$(mktemp -d) +trap 'rm -rf "$context"' EXIT +cp "$cargo_target_dir/release/asap-stream-bench" "$context/" +strip --strip-debug "$context/asap-stream-bench" +"${docker_command[@]}" build -q -f benchmarks/stream.Dockerfile -t asap-stream-benchmark:local "$context" +"${docker_command[@]}" compose -f demos/kll-grafana/compose.yaml up -d +python3 benchmarks/run-streaming.py --demo --docker-command "${DOCKER_COMMAND:-docker}" \ + --output benchmark-results/demo --scenarios exact kll --window-points 65536 \ + --rolling-windows 6 --generator-cores 2 "$@" diff --git a/demos/kll-grafana/verify-streaming.py b/demos/kll-grafana/verify-streaming.py new file mode 100644 index 0000000..2940c2f --- /dev/null +++ b/demos/kll-grafana/verify-streaming.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +"""Run the multi-process demo briefly and verify its metrics in Prometheus and Grafana. + +Requires the asap-stream-benchmark:local image (built by run-demo.sh) and the +compose stack. Extra arguments are passed to benchmarks/run-streaming.py. +""" +import argparse +import json +import subprocess +import sys +import time +import urllib.parse +import urllib.request +from pathlib import Path + +REPO = Path(__file__).resolve().parents[2] +PROCESSORS = ('branch_a', 'branch_b', 'merge', 'estimate') + + +def query(base, expression, grafana=False): + path = '/api/v1/query?' + urllib.parse.urlencode({'query': expression}) + if grafana: + path = '/api/datasources/proxy/uid/asap-prometheus' + path + with urllib.request.urlopen(base.rstrip('/') + path, timeout=10) as response: + payload = json.load(response) + assert payload['status'] == 'success', payload + return payload['data']['result'] + + +def values(base, expression, keys, grafana=False): + return {tuple(row['metric'].get(k, '') for k in keys): float(row['value'][1]) + for row in query(base, expression, grafana)} + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--prometheus', default='http://localhost:9090') + parser.add_argument('--grafana', default='http://localhost:3000') + parser.add_argument('--docker-command', default='docker') + parser.add_argument('--scenarios', nargs='+', default=['exact', 'kll']) + parser.add_argument('--tolerance', type=float, default=0.05, + help='maximum relative error for sketch scenarios') + args, extra = parser.parse_known_args() + demo = subprocess.Popen([ + sys.executable, str(REPO / 'benchmarks/run-streaming.py'), '--demo', + '--docker-command', args.docker_command, '--output', str(REPO / 'benchmark-results/demo-verify'), + '--scenarios', *args.scenarios, '--window-points', '65536', '--rolling-windows', '3', + '--generator-cores', '1', '--demo-rate', '50000', '--run-seconds', '60', *extra]) + try: + deadline = time.monotonic() + 240 + while True: + assert demo.poll() is None, 'demo exited before metrics were verified' + assert time.monotonic() < deadline, 'demo metrics did not appear' + windows = values(args.prometheus, 'asap_demo_completed_windows', ['scenario']) + rolling = values(args.prometheus, 'asap_demo_estimate{window_kind="rolling"}', ['scenario']) + if all(windows.get((s,), 0) >= 4 for s in args.scenarios) and len(rolling) == len(args.scenarios): + break + time.sleep(2) + throughput = values(args.prometheus, 'asap_demo_validated_signals_per_second', ['scenario']) + assert all(throughput[(s,)] > 0 for s in args.scenarios), throughput + cpu = values(args.prometheus, 'asap_demo_cpu_cores', ['scenario', 'component']) + assert all((s, c) in cpu for s in args.scenarios for c in PROCESSORS), cpu + errors = values(args.prometheus, 'asap_demo_relative_error', + ['scenario', 'window_kind', 'quantile']) + assert {(s, k, q) for s in args.scenarios for k in ('tumbling', 'rolling') + for q in ('0.5', '0.99')} <= set(errors), errors + for (scenario, _, _), error in errors.items(): + assert error <= (0.0 if scenario == 'exact' else args.tolerance), errors + expression = 'asap_demo_estimate' + keys = ['scenario', 'window_kind', 'quantile'] + assert set(values(args.prometheus, expression, keys)) == set( + values(args.grafana, expression, keys, grafana=True)) + with urllib.request.urlopen(args.grafana.rstrip('/') + '/api/dashboards/uid/asap-kll-stream', + timeout=10) as response: + dashboard = json.load(response)['dashboard'] + assert any('asap_demo_validated_signals_per_second' in t['expr'] + for p in dashboard['panels'] for t in p['targets']) + print('throughput (signals/s):', {k[0]: round(v) for k, v in throughput.items()}) + print('relative error:', {'/'.join(k): round(v, 5) for k, v in sorted(errors.items())}) + print('Verified per-process throughput/CPU, tumbling and rolling accuracy, and Grafana queries.') + finally: + demo.wait(timeout=180) + assert demo.returncode == 0, 'demo returned an error' + + +if __name__ == '__main__': + main() diff --git a/demos/kll-grafana/verify.py b/demos/kll-grafana/verify.py new file mode 100644 index 0000000..fead757 --- /dev/null +++ b/demos/kll-grafana/verify.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +"""Run the real demo and verify its estimates through Prometheus and Grafana.""" +import argparse +import json +import re +import subprocess +import time +import urllib.parse +import urllib.request + + +def get(base, path): + with urllib.request.urlopen(base.rstrip('/') + path, timeout=10) as response: + return json.load(response) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('binary', help='path to the built asap-otap-demo binary') + parser.add_argument('--prometheus', default='http://localhost:9090') + parser.add_argument('--grafana', default='http://localhost:3000') + args = parser.parse_args() + output = subprocess.check_output([ + args.binary, '--prometheus-otlp-endpoint', + args.prometheus.rstrip('/') + '/api/v1/otlp/v1/metrics', + ], text=True) + print(output, end='') + expected = {q: float(v) for q, v in re.findall( + r'result metric=request.duration.estimate quantile=(\S+) value=(\S+)', output)} + assert set(expected) == {'0.5', '0.99'}, expected + query = '/api/v1/query?' + urllib.parse.urlencode({'query': 'request_duration_estimate{quantile=~"0.5|0.99"}'}) + for base, path in [ + (args.prometheus, query), + (args.grafana, '/api/datasources/proxy/uid/asap-prometheus' + query), + ]: + for attempt in range(30): + response = get(base, path) + assert response['status'] == 'success', response + actual = {row['metric']['quantile']: float(row['value'][1]) + for row in response['data']['result']} + if set(actual) == set(expected): + break + time.sleep(1) + assert set(actual) == set(expected), actual + for quantile, value in expected.items(): + assert abs(actual[quantile] - value) <= 0.001, (actual, expected) + dashboard = get(args.grafana, '/api/dashboards/uid/asap-kll')['dashboard'] + assert dashboard['panels'][0]['targets'][0]['expr'].startswith('request_duration_estimate') + print('Verified p50/p99 in Prometheus, Grafana datasource queries, and provisioned dashboard.') + + +if __name__ == '__main__': + main() diff --git a/docs/demo-guide.md b/docs/demo-guide.md index 5985e4f..ef48d22 100644 --- a/docs/demo-guide.md +++ b/docs/demo-guide.md @@ -68,3 +68,86 @@ cargo test --features otap-engine --test otap_pipeline_e2e The runnable binary additionally asserts that the final output contains p50 and p99 after the four child processes finish successfully. + +For a continuous stream with one pinned process per processor, rolling +quantiles, and a live KLL vs exact comparison, see +[the multi-process streaming KLL demo](./kll-windows.md). + +## Write KLL estimates to Prometheus and view them in Grafana + +The optional export sends the validated estimator output as OTLP/HTTP protobuf +straight to Prometheus. Prometheus stores the gauges and Grafana queries them. +It requires `curl` (7.76 or newer) on the machine running the demo, plus Docker +Compose for the bundled backend. No additional collector is required. + +From the repository root: + +```sh +docker compose -f demos/kll-grafana/compose.yaml up -d +cd asap-precompute-rs +cargo run --bin asap-otap-demo --features otap-engine -- \ + --prometheus-otlp-endpoint http://localhost:9090/api/v1/otlp/v1/metrics +``` + +Open [the KLL dashboard](http://localhost:3000/d/asap-kll) (anonymous viewer +access is enabled for this local demo). The dashboard provisions its Prometheus +datasource automatically and displays p50/p99 using: + +```promql +request_duration_estimate{quantile=~"0.5|0.99"} +``` + +The [Prometheus OTLP receiver](https://prometheus.io/docs/guides/opentelemetry/) +is enabled with `--web.enable-otlp-receiver`. Its default translation converts +`request.duration.estimate` to `request_duration_estimate` and retains the +`quantile` label. [Grafana provisioning](https://grafana.com/docs/grafana/latest/administration/provisioning/) +connects that datasource to the dashboard. + +This is a finite demo: one run writes one sample per quantile. Repeat the +command to add samples to the time series. Instant queries show the latest +sample for Prometheus's default five-minute lookback; the history panel shows +previous samples within the selected time range. Values use the demo input's +units (the sequential input is synthetic). The outbound copy uses the current +wall-clock timestamp, so the synthetic 1970-era window remains available in +`out.otlp` and debug traces while the dashboard shows the completed estimate +now. Original gauge values, resource attributes, and series labels are preserved; +window-boundary metadata is excluded from Prometheus labels to keep series stable. +Series labels stored on OTAP scopes (including `quantile`) are also copied to +gauge point attributes, which Prometheus uses to identify separate series. +Only the `kll` scenario supports this option. HTTP errors, invalid OTLP +responses, and partial rejection fail the command; no automatic retry can +silently duplicate a partially accepted write. + +To check the entire path after building the binary, run from the repository root: + +```sh +python3 demos/kll-grafana/verify.py asap-precompute-rs/target/debug/asap-otap-demo +``` + +The check compares the printed estimates with Prometheus results and Grafana's +queries through its configured datasource, and verifies the dashboard exists. +Set `PROMETHEUS_PORT` and `GRAFANA_PORT` when starting Compose if the default +ports are occupied, and supply matching `--prometheus` / `--grafana` URLs to +the check. Services bind to localhost and data persists in Compose volumes. +Stop them with `docker compose -f demos/kll-grafana/compose.yaml down`; add `-v` +when you want to remove the demo's stored data. + +## Before and after example + +Without the export option, the demo prints estimates and retains local OTLP +files and debug traces. It does not write samples to Prometheus: + +![Console output without Prometheus export](./images/kll-before-console.png) + +This image renders captured console output from a real run; process launch +lines and trace directory paths are omitted for readability. + +With `--prometheus-otlp-endpoint`, the same estimated values are stored as +separate quantile series and appear in the provisioned Grafana dashboard: + +![Grafana dashboard after exporting KLL estimates](./images/kll-after-grafana.png) + +The screenshot is from the running Grafana demo. This run produced p50 = 100 +and p99 = 198 for the synthetic sequential input. Prometheus and Grafana +queries returned those same values. The history panel uses Prometheus range +queries; displayed steps can repeat the latest sample between demo runs. diff --git a/docs/images/kll-after-grafana.png b/docs/images/kll-after-grafana.png new file mode 100644 index 0000000..9394e90 Binary files /dev/null and b/docs/images/kll-after-grafana.png differ diff --git a/docs/images/kll-before-console.png b/docs/images/kll-before-console.png new file mode 100644 index 0000000..c379be8 Binary files /dev/null and b/docs/images/kll-before-console.png differ diff --git a/docs/images/kll-streaming-grafana.png b/docs/images/kll-streaming-grafana.png new file mode 100644 index 0000000..3bb9747 Binary files /dev/null and b/docs/images/kll-streaming-grafana.png differ diff --git a/docs/images/kll-streaming-throughput.png b/docs/images/kll-streaming-throughput.png new file mode 100644 index 0000000..8999486 Binary files /dev/null and b/docs/images/kll-streaming-throughput.png differ diff --git a/docs/kll-windows.md b/docs/kll-windows.md new file mode 100644 index 0000000..ad70005 --- /dev/null +++ b/docs/kll-windows.md @@ -0,0 +1,224 @@ +# Continuous KLL quantiles: multi-process streaming demo + +The demo runs the exact and KLL pipelines side by side on the same traffic. +Every processor is its own OS process in its own container, pinned to one +CPU, and every hop is standard OTLP/HTTP over TCP. It is the +[sustained streaming benchmark](../benchmarks/STREAMING.md) topology, kept +running at one offered rate until Ctrl+C, with live metrics in Grafana. + +```mermaid +flowchart LR + GA["generator A"] --> BA["branch A
1 CPU"] + GB["generator B"] --> BB["branch B
1 CPU"] + BA -->|"OTLP/HTTP
KLL pane or sorted run"| M["merge
1 CPU"] + BB --> M + M --> E["estimate
1 CPU
tumbling + rolling p50/p99"] + E --> V["validating backend"] + V -. "control /stats" .-> D["run-streaming.py --demo
:9465/metrics"] + D --> P[Prometheus] --> G[Grafana] +``` + +Each branch builds one KLL sketch (`k=400`) per count-aligned window and +sends it as an ASAPv1 envelope; merge combines the two sources and estimate +answers p50/p99. Exact sends every sorted value instead. With +`--rolling-windows N`, estimate also merges the last N window sketches (exact +re-sorts the last N windows' values) for rolling p50/p99. The backend checks +every answer against the exact quantiles of the replayed corpus: exact must +match, KLL must be within 5%. + +## Run the demo + +Linux with Docker and Docker Compose. From the repository root: + +```sh +demos/kll-grafana/run-demo.sh # Ctrl+C stops it +demos/kll-grafana/run-demo.sh --demo-rate 100000 # signals/s per source +``` + +The script builds `asap-stream-bench`, the benchmark image, and the +Prometheus/Grafana stack, then runs both pipelines with 65,536-observation +windows, six-window rolling results, and two generator CPUs per source. Each +pipeline needs 2 + 4 + 1 CPUs. Extra arguments go to +`benchmarks/run-streaming.py`. Open +. + +The dashboard shows, per scenario: + +- backend-validated throughput, and processor CPU-seconds per million signals + (branches + merge + estimate); +- CPU cores, data-network egress and resident memory of each pinned process; +- tumbling and rolling p50/p99 against the exact reference, and their + relative error. + +Rates are averaged over the last 30 seconds; windows complete in bursts. Every +window replays the same deterministic corpus, so the quantile lines are flat +by design: the moving signals are throughput and resource use. + +![KLL vs exact multi-process streaming dashboard](images/kll-streaming-grafana.png) + +The screenshot is from a 56-core x86 server at 50,000 +signals/s per source. KLL keeps up with the offered 100,000 signals/s using +0.96 processor CPU-seconds per million signals. Exact uses 23 and is +back-pressured to about 68,000 signals/s although no exact process exceeds +0.8 CPU cores, consistent with every value crossing merge in 1,024-value +requests with one request in flight per hop. See +[benchmark results](#benchmark-results). + +## Benchmark results + +`benchmarks/run-streaming.py --scenarios raw exact kll --window-points 65536 +--repetitions 3` on a 56-core x86 server (Ubuntu 22.04), four generator +CPUs per source, one dedicated CPU per processor. Median backend-validated +signals/s over three runs (both sources combined): + +| Offered | Raw | Exact | KLL | +| ---: | ---: | ---: | ---: | +| 20,000 | 21,825 | 21,797 | 21,800 | +| 50,000 | 48,025 | 47,962 | 47,992 | +| 100,000 | 100,281 | 61,038 | 100,249 | +| 200,000 | 200,542 | 61,033 | 200,486 | +| 400,000 | 296,472 | 61,042 | 396,609 | +| 600,000 | 300,830 | 52,322 | 601,453 | +| 800,000 | 305,168 | 47,961 | 797,614 | + +(At 20,000 offered, a run's 30-second interval rounds to whole windows.) + +![Validated throughput vs offered load](images/kll-streaming-throughput.png) + +Exact plateaus near 61,000 signals/s; KLL follows the offered load to the +highest rate tested, and raw forwarding saturates near 300,000. By the +[capacity criteria](../benchmarks/CAPACITY_DESIGN.md) (delivery, backlog and +p99 window latency), the largest sustainable offered rate per source is 10,000 +for exact, 100,000 for raw and at least 400,000 for KLL. Exact already +misses the latency bound at 25,000 per source (p99 6.0 s). + +Summed over branch A, branch B, merge and estimate, at the same offered load: + +| Offered per source | Exact CPU cores | KLL CPU cores | Exact egress | KLL egress | +| ---: | ---: | ---: | ---: | ---: | +| 10,000 | 0.63 | 0.031 | 36.7 Mbit/s | 0.1 Mbit/s | +| 25,000 | 1.41 | 0.063 | 84.4 Mbit/s | 0.2 Mbit/s | +| 50,000 | 1.69 (back-pressured) | 0.117 | 101.6 Mbit/s | 0.4 Mbit/s | + +KLL answers stayed within the backend's 5% tolerance in every window +(p50 0.5%, p99 1.6% relative error); exact answers matched exactly. + +The Compose stack uses host networking so Prometheus, Grafana and the metrics +endpoint listen only on `127.0.0.1` (ports 9090, 3000 and 9465; +`PROMETHEUS_PORT` and `GRAFANA_PORT` change the first two). The finite +`asap-otap-demo` export described in the [demo guide](./demo-guide.md) still +pushes to the same Prometheus over OTLP. + +## Configure an OTAP rolling estimator + +This processor consumes scalar observations directly, or a stream of complete +ASAPv1 KLL panes of `window_slide` width: + +```yaml +type: urn:asap:processor:asap_sketches +config: + sketch_type: kll + encoding: Msgpack + window_size: 300s + window_slide: 5s + output_metric_name: request.duration.rolling_estimate + agg_id: 7 + sketch_params: { k: 400 } + transmit_sketch: false + quantiles: [0.5, 0.99] +``` + +An upstream pane creator uses `window_size: 5s`, `window_slide: 5s`, and +`transmit_sketch: true`. Equal size and slide produce disjoint tumbling panes +that can be merged downstream. Overlapping sliding windows only emit scalar +estimates, preventing their overlapping sketches from being counted twice. +Omitting `window_slide` retains the original tumbling runtime behavior. + +Pane boundaries align to Unix time and are half-open `[start, end)`. The +lookback must be an integer multiple of the pane width, and both durations +must be whole milliseconds. Sliding is currently KLL-only with full Msgpack +encoding, no delta transmission, and no late scalar updates to closed panes. +The first results use the observations available during warmup; a complete +lookback becomes available after the process has run for that duration. + +For upstream sketch input, the estimator waits for a complete pane rather than +letting a wall-clock timer race ahead of delivery. An upstream producer must +provide each complete pane once. If several producers contribute to a pane, +merge their contributions before publishing that complete pane to the sliding +processor; this change does not introduce a distributed completeness barrier +or duplicate-delivery detection. A delayed pane within retained history can be +merged, but a previously emitted range is not retroactively republished. + +## Query retained tumbling panes in Rust + +The host-neutral `KllWindowStore` supports raw observations and merging full +upstream panes, plus reusable aligned range queries: + +```rust +use std::time::Duration; +use asap_precompute_rs::{Encoding, PrecomputeConfig, SketchType}; +use asap_precompute_rs::kll_windows::KllWindowStore; + +let config = PrecomputeConfig { + sketch_type: SketchType::KLLSketch, + encoding: Encoding::Msgpack, + quantiles: vec![0.5, 0.99], + ..Default::default() +}; +let mut panes = KllWindowStore::new( + config, Duration::from_secs(5), Duration::from_secs(300), +)?; +// panes.observe(&observation)?; +// panes.merge_pane(&upstream_full_pane)?; +panes.advance(300_000); +let merged_sketches = panes.query(0, 300_000)?; +let quantile_gauges = panes.quantile_over_time(0, 300_000, &[0.5, 0.95, 0.99])?; +``` + +`query` returns one full merged KLL envelope per series; `quantile_over_time` +merges first and estimates the requested quantiles, which may vary per query. +Neither query consumes stored panes. Bounds must align to pane boundaries and +exclude open or expired panes. Unaligned ranges are rejected: a compacted +sketch cannot split a partially overlapping pane into exact observations. +Quantile accuracy remains approximate according to KLL's rank error, even +when temporal selection is exact at pane boundaries. + +The store retains at most `retention / pane_width` completed panes plus the +active pane. Idle `advance` calls expire old state, and a configured +`max_series` cap applies across retained panes (new series are rejected at the +cap). Retained state is in memory; restarts discard it. The ASAPv1 KLL wrapper +retains compacted items without a raw-observation replay history. + +This differs from applying PromQL `quantile_over_time` to an exported p99 +series: that would compute a quantile of earlier p99 estimates. Here the KLLs +are merged with their compaction weights, so the result estimates the quantile +of all original observations in the selected panes. + +## Verification + +With the image built and the Compose stack up (`run-demo.sh` does both): + +```sh +python3 demos/kll-grafana/verify-streaming.py +``` + +This runs both pipelines for 60 seconds and checks per-process throughput and +CPU series for all four processors, tumbling and rolling results within +tolerance (exact: zero error), and that Grafana's datasource and the +provisioned dashboard return the same series. Pass `--docker-command +'sudo -n docker'` when Docker needs sudo. + +The runtime tests exercise aligned range merging, compacted unequal-size pane +weights, series identity, expiration, idle gaps, late data, invalid ranges, +upstream completion timing, and configuration changes: + +```sh +cargo test --manifest-path asap-precompute-rs/Cargo.toml --test kll_windows +``` + +The benchmark model tests cover rolling estimates over retained windows: + +```sh +cargo test --manifest-path asap-precompute-rs/Cargo.toml \ + --features stream-benchmark --bin asap-stream-bench +```