From 3fb4dcd52e3de48c57dc8087672d4bb5848e9a40 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 27 Jun 2026 09:47:30 +0800 Subject: [PATCH] refactor: route cluster control plane readiness (#3935) --- crates/ecstore/src/api/mod.rs | 5 +- crates/ecstore/src/cluster/control_plane.rs | 79 +++++++++++++++ crates/ecstore/src/cluster/mod.rs | 5 +- docs/architecture/migration-progress.md | 82 ++++++++++++--- rustfs/src/admin/handlers/cluster_snapshot.rs | 99 ++++++++++++++++++- rustfs/src/admin/storage_api.rs | 4 +- rustfs/src/app/admin_usecase.rs | 16 ++- rustfs/src/cluster_snapshot.rs | 7 +- rustfs/src/server/mod.rs | 2 +- rustfs/src/server/readiness.rs | 83 +++++++++++----- rustfs/src/storage/storage_api.rs | 4 +- rustfs/src/storage_api.rs | 2 +- scripts/check_architecture_migration_rules.sh | 13 +++ 13 files changed, 343 insertions(+), 58 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 7a40199d7..0ce9920c4 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -49,9 +49,10 @@ pub mod cluster { pub use crate::cluster::{ ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, - ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot, local_node_storage_snapshot_from_membership, + ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, + ClusterRpcChannelSnapshot, ClusterRpcPlane, ClusterRpcTransport, local_node_storage_snapshot_from_membership, membership_snapshot_from_endpoint_pools, peer_health_snapshot_from_membership, pool_state_snapshot_from_endpoint_pools, - topology_snapshot_from_endpoint_pools, topology_snapshot_from_endpoint_pools_with_capabilities, + rpc_boundary_snapshot, topology_snapshot_from_endpoint_pools, topology_snapshot_from_endpoint_pools_with_capabilities, }; } diff --git a/crates/ecstore/src/cluster/control_plane.rs b/crates/ecstore/src/cluster/control_plane.rs index e8bfa877d..fa263c7f7 100644 --- a/crates/ecstore/src/cluster/control_plane.rs +++ b/crates/ecstore/src/cluster/control_plane.rs @@ -32,6 +32,8 @@ const FAILURE_DOMAIN_NOT_REPORTED: &str = "failure domain labels not reported by const NUMA_NOT_WIRED: &str = "NUMA topology not wired into runtime"; const PROFILING_NOT_WIRED: &str = "profiling capability not wired into ECStore"; const PEER_HEALTH_NOT_REPORTED: &str = "peer health not reported by endpoints"; +const CONTROL_RPC_SEPARATED: &str = "control RPC remains on the gRPC control plane"; +const DATA_STREAM_RPC_SEPARATED: &str = "remote disk data streams remain on the internode data transport"; #[derive(Debug, Clone)] pub struct ClusterControlPlane { @@ -65,6 +67,10 @@ impl ClusterControlPlane { peer_health_snapshot_from_membership(&membership) } + pub fn rpc_boundary_snapshot(&self) -> ClusterRpcBoundarySnapshot { + rpc_boundary_snapshot() + } + pub fn read_snapshot(&self) -> ClusterControlPlaneSnapshot { let membership = self.membership_snapshot(); @@ -73,6 +79,7 @@ impl ClusterControlPlane { pool_state: self.pool_state_snapshot(), local_storage: local_node_storage_snapshot_from_membership(&membership), peer_health: peer_health_snapshot_from_membership(&membership), + rpc_boundary: self.rpc_boundary_snapshot(), membership, } } @@ -84,6 +91,7 @@ pub struct ClusterControlPlaneSnapshot { pub pool_state: ClusterPoolStateSnapshot, pub local_storage: ClusterLocalNodeStorageSnapshot, pub peer_health: ClusterPeerHealthSnapshot, + pub rpc_boundary: ClusterRpcBoundarySnapshot, pub membership: ClusterMembershipSnapshot, } @@ -154,6 +162,32 @@ pub struct ClusterPeerHealth { pub status: CapabilityStatus, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ClusterRpcBoundarySnapshot { + pub control_channels: Vec, + pub data_channels: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterRpcChannelSnapshot { + pub name: String, + pub plane: ClusterRpcPlane, + pub transport: ClusterRpcTransport, + pub status: CapabilityStatus, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClusterRpcPlane { + Control, + Data, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClusterRpcTransport { + Grpc, + InternodeDataTransport, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] pub enum ClusterEndpointType { Path, @@ -313,6 +347,35 @@ pub fn peer_health_snapshot_from_membership(membership: &ClusterMembershipSnapsh } } +pub fn rpc_boundary_snapshot() -> ClusterRpcBoundarySnapshot { + ClusterRpcBoundarySnapshot { + control_channels: ["metadata", "lock", "health", "administrative"] + .into_iter() + .map(|name| rpc_channel_snapshot(name, ClusterRpcPlane::Control, ClusterRpcTransport::Grpc, CONTROL_RPC_SEPARATED)) + .collect(), + data_channels: vec![rpc_channel_snapshot( + "remote_disk_stream", + ClusterRpcPlane::Data, + ClusterRpcTransport::InternodeDataTransport, + DATA_STREAM_RPC_SEPARATED, + )], + } +} + +fn rpc_channel_snapshot( + name: &str, + plane: ClusterRpcPlane, + transport: ClusterRpcTransport, + reason: &'static str, +) -> ClusterRpcChannelSnapshot { + ClusterRpcChannelSnapshot { + name: name.to_owned(), + plane, + transport, + status: CapabilityStatus::supported().with_reason(reason), + } +} + fn topology_sets_from_endpoints( pool_index: usize, drives_per_set: usize, @@ -552,9 +615,25 @@ mod tests { assert_eq!(snapshot.pool_state.pools[0].endpoint_count, 4); assert_eq!(snapshot.local_storage.nodes.len(), 2); assert_eq!(snapshot.peer_health.peers.len(), 3); + assert_eq!(snapshot.rpc_boundary.control_channels.len(), 4); + assert_eq!(snapshot.rpc_boundary.data_channels.len(), 1); assert_eq!(node_ids, BTreeSet::from([LOCAL_NODE_ID, "node1.example:9000", "node2.example:9000"])); } + #[test] + fn rpc_boundary_snapshot_keeps_control_rpc_separate_from_data_streams() { + let snapshot = rpc_boundary_snapshot(); + + assert!( + snapshot + .control_channels + .iter() + .all(|channel| channel.plane == ClusterRpcPlane::Control && channel.transport == ClusterRpcTransport::Grpc) + ); + assert_eq!(snapshot.data_channels[0].plane, ClusterRpcPlane::Data); + assert_eq!(snapshot.data_channels[0].transport, ClusterRpcTransport::InternodeDataTransport); + } + fn sample_path_endpoint_pools() -> EndpointServerPools { let endpoints = (0..4) .map(|index| { diff --git a/crates/ecstore/src/cluster/mod.rs b/crates/ecstore/src/cluster/mod.rs index 98ff9084b..b7c91dea6 100644 --- a/crates/ecstore/src/cluster/mod.rs +++ b/crates/ecstore/src/cluster/mod.rs @@ -18,7 +18,8 @@ pub(crate) mod rpc; pub use control_plane::{ ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, - ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot, local_node_storage_snapshot_from_membership, - membership_snapshot_from_endpoint_pools, peer_health_snapshot_from_membership, pool_state_snapshot_from_endpoint_pools, + ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, ClusterRpcChannelSnapshot, + ClusterRpcPlane, ClusterRpcTransport, local_node_storage_snapshot_from_membership, membership_snapshot_from_endpoint_pools, + peer_health_snapshot_from_membership, pool_state_snapshot_from_endpoint_pools, rpc_boundary_snapshot, topology_snapshot_from_endpoint_pools, topology_snapshot_from_endpoint_pools_with_capabilities, }; diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 6d1b45e59..d3f82ff0e 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,23 +5,31 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-ecstore-services-domain-batch` +- Branch: `overtrue/arch-cluster-readonly-control-plane-batch` - Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124/API-125/API-126/API-127/API-128/API-129/API-130/API-131/API-132/API-133/API-134/API-135/API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153/API-154/API-155/API-156/API-157/API-158/API-159/API-160/API-161/API-162/API-163/API-164/API-165/API-166/API-167/API-168/API-169/API-170/API-171/API-172/API-173/API-174/API-175/API-176/API-177/API-178/API-179/API-180/API-181/API-182/API-183/API-184/API-185/API-186/API-187/API-188/API-189/API-190/API-191/API-192/API-193/API-194/API-195/API-196/API-197/API-198/API-199/API-200/API-201/API-202/API-203/API-204/API-205/API-206/API-207/API-208/API-209/API-210/API-211/API-212/API-213/API-214/API-215/API-216/API-217/API-218/API-219/API-220/API-221/API-222/API-223/API-224/API-225/API-226/API-227/API-228/API-229/API-230/API-231/API-232/API-233/API-234/API-235/API-236/API-237/API-238/API-239/API-240/API-241/API-242/API-243/API-244/API-245/API-246/API-247/API-248/API-249/API-250/API-251/API-252/API-253/API-254/CTX-002`. - Current baseline also includes API-255 from PR #3923, API-256 from PR #3925, and CFG-009 from PR #3927. -- Current phase PR: E-035/E-008 ECStore services domain batch. -- Based on: E-034 branch while PR #3932 and follow-up ECStore owner PRs are - pending; rebase onto `origin/main` after E-019 through E-034 merge before - opening this PR. -- PR type for this branch: `pure-move`. -- Runtime behavior changes: none expected for E-035/E-008; production changes - only move the ECStore rebalance and tier owner modules under `services` while - retaining the public `api::rebalance` and `api::tier` facades. -- Rust code changes: move the ECStore root `rebalance` and `tier` directories - into `services`, remove the root `lib.rs` declarations, and migrate - crate-internal imports to `services::{rebalance,tier}` without function-body - changes. -- CI/script changes: reject restoring ECStore root `store.rs`, `set_disk.rs`, +- Current phase PR: C-007/C-009 cluster read-only control-plane consumer, RPC + boundary, and readiness invariant batch. +- Based on: E-035/E-008 branch while PR #3933 and the E-031 through E-035 + follow-up ECStore owner batch are pending; rebase onto `origin/main` after + those PRs merge before opening this PR. +- PR type for this branch: `contract`. +- Runtime behavior changes: admin readiness reads keep the existing readiness + report collector and, when endpoint context is available, project that same + report through the read-only cluster snapshot runtime status before returning + readiness; `FullReady` publication now uses the existing storage, IAM, and + lock quorum readiness invariant. +- Rust code changes: add a read-only cluster RPC boundary snapshot that models + metadata/lock/health/admin control RPC as gRPC and remote-disk streams as the + internode data transport, expose it through the ECStore public cluster facade, + include it in RustFS cluster snapshots and admin JSON views, route admin + readiness usecase reads through the snapshot runtime status using the existing + readiness report collector, and require lock quorum before publishing + `FullReady`. +- CI/script changes: guard that the ECStore RPC boundary snapshot and + internode data-transport control-plane separation note remain present; reject + restoring ECStore root `store.rs`, `set_disk.rs`, `store_list_objects.rs`, `store_utils.rs`, `store_init.rs`, `disks_layout.rs`, `endpoints.rs`, `storage_api_contracts.rs`, `batch_processor.rs`, `event_notification.rs`, `metrics_realtime.rs`, or @@ -92,7 +100,8 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block codec or coding modules after E-033/E-004, and reject restoring ECStore root RPC facade modules or ECStore-internal `crate::rpc` consumers after E-034, and reject restoring ECStore root rebalance/tier service-domain modules or - ECStore-internal `crate::rebalance`/`crate::tier` consumers after E-035/E-008. + ECStore-internal `crate::rebalance`/`crate::tier` consumers after E-035/E-008, + and reject losing the C-009 cluster RPC boundary model. ## Phase 0 Tasks @@ -1386,6 +1395,34 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block - Verification: ECStore peer-health tests, compile coverage, formatting, diff hygiene, risk scan, pre-commit quality gate, and three-expert review. +- [x] `C-007` Route admin readiness reads to snapshot. + - Completed slice: route the admin readiness usecase through the read-only + cluster snapshot runtime status when endpoint context is available, while + retaining the existing readiness report collector. + - Acceptance: admin readiness reads use the same runtime status that the + cluster snapshot exposes; systems without endpoint context keep the old + fallback. + - Must preserve: health endpoint semantics, readiness metrics, startup + readiness publication, storage/IAM/lock readiness calculations, and all + S3/RPC gate behavior. + - Verification: focused RustFS cluster snapshot/admin usecase tests, compile + coverage, migration guard, formatting, diff hygiene, risk scan, PR quality + gate, and three-expert review. + +- [x] `C-009` Model control RPC separately. + - Completed slice: add a read-only RPC boundary snapshot that keeps + metadata, lock, health, and administrative control RPC on the gRPC control + plane while modeling remote-disk streams as internode data transport. + - Acceptance: cluster snapshots expose separate control and data channel + groups, and migration rules fail if the RPC boundary model or + data-transport separation note is removed. + - Must preserve: internode data stream behavior, remote lock behavior, peer + health behavior, metadata/admin RPC behavior, and all transport selection + defaults. + - Verification: ECStore cluster tests, RustFS cluster snapshot handler tests, + compile coverage, migration guard, formatting, diff hygiene, risk scan, PR + quality gate, and three-expert review. + - [x] `TEST-PRTYPE-001` Check PR type enum consistency. - Acceptance: `./scripts/check_architecture_migration_rules.sh` parses the allowed PR types from [`crate-boundaries.md`](crate-boundaries.md) and fails @@ -9346,6 +9383,21 @@ Notes: public APIs, boxed public errors, production println/eprintln, or relaxed ordering introduced in changed Rust files. +- Issue #660 C-007/C-009 current slice: + - `cargo check -p rustfs-ecstore --lib`: passed. + - `cargo check -p rustfs --lib`: passed. + - `cargo test -p rustfs-ecstore --lib cluster`: passed, 98 passed. + - `cargo test -p rustfs --lib cluster_snapshot`: passed, 9 passed including + admin cluster snapshot handler coverage. + - `cargo fmt --all --check`: passed. + - `bash -n scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `git diff --check overtrue/arch-ecstore-services-domain-batch...HEAD`: + passed. + - Rust risk scan: passed; no new unwrap/expect, numeric casts, string error + public APIs, boxed public errors, production println/eprintln, or relaxed + ordering introduced in changed Rust files. + ## Handoff Notes - Continue with larger consumer-migration batches outside the cleaned diff --git a/rustfs/src/admin/handlers/cluster_snapshot.rs b/rustfs/src/admin/handlers/cluster_snapshot.rs index f4a1eac04..63db2ac84 100644 --- a/rustfs/src/admin/handlers/cluster_snapshot.rs +++ b/rustfs/src/admin/handlers/cluster_snapshot.rs @@ -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, pub topology: Option, pub peer_health: Option, + pub rpc_boundary: Option, pub workload_admission: Option, pub runtime: Option, } @@ -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, pub runtime_status: ClusterRuntimeStatusView, @@ -167,6 +171,7 @@ impl From 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 for ClusterPeerHealthItemView { } } +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterRpcBoundaryView { + pub control_channels: Vec, + pub data_channels: Vec, +} + +impl From 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 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..] diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index 0b67917c2..c69b1cc63 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -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, diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index dcdd34cb7..dd37973f8 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -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 { diff --git a/rustfs/src/cluster_snapshot.rs b/rustfs/src/cluster_snapshot.rs index 4314508e6..e647d1726 100644 --- a/rustfs/src/cluster_snapshot.rs +++ b/rustfs/src/cluster_snapshot.rs @@ -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, diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index 3055af39e..8769662ab 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -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; diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 48987490a..b3598ada4 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -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; } } diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 793c1898c..5893ce6c0 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -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, }; } diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 6f98f14d0..188c327e1 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -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, }; } } diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 02e1edfc8..7471b37b2 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -3018,6 +3018,19 @@ if [[ -s "$ECSTORE_ROOT_RPC_IMPL_HITS_FILE" ]]; then report_failure "ECStore internal RPC consumers must use the cluster/rpc owner path: $(paste -sd '; ' "$ECSTORE_ROOT_RPC_IMPL_HITS_FILE")" fi +require_source_contains \ + "crates/ecstore/src/cluster/control_plane.rs" \ + "pub struct ClusterRpcBoundarySnapshot" \ + "ECStore cluster RPC boundary snapshot" +require_source_contains \ + "crates/ecstore/src/cluster/control_plane.rs" \ + "ClusterRpcTransport::Grpc" \ + "ECStore cluster control RPC transport model" +require_source_contains \ + "crates/ecstore/src/cluster/rpc/internode_data_transport.rs" \ + "Internode metadata, lock, health, and administrative calls remain on the" \ + "ECStore internode data transport control-plane separation note" + ( cd "$ROOT_DIR" {