gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-06 21:33:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: route cluster control plane readiness (#3935)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-27 09:47:30 +08:00
committed by GitHub
parent 0a5b1b1b3a
commit 3fb4dcd52e
13 changed files with 343 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/cluster_snapshot.rs
+97 -2
View File
@@ -20,7 +20,7 @@ use crate::admin::{
storage_api::cluster::{
ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot,
ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState,
ClusterPoolStateSnapshot,
ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, ClusterRpcChannelSnapshot, ClusterRpcPlane, ClusterRpcTransport,
},
system,
};
@@ -60,6 +60,7 @@ pub(crate) struct ClusterSnapshotDiscoveryResponse {
pub summary: Option<CapabilityStatus>,
pub topology: Option<CapabilityStatus>,
pub peer_health: Option<CapabilityStatus>,
pub rpc_boundary: Option<CapabilityStatus>,
pub workload_admission: Option<CapabilityStatus>,
pub runtime: Option<CapabilityStatus>,
}
@@ -124,6 +125,7 @@ pub(crate) async fn build_cluster_snapshot_discovery_response() -> ClusterSnapsh
summary: Some(summary.actionable_pressure.clone()),
topology: Some(summary.topology),
peer_health: Some(summary.peer_health),
rpc_boundary: Some(summary.rpc_boundary),
workload_admission: Some(summary.workload_admission),
runtime: Some(summary.runtime),
}
@@ -133,6 +135,7 @@ pub(crate) async fn build_cluster_snapshot_discovery_response() -> ClusterSnapsh
summary: None,
topology: None,
peer_health: None,
rpc_boundary: None,
workload_admission: None,
runtime: None,
},
@@ -149,6 +152,7 @@ pub(crate) struct ClusterSnapshotView {
pub pool_state: ClusterPoolStateView,
pub local_storage: ClusterLocalStorageView,
pub peer_health: ClusterPeerHealthView,
pub rpc_boundary: ClusterRpcBoundaryView,
pub observability: ObservabilitySnapshot,
pub workload_admission: Vec<WorkloadAdmissionView>,
pub runtime_status: ClusterRuntimeStatusView,
@@ -167,6 +171,7 @@ impl From<ClusterReadOnlySnapshot> for ClusterSnapshotView {
pool_state: ClusterPoolStateView::from(snapshot.pool_state),
local_storage: ClusterLocalStorageView::from(snapshot.local_storage),
peer_health: ClusterPeerHealthView::from(snapshot.peer_health),
rpc_boundary: ClusterRpcBoundaryView::from(snapshot.rpc_boundary),
observability: snapshot.observability,
workload_admission: workload_admission_views(snapshot.workload_admission),
runtime_status: ClusterRuntimeStatusView::from(snapshot.runtime_status),
@@ -181,6 +186,7 @@ pub(crate) struct ClusterSnapshotSummary {
pub topology: CapabilityStatus,
pub membership: CapabilityStatus,
pub peer_health: CapabilityStatus,
pub rpc_boundary: CapabilityStatus,
pub observability: CapabilityStatus,
pub workload_admission: CapabilityStatus,
pub actionable_pressure: CapabilityStatus,
@@ -191,6 +197,7 @@ impl From<&ClusterReadOnlySnapshot> for ClusterSnapshotSummary {
let topology = summarize_topology(snapshot);
let membership = summarize_membership(snapshot);
let peer_health = summarize_peer_health(snapshot);
let rpc_boundary = summarize_rpc_boundary(snapshot);
let observability = summarize_observability(snapshot);
let workload_admission = summarize_workload_admission(snapshot);
let actionable_pressure = if cluster_has_actionable_pressure(snapshot) {
@@ -204,6 +211,7 @@ impl From<&ClusterReadOnlySnapshot> for ClusterSnapshotSummary {
topology,
membership,
peer_health,
rpc_boundary,
observability,
workload_admission,
actionable_pressure,
@@ -372,6 +380,44 @@ impl From<ClusterPeerHealth> for ClusterPeerHealthItemView {
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub(crate) struct ClusterRpcBoundaryView {
pub control_channels: Vec<ClusterRpcChannelView>,
pub data_channels: Vec<ClusterRpcChannelView>,
}
impl From<ClusterRpcBoundarySnapshot> for ClusterRpcBoundaryView {
fn from(snapshot: ClusterRpcBoundarySnapshot) -> Self {
Self {
control_channels: snapshot
.control_channels
.into_iter()
.map(ClusterRpcChannelView::from)
.collect(),
data_channels: snapshot.data_channels.into_iter().map(ClusterRpcChannelView::from).collect(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub(crate) struct ClusterRpcChannelView {
pub name: String,
pub plane: &'static str,
pub transport: &'static str,
pub status: CapabilityStatus,
}
impl From<ClusterRpcChannelSnapshot> for ClusterRpcChannelView {
fn from(channel: ClusterRpcChannelSnapshot) -> Self {
Self {
name: channel.name,
plane: rpc_plane_label(channel.plane),
transport: rpc_transport_label(channel.transport),
status: channel.status,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub(crate) struct WorkloadAdmissionView {
pub class: &'static str,
@@ -427,6 +473,20 @@ fn endpoint_type_label(endpoint_type: ClusterEndpointType) -> &'static str {
}
}
fn rpc_plane_label(plane: ClusterRpcPlane) -> &'static str {
match plane {
ClusterRpcPlane::Control => "control",
ClusterRpcPlane::Data => "data",
}
}
fn rpc_transport_label(transport: ClusterRpcTransport) -> &'static str {
match transport {
ClusterRpcTransport::Grpc => "grpc",
ClusterRpcTransport::InternodeDataTransport => "internode_data_transport",
}
}
fn workload_class_label(class: WorkloadClass) -> &'static str {
class.as_str()
}
@@ -509,6 +569,16 @@ fn summarize_peer_health(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus
}
}
fn summarize_rpc_boundary(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus {
let has_control = !snapshot.rpc_boundary.control_channels.is_empty();
let has_data = !snapshot.rpc_boundary.data_channels.is_empty();
if has_control && has_data {
CapabilityStatus::supported().with_reason("cluster control RPC and data streams are modeled as separate planes")
} else {
CapabilityStatus::unknown().with_reason("cluster RPC boundary snapshot is incomplete")
}
}
fn summarize_workload_admission(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus {
let entries = snapshot.workload_admission.entries();
if entries.is_empty() {
@@ -595,7 +665,7 @@ mod tests {
use crate::admin::storage_api::cluster::{
ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot,
ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState,
ClusterPoolStateSnapshot,
ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, ClusterRpcChannelSnapshot, ClusterRpcPlane, ClusterRpcTransport,
};
use crate::cluster_snapshot::{ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot};
use crate::server::{DependencyReadiness, ReadinessDegradedReason};
@@ -632,6 +702,7 @@ mod tests {
assert_eq!(response.summary, None);
assert_eq!(response.topology, None);
assert_eq!(response.peer_health, None);
assert_eq!(response.rpc_boundary, None);
assert_eq!(response.workload_admission, None);
assert_eq!(response.runtime, None);
}
@@ -684,6 +755,7 @@ mod tests {
status: CapabilityStatus::unknown().with_reason("peer state unavailable"),
}],
},
rpc_boundary: sample_rpc_boundary_snapshot(),
observability: ObservabilitySnapshot::default(),
workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![
WorkloadAdmissionSnapshot::new(WorkloadClass::Repair, AdmissionState::Unknown)
@@ -707,8 +779,12 @@ mod tests {
assert_eq!(value["membership"]["drives"][0]["endpoint_type"], "url");
assert_eq!(value["workload_admission"][0]["class"], "repair");
assert_eq!(value["workload_admission"][0]["state"], "unknown");
assert_eq!(value["rpc_boundary"]["control_channels"][0]["name"], "metadata");
assert_eq!(value["rpc_boundary"]["control_channels"][0]["transport"], "grpc");
assert_eq!(value["rpc_boundary"]["data_channels"][0]["transport"], "internode_data_transport");
assert_eq!(value["runtime_status"]["state"], "degraded");
assert_eq!(value["summary"]["runtime"]["state"], "unknown");
assert_eq!(value["summary"]["rpc_boundary"]["state"], "supported");
assert_eq!(value["runtime_status"]["degraded_reasons"][0], "storage_and_lock_unavailable");
assert_eq!(value["actionable_pressure"], true);
}
@@ -721,6 +797,7 @@ mod tests {
pool_state: ClusterPoolStateSnapshot::default(),
local_storage: ClusterLocalNodeStorageSnapshot::default(),
peer_health: ClusterPeerHealthSnapshot::default(),
rpc_boundary: sample_rpc_boundary_snapshot(),
observability: ObservabilitySnapshot::default(),
workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![WorkloadAdmissionSnapshot::new(
WorkloadClass::ForegroundRead,
@@ -741,10 +818,28 @@ mod tests {
assert_eq!(summary.runtime.state, CapabilityState::Supported);
assert_eq!(summary.membership.state, CapabilityState::Unknown);
assert_eq!(summary.peer_health.state, CapabilityState::Unknown);
assert_eq!(summary.rpc_boundary.state, CapabilityState::Supported);
assert_eq!(summary.workload_admission.state, CapabilityState::Supported);
assert_eq!(summary.actionable_pressure.state, CapabilityState::Disabled);
}
fn sample_rpc_boundary_snapshot() -> ClusterRpcBoundarySnapshot {
ClusterRpcBoundarySnapshot {
control_channels: vec![ClusterRpcChannelSnapshot {
name: "metadata".to_string(),
plane: ClusterRpcPlane::Control,
transport: ClusterRpcTransport::Grpc,
status: CapabilityStatus::supported().with_reason("control RPC remains on gRPC"),
}],
data_channels: vec![ClusterRpcChannelSnapshot {
name: "remote_disk_stream".to_string(),
plane: ClusterRpcPlane::Data,
transport: ClusterRpcTransport::InternodeDataTransport,
status: CapabilityStatus::supported().with_reason("remote disk data streams remain separate"),
}],
}
}
fn extract_block_between_markers(src: &str, start: &str, end: &str) -> String {
let start_index = src.find(start).expect("start marker should exist");
let end_index = src[start_index..]
rustfs/src/admin/storage_api.rs
+2 -2
View File
@@ -36,7 +36,7 @@ pub(crate) mod ecstore_cluster {
pub(crate) use crate::storage::storage_api::ecstore_cluster::{
ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot,
ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState,
ClusterPoolStateSnapshot,
ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, ClusterRpcChannelSnapshot, ClusterRpcPlane, ClusterRpcTransport,
};
}
@@ -447,7 +447,7 @@ pub(crate) mod cluster {
pub(crate) use super::ecstore_cluster::{
ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot,
ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState,
ClusterPoolStateSnapshot,
ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, ClusterRpcChannelSnapshot, ClusterRpcPlane, ClusterRpcTransport,
};
pub(crate) use super::storage_contracts::{
CapabilitySnapshotError, CapabilityState, CapabilityStatus, ObservabilitySnapshot, ObservabilitySnapshotProvider,
rustfs/src/app/admin_usecase.rs
+13 -3
View File
@@ -28,9 +28,12 @@ use crate::app::runtime_sources::{
AppContext, current_app_context, resolve_endpoints_handle, resolve_object_store_handle_for_context,
};
use crate::capacity::resolve_admin_used_capacity;
use crate::cluster_snapshot::{ClusterReadOnlySnapshot, collect_cluster_read_only_snapshot};
use crate::cluster_snapshot::{
ClusterReadOnlySnapshot, ClusterRuntimeStatusSnapshot, cluster_read_only_snapshot_from_endpoint_pools,
collect_cluster_read_only_snapshot,
};
use crate::error::ApiError;
use crate::server::{DependencyReadiness, collect_dependency_readiness as collect_runtime_dependency_readiness};
use crate::server::{DependencyReadiness, collect_dependency_readiness_report as collect_runtime_dependency_readiness_report};
use rustfs_data_usage::DataUsageInfo;
use rustfs_madmin::{InfoMessage, StorageInfo};
use s3s::S3ErrorCode;
@@ -579,7 +582,14 @@ impl DefaultAdminUsecase {
}
pub async fn execute_collect_dependency_readiness(&self) -> DependencyReadiness {
collect_runtime_dependency_readiness().await
let report = collect_runtime_dependency_readiness_report().await;
if let Some(endpoint_pools) = resolve_endpoints_handle() {
let runtime_status = ClusterRuntimeStatusSnapshot::from_readiness_report(report);
return cluster_read_only_snapshot_from_endpoint_pools(&endpoint_pools, runtime_status)
.runtime_status
.readiness;
}
report.readiness
}
pub async fn execute_collect_cluster_read_only_snapshot(&self) -> Option<ClusterReadOnlySnapshot> {
rustfs/src/cluster_snapshot.rs
+6 -1
View File
@@ -21,7 +21,7 @@ use crate::storage_api::cluster::contract::observability::ObservabilitySnapshot;
use crate::storage_api::cluster::contract::topology::TopologySnapshot;
use crate::storage_api::cluster::control_plane::{
ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot,
ClusterPeerHealthSnapshot, ClusterPoolStateSnapshot,
ClusterPeerHealthSnapshot, ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot,
};
use crate::workload_admission::workload_admission_registry_snapshot;
use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot};
@@ -33,6 +33,7 @@ pub struct ClusterReadOnlySnapshot {
pub pool_state: ClusterPoolStateSnapshot,
pub local_storage: ClusterLocalNodeStorageSnapshot,
pub peer_health: ClusterPeerHealthSnapshot,
pub rpc_boundary: ClusterRpcBoundarySnapshot,
pub observability: ObservabilitySnapshot,
pub workload_admission: WorkloadAdmissionRegistrySnapshot,
pub runtime_status: ClusterRuntimeStatusSnapshot,
@@ -94,6 +95,7 @@ pub fn cluster_read_only_snapshot_from_control_plane(
pool_state: control_plane.pool_state,
local_storage: control_plane.local_storage,
peer_health: control_plane.peer_health,
rpc_boundary: control_plane.rpc_boundary,
observability: runtime_observability_snapshot(),
workload_admission: workload_admission_registry_snapshot(),
runtime_status,
@@ -165,6 +167,8 @@ mod tests {
assert_eq!(snapshot.pool_state.pools[0].endpoint_count, 4);
assert_eq!(snapshot.local_storage.nodes.len(), 1);
assert_eq!(snapshot.peer_health.peers.len(), 2);
assert_eq!(snapshot.rpc_boundary.control_channels.len(), 4);
assert_eq!(snapshot.rpc_boundary.data_channels.len(), 1);
assert_eq!(snapshot.runtime_status.state, ClusterRuntimeReadinessState::Degraded);
assert_eq!(
snapshot.runtime_status.degraded_reasons,
@@ -193,6 +197,7 @@ mod tests {
pool_state: ClusterPoolStateSnapshot::default(),
local_storage: ClusterLocalNodeStorageSnapshot::default(),
peer_health: ClusterPeerHealthSnapshot::default(),
rpc_boundary: ClusterRpcBoundarySnapshot::default(),
observability: ObservabilitySnapshot::default(),
workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![WorkloadAdmissionSnapshot::new(
WorkloadClass::ForegroundRead,
rustfs/src/server/mod.rs
+1 -1
View File
@@ -67,7 +67,7 @@ pub(crate) use readiness::DependencyReadiness;
pub(crate) use readiness::DependencyReadinessReport;
pub(crate) use readiness::ReadinessDegradedReason;
pub(crate) use readiness::ReadinessGateLayer;
pub(crate) use readiness::collect_dependency_readiness;
pub(crate) use readiness::collect_dependency_readiness_report;
pub(crate) use readiness::collect_node_readiness_report;
pub use readiness::publish_ready_when_runtime_ready;
pub(crate) use readiness::snapshot_dependency_readiness_report;
rustfs/src/server/readiness.rs
+56 -27
View File
@@ -222,6 +222,7 @@ pub async fn publish_ready_when_runtime_ready(
target: "rustfs::server::readiness",
storage_ready = dependency_readiness.storage_ready,
iam_ready = dependency_readiness.iam_ready,
lock_quorum_ready = dependency_readiness.lock_quorum_ready,
"Runtime node readiness reached; publishing ready state"
);
},
@@ -558,10 +559,6 @@ fn dependency_readiness_report_from_readiness(readiness: DependencyReadiness) ->
}
}
pub async fn collect_dependency_readiness() -> DependencyReadiness {
collect_dependency_readiness_report().await.readiness
}
pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport {
let iam_ready_raw = runtime_sources::iam_ready();
let storage_ready = if let Some(cached) = load_cached_storage_readiness().await {
@@ -595,7 +592,7 @@ pub async fn collect_node_readiness_report() -> DependencyReadinessReport {
let readiness = DependencyReadiness {
storage_ready: runtime_sources::object_store_handle().is_some(),
iam_ready: runtime_sources::iam_ready(),
lock_quorum_ready: true,
lock_quorum_ready: collect_lock_quorum_status().await.ready,
};
let report = dependency_readiness_report_from_readiness(readiness);
record_readiness_report(&report);
@@ -813,7 +810,7 @@ where
loop {
let readiness = load_readiness().await;
if readiness.storage_ready && readiness.iam_ready {
if readiness.storage_ready && readiness.iam_ready && readiness.lock_quorum_ready {
on_ready(readiness);
return Ok(());
}
@@ -970,11 +967,11 @@ mod tests {
}
#[tokio::test]
async fn wait_for_runtime_readiness_with_publishes_ready_without_lock_quorum() {
async fn wait_for_runtime_readiness_with_does_not_publish_ready_without_lock_quorum() {
let readiness = GlobalReadiness::new();
let state_manager = ServiceStateManager::new();
let result = wait_for_runtime_readiness_with(
let err = wait_for_runtime_readiness_with(
Duration::ZERO,
Duration::from_millis(1),
|| {
@@ -989,9 +986,37 @@ mod tests {
state_manager.update(ServiceState::Ready);
},
)
.await
.expect_err("startup readiness should require lock quorum");
assert!(err.to_string().contains("lock_quorum_ready=false"));
assert!(!readiness.is_ready());
assert_eq!(state_manager.current_state(), ServiceState::Starting);
}
#[tokio::test]
async fn wait_for_runtime_readiness_with_publishes_ready_when_dependencies_are_ready() {
let readiness = GlobalReadiness::new();
let state_manager = ServiceStateManager::new();
let result = wait_for_runtime_readiness_with(
Duration::ZERO,
Duration::from_millis(1),
|| {
future::ready(DependencyReadiness {
storage_ready: true,
iam_ready: true,
lock_quorum_ready: true,
})
},
|_| {
readiness.mark_stage(rustfs_common::SystemStage::FullReady);
state_manager.update(ServiceState::Ready);
},
)
.await;
assert!(result.is_ok(), "lock quorum must not block node readiness publication");
assert!(result.is_ok(), "all runtime dependencies should publish readiness");
assert!(readiness.is_ready());
assert_eq!(state_manager.current_state(), ServiceState::Ready);
}
@@ -1264,30 +1289,34 @@ mod tests {
}
#[tokio::test]
#[serial]
async fn lock_quorum_status_cache_roundtrip() {
let cache = lock_quorum_status_cache();
{
let mut guard = cache.lock().await;
*guard = None;
}
async_with_vars([(rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000"))], async {
let cache = lock_quorum_status_cache();
{
let mut guard = cache.lock().await;
*guard = None;
}
update_lock_quorum_status_cache(LockQuorumStatus {
ready: true,
connected_clients: 2,
total_clients: 3,
required_quorum: 2,
})
.await;
let cached = load_cached_lock_quorum_status().await;
assert_eq!(
cached,
Some(LockQuorumStatus {
update_lock_quorum_status_cache(LockQuorumStatus {
ready: true,
connected_clients: 2,
total_clients: 3,
required_quorum: 2,
})
);
.await;
let cached = load_cached_lock_quorum_status().await;
assert_eq!(
cached,
Some(LockQuorumStatus {
ready: true,
connected_clients: 2,
total_clients: 3,
required_quorum: 2,
})
);
})
.await;
}
}
rustfs/src/storage/storage_api.rs
+2 -2
View File
@@ -334,8 +334,8 @@ pub(crate) mod ecstore_cluster {
pub(crate) use rustfs_ecstore::api::cluster::{
ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage,
ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth,
ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot,
topology_snapshot_from_endpoint_pools_with_capabilities,
ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot,
ClusterRpcChannelSnapshot, ClusterRpcPlane, ClusterRpcTransport, topology_snapshot_from_endpoint_pools_with_capabilities,
};
}
rustfs/src/storage_api.rs
+1 -1
View File
@@ -52,7 +52,7 @@ pub(crate) mod cluster {
pub(crate) mod control_plane {
pub(crate) use crate::storage::storage_api::ecstore_cluster::{
ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot,
ClusterPeerHealthSnapshot, ClusterPoolStateSnapshot,
ClusterPeerHealthSnapshot, ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot,
};
}
}