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
96const BACKPRESSURE_THRESHOLD: f64 = 0.8;
97
98#[must_use]
99#[allow(clippy::cast_precision_loss)]
100pub(crate) fn is_backpressured(pending: usize, capacity: usize) -> bool {
101    capacity > 0 && (pending as f64 / capacity as f64) > BACKPRESSURE_THRESHOLD
102}
103
104#[must_use]
105#[allow(clippy::cast_precision_loss)]
106pub(crate) fn utilization(pending: usize, capacity: usize) -> f64 {
107    if capacity == 0 {
108        0.0
109    } else {
110        pending as f64 / capacity as f64
111    }
112}
113
114#[cfg(test)]
115mod tests {
116    use super::*;
117
118    #[test]
119    fn test_pipeline_state_display() {
120        assert_eq!(PipelineState::Created.to_string(), "Created");
121        assert_eq!(PipelineState::Starting.to_string(), "Starting");
122        assert_eq!(PipelineState::Running.to_string(), "Running");
123        assert_eq!(PipelineState::ShuttingDown.to_string(), "ShuttingDown");
124        assert_eq!(PipelineState::Stopped.to_string(), "Stopped");
125        assert_eq!(PipelineState::Faulted.to_string(), "Faulted");
126    }
127
128    #[test]
129    fn test_pipeline_state_equality() {
130        assert_eq!(PipelineState::Running, PipelineState::Running);
131        assert_ne!(PipelineState::Created, PipelineState::Running);
132    }
133
134    #[test]
135    fn test_backpressure_detection() {
136        // Empty buffer: not backpressured
137        assert!(!is_backpressured(0, 100));
138        // 50% full: not backpressured
139        assert!(!is_backpressured(50, 100));
140        // 80% full: not backpressured (threshold is >0.8, not >=)
141        assert!(!is_backpressured(80, 100));
142        // 81% full: backpressured
143        assert!(is_backpressured(81, 100));
144        // Full: backpressured
145        assert!(is_backpressured(100, 100));
146        // Zero capacity: not backpressured
147        assert!(!is_backpressured(0, 0));
148    }
149
150    #[test]
151    fn test_utilization() {
152        assert!((utilization(0, 100) - 0.0).abs() < f64::EPSILON);
153        assert!((utilization(50, 100) - 0.5).abs() < f64::EPSILON);
154        assert!((utilization(100, 100) - 1.0).abs() < f64::EPSILON);
155        assert!((utilization(0, 0) - 0.0).abs() < f64::EPSILON);
156    }
157
158    #[test]
159    fn test_pipeline_metrics_clone() {
160        let m = PipelineMetrics {
161            total_events_ingested: 100,
162            total_events_emitted: 50,
163            total_events_dropped: 0,
164            total_cycles: 10,
165            total_batches: 5,
166            uptime: Duration::from_secs(60),
167            state: PipelineState::Running,
168            source_count: 2,
169            stream_count: 1,
170            sink_count: 1,
171            pipeline_watermark: i64::MIN,
172            mv_updates: 0,
173            mv_bytes_stored: 0,
174        };
175        let m2 = m.clone();
176        assert_eq!(m2.total_events_ingested, 100);
177        assert_eq!(m2.state, PipelineState::Running);
178    }
179
180    #[test]
181    fn test_source_metrics_debug() {
182        let m = SourceMetrics {
183            name: "trades".to_string(),
184            total_events: 1000,
185            pending: 50,
186            capacity: 1024,
187            is_backpressured: false,
188            watermark: 12345,
189            utilization: 0.05,
190        };
191        let dbg = format!("{m:?}");
192        assert!(dbg.contains("trades"));
193        assert!(dbg.contains("1000"));
194    }
195
196    #[test]
197    fn test_stream_metrics_with_sql() {
198        let m = StreamMetrics {
199            name: "avg_price".to_string(),
200            total_events: 500,
201            sql: Some("SELECT symbol, AVG(price) FROM trades GROUP BY symbol".to_string()),
202        };
203        assert_eq!(
204            m.sql.as_deref(),
205            Some("SELECT symbol, AVG(price) FROM trades GROUP BY symbol")
206        );
207    }
208}
laminar_db/metrics.rs

laminar_db/
metrics.rs
