laminar_db/
metrics_api.rs

1//! Pipeline metrics and state query methods for `LaminarDB`.
2//!
3//! Reopens `impl LaminarDB` to keep the main `db.rs` focused on dispatch.
4
5use std::sync::Arc;
6
7use crate::db::{DbState, LaminarDB};
8use crate::error::DbError;
9
10impl LaminarDB {
11    /// Time elapsed since the database was created.
12    #[must_use]
13    pub fn uptime(&self) -> std::time::Duration {
14        self.start_time.elapsed()
15    }
16
17    /// Inject prometheus engine metrics. Called once at startup before `start()`.
18    pub fn set_engine_metrics(&self, metrics: Arc<crate::engine_metrics::EngineMetrics>) {
19        *self.engine_metrics.lock() = Some(metrics);
20    }
21
22    /// Inject a shared Prometheus registry for connector-level metrics.
23    ///
24    /// Called once at startup, after the registry is constructed but before
25    /// `start()`. Connectors created after this call will register their
26    /// metrics on this registry so they appear in the scrape output.
27    ///
28    /// # Errors
29    ///
30    /// Returns an error after the database leaves `Created` or when a registry
31    /// was already installed.
32    pub fn set_prometheus_registry(
33        &self,
34        registry: Arc<prometheus::Registry>,
35    ) -> Result<(), DbError> {
36        let _startup = self.startup_attempt.lock();
37        let mut slot = self.prometheus_registry.lock();
38        let state = DbState::load(&self.state);
39        if state != DbState::Created {
40            return Err(DbError::InvalidOperation(format!(
41                "Prometheus registry can only be installed while the database is Created, not {state:?}"
42            )));
43        }
44        if slot.is_some() {
45            return Err(DbError::InvalidOperation(
46                "Prometheus registry is already installed".into(),
47            ));
48        }
49        *slot = Some(registry);
50        Ok(())
51    }
52
53    /// Get the engine metrics if set.
54    #[must_use]
55    pub fn engine_metrics(&self) -> Option<Arc<crate::engine_metrics::EngineMetrics>> {
56        self.engine_metrics.lock().clone()
57    }
58
59    /// Read process-local committed subscription output-log health without storage I/O.
60    #[cfg(feature = "cluster")]
61    #[must_use]
62    pub fn cluster_subscription_output_health(
63        &self,
64    ) -> Option<crate::metrics::ClusterSubscriptionOutputHealth> {
65        let metrics = self.engine_metrics.lock();
66        let metrics = &metrics.as_ref()?.cluster_subscription;
67        Some(crate::metrics::ClusterSubscriptionOutputHealth {
68            active_readers: nonnegative_gauge(metrics.active_readers.get()),
69            pending_bytes: nonnegative_gauge(metrics.pending_bytes.get()),
70            retained_bytes: nonnegative_gauge(metrics.retained_bytes.get()),
71            orphan_bytes: nonnegative_gauge(metrics.orphan_bytes.get()),
72            open_failures: metrics.open_failures_total.get(),
73            segment_write_failures: metrics.segment_write_failures_total.get(),
74            manifest_failures: metrics.manifest_failures_total.get(),
75            integrity_failures: metrics.integrity_failures_total.get(),
76            stale_writer_rejections: metrics.stale_writer_rejections_total.get(),
77            sequence_gaps: metrics.sequence_gaps_total.get(),
78            lag_disconnects: metrics.gateway_lag_disconnects_total.get(),
79        })
80    }
81
82    /// Read one bounded, process-bound page of local cluster checkpoint barrier timings.
83    ///
84    /// This is an in-memory diagnostic read. It performs no checkpoint, state-backend, network,
85    /// filesystem, or object-store operation. A continuation cursor is the pair of
86    /// `expected_process` and `after_sequence`; callers must retain both from the prior page.
87    /// Snapshot counters may advance immediately after this call, so certification must finish
88    /// with a quiescent repeated read rather than treating one page as a stable cut.
89    ///
90    /// # Errors
91    /// Returns an error when a continuation omits or mismatches its process identity, live process
92    /// authority cannot be sampled without waiting, authority changes during the read, or the
93    /// bounded ledger snapshot rejects its cursor/page request.
94    #[cfg(feature = "cluster")]
95    pub fn checkpoint_barrier_timing_snapshot(
96        &self,
97        expected_process: Option<laminar_core::cluster::control::LocalProcessAuthorityIdentity>,
98        after_sequence: u64,
99        limit: usize,
100    ) -> Result<
101        crate::checkpoint_timing::CheckpointBarrierTimingPage,
102        crate::checkpoint_timing::CheckpointBarrierTimingReadError,
103    > {
104        use crate::checkpoint_timing::{
105            CheckpointBarrierTimingPage, CheckpointBarrierTimingReadError,
106        };
107
108        if after_sequence != 0 && expected_process.is_none() {
109            return Err(CheckpointBarrierTimingReadError::ProcessIdentityRequired);
110        }
111        let controller = self
112            .cluster_controller
113            .try_lock()
114            .and_then(|controller| controller.clone())
115            .ok_or(CheckpointBarrierTimingReadError::ProcessIdentityUnavailable)?;
116        let before = controller
117            .try_live_local_process_authority_identity()
118            .map_err(|_| CheckpointBarrierTimingReadError::ProcessIdentityUnavailable)?;
119        if let Some(expected) = expected_process {
120            if expected != before {
121                return Err(CheckpointBarrierTimingReadError::ProcessIdentityMismatch {
122                    expected,
123                    actual: before,
124                });
125            }
126        }
127
128        let snapshot = self
129            .checkpoint_barrier_timings
130            .snapshot_after(after_sequence, limit)?;
131        let after = controller
132            .try_live_local_process_authority_identity()
133            .map_err(|_| CheckpointBarrierTimingReadError::ProcessIdentityUnavailable)?;
134        if before != after {
135            return Err(CheckpointBarrierTimingReadError::ProcessIdentityChanged { before, after });
136        }
137        if let Some(actual) = snapshot.process {
138            if actual != before {
139                return Err(CheckpointBarrierTimingReadError::LedgerProcessMismatch {
140                    expected: before,
141                    actual,
142                });
143            }
144        }
145        debug_assert!(snapshot
146            .records
147            .iter()
148            .all(|record| record.process == before));
149        Ok(CheckpointBarrierTimingPage {
150            process: before,
151            snapshot,
152        })
153    }
154
155    /// Get the current pipeline state as a string.
156    pub fn pipeline_state(&self) -> &'static str {
157        match DbState::load(&self.state) {
158            DbState::Created => "Created",
159            DbState::Starting => "Starting",
160            DbState::Running => "Running",
161            DbState::ShuttingDown => "ShuttingDown",
162            DbState::Stopped => "Stopped",
163            DbState::Faulted => "Faulted",
164        }
165    }
166
167    /// The last runtime-fault or terminal resource-exhaustion reason. Populated when state is
168    /// `Faulted`; cleared on a clean start.
169    #[must_use]
170    pub fn last_fault(&self) -> Option<String> {
171        self.last_fault.lock().clone()
172    }
173
174    /// Get a pipeline-wide metrics snapshot.
175    ///
176    /// Reads prometheus engine metrics and catalog sizes to produce a
177    /// point-in-time view of pipeline health.
178    #[must_use]
179    #[allow(clippy::cast_sign_loss)]
180    pub fn metrics(&self) -> crate::metrics::PipelineMetrics {
181        let guard = self.engine_metrics.lock();
182        let (ingested, emitted, dropped, cycles, batches, mv_updates, mv_bytes) =
183            if let Some(ref m) = *guard {
184                (
185                    m.events_ingested.get(),
186                    m.events_emitted.get(),
187                    m.events_dropped.get(),
188                    m.cycles.get(),
189                    m.batches.get(),
190                    m.mv_updates.get(),
191                    m.mv_bytes_stored.get() as u64,
192                )
193            } else {
194                (0, 0, 0, 0, 0, 0, 0)
195            };
196        crate::metrics::PipelineMetrics {
197            total_events_ingested: ingested,
198            total_events_emitted: emitted,
199            total_events_dropped: dropped,
200            total_cycles: cycles,
201            total_batches: batches,
202            uptime: self.start_time.elapsed(),
203            state: self.pipeline_state_enum(),
204            source_count: self.catalog.list_sources().len(),
205            stream_count: self.catalog.list_streams().len(),
206            sink_count: self.catalog.list_sinks().len(),
207            pipeline_watermark: self.pipeline_watermark(),
208            mv_updates,
209            mv_bytes_stored: mv_bytes,
210        }
211    }
212
213    /// Get metrics for a single source by name.
214    #[must_use]
215    pub fn source_metrics(&self, name: &str) -> Option<crate::metrics::SourceMetrics> {
216        let entry = self.catalog.get_source(name)?;
217        let pending = entry.source.pending();
218        let capacity = entry.source.capacity();
219        Some(crate::metrics::SourceMetrics {
220            name: entry.name.clone(),
221            total_events: entry.source.sequence(),
222            pending,
223            capacity,
224            is_backpressured: crate::metrics::is_backpressured(pending, capacity),
225            watermark: entry.source.current_watermark(),
226            utilization: crate::metrics::utilization(pending, capacity),
227        })
228    }
229
230    /// Get metrics for all registered sources.
231    #[must_use]
232    pub fn all_source_metrics(&self) -> Vec<crate::metrics::SourceMetrics> {
233        self.catalog
234            .list_sources()
235            .iter()
236            .filter_map(|name| self.source_metrics(name))
237            .collect()
238    }
239
240    /// Get metrics for a single stream by name.
241    #[must_use]
242    pub fn stream_metrics(&self, name: &str) -> Option<crate::metrics::StreamMetrics> {
243        let entry = self.catalog.get_stream_entry(name)?;
244        let sql = self
245            .connector_manager
246            .lock()
247            .streams()
248            .get(name)
249            .map(|reg| reg.query_sql.clone());
250        Some(crate::metrics::StreamMetrics {
251            name: entry.name.clone(),
252            total_events: entry.emitted_rows(),
253            sql,
254        })
255    }
256
257    /// Get metrics for all registered streams.
258    #[must_use]
259    pub fn all_stream_metrics(&self) -> Vec<crate::metrics::StreamMetrics> {
260        self.catalog
261            .list_streams()
262            .iter()
263            .filter_map(|name| self.stream_metrics(name))
264            .collect()
265    }
266
267    /// Get the total number of events processed (ingested + emitted).
268    #[must_use]
269    pub fn total_events_processed(&self) -> u64 {
270        let guard = self.engine_metrics.lock();
271        if let Some(ref m) = *guard {
272            m.events_ingested.get() + m.events_emitted.get()
273        } else {
274            0
275        }
276    }
277
278    /// Returns the global pipeline watermark (minimum across all source watermarks).
279    ///
280    /// Returns `i64::MIN` if no watermark-enabled sources exist or no events
281    /// have been processed.
282    #[must_use]
283    pub fn pipeline_watermark(&self) -> i64 {
284        self.pipeline_watermark
285            .load(std::sync::atomic::Ordering::Relaxed)
286    }
287
288    pub(crate) fn pipeline_state_enum(&self) -> crate::metrics::PipelineState {
289        match DbState::load(&self.state) {
290            DbState::Created => crate::metrics::PipelineState::Created,
291            DbState::Starting => crate::metrics::PipelineState::Starting,
292            DbState::Running => crate::metrics::PipelineState::Running,
293            DbState::ShuttingDown => crate::metrics::PipelineState::ShuttingDown,
294            DbState::Stopped => crate::metrics::PipelineState::Stopped,
295            DbState::Faulted => crate::metrics::PipelineState::Faulted,
296        }
297    }
298
299    /// Cancel a running query by ID.
300    ///
301    /// Marks the query as inactive in the catalog. Future subscription
302    /// polls for this query will receive no more data.
303    ///
304    /// # Errors
305    ///
306    /// Returns `DbError` if the query is not found.
307    pub fn cancel_query(&self, query_id: u64) -> Result<(), DbError> {
308        if self.catalog.deactivate_query(query_id) {
309            Ok(())
310        } else {
311            Err(DbError::QueryNotFound(query_id.to_string()))
312        }
313    }
314
315    /// Get the number of registered sources.
316    pub fn source_count(&self) -> usize {
317        self.catalog.list_sources().len()
318    }
319
320    /// Get the number of registered sinks.
321    pub fn sink_count(&self) -> usize {
322        self.catalog.list_sinks().len()
323    }
324
325    /// Returns checkpoint statistics if available (non-blocking).
326    ///
327    /// Uses `try_lock()` on the coordinator mutex. Returns `None` if
328    /// the coordinator is not initialized or the lock is contended.
329    pub fn checkpoint_stats_nonblocking(
330        &self,
331    ) -> Option<crate::checkpoint_coordinator::CheckpointStats> {
332        let guard = self.coordinator.try_lock().ok()?;
333        guard
334            .as_ref()
335            .map(crate::checkpoint_coordinator::CheckpointCoordinator::stats)
336    }
337
338    /// Get the number of active queries.
339    pub fn active_query_count(&self) -> usize {
340        self.catalog
341            .list_queries()
342            .iter()
343            .filter(|(_, _, active)| *active)
344            .count()
345    }
346}
347
348#[cfg(feature = "cluster")]
349fn nonnegative_gauge(value: i64) -> u64 {
350    u64::try_from(value).unwrap_or(0)
351}
352
353#[cfg(test)]
354mod tests {
355    use super::*;
356
357    #[cfg(feature = "cluster")]
358    async fn cluster_db_with_live_process() -> (
359        Arc<LaminarDB>,
360        laminar_core::cluster::control::LocalProcessAuthorityIdentity,
361    ) {
362        use laminar_core::cluster::control::{
363            ClusterController, ClusterKv, LeaseDeadline, ProcessLeaseAuthority, ProcessLeaseOutcome,
364        };
365        use laminar_core::cluster::discovery::{NodeId, NodeInfo};
366
367        let node = NodeId(71);
368        let boot = uuid::Uuid::from_u128(71);
369        let kv: Arc<dyn ClusterKv> =
370            Arc::new(laminar_core::cluster::control::InMemoryKv::new(node));
371        let (_members_tx, members_rx) = tokio::sync::watch::channel(Vec::<NodeInfo>::new());
372        let controller = Arc::new(ClusterController::new_with_recovery_incarnation(
373            node,
374            Arc::clone(&kv),
375            kv,
376            None,
377            members_rx,
378            boot,
379        ));
380        controller
381            .set_process_lease_deadline(Arc::new(LeaseDeadline::live_for(
382                std::time::Duration::from_secs(30),
383            )))
384            .unwrap();
385        let authority = Arc::new(
386            ProcessLeaseAuthority::new(
387                Arc::new(object_store::memory::InMemory::new()),
388                std::time::Duration::from_secs(30),
389            )
390            .unwrap(),
391        );
392        let ProcessLeaseOutcome::Acquired(lease) = authority
393            .store_for(node)
394            .try_acquire(boot, 0)
395            .await
396            .unwrap()
397        else {
398            panic!("empty process authority must grant the test process");
399        };
400        controller.set_process_lease_authority(authority).unwrap();
401        controller
402            .publish_leased_recovery_incarnation(&lease)
403            .await
404            .unwrap();
405        let process = controller
406            .try_live_local_process_authority_identity()
407            .unwrap();
408
409        let db = Arc::new(
410            LaminarDB::open_with_config_and_vars_and_rules(
411                crate::config::LaminarConfig::default(),
412                std::collections::HashMap::new(),
413                &[],
414                None,
415                crate::db::RuntimeMode::Cluster,
416            )
417            .unwrap(),
418        );
419        db.set_cluster_controller(controller).unwrap();
420        (db, process)
421    }
422
423    #[cfg(feature = "cluster")]
424    fn timing_observation(
425        process: laminar_core::cluster::control::LocalProcessAuthorityIdentity,
426        checkpoint_id: u64,
427    ) -> crate::checkpoint_timing::CheckpointBarrierTimingObservation {
428        crate::checkpoint_timing::CheckpointBarrierTimingObservation {
429            process,
430            attempt: laminar_core::checkpoint::CheckpointAttempt::canonical(checkpoint_id),
431            role: crate::checkpoint_timing::CheckpointBarrierRole::Follower,
432            assignment_version: 3,
433            assignment_digest: [9; 32],
434            pipeline_stall_ns: 100,
435            local_barrier_ns: 60,
436            aligned_resume_ns: Some(20),
437            durable_tail_handoff: true,
438            deadline_exhausted: false,
439        }
440    }
441
442    #[test]
443    fn prometheus_registry_is_single_assignment() {
444        let db = LaminarDB::open().unwrap();
445        db.set_prometheus_registry(Arc::new(prometheus::Registry::new()))
446            .unwrap();
447
448        let error = db
449            .set_prometheus_registry(Arc::new(prometheus::Registry::new()))
450            .unwrap_err()
451            .to_string();
452
453        assert!(error.contains("already installed"), "{error}");
454    }
455
456    #[test]
457    fn prometheus_registry_cannot_change_after_start_begins() {
458        let db = LaminarDB::open().unwrap();
459        DbState::Starting.store(&db.state);
460
461        let error = db
462            .set_prometheus_registry(Arc::new(prometheus::Registry::new()))
463            .unwrap_err()
464            .to_string();
465
466        DbState::Created.store(&db.state);
467        assert!(error.contains("Created"), "{error}");
468    }
469
470    #[test]
471    fn prometheus_registry_install_is_serialized_with_startup_claim() {
472        let db = Arc::new(LaminarDB::open().unwrap());
473        let startup = db.startup_attempt.lock();
474        let (started_tx, started_rx) = std::sync::mpsc::sync_channel(1);
475        let installing = Arc::clone(&db);
476        let install = std::thread::spawn(move || {
477            started_tx.send(()).unwrap();
478            installing.set_prometheus_registry(Arc::new(prometheus::Registry::new()))
479        });
480        started_rx.recv().unwrap();
481        DbState::Starting.store(&db.state);
482        drop(startup);
483
484        let error = install.join().unwrap().unwrap_err().to_string();
485
486        DbState::Created.store(&db.state);
487        assert!(error.contains("Created"), "{error}");
488    }
489
490    #[cfg(feature = "cluster")]
491    #[test]
492    fn cluster_subscription_health_is_aggregated_without_identity_labels() {
493        let db = LaminarDB::open().unwrap();
494        let registry = prometheus::Registry::new();
495        let metrics = Arc::new(crate::engine_metrics::EngineMetrics::new(&registry));
496        metrics.cluster_subscription.active_readers.set(2);
497        metrics.cluster_subscription.pending_bytes.set(128);
498        metrics.cluster_subscription.integrity_failures_total.inc();
499        db.set_engine_metrics(metrics);
500
501        let health = db.cluster_subscription_output_health().unwrap();
502        assert_eq!(health.active_readers, 2);
503        assert_eq!(health.pending_bytes, 128);
504        assert_eq!(health.integrity_failures, 1);
505        assert_eq!(health.retained_bytes, 0);
506    }
507
508    #[cfg(feature = "cluster")]
509    #[tokio::test]
510    async fn checkpoint_timing_cursor_is_bound_to_the_live_process() {
511        use crate::checkpoint_timing::CheckpointBarrierTimingReadError;
512
513        let (db, process) = cluster_db_with_live_process().await;
514        assert!(db
515            .checkpoint_barrier_timings
516            .try_record(timing_observation(process, 1)));
517        assert!(db
518            .checkpoint_barrier_timings
519            .try_record(timing_observation(process, 2)));
520
521        assert_eq!(
522            db.checkpoint_barrier_timing_snapshot(None, 1, 1),
523            Err(CheckpointBarrierTimingReadError::ProcessIdentityRequired)
524        );
525        let mut stale_process = process;
526        stale_process.process_term += 1;
527        assert_eq!(
528            db.checkpoint_barrier_timing_snapshot(Some(stale_process), 1, 1),
529            Err(CheckpointBarrierTimingReadError::ProcessIdentityMismatch {
530                expected: stale_process,
531                actual: process,
532            })
533        );
534
535        let first = db.checkpoint_barrier_timing_snapshot(None, 0, 1).unwrap();
536        assert_eq!(first.process, process);
537        assert_eq!(first.snapshot.process, Some(process));
538        assert_eq!(first.snapshot.records[0].sequence, 1);
539        let second = db
540            .checkpoint_barrier_timing_snapshot(Some(process), 1, 1)
541            .unwrap();
542        assert_eq!(second.snapshot.records[0].sequence, 2);
543    }
544
545    #[cfg(feature = "cluster")]
546    #[tokio::test]
547    async fn checkpoint_timing_read_rejects_a_foreign_ledger_domain() {
548        use crate::checkpoint_timing::CheckpointBarrierTimingReadError;
549
550        let (db, process) = cluster_db_with_live_process().await;
551        let mut foreign = process;
552        foreign.process_term += 1;
553        assert!(db
554            .checkpoint_barrier_timings
555            .try_record(timing_observation(foreign, 1)));
556
557        assert_eq!(
558            db.checkpoint_barrier_timing_snapshot(None, 0, 1),
559            Err(CheckpointBarrierTimingReadError::LedgerProcessMismatch {
560                expected: process,
561                actual: foreign,
562            })
563        );
564    }
565}
laminar_db/metrics_api.rs

laminar_db/
metrics_api.rs
