linkrs_metrics/manager/
checkpoint.rs

1//! Checkpoint and dirty-page observability.
2use serde::{Deserialize, Serialize};
3use std::collections::HashMap;
4use std::time::Duration;
5
6use super::core::StatsManager;
7use super::metric_type::MetricType;
8
9/// Checkpoint summary for handlers.
10#[derive(Debug, Clone, Serialize, Deserialize)]
11pub struct CheckpointSnapshot {
12    pub trigger_count: u64,
13    pub success_count: u64,
14    pub failure_count: u64,
15    pub avg_duration_us: f64,
16    pub data_flushed_bytes: u64,
17    pub wal_files_truncated: u64,
18    pub triggered_by_wal_size: u64,
19    pub triggered_by_interval: u64,
20    pub triggered_explicit: u64,
21    pub deduplicated: u64,
22    pub blocked: u64,
23}
24
25/// Reason a checkpoint was triggered.
26#[derive(Debug, Clone, Copy, PartialEq, Eq)]
27pub enum CheckpointTriggerReason {
28    WalSizeExceeded,
29    TimeSinceLastCheckpoint,
30    Explicit,
31}
32
33impl StatsManager {
34    /// Record a successful checkpoint operation.
35    pub fn record_checkpoint_success(
36        &self,
37        duration: Duration,
38        bytes_flushed: u64,
39        wal_files_truncated: u64,
40    ) {
41        self.add_value(MetricType::CheckpointSuccessCount);
42        self.add_value_with_amount(
43            MetricType::CheckpointDurationUs,
44            duration.as_micros() as u64,
45        );
46        self.add_value_with_amount(MetricType::CheckpointDataFlushedBytes, bytes_flushed);
47        self.add_value_with_amount(MetricType::CheckpointWalFilesTruncated, wal_files_truncated);
48    }
49
50    /// Record a failed checkpoint operation.
51    pub fn record_checkpoint_failure(&self) {
52        self.add_value(MetricType::CheckpointFailureCount);
53    }
54
55    /// Record a checkpoint trigger event.
56    pub fn record_checkpoint_trigger(&self, reason: CheckpointTriggerReason, age: Duration) {
57        self.add_value(MetricType::CheckpointTriggerCount);
58        self.add_value_with_amount(MetricType::CheckpointTriggerAgeSecs, age.as_secs());
59        match reason {
60            CheckpointTriggerReason::WalSizeExceeded => {
61                self.add_value(MetricType::CheckpointTriggeredByWalSize);
62            }
63            CheckpointTriggerReason::TimeSinceLastCheckpoint => {
64                self.add_value(MetricType::CheckpointTriggeredByInterval);
65            }
66            CheckpointTriggerReason::Explicit => {
67                self.add_value(MetricType::CheckpointTriggeredExplicit);
68            }
69        }
70    }
71
72    /// Record a deduplicated checkpoint request.
73    pub fn record_checkpoint_deduplicated(&self) {
74        self.add_value(MetricType::CheckpointRequestsDeduplicated);
75    }
76
77    /// Record a checkpoint request that found another checkpoint in progress.
78    pub fn record_checkpoint_blocked(&self) {
79        self.add_value(MetricType::CheckpointRequestsBlocked);
80    }
81
82    pub fn checkpoint_snapshot(&self) -> CheckpointSnapshot {
83        let v = |m: MetricType| self.get_value(m).unwrap_or(0);
84        let success = v(MetricType::CheckpointSuccessCount);
85        CheckpointSnapshot {
86            trigger_count: v(MetricType::CheckpointTriggerCount),
87            success_count: success,
88            failure_count: v(MetricType::CheckpointFailureCount),
89            avg_duration_us: if success == 0 {
90                0.0
91            } else {
92                v(MetricType::CheckpointDurationUs) as f64 / success as f64
93            },
94            data_flushed_bytes: v(MetricType::CheckpointDataFlushedBytes),
95            wal_files_truncated: v(MetricType::CheckpointWalFilesTruncated),
96            triggered_by_wal_size: v(MetricType::CheckpointTriggeredByWalSize),
97            triggered_by_interval: v(MetricType::CheckpointTriggeredByInterval),
98            triggered_explicit: v(MetricType::CheckpointTriggeredExplicit),
99            deduplicated: v(MetricType::CheckpointRequestsDeduplicated),
100            blocked: v(MetricType::CheckpointRequestsBlocked),
101        }
102    }
103
104    /// Get all checkpoint-related metrics.
105    pub fn get_checkpoint_metrics(&self) -> HashMap<MetricType, u64> {
106        let mut out = HashMap::new();
107        for mt in [
108            MetricType::CheckpointTriggerCount,
109            MetricType::CheckpointSuccessCount,
110            MetricType::CheckpointFailureCount,
111            MetricType::CheckpointDurationUs,
112            MetricType::CheckpointDataFlushedBytes,
113            MetricType::CheckpointWalFilesTruncated,
114            MetricType::CheckpointTriggerAgeSecs,
115            MetricType::CheckpointTriggeredByWalSize,
116            MetricType::CheckpointTriggeredByInterval,
117            MetricType::CheckpointTriggeredExplicit,
118            MetricType::CheckpointRequestsDeduplicated,
119            MetricType::CheckpointRequestsBlocked,
120            MetricType::CheckpointStrategyIncremental,
121            MetricType::CheckpointStrategyHybrid,
122            MetricType::CheckpointStrategyFull,
123            MetricType::CheckpointIncrementalDurationUs,
124            MetricType::CheckpointIncrementalBytesFlushed,
125        ] {
126            if let Some(v) = self.get_value(mt) {
127                out.insert(mt, v);
128            }
129        }
130        out
131    }
132
133    pub fn record_dirty_pages(&self, dirty: u64, total: u64) {
134        self.set_value(MetricType::DirtyPagesCount, dirty);
135        self.set_value(MetricType::DirtyPagesTotal, total);
136        let ratio_permille = if total == 0 {
137            0
138        } else {
139            (dirty as f64 / total as f64 * 1000.0) as u64
140        };
141        self.set_value(MetricType::DirtyPageRatioPermille, ratio_permille);
142    }
143
144    pub fn record_checkpoint_strategy_by_name(&self, strategy: &str) {
145        match strategy {
146            "incremental" => self.add_value(MetricType::CheckpointStrategyIncremental),
147            "hybrid" => self.add_value(MetricType::CheckpointStrategyHybrid),
148            "full" => self.add_value(MetricType::CheckpointStrategyFull),
149            _ => {}
150        }
151    }
152
153    pub fn record_incremental_checkpoint(&self, duration: Duration, bytes: u64) {
154        self.add_value_with_amount(
155            MetricType::CheckpointIncrementalDurationUs,
156            duration.as_micros() as u64,
157        );
158        self.add_value_with_amount(MetricType::CheckpointIncrementalBytesFlushed, bytes);
159    }
160
161    pub fn get_dirty_page_metrics(&self) -> HashMap<MetricType, u64> {
162        let mut out = HashMap::new();
163        for mt in [
164            MetricType::DirtyPagesTotal,
165            MetricType::DirtyPagesCount,
166            MetricType::DirtyPageRatioPermille,
167        ] {
168            if let Some(v) = self.get_value(mt) {
169                out.insert(mt, v);
170            }
171        }
172        out
173    }
174}
linkrs_metrics/manager/checkpoint.rs

linkrs_metrics/manager/
checkpoint.rs
