orlando-cluster 0.1.0

A virtual actor framework in Rust, inspired by Microsoft Orleans.
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

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use arc_swap::ArcSwap;
use tokio::sync::watch;

use orlando_core::ClusterId;

use crate::connection_pool::ConnectionPool;

/// Configuration for a multi-cluster deployment.
#[derive(Debug, Clone)]
pub struct MultiClusterConfig {
    /// This cluster's identity.
    pub cluster_id: ClusterId,
    /// Gateway endpoints of peer clusters (cluster_id -> "host:port").
    pub peers: HashMap<ClusterId, String>,
    /// How often to health-check peer clusters.
    pub health_check_interval: Duration,
}

impl MultiClusterConfig {
    pub fn new(cluster_id: impl Into<String>) -> Self {
        Self {
            cluster_id: ClusterId::new(cluster_id),
            peers: HashMap::new(),
            health_check_interval: Duration::from_secs(10),
        }
    }

    pub fn peer(mut self, cluster_id: impl Into<String>, endpoint: impl Into<String>) -> Self {
        self.peers
            .insert(ClusterId::new(cluster_id), endpoint.into());
        self
    }

    pub fn health_check_interval(mut self, interval: Duration) -> Self {
        self.health_check_interval = interval;
        self
    }
}

/// Status of a peer cluster.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PeerStatus {
    Healthy,
    Unreachable,
    Unknown,
}

/// Tracks health of peer clusters via periodic gRPC probes.
pub struct ClusterHealth {
    config: MultiClusterConfig,
    pool: Arc<ConnectionPool>,
    statuses: Arc<ArcSwap<HashMap<ClusterId, PeerStatus>>>,
    shutdown_rx: watch::Receiver<bool>,
}

impl ClusterHealth {
    pub fn new(
        config: MultiClusterConfig,
        pool: Arc<ConnectionPool>,
        shutdown_rx: watch::Receiver<bool>,
    ) -> Self {
        let initial: HashMap<ClusterId, PeerStatus> = config
            .peers
            .keys()
            .map(|id| (id.clone(), PeerStatus::Unknown))
            .collect();

        Self {
            config,
            pool,
            statuses: Arc::new(ArcSwap::from_pointee(initial)),
            shutdown_rx,
        }
    }

    /// Get the current status of a peer cluster.
    pub fn status(&self, cluster_id: &ClusterId) -> PeerStatus {
        self.statuses
            .load()
            .get(cluster_id)
            .cloned()
            .unwrap_or(PeerStatus::Unknown)
    }

    /// Get a snapshot of all peer statuses.
    pub fn all_statuses(&self) -> Arc<HashMap<ClusterId, PeerStatus>> {
        self.statuses.load_full()
    }

    /// Get the endpoint for a peer cluster.
    pub fn peer_endpoint(&self, cluster_id: &ClusterId) -> Option<String> {
        self.config.peers.get(cluster_id).cloned()
    }

    /// Run the health check loop (call via `tokio::spawn`).
    pub async fn run(mut self) {
        loop {
            tokio::select! {
                _ = tokio::time::sleep(self.config.health_check_interval) => {}
                _ = self.shutdown_rx.changed() => {
                    tracing::debug!("cluster health checker shutting down");
                    return;
                }
            }

            let mut new_statuses = HashMap::new();
            for (cluster_id, endpoint) in &self.config.peers {
                let status = match self.pool.get_membership(endpoint).await {
                    Ok(mut client) => {
                        match tokio::time::timeout(
                            Duration::from_secs(5),
                            client.get_members(crate::proto::GetMembersRequest {}),
                        )
                        .await
                        {
                            Ok(Ok(_)) => PeerStatus::Healthy,
                            _ => PeerStatus::Unreachable,
                        }
                    }
                    Err(_) => PeerStatus::Unreachable,
                };

                if status != self.status(cluster_id) {
                    tracing::info!(
                        cluster = %cluster_id,
                        ?status,
                        "peer cluster status changed"
                    );
                }
                new_statuses.insert(cluster_id.clone(), status);
            }
            self.statuses.store(Arc::new(new_statuses));
        }
    }
}