From 9688f44efbc842da3e163bca7cfb13f02564faa0 Mon Sep 17 00:00:00 2001 From: Karakatiza666 Date: Thu, 16 Jul 2026 13:58:38 +0000 Subject: [PATCH 1/2] [pipeline-manager] Add connectors latency to time series stream [web-console] Add pipeline latency graph to Performance tab Signed-off-by: Karakatiza666 --- crates/adapters/src/controller.rs | 5 + crates/adapters/src/controller/stats.rs | 120 +++++++++ crates/feldera-types/src/time_series.rs | 18 ++ crates/storage/src/histogram.rs | 66 ++++- .../pipelines/PipelineLatencyGraph.svelte | 235 ++++++++++++++++++ .../pipelines/editor/TabPerformance.svelte | 17 +- .../src/lib/functions/format.spec.ts | 29 ++- .../web-console/src/lib/functions/format.ts | 24 ++ .../src/lib/functions/pipelineMetrics.spec.ts | 67 ++++- .../src/lib/functions/pipelineMetrics.ts | 37 +++ .../src/lib/services/manager/index.ts | 8 +- .../src/lib/services/manager/sdk.gen.ts | 24 -- .../src/lib/services/manager/types.gen.ts | 166 ++++++++++--- .../src/lib/types/pipelineManager.ts | 18 ++ openapi.json | 28 +++ 15 files changed, 790 insertions(+), 72 deletions(-) create mode 100644 js-packages/web-console/src/lib/components/layout/pipelines/PipelineLatencyGraph.svelte diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index e0a99ec77f5..f14cf32b51d 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -5470,6 +5470,7 @@ impl StatisticsThread { }; // Update time series.. + let latency = controller_status.latency_percentiles(); let sample = SampleStatistics { time: Utc::now(), total_processed_records: controller_status @@ -5477,6 +5478,10 @@ impl StatisticsThread { .num_total_processed_records(), memory_bytes: process_rss_bytes().unwrap_or_default(), storage_bytes, + processing_latency_p50_micros: latency.processing_p50, + processing_latency_p99_micros: latency.processing_p99, + completion_latency_p50_micros: latency.completion_p50, + completion_latency_p99_micros: latency.completion_p99, }; let mut time_series = controller_status.time_series.lock().unwrap(); if time_series.len() >= 60 { diff --git a/crates/adapters/src/controller/stats.rs b/crates/adapters/src/controller/stats.rs index a021c8b42aa..936b88399c9 100644 --- a/crates/adapters/src/controller/stats.rs +++ b/crates/adapters/src/controller/stats.rs @@ -751,6 +751,47 @@ impl ControllerStatus { self.inputs.read_recursive() } + /// Latency percentiles (microseconds) across input connectors, for the + /// time-series graph. Each connector contributes its own median latency, + /// we then take p50 and p99 across connectors, for both + /// processing (ingest to processed) and completion (ingest to all outputs). + /// `None` when no connector has samples. + pub fn latency_percentiles(&self) -> LatencyPercentiles { + let inputs = self.input_status(); + let mut processing = Vec::with_capacity(inputs.len()); + let mut completion = Vec::with_capacity(inputs.len()); + for ep in inputs.values() { + if let Some(median) = ep + .metrics + .processing_latency_micros_histogram + .lock() + .unwrap() + .snapshot() + .quantile(0.5) + { + processing.push(median); + } + if let Some(median) = ep + .metrics + .completion_latency_micros_histogram + .lock() + .unwrap() + .snapshot() + .quantile(0.5) + { + completion.push(median); + } + } + processing.sort_unstable(); + completion.sort_unstable(); + LatencyPercentiles { + processing_p50: percentile_of_sorted(&processing, 0.5), + processing_p99: percentile_of_sorted(&processing, 0.99), + completion_p50: percentile_of_sorted(&completion, 0.5), + completion_p99: percentile_of_sorted(&completion, 0.99), + } + } + /// Register connector-specific metrics for an input endpoint. pub fn set_input_custom_metrics( &self, @@ -1525,6 +1566,25 @@ impl InputEndpointMetrics { } } +/// Latency percentiles for the time series, in microseconds. See +/// [`ControllerStatus::latency_percentiles`]. +#[derive(Clone, Copy, Debug, Default)] +pub struct LatencyPercentiles { + pub processing_p50: Option, + pub processing_p99: Option, + pub completion_p50: Option, + pub completion_p99: Option, +} + +/// Returns `q`-quantile of an ascending-sorted slice (ceil-rank). `None` if empty. +fn percentile_of_sorted(sorted: &[u64], q: f64) -> Option { + if sorted.is_empty() { + return None; + } + let rank = ((q.clamp(0.0, 1.0) * sorted.len() as f64).ceil() as usize).clamp(1, sorted.len()); + Some(sorted[rank - 1]) +} + #[derive(Debug)] pub struct StepResults { pub amt: BufferSize, @@ -2800,3 +2860,63 @@ impl OutputEndpointStatus { self.metrics.total_processed_steps.load(Ordering::Acquire) } } + +#[cfg(test)] +mod latency_tests { + use super::percentile_of_sorted; + + #[test] + fn percentile_empty() { + assert_eq!(percentile_of_sorted(&[], 0.5), None); + assert_eq!(percentile_of_sorted(&[], 0.99), None); + } + + #[test] + fn percentile_single() { + assert_eq!(percentile_of_sorted(&[42], 0.5), Some(42)); + assert_eq!(percentile_of_sorted(&[42], 0.99), Some(42)); + } + + #[test] + fn percentile_uniform() { + let v: Vec = (1..=100).collect(); + // ceil(0.5*100)=50 -> 1-based rank 50 -> value 50. + assert_eq!(percentile_of_sorted(&v, 0.5), Some(50)); + // ceil(0.99*100)=99 -> value 99. + assert_eq!(percentile_of_sorted(&v, 0.99), Some(99)); + } + + #[test] + fn percentile_few_slow_connectors() { + // 97 fast connectors at 1ms, 3 slow at 500ms (values in micros). + let mut v = vec![1_000u64; 97]; + v.extend([500_000, 500_000, 500_000]); + v.sort_unstable(); + // p50 stays with the fast majority. + assert_eq!(percentile_of_sorted(&v, 0.5), Some(1_000)); + // p99 (rank 99) reaches the slow tail: surfaces the laggards. + assert_eq!(percentile_of_sorted(&v, 0.99), Some(500_000)); + } + + #[test] + fn percentile_broad_degradation() { + // 80 fast at 1ms, 20 slow at 100ms: ~20% degraded. + let mut v = vec![1_000u64; 80]; + v.extend(std::iter::repeat(100_000).take(20)); + v.sort_unstable(); + // Both p50 (typical) stays fast and p99 catches the slow band. + assert_eq!(percentile_of_sorted(&v, 0.5), Some(1_000)); + assert_eq!(percentile_of_sorted(&v, 0.99), Some(100_000)); + } + + #[test] + fn percentile_single_lone_outlier_escapes_p99() { + // Documents the known limit: one slow connector of 100 sits above p99. + let mut v = vec![1_000u64; 99]; + v.push(900_000); + v.sort_unstable(); + // p99 (rank 99) is still the fast majority; only max would catch it. + assert_eq!(percentile_of_sorted(&v, 0.99), Some(1_000)); + assert_eq!(percentile_of_sorted(&v, 1.0), Some(900_000)); + } +} diff --git a/crates/feldera-types/src/time_series.rs b/crates/feldera-types/src/time_series.rs index b3c9831aa24..143b5da0a63 100644 --- a/crates/feldera-types/src/time_series.rs +++ b/crates/feldera-types/src/time_series.rs @@ -34,4 +34,22 @@ pub struct SampleStatistics { /// Storage usage in bytes. #[serde(rename = "s")] pub storage_bytes: u64, + + /// Processing latency (ingest to circuit-processed), microseconds: + /// p50 across connectors of each connector's median. Absent without samples. + #[serde(rename = "pp50", skip_serializing_if = "Option::is_none", default)] + pub processing_latency_p50_micros: Option, + + /// Processing latency, microseconds: p99 across connectors. + #[serde(rename = "pp99", skip_serializing_if = "Option::is_none", default)] + pub processing_latency_p99_micros: Option, + + /// Completion latency (ingest to all outputs pushed), microseconds: + /// p50 across connectors. + #[serde(rename = "cp50", skip_serializing_if = "Option::is_none", default)] + pub completion_latency_p50_micros: Option, + + /// Completion latency, microseconds: p99 across connectors. + #[serde(rename = "cp99", skip_serializing_if = "Option::is_none", default)] + pub completion_latency_p99_micros: Option, } diff --git a/crates/storage/src/histogram.rs b/crates/storage/src/histogram.rs index f4de7018bd7..fc777592c9d 100644 --- a/crates/storage/src/histogram.rs +++ b/crates/storage/src/histogram.rs @@ -115,6 +115,25 @@ impl ExponentialHistogramSnapshot { pub fn sum(&self) -> u64 { self.sum } + + /// Approximate `q`-quantile (`q` in `0.0..=1.0`), as the lower bound of the + /// containing bucket. `None` if empty. + pub fn quantile(&self, q: f64) -> Option { + let total: u64 = self.buckets.iter().sum(); + if total == 0 { + return None; + } + let rank = ((q.clamp(0.0, 1.0) * total as f64).ceil() as u64).clamp(1, total); + let mut cumulative = 0u64; + for (index, count) in self.buckets.iter().enumerate() { + cumulative += *count; + if cumulative >= rank { + return Some(*bucket_to_range(index).start()); + } + } + // Unreachable: `cumulative` reaches `total >= rank` in the loop. + Some(*bucket_to_range(N_BUCKETS - 1).start()) + } } pub struct Bucket { @@ -278,7 +297,52 @@ impl SlidingHistogram { #[cfg(test)] mod test { - use crate::histogram::{N_BUCKETS, bucket_to_range, number_to_bucket}; + use crate::histogram::{ExponentialHistogram, N_BUCKETS, bucket_to_range, number_to_bucket}; + + #[test] + fn quantile_empty() { + let hist = ExponentialHistogram::new(); + assert_eq!(hist.snapshot().quantile(0.5), None); + assert_eq!(hist.snapshot().quantile(0.99), None); + } + + #[test] + fn quantile_single_sample() { + let hist = ExponentialHistogram::new(); + hist.record(42u64); + // 42 lands in the bucket for [40, 49]; the lower bound is 40. + assert_eq!(hist.snapshot().quantile(0.0), Some(40)); + assert_eq!(hist.snapshot().quantile(0.5), Some(40)); + assert_eq!(hist.snapshot().quantile(1.0), Some(40)); + } + + #[test] + fn quantile_all_same_bucket() { + let hist = ExponentialHistogram::new(); + for _ in 0..100 { + hist.record(5u64); + } + // Values 0..=9 each occupy their own bucket, so 5 is exact. + assert_eq!(hist.snapshot().quantile(0.5), Some(5)); + assert_eq!(hist.snapshot().quantile(0.99), Some(5)); + } + + #[test] + fn quantile_known_distribution() { + let hist = ExponentialHistogram::new(); + // 99 fast samples (bucket [0,0]) and 1 slow sample (bucket [900,999]). + for _ in 0..99 { + hist.record(0u64); + } + hist.record(950u64); + let snap = hist.snapshot(); + // p50 stays with the fast majority. + assert_eq!(snap.quantile(0.5), Some(0)); + // p99 (ceil(0.99*100)=99th value) is still the last fast sample. + assert_eq!(snap.quantile(0.99), Some(0)); + // p100 reaches the lone slow sample: bucket [900, 999], lower bound 900. + assert_eq!(snap.quantile(1.0), Some(900)); + } #[test] fn buckets() { diff --git a/js-packages/web-console/src/lib/components/layout/pipelines/PipelineLatencyGraph.svelte b/js-packages/web-console/src/lib/components/layout/pipelines/PipelineLatencyGraph.svelte new file mode 100644 index 00000000000..097c79d0b0f --- /dev/null +++ b/js-packages/web-console/src/lib/components/layout/pipelines/PipelineLatencyGraph.svelte @@ -0,0 +1,235 @@ + + +
+ +
+
+ {#if compactTitle} + Latency: {formatDuration(currentProcessing)} + {formatDuration(completionDelta)} + {:else} + Latency: {formatDuration(currentProcessing)} processing + {formatDuration(completionDelta)} completion + {/if} +
+ +
+ + + {#key pipelineName} + (ref = init(dom, theme, opts))} {options} /> + {/key} +
diff --git a/js-packages/web-console/src/lib/components/pipelines/editor/TabPerformance.svelte b/js-packages/web-console/src/lib/components/pipelines/editor/TabPerformance.svelte index 2dd2a0694b6..9b3103b3e0e 100644 --- a/js-packages/web-console/src/lib/components/pipelines/editor/TabPerformance.svelte +++ b/js-packages/web-console/src/lib/components/pipelines/editor/TabPerformance.svelte @@ -7,6 +7,7 @@