openraft 0.10.0-alpha.18

Advanced Raft consensus
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144

//! Metrics recording trait for external metrics collection.
//!
//! This module defines the [`MetricsRecorder`] trait that allows applications
//! to plug in custom metrics collection backends. The trait provides a minimal
//! interface for recording Raft internal events.
//!
//! # Example
//!
//! ```ignore
//! use std::sync::Arc;
//! use openraft::metrics::MetricsRecorder;
//!
//! #[derive(Debug)]
//! struct MyRecorder;
//!
//! impl MetricsRecorder for MyRecorder {
//!     fn record_apply_batch(&self, entry_count: u64) { /* ... */ }
//!     fn record_append_batch(&self, entry_count: u64) { /* ... */ }
//!     fn record_write_batch(&self, entry_count: u64) { /* ... */ }
//!     fn set_current_term(&self, term: u64) { /* ... */ }
//!     fn set_last_log_index(&self, index: u64) { /* ... */ }
//!     fn set_committed_index(&self, index: u64) { /* ... */ }
//!     fn set_applied_index(&self, index: u64) { /* ... */ }
//!     fn set_snapshot_index(&self, index: u64) { /* ... */ }
//!     fn set_purged_index(&self, index: u64) { /* ... */ }
//!     fn set_server_state(&self, state: u8) { /* ... */ }
//!     fn increment_vote(&self) { /* ... */ }
//!     fn increment_heartbeat(&self) { /* ... */ }
//!     fn increment_append(&self) { /* ... */ }
//! }
//!
//! // Install in Raft instance
//! raft.set_metrics_recorder(Some(Arc::new(MyRecorder)));
//! ```

use crate::RaftTypeConfig;
use crate::core::ServerState;
use crate::metrics::RaftMetrics;
use crate::vote::RaftTerm;

/// Trait for recording Raft metrics to external systems.
///
/// Implement this trait to collect metrics from Raft internals and export them
/// to your preferred metrics backend (e.g., Prometheus, OpenTelemetry, StatsD).
///
/// All methods are called from the RaftCore task and should return quickly to
/// avoid blocking Raft operations.
pub trait MetricsRecorder: Send + Sync + std::fmt::Debug {
    // --- Histograms (batch sizes) ---

    /// Record a batch of log entries being applied to the state machine.
    fn record_apply_batch(&self, entry_count: u64);

    /// Record a batch of log entries being appended to storage.
    fn record_append_batch(&self, entry_count: u64);

    /// Record a batch of client write entries merged together.
    fn record_write_batch(&self, entry_count: u64);

    // --- Gauges (current state) ---

    /// Set the current Raft term.
    fn set_current_term(&self, term: u64);

    /// Set the index of the last log entry.
    fn set_last_log_index(&self, index: u64);

    /// Set the index of the last committed log entry.
    fn set_committed_index(&self, index: u64);

    /// Set the index of the last applied log entry.
    fn set_applied_index(&self, index: u64);

    /// Set the index of the last snapshot.
    fn set_snapshot_index(&self, index: u64);

    /// Set the index of the last purged log entry.
    fn set_purged_index(&self, index: u64);

    /// Set the server state.
    ///
    /// States: 0=Learner, 1=Follower, 2=Candidate, 3=Leader, 4=Shutdown
    fn set_server_state(&self, state: u8);

    // --- Counters (operations) ---

    /// Increment the vote operation counter.
    fn increment_vote(&self);

    /// Increment the heartbeat operation counter.
    fn increment_heartbeat(&self);

    /// Increment the append entries operation counter.
    fn increment_append(&self);
}

/// Forward gauge metrics from `RaftMetrics` to a `MetricsRecorder`.
///
/// This helper function extracts the relevant fields from `RaftMetrics` and
/// calls the corresponding setter methods on the recorder. It only forwards
/// gauge metrics (current state values), not counters or histograms.
///
/// # Example
///
/// ```ignore
/// use openraft::metrics::{RaftMetrics, MetricsRecorder, forward_metrics};
///
/// fn on_metrics_update<C: RaftTypeConfig>(metrics: &RaftMetrics<C>, recorder: &dyn MetricsRecorder) {
///     forward_metrics(metrics, recorder);
/// }
/// ```
pub fn forward_metrics<C: RaftTypeConfig>(metrics: &RaftMetrics<C>, recorder: &dyn MetricsRecorder) {
    if let Some(term) = metrics.current_term.as_u64() {
        recorder.set_current_term(term);
    }

    if let Some(index) = metrics.last_log_index {
        recorder.set_last_log_index(index);
    }

    if let Some(ref committed) = metrics.committed {
        recorder.set_committed_index(committed.index());
    }

    if let Some(ref applied) = metrics.last_applied {
        recorder.set_applied_index(applied.index());
    }

    if let Some(ref snapshot) = metrics.snapshot {
        recorder.set_snapshot_index(snapshot.index());
    }

    if let Some(ref purged) = metrics.purged {
        recorder.set_purged_index(purged.index());
    }

    recorder.set_server_state(match metrics.state {
        ServerState::Learner => 0,
        ServerState::Follower => 1,
        ServerState::Candidate => 2,
        ServerState::Leader => 3,
        ServerState::Shutdown => 4,
    });
}