sz_orm_rw/
replication_lag.rs

1//! 复制延迟监控
2//!
3//! 跟踪 master-slave 复制延迟，延迟过大时将读请求路由到 master（强一致读）。
4//! 延迟数据由外部探针定期采集并上报。
5
6use serde::{Deserialize, Serialize};
7use std::collections::HashMap;
8use std::sync::Mutex;
9use std::time::Instant;
10
11/// 单个 slave 的复制延迟快照
12#[derive(Debug, Clone, Serialize, Deserialize)]
13pub struct ReplicationLagSnapshot {
14    /// slave 地址
15    pub slave: String,
16    /// 复制延迟（秒）
17    pub lag_seconds: u64,
18    /// 采集时间戳（从实例创建起经过的秒数）
19    pub collected_at_secs: u64,
20}
21
22impl ReplicationLagSnapshot {
23    /// 创建快照
24    pub fn new(slave: &str, lag_seconds: u64, collected_at_secs: u64) -> Self {
25        Self {
26            slave: slave.to_string(),
27            lag_seconds,
28            collected_at_secs,
29        }
30    }
31
32    /// 延迟是否超过阈值
33    pub fn exceeds(&self, threshold_secs: u64) -> bool {
34        self.lag_seconds > threshold_secs
35    }
36
37    /// 延迟是否在可接受范围内
38    pub fn is_acceptable(&self, threshold_secs: u64) -> bool {
39        self.lag_seconds <= threshold_secs
40    }
41}
42
43/// 复制延迟监控器
44///
45/// 维护每个 slave 的最新延迟快照，提供基于延迟的路由决策。
46pub struct ReplicationLagMonitor {
47    /// 延迟阈值（秒），超过此值的 slave 被视为"延迟过大"
48    threshold_secs: u64,
49    /// 各 slave 的延迟历史（保留最近 N 条）
50    history: Mutex<HashMap<String, Vec<ReplicationLagSnapshot>>>,
51    /// 最大历史保留条数
52    max_history: usize,
53    /// 实例创建时间
54    started: Instant,
55}
56
57impl ReplicationLagMonitor {
58    /// 创建监控器
59    pub fn new(threshold_secs: u64) -> Self {
60        Self {
61            threshold_secs,
62            history: Mutex::new(HashMap::new()),
63            max_history: 10,
64            started: Instant::now(),
65        }
66    }
67
68    /// 设置最大历史保留条数
69    pub fn with_max_history(mut self, max: usize) -> Self {
70        self.max_history = max.max(1);
71        self
72    }
73
74    /// 上报一次延迟采样
75    pub fn report(&self, slave: &str, lag_seconds: u64) {
76        let now = self.started.elapsed().as_secs();
77        let snap = ReplicationLagSnapshot::new(slave, lag_seconds, now);
78        if let Ok(mut history) = self.history.lock() {
79            let entries = history.entry(slave.to_string()).or_default();
80            entries.push(snap);
81            if entries.len() > self.max_history {
82                entries.remove(0);
83            }
84        }
85    }
86
87    /// 获取 slave 的最新延迟
88    pub fn latest_lag(&self, slave: &str) -> Option<u64> {
89        match self.history.lock() {
90            Ok(history) => history
91                .get(slave)
92                .and_then(|v| v.last())
93                .map(|s| s.lag_seconds),
94            Err(_) => None,
95        }
96    }
97
98    /// 判断 slave 是否延迟过大
99    pub fn is_lagging(&self, slave: &str) -> bool {
100        match self.latest_lag(slave) {
101            Some(lag) => lag > self.threshold_secs,
102            None => false,
103        }
104    }
105
106    /// 判断 slave 是否可以安全读取
107    pub fn is_safe_to_read(&self, slave: &str) -> bool {
108        !self.is_lagging(slave)
109    }
110
111    /// 返回所有延迟过大的 slave
112    pub fn lagging_slaves(&self) -> Vec<String> {
113        match self.history.lock() {
114            Ok(history) => history
115                .iter()
116                .filter(|(_, v)| {
117                    v.last()
118                        .map(|s| s.lag_seconds > self.threshold_secs)
119                        .unwrap_or(false)
120                })
121                .map(|(k, _)| k.clone())
122                .collect(),
123            Err(_) => Vec::new(),
124        }
125    }
126
127    /// 返回所有安全可读的 slave
128    pub fn safe_slaves(&self) -> Vec<String> {
129        match self.history.lock() {
130            Ok(history) => history
131                .iter()
132                .filter(|(_, v)| {
133                    v.last()
134                        .map(|s| s.lag_seconds <= self.threshold_secs)
135                        .unwrap_or(true)
136                })
137                .map(|(k, _)| k.clone())
138                .collect(),
139            Err(_) => Vec::new(),
140        }
141    }
142
143    /// 获取 slave 的延迟历史
144    pub fn history(&self, slave: &str) -> Vec<ReplicationLagSnapshot> {
145        match self.history.lock() {
146            Ok(history) => history.get(slave).cloned().unwrap_or_default(),
147            Err(_) => Vec::new(),
148        }
149    }
150
151    /// 计算 slave 的平均延迟（基于历史）
152    pub fn avg_lag(&self, slave: &str) -> Option<u64> {
153        match self.history.lock() {
154            Ok(history) => {
155                let entries = history.get(slave)?;
156                if entries.is_empty() {
157                    return None;
158                }
159                let total: u64 = entries.iter().map(|s| s.lag_seconds).sum();
160                Some(total / entries.len() as u64)
161            }
162            Err(_) => None,
163        }
164    }
165
166    /// 计算 slave 的最大延迟（基于历史）
167    pub fn max_lag(&self, slave: &str) -> Option<u64> {
168        match self.history.lock() {
169            Ok(history) => history
170                .get(slave)
171                .and_then(|v| v.iter().map(|s| s.lag_seconds).max()),
172            Err(_) => None,
173        }
174    }
175
176    /// 获取延迟阈值
177    pub fn threshold(&self) -> u64 {
178        self.threshold_secs
179    }
180
181    /// 重置 slave 的历史
182    pub fn reset(&self, slave: &str) {
183        if let Ok(mut history) = self.history.lock() {
184            history.remove(slave);
185        }
186    }
187
188    /// 重置所有 slave 的历史
189    pub fn reset_all(&self) {
190        if let Ok(mut history) = self.history.lock() {
191            history.clear();
192        }
193    }
194
195    /// 从候选列表中选择延迟最小的 slave
196    ///
197    /// 返回 None 表示候选列表为空或无延迟数据。
198    pub fn select_least_lag<'a>(&self, candidates: &'a [String]) -> Option<&'a str> {
199        let history = self.history.lock().ok()?;
200        let mut best: Option<(&str, u64)> = None;
201        for slave in candidates {
202            if let Some(entries) = history.get(slave) {
203                if let Some(latest) = entries.last() {
204                    match best {
205                        None => best = Some((slave.as_str(), latest.lag_seconds)),
206                        Some((_, b_lag)) if latest.lag_seconds < b_lag => {
207                            best = Some((slave.as_str(), latest.lag_seconds))
208                        }
209                        _ => {}
210                    }
211                }
212            }
213        }
214        best.map(|(s, _)| s)
215    }
216
217    /// 计算 slave 的最小延迟（基于历史）
218    pub fn min_lag(&self, slave: &str) -> Option<u64> {
219        match self.history.lock() {
220            Ok(history) => history
221                .get(slave)
222                .and_then(|v| v.iter().map(|s| s.lag_seconds).min()),
223            Err(_) => None,
224        }
225    }
226
227    /// 返回所有已知 slave 列表
228    pub fn all_slaves(&self) -> Vec<String> {
229        match self.history.lock() {
230            Ok(history) => history.keys().cloned().collect(),
231            Err(_) => Vec::new(),
232        }
233    }
234
235    /// 已知 slave 数量
236    pub fn slave_count(&self) -> usize {
237        match self.history.lock() {
238            Ok(history) => history.len(),
239            Err(_) => 0,
240        }
241    }
242
243    /// 延迟趋势：比较最近两次采样
244    ///
245    /// 返回正数表示延迟上升，负数表示下降，0 表示稳定或数据不足。
246    pub fn lag_trend(&self, slave: &str) -> i64 {
247        match self.history.lock() {
248            Ok(history) => {
249                if let Some(entries) = history.get(slave) {
250                    if entries.len() < 2 {
251                        return 0;
252                    }
253                    let last = entries.last().unwrap().lag_seconds as i64;
254                    let prev = entries[entries.len() - 2].lag_seconds as i64;
255                    last - prev
256                } else {
257                    0
258                }
259            }
260            Err(_) => 0,
261        }
262    }
263
264    /// 生成汇总报告字符串
265    pub fn summary(&self) -> String {
266        let history = match self.history.lock() {
267            Ok(h) => h,
268            Err(_) => return "ReplicationLagMonitor: lock poisoned".to_string(),
269        };
270        let mut out = format!(
271            "ReplicationLagMonitor: {} slave(s), threshold={}s\n",
272            history.len(),
273            self.threshold_secs
274        );
275        for (slave, entries) in history.iter() {
276            let latest = entries.last().map(|s| s.lag_seconds).unwrap_or(0);
277            let status = if latest > self.threshold_secs {
278                "LAGGING"
279            } else {
280                "OK"
281            };
282            out.push_str(&format!("  {} : lag={}s [{}]\n", slave, latest, status));
283        }
284        out
285    }
286}
287
288#[cfg(test)]
289mod tests {
290    use super::*;
291
292    #[test]
293    fn test_snapshot_exceeds() {
294        let snap = ReplicationLagSnapshot::new("s1", 10, 0);
295        assert!(snap.exceeds(5));
296        assert!(!snap.exceeds(10));
297        assert!(!snap.exceeds(15));
298    }
299
300    #[test]
301    fn test_snapshot_is_acceptable() {
302        let snap = ReplicationLagSnapshot::new("s1", 5, 0);
303        assert!(snap.is_acceptable(5));
304        assert!(snap.is_acceptable(10));
305        assert!(!snap.is_acceptable(3));
306    }
307
308    #[test]
309    fn test_monitor_new_has_threshold() {
310        let m = ReplicationLagMonitor::new(30);
311        assert_eq!(m.threshold(), 30);
312    }
313
314    #[test]
315    fn test_report_and_latest_lag() {
316        let m = ReplicationLagMonitor::new(10);
317        m.report("s1", 5);
318        assert_eq!(m.latest_lag("s1"), Some(5));
319        m.report("s1", 8);
320        assert_eq!(m.latest_lag("s1"), Some(8));
321    }
322
323    #[test]
324    fn test_latest_lag_unknown_slave() {
325        let m = ReplicationLagMonitor::new(10);
326        assert_eq!(m.latest_lag("ghost"), None);
327    }
328
329    #[test]
330    fn test_is_lagging() {
331        let m = ReplicationLagMonitor::new(10);
332        m.report("s1", 5);
333        assert!(!m.is_lagging("s1"));
334        m.report("s1", 15);
335        assert!(m.is_lagging("s1"));
336    }
337
338    #[test]
339    fn test_is_safe_to_read() {
340        let m = ReplicationLagMonitor::new(10);
341        m.report("s1", 5);
342        assert!(m.is_safe_to_read("s1"));
343        m.report("s1", 20);
344        assert!(!m.is_safe_to_read("s1"));
345    }
346
347    #[test]
348    fn test_is_safe_to_read_no_data() {
349        let m = ReplicationLagMonitor::new(10);
350        assert!(m.is_safe_to_read("unknown"), "no data should be safe");
351    }
352
353    #[test]
354    fn test_lagging_slaves() {
355        let m = ReplicationLagMonitor::new(10);
356        m.report("s1", 5);
357        m.report("s2", 15);
358        m.report("s3", 20);
359        let mut lagging = m.lagging_slaves();
360        lagging.sort();
361        assert_eq!(lagging, vec!["s2".to_string(), "s3".to_string()]);
362    }
363
364    #[test]
365    fn test_safe_slaves() {
366        let m = ReplicationLagMonitor::new(10);
367        m.report("s1", 5);
368        m.report("s2", 15);
369        let safe = m.safe_slaves();
370        assert_eq!(safe, vec!["s1".to_string()]);
371    }
372
373    #[test]
374    fn test_history_retention() {
375        let m = ReplicationLagMonitor::new(10).with_max_history(3);
376        m.report("s1", 1);
377        m.report("s1", 2);
378        m.report("s1", 3);
379        m.report("s1", 4);
380        assert_eq!(m.history("s1").len(), 3);
381        assert_eq!(m.latest_lag("s1"), Some(4));
382    }
383
384    #[test]
385    fn test_avg_lag() {
386        let m = ReplicationLagMonitor::new(100);
387        m.report("s1", 10);
388        m.report("s1", 20);
389        m.report("s1", 30);
390        assert_eq!(m.avg_lag("s1"), Some(20));
391    }
392
393    #[test]
394    fn test_avg_lag_no_data() {
395        let m = ReplicationLagMonitor::new(100);
396        assert_eq!(m.avg_lag("ghost"), None);
397    }
398
399    #[test]
400    fn test_max_lag() {
401        let m = ReplicationLagMonitor::new(100);
402        m.report("s1", 10);
403        m.report("s1", 50);
404        m.report("s1", 20);
405        assert_eq!(m.max_lag("s1"), Some(50));
406    }
407
408    #[test]
409    fn test_reset() {
410        let m = ReplicationLagMonitor::new(10);
411        m.report("s1", 5);
412        m.reset("s1");
413        assert_eq!(m.latest_lag("s1"), None);
414    }
415
416    #[test]
417    fn test_reset_all() {
418        let m = ReplicationLagMonitor::new(10);
419        m.report("s1", 5);
420        m.report("s2", 10);
421        m.reset_all();
422        assert_eq!(m.latest_lag("s1"), None);
423        assert_eq!(m.latest_lag("s2"), None);
424    }
425
426    #[test]
427    fn test_select_least_lag() {
428        let m = ReplicationLagMonitor::new(100);
429        m.report("s1", 20);
430        m.report("s2", 5);
431        m.report("s3", 15);
432        let candidates = vec!["s1".to_string(), "s2".to_string(), "s3".to_string()];
433        let selected = m.select_least_lag(&candidates);
434        assert_eq!(selected, Some("s2"));
435    }
436
437    #[test]
438    fn test_select_least_lag_no_data() {
439        let m = ReplicationLagMonitor::new(100);
440        let candidates = vec!["s1".to_string()];
441        assert_eq!(m.select_least_lag(&candidates), None);
442    }
443
444    #[test]
445    fn test_select_least_lag_empty_candidates() {
446        let m = ReplicationLagMonitor::new(100);
447        let candidates: Vec<String> = vec![];
448        assert_eq!(m.select_least_lag(&candidates), None);
449    }
450
451    #[test]
452    fn test_with_max_history_clamped_to_one() {
453        let m = ReplicationLagMonitor::new(10).with_max_history(0);
454        m.report("s1", 5);
455        assert_eq!(m.history("s1").len(), 1);
456    }
457
458    #[test]
459    fn test_min_lag() {
460        let m = ReplicationLagMonitor::new(100);
461        m.report("s1", 30);
462        m.report("s1", 10);
463        m.report("s1", 50);
464        assert_eq!(m.min_lag("s1"), Some(10));
465    }
466
467    #[test]
468    fn test_min_lag_no_data() {
469        let m = ReplicationLagMonitor::new(100);
470        assert_eq!(m.min_lag("ghost"), None);
471    }
472
473    #[test]
474    fn test_all_slaves() {
475        let m = ReplicationLagMonitor::new(100);
476        m.report("s1", 5);
477        m.report("s2", 10);
478        let mut slaves = m.all_slaves();
479        slaves.sort();
480        assert_eq!(slaves, vec!["s1".to_string(), "s2".to_string()]);
481    }
482
483    #[test]
484    fn test_slave_count() {
485        let m = ReplicationLagMonitor::new(100);
486        m.report("s1", 5);
487        m.report("s2", 10);
488        assert_eq!(m.slave_count(), 2);
489    }
490
491    #[test]
492    fn test_lag_trend_rising() {
493        let m = ReplicationLagMonitor::new(100);
494        m.report("s1", 5);
495        m.report("s1", 10);
496        assert_eq!(m.lag_trend("s1"), 5);
497    }
498
499    #[test]
500    fn test_lag_trend_falling() {
501        let m = ReplicationLagMonitor::new(100);
502        m.report("s1", 20);
503        m.report("s1", 10);
504        assert_eq!(m.lag_trend("s1"), -10);
505    }
506
507    #[test]
508    fn test_lag_trend_stable() {
509        let m = ReplicationLagMonitor::new(100);
510        m.report("s1", 10);
511        m.report("s1", 10);
512        assert_eq!(m.lag_trend("s1"), 0);
513    }
514
515    #[test]
516    fn test_lag_trend_single_sample() {
517        let m = ReplicationLagMonitor::new(100);
518        m.report("s1", 10);
519        assert_eq!(m.lag_trend("s1"), 0);
520    }
521
522    #[test]
523    fn test_summary_contains_slave() {
524        let m = ReplicationLagMonitor::new(10);
525        m.report("s1", 5);
526        m.report("s2", 20);
527        let s = m.summary();
528        assert!(s.contains("s1"));
529        assert!(s.contains("s2"));
530        assert!(s.contains("OK"));
531        assert!(s.contains("LAGGING"));
532    }
533}
sz_orm_rw/replication_lag.rs

sz_orm_rw/
replication_lag.rs
