laminar_db/
metrics.rs

1//! Pipeline observability metrics.
2
3use std::time::Duration;
4
5/// The state of a streaming pipeline.
6#[derive(Debug, Clone, Copy, PartialEq, Eq)]
7pub enum PipelineState {
8    /// Created but not started.
9    Created,
10    /// Starting.
11    Starting,
12    /// Processing events.
13    Running,
14    /// Gracefully shutting down.
15    ShuttingDown,
16    /// Stopped.
17    Stopped,
18    /// Compute thread crashed (operator panic); recoverable via restart.
19    Faulted,
20}
21
22impl std::fmt::Display for PipelineState {
23    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
24        match self {
25            Self::Created => write!(f, "Created"),
26            Self::Starting => write!(f, "Starting"),
27            Self::Running => write!(f, "Running"),
28            Self::ShuttingDown => write!(f, "ShuttingDown"),
29            Self::Stopped => write!(f, "Stopped"),
30            Self::Faulted => write!(f, "Faulted"),
31        }
32    }
33}
34
35/// Pipeline-wide metrics snapshot.
36#[derive(Debug, Clone)]
37pub struct PipelineMetrics {
38    /// Events ingested.
39    pub total_events_ingested: u64,
40    /// Events emitted.
41    pub total_events_emitted: u64,
42    /// Events dropped.
43    pub total_events_dropped: u64,
44    /// Cycles.
45    pub total_cycles: u64,
46    /// Batches.
47    pub total_batches: u64,
48    /// Uptime.
49    pub uptime: Duration,
50    /// State.
51    pub state: PipelineState,
52    /// Sources.
53    pub source_count: usize,
54    /// Streams.
55    pub stream_count: usize,
56    /// Sinks.
57    pub sink_count: usize,
58    /// Min watermark across all sources.
59    pub pipeline_watermark: i64,
60    /// MV updates.
61    pub mv_updates: u64,
62    /// Approximate MV bytes.
63    pub mv_bytes_stored: u64,
64}
65
66/// Metrics for a single registered source.
67#[derive(Debug, Clone)]
68pub struct SourceMetrics {
69    /// Name.
70    pub name: String,
71    /// Total events (sequence number).
72    pub total_events: u64,
73    /// Buffered events.
74    pub pending: usize,
75    /// Capacity.
76    pub capacity: usize,
77    /// >80% full.
78    pub is_backpressured: bool,
79    /// Watermark.
80    pub watermark: i64,
81    /// 0.0..1.0.
82    pub utilization: f64,
83}
84
85/// Metrics for a single registered stream.
86#[derive(Debug, Clone)]
87pub struct StreamMetrics {
88    /// Name.
89    pub name: String,
90    /// Total rows emitted by this stream since it started.
91    pub total_events: u64,
92    /// Defining SQL query.
93    pub sql: Option<String>,
94}
95
96/// Process-local, bounded-cardinality health snapshot for committed cluster subscriptions.
97#[cfg(feature = "cluster")]
98#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
99pub struct ClusterSubscriptionOutputHealth {
100    /// Readers currently attached to this gateway.
101    pub active_readers: u64,
102    /// In-memory output bytes awaiting checkpoint disposition.
103    pub pending_bytes: u64,
104    /// Bytes reachable from retained committed history.
105    pub retained_bytes: u64,
106    /// Grace-held unreachable bytes found by the latest orphan scan.
107    pub orphan_bytes: u64,
108    /// Failed opens since this process started.
109    pub open_failures: u64,
110    /// Segment write failures since this process started.
111    pub segment_write_failures: u64,
112    /// Manifest failures since this process started.
113    pub manifest_failures: u64,
114    /// Integrity failures since this process started.
115    pub integrity_failures: u64,
116    /// Stale writer rejections since this process started.
117    pub stale_writer_rejections: u64,
118    /// Partition sequence gaps since this process started.
119    pub sequence_gaps: u64,
120    /// Bounded-lag gateway disconnects since this process started.
121    pub lag_disconnects: u64,
122}
123
124const BACKPRESSURE_THRESHOLD: f64 = 0.8;
125
126#[must_use]
127#[allow(clippy::cast_precision_loss)]
128pub(crate) fn is_backpressured(pending: usize, capacity: usize) -> bool {
129    capacity > 0 && (pending as f64 / capacity as f64) > BACKPRESSURE_THRESHOLD
130}
131
132#[must_use]
133#[allow(clippy::cast_precision_loss)]
134pub(crate) fn utilization(pending: usize, capacity: usize) -> f64 {
135    if capacity == 0 {
136        0.0
137    } else {
138        pending as f64 / capacity as f64
139    }
140}
141
142#[cfg(test)]
143mod tests {
144    use super::*;
145
146    #[test]
147    fn test_pipeline_state_display() {
148        assert_eq!(PipelineState::Created.to_string(), "Created");
149        assert_eq!(PipelineState::Starting.to_string(), "Starting");
150        assert_eq!(PipelineState::Running.to_string(), "Running");
151        assert_eq!(PipelineState::ShuttingDown.to_string(), "ShuttingDown");
152        assert_eq!(PipelineState::Stopped.to_string(), "Stopped");
153        assert_eq!(PipelineState::Faulted.to_string(), "Faulted");
154    }
155
156    #[test]
157    fn test_pipeline_state_equality() {
158        assert_eq!(PipelineState::Running, PipelineState::Running);
159        assert_ne!(PipelineState::Created, PipelineState::Running);
160    }
161
162    #[test]
163    fn test_backpressure_detection() {
164        // Empty buffer: not backpressured
165        assert!(!is_backpressured(0, 100));
166        // 50% full: not backpressured
167        assert!(!is_backpressured(50, 100));
168        // 80% full: not backpressured (threshold is >0.8, not >=)
169        assert!(!is_backpressured(80, 100));
170        // 81% full: backpressured
171        assert!(is_backpressured(81, 100));
172        // Full: backpressured
173        assert!(is_backpressured(100, 100));
174        // Zero capacity: not backpressured
175        assert!(!is_backpressured(0, 0));
176    }
177
178    #[test]
179    fn test_utilization() {
180        assert!((utilization(0, 100) - 0.0).abs() < f64::EPSILON);
181        assert!((utilization(50, 100) - 0.5).abs() < f64::EPSILON);
182        assert!((utilization(100, 100) - 1.0).abs() < f64::EPSILON);
183        assert!((utilization(0, 0) - 0.0).abs() < f64::EPSILON);
184    }
185
186    #[test]
187    fn test_pipeline_metrics_clone() {
188        let m = PipelineMetrics {
189            total_events_ingested: 100,
190            total_events_emitted: 50,
191            total_events_dropped: 0,
192            total_cycles: 10,
193            total_batches: 5,
194            uptime: Duration::from_secs(60),
195            state: PipelineState::Running,
196            source_count: 2,
197            stream_count: 1,
198            sink_count: 1,
199            pipeline_watermark: i64::MIN,
200            mv_updates: 0,
201            mv_bytes_stored: 0,
202        };
203        let m2 = m.clone();
204        assert_eq!(m2.total_events_ingested, 100);
205        assert_eq!(m2.state, PipelineState::Running);
206    }
207
208    #[test]
209    fn test_source_metrics_debug() {
210        let m = SourceMetrics {
211            name: "trades".to_string(),
212            total_events: 1000,
213            pending: 50,
214            capacity: 1024,
215            is_backpressured: false,
216            watermark: 12345,
217            utilization: 0.05,
218        };
219        let dbg = format!("{m:?}");
220        assert!(dbg.contains("trades"));
221        assert!(dbg.contains("1000"));
222    }
223
224    #[test]
225    fn test_stream_metrics_with_sql() {
226        let m = StreamMetrics {
227            name: "avg_price".to_string(),
228            total_events: 500,
229            sql: Some("SELECT symbol, AVG(price) FROM trades GROUP BY symbol".to_string()),
230        };
231        assert_eq!(
232            m.sql.as_deref(),
233            Some("SELECT symbol, AVG(price) FROM trades GROUP BY symbol")
234        );
235    }
236}
laminar_db/metrics.rs

laminar_db/
metrics.rs
