nodedb_cluster/
cluster_info.rs

1// SPDX-License-Identifier: BUSL-1.1
2
3//! Observability view of the cluster — snapshot types, trait for
4//! querying per-group Raft status, and the `ClusterObserver` handle
5//! bundled into `SharedState` for HTTP / metrics readers.
6//!
7//! The split between `ClusterObserver` (operational aggregation) and
8//! the individual sources it pulls from (`ClusterTopology`,
9//! `RoutingTable`, `ClusterLifecycleTracker`, `GroupStatusProvider`)
10//! keeps every consumer parameter-free: an HTTP handler takes a
11//! single `Arc<ClusterObserver>` and gets a complete, serialisable
12//! snapshot of everything the cluster surface exposes.
13
14use std::sync::{Arc, RwLock, Weak};
15
16use serde::{Deserialize, Serialize};
17
18use crate::forward::PlanExecutor;
19use crate::lifecycle_state::{ClusterLifecycleState, ClusterLifecycleTracker};
20use crate::multi_raft::GroupStatus;
21use crate::raft_loop::{CommitApplier, RaftLoop};
22use crate::routing::RoutingTable;
23use crate::topology::ClusterTopology;
24
25/// Read-only accessor for per-group Raft status.
26///
27/// Implemented for every `RaftLoop` via a blanket impl so the main
28/// binary can coerce `Arc<RaftLoop<...>>` to `Arc<dyn
29/// GroupStatusProvider + Send + Sync>` without thinking about the
30/// `CommitApplier` / `PlanExecutor` type parameters.
31pub trait GroupStatusProvider: Send + Sync {
32    /// Current status of every Raft group hosted on this node.
33    fn group_statuses(&self) -> Vec<GroupStatus>;
34}
35
36impl<A, P> GroupStatusProvider for RaftLoop<A, P>
37where
38    A: CommitApplier,
39    P: PlanExecutor,
40{
41    fn group_statuses(&self) -> Vec<GroupStatus> {
42        RaftLoop::group_statuses(self)
43    }
44}
45
46/// Aggregated observability handle for the cluster.
47///
48/// Stored in `SharedState::cluster_observer` as an
49/// `Arc<ClusterObserver>` so HTTP route handlers and the metrics
50/// endpoint can build snapshots without threading four separate
51/// handles through every call.
52///
53/// Construction is done exactly once — after `start_cluster` has
54/// returned and `start_raft` has built the `RaftLoop`. See
55/// `nodedb::control::cluster::start_raft` for the wiring.
56pub struct ClusterObserver {
57    /// This node's id.
58    pub node_id: u64,
59    /// Lifecycle phase tracker (shared with `start_cluster`).
60    pub lifecycle: ClusterLifecycleTracker,
61    /// Shared cluster topology.
62    pub topology: Arc<RwLock<ClusterTopology>>,
63    /// Shared routing table.
64    pub routing: Arc<RwLock<RoutingTable>>,
65    /// Type-erased per-group status provider backed by `RaftLoop`.
66    ///
67    /// Held as a `Weak` on purpose: the owner of this observer
68    /// (`SharedState`) is itself kept alive transitively by the
69    /// `RaftLoop`, so a strong reference here would close a cycle that
70    /// pins both forever and blocks clean shutdown. The loop is kept
71    /// alive by its own spawned tasks; `upgrade` therefore succeeds
72    /// throughout normal operation and only fails once the loop has been
73    /// dropped on shutdown, where an empty status snapshot is correct.
74    pub group_status: Weak<dyn GroupStatusProvider + Send + Sync>,
75}
76
77impl ClusterObserver {
78    pub fn new(
79        node_id: u64,
80        lifecycle: ClusterLifecycleTracker,
81        topology: Arc<RwLock<ClusterTopology>>,
82        routing: Arc<RwLock<RoutingTable>>,
83        group_status: Weak<dyn GroupStatusProvider + Send + Sync>,
84    ) -> Self {
85        Self {
86            node_id,
87            lifecycle,
88            topology,
89            routing,
90            group_status,
91        }
92    }
93
94    /// Build a complete `ClusterInfoSnapshot` for rendering via
95    /// `/cluster/status` or `/metrics`.
96    pub fn snapshot(&self) -> ClusterInfoSnapshot {
97        let lifecycle = self.lifecycle.current();
98        let peers: Vec<PeerSnapshot> = {
99            let topo = self.topology.read().unwrap_or_else(|p| p.into_inner());
100            topo.all_nodes()
101                .map(|n| PeerSnapshot {
102                    node_id: n.node_id,
103                    addr: n.addr.clone(),
104                    state: format!("{:?}", n.state),
105                })
106                .collect()
107        };
108
109        // Merge Raft group status (from the live RaftLoop) with the
110        // routing-table members/learners view. The routing table is
111        // authoritative for "who is supposed to be in this group";
112        // the RaftLoop is authoritative for "who is leader / what's
113        // the commit index right now".
114        let routing_snapshot: Vec<(u64, Vec<u64>, Vec<u64>)> = {
115            let rt = self.routing.read().unwrap_or_else(|p| p.into_inner());
116            rt.group_members()
117                .iter()
118                .map(|(&gid, info)| (gid, info.members.clone(), info.learners.clone()))
119                .collect()
120        };
121
122        // `upgrade` fails only after the backing `RaftLoop` has been
123        // dropped on shutdown; an empty snapshot ("no groups hosted")
124        // is the correct observability answer there, never a panic.
125        let raft_groups = self
126            .group_status
127            .upgrade()
128            .map(|gs| gs.group_statuses())
129            .unwrap_or_default();
130        let groups: Vec<GroupSnapshot> = raft_groups
131            .into_iter()
132            .map(|gs| {
133                let (members, learners) = routing_snapshot
134                    .iter()
135                    .find(|(gid, _, _)| *gid == gs.group_id)
136                    .map(|(_, m, l)| (m.clone(), l.clone()))
137                    .unwrap_or_default();
138                GroupSnapshot {
139                    group_id: gs.group_id,
140                    role: gs.role,
141                    leader_id: gs.leader_id,
142                    term: gs.term,
143                    commit_index: gs.commit_index,
144                    last_applied: gs.last_applied,
145                    member_count: gs.member_count,
146                    vshard_count: gs.vshard_count,
147                    members,
148                    learners,
149                }
150            })
151            .collect();
152
153        ClusterInfoSnapshot {
154            node_id: self.node_id,
155            lifecycle,
156            peers,
157            groups,
158        }
159    }
160}
161
162/// Serialisable snapshot of the full cluster observability surface.
163///
164/// This is the JSON shape returned by `GET /cluster/status` and the
165/// source-of-truth for the Prometheus gauges.
166#[derive(Debug, Clone, Serialize, Deserialize)]
167pub struct ClusterInfoSnapshot {
168    /// This node's id.
169    pub node_id: u64,
170    /// Current lifecycle phase.
171    pub lifecycle: ClusterLifecycleState,
172    /// Every peer known to this node.
173    pub peers: Vec<PeerSnapshot>,
174    /// Every Raft group hosted on this node.
175    pub groups: Vec<GroupSnapshot>,
176}
177
178impl ClusterInfoSnapshot {
179    /// Convenience accessor used by the metrics endpoint.
180    pub fn lifecycle_label(&self) -> &'static str {
181        self.lifecycle.label()
182    }
183
184    /// Number of peers in the topology snapshot. Drives the
185    /// `nodedb_cluster_members` Prometheus gauge.
186    pub fn members_count(&self) -> usize {
187        self.peers.len()
188    }
189
190    /// Number of Raft groups hosted locally.
191    pub fn groups_count(&self) -> usize {
192        self.groups.len()
193    }
194}
195
196/// One peer entry rendered in `/cluster/status`.
197#[derive(Debug, Clone, Serialize, Deserialize)]
198pub struct PeerSnapshot {
199    pub node_id: u64,
200    pub addr: String,
201    pub state: String,
202}
203
204/// One Raft group entry rendered in `/cluster/status`.
205#[derive(Debug, Clone, Serialize, Deserialize)]
206pub struct GroupSnapshot {
207    pub group_id: u64,
208    pub role: String,
209    pub leader_id: u64,
210    pub term: u64,
211    pub commit_index: u64,
212    pub last_applied: u64,
213    pub member_count: usize,
214    pub vshard_count: usize,
215    pub members: Vec<u64>,
216    pub learners: Vec<u64>,
217}
218
219#[cfg(test)]
220mod tests {
221    use super::*;
222    use crate::topology::{NodeInfo, NodeState};
223
224    /// Fake provider that returns a canned list — lets us test the
225    /// snapshot builder end-to-end without spinning up a real
226    /// `MultiRaft`.
227    struct FakeProvider(Vec<GroupStatus>);
228    impl GroupStatusProvider for FakeProvider {
229        fn group_statuses(&self) -> Vec<GroupStatus> {
230            self.0.clone()
231        }
232    }
233
234    /// Returns the observer plus the strong provider it holds weakly. Tests
235    /// must keep the returned `Arc` alive for the duration they call
236    /// `snapshot()`, or the observer's `Weak` upgrade would fail and every
237    /// group would render as empty.
238    fn make_observer(
239        lifecycle: ClusterLifecycleTracker,
240        peers: Vec<NodeInfo>,
241        routing: RoutingTable,
242        raft_groups: Vec<GroupStatus>,
243    ) -> (ClusterObserver, Arc<dyn GroupStatusProvider + Send + Sync>) {
244        let mut topology = ClusterTopology::new();
245        for p in peers {
246            topology.add_node(p);
247        }
248        let provider: Arc<dyn GroupStatusProvider + Send + Sync> =
249            Arc::new(FakeProvider(raft_groups));
250        let observer = ClusterObserver::new(
251            1,
252            lifecycle,
253            Arc::new(RwLock::new(topology)),
254            Arc::new(RwLock::new(routing)),
255            Arc::downgrade(&provider),
256        );
257        (observer, provider)
258    }
259
260    fn gs(group_id: u64, role: &str, leader: u64) -> GroupStatus {
261        GroupStatus {
262            group_id,
263            role: role.into(),
264            leader_id: leader,
265            term: 1,
266            commit_index: 5,
267            last_applied: 5,
268            last_log_index: 5,
269            snapshot_index: 0,
270            member_count: 3,
271            learner_count: 0,
272            vshard_count: 512,
273        }
274    }
275
276    #[test]
277    fn snapshot_renders_full_state() {
278        let lifecycle = ClusterLifecycleTracker::new();
279        lifecycle.to_ready(3);
280
281        let peers = vec![
282            NodeInfo::new(1, "10.0.0.1:9400".parse().unwrap(), NodeState::Active),
283            NodeInfo::new(2, "10.0.0.2:9400".parse().unwrap(), NodeState::Active),
284            NodeInfo::new(3, "10.0.0.3:9400".parse().unwrap(), NodeState::Active),
285        ];
286
287        let mut routing = RoutingTable::uniform(2, &[1, 2, 3], 3);
288        // Inject a learner to verify it lands in the snapshot.
289        routing.add_group_learner(0, 4);
290
291        let raft_groups = vec![gs(0, "Leader", 1), gs(1, "Follower", 2)];
292
293        let (observer, _provider) = make_observer(lifecycle, peers, routing, raft_groups);
294        let snap = observer.snapshot();
295
296        assert_eq!(snap.node_id, 1);
297        assert_eq!(snap.lifecycle_label(), "ready");
298        assert_eq!(snap.members_count(), 3);
299        assert_eq!(snap.groups_count(), 2);
300
301        // Group 0 should carry both members (1,2,3) and the injected
302        // learner (4).
303        let g0 = snap
304            .groups
305            .iter()
306            .find(|g| g.group_id == 0)
307            .expect("group 0 present");
308        assert_eq!(g0.role, "Leader");
309        assert_eq!(g0.leader_id, 1);
310        assert!(g0.members.contains(&1));
311        assert!(g0.learners.contains(&4));
312
313        // Peer snapshots preserve addresses.
314        let addrs: Vec<&str> = snap.peers.iter().map(|p| p.addr.as_str()).collect();
315        assert!(addrs.contains(&"10.0.0.1:9400"));
316        assert!(addrs.contains(&"10.0.0.3:9400"));
317    }
318
319    #[test]
320    fn snapshot_without_groups() {
321        // A node whose RaftLoop reports zero groups (not in cluster
322        // mode, but we got a stub observer). Topology still rendered.
323        let lifecycle = ClusterLifecycleTracker::new();
324        lifecycle.to_bootstrapping();
325        let peers = vec![NodeInfo::new(
326            1,
327            "127.0.0.1:9400".parse().unwrap(),
328            NodeState::Active,
329        )];
330        let routing = RoutingTable::uniform(1, &[1], 1);
331        let (observer, _provider) = make_observer(lifecycle, peers, routing, vec![]);
332
333        let snap = observer.snapshot();
334        assert_eq!(snap.lifecycle_label(), "bootstrapping");
335        assert_eq!(snap.members_count(), 1);
336        assert_eq!(snap.groups_count(), 0);
337    }
338
339    #[test]
340    fn snapshot_is_json_roundtrippable() {
341        // The shape is consumed by clients, so serde must round-trip
342        // losslessly for every variant. This keeps downstream tooling
343        // honest as fields are added.
344        let lifecycle = ClusterLifecycleTracker::new();
345        lifecycle.to_joining(2);
346        let peers = vec![NodeInfo::new(
347            1,
348            "127.0.0.1:9400".parse().unwrap(),
349            NodeState::Active,
350        )];
351        let routing = RoutingTable::uniform(1, &[1], 1);
352        let (observer, _provider) =
353            make_observer(lifecycle, peers, routing, vec![gs(0, "Leader", 1)]);
354        let snap = observer.snapshot();
355
356        let json = serde_json::to_string(&snap).expect("serialize");
357        let back: ClusterInfoSnapshot = serde_json::from_str(&json).expect("deserialize");
358        assert_eq!(back.node_id, snap.node_id);
359        assert_eq!(back.peers.len(), 1);
360        assert_eq!(back.groups.len(), 1);
361        // Joining preserves the attempt field through serde.
362        match back.lifecycle {
363            ClusterLifecycleState::Joining { attempt } => assert_eq!(attempt, 2),
364            other => panic!("expected Joining, got {other:?}"),
365        }
366    }
367}
nodedb_cluster/cluster_info.rs

nodedb_cluster/
cluster_info.rs
