Add multi-process KLL vs exact streaming demo with Prometheus and Grafana

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

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",

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"

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<String>,
     output_dir: Option<PathBuf>,
     result_manifest: Option<PathBuf>,
     points_per_source: u64,
@@ -766,6 +770,7 @@ struct ParentOptions {
 
 fn parent_options(args: &[std::ffi::OsString]) -> Result<ParentOptions, String> {
     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<ParentOptions, String>
             .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<ParentOptions, String>
         }
         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(),

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<Self, String> {
@@ -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<u64, u64>,
+    latest_results: Vec<model::ResultPoint>,
 }
 struct Shared {
     stats: Mutex<Stats>,
@@ -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<OtapPdata> 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<ExporterWrapper<OtapPdata>, 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,

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<u64>) -> Result<Vec<u8>, 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<u64>) -> 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());
+    }
+}