From c6ecfae39e31986d9095e08c11a53cbf62985f15 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 27 Jun 2026 05:54:25 +0800 Subject: [PATCH] refactor: move ecstore owner layout modules (#3932) --- crates/ecstore/src/cluster/control_plane.rs | 635 ++++++++++++++++++ crates/ecstore/src/cluster/mod.rs | 626 +---------------- crates/ecstore/src/{ => core}/pools.rs | 0 crates/ecstore/src/{ => core}/pools_test.rs | 0 crates/ecstore/src/{ => core}/sets.rs | 44 +- crates/ecstore/src/{ => core}/store_test.rs | 0 .../backpressure.rs} | 0 .../mod.rs} | 0 .../src/{data_usage.rs => data_usage/mod.rs} | 0 .../{ => diagnostics}/admin_server_info.rs | 0 .../get.rs} | 0 crates/ecstore/src/{error.rs => error/mod.rs} | 0 crates/ecstore/src/{ => io_support}/bitrot.rs | 0 .../ecstore/src/{ => io_support}/compress.rs | 0 crates/ecstore/src/{ => io_support}/rio.rs | 0 .../disks_layout_facade.rs} | 0 .../endpoints_facade.rs} | 0 crates/ecstore/src/lib.rs | 18 + .../src/{rebalance.rs => rebalance/mod.rs} | 0 crates/ecstore/src/{ => runtime}/global.rs | 0 .../sources.rs} | 0 .../src/{ => services}/batch_processor.rs | 0 .../src/{ => services}/event_notification.rs | 0 .../src/{ => services}/metrics_realtime.rs | 0 .../src/{ => services}/notification_sys.rs | 0 .../mod.rs} | 0 docs/architecture/migration-progress.md | 219 +++++- scripts/check_architecture_migration_rules.sh | 150 ++++- 28 files changed, 1050 insertions(+), 642 deletions(-) create mode 100644 crates/ecstore/src/cluster/control_plane.rs rename crates/ecstore/src/{ => core}/pools.rs (100%) rename crates/ecstore/src/{ => core}/pools_test.rs (100%) rename crates/ecstore/src/{ => core}/sets.rs (96%) rename crates/ecstore/src/{ => core}/store_test.rs (100%) rename crates/ecstore/src/{data_movement_backpressure.rs => data_movement/backpressure.rs} (100%) rename crates/ecstore/src/{data_movement.rs => data_movement/mod.rs} (100%) rename crates/ecstore/src/{data_usage.rs => data_usage/mod.rs} (100%) rename crates/ecstore/src/{ => diagnostics}/admin_server_info.rs (100%) rename crates/ecstore/src/{get_diagnostics.rs => diagnostics/get.rs} (100%) rename crates/ecstore/src/{error.rs => error/mod.rs} (100%) rename crates/ecstore/src/{ => io_support}/bitrot.rs (100%) rename crates/ecstore/src/{ => io_support}/compress.rs (100%) rename crates/ecstore/src/{ => io_support}/rio.rs (100%) rename crates/ecstore/src/{disks_layout.rs => layout/disks_layout_facade.rs} (100%) rename crates/ecstore/src/{endpoints.rs => layout/endpoints_facade.rs} (100%) rename crates/ecstore/src/{rebalance.rs => rebalance/mod.rs} (100%) rename crates/ecstore/src/{ => runtime}/global.rs (100%) rename crates/ecstore/src/{runtime_sources.rs => runtime/sources.rs} (100%) rename crates/ecstore/src/{ => services}/batch_processor.rs (100%) rename crates/ecstore/src/{ => services}/event_notification.rs (100%) rename crates/ecstore/src/{ => services}/metrics_realtime.rs (100%) rename crates/ecstore/src/{ => services}/notification_sys.rs (100%) rename crates/ecstore/src/{storage_api_contracts.rs => storage_api_contracts/mod.rs} (100%) diff --git a/crates/ecstore/src/cluster/control_plane.rs b/crates/ecstore/src/cluster/control_plane.rs new file mode 100644 index 000000000..3e790d476 --- /dev/null +++ b/crates/ecstore/src/cluster/control_plane.rs @@ -0,0 +1,635 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::{BTreeMap, BTreeSet}; + +use crate::storage_api_contracts::topology::{ + CapabilityStatus, DiskCapabilities, TopologyCapabilities, TopologyDisk, TopologyLabels, TopologyPool, TopologySet, + TopologySnapshot, +}; + +use crate::{ + endpoints::EndpointServerPools, + layout::endpoint::{Endpoint, EndpointType}, +}; + +const ENDPOINT_TYPE_LABEL: &str = "endpoint_type"; +const LOCAL_ENDPOINT_LABEL: &str = "local"; +const LOCAL_NODE_ID: &str = "local"; +const STORAGE_MEDIA_NOT_REPORTED: &str = "storage media not reported by endpoints"; +const FAILURE_DOMAIN_NOT_REPORTED: &str = "failure domain labels not reported by endpoints"; +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"; + +#[derive(Debug, Clone)] +pub struct ClusterControlPlane { + endpoint_pools: EndpointServerPools, +} + +impl ClusterControlPlane { + pub fn new(endpoint_pools: EndpointServerPools) -> Self { + Self { endpoint_pools } + } + + pub fn topology_snapshot(&self) -> TopologySnapshot { + topology_snapshot_from_endpoint_pools(&self.endpoint_pools) + } + + pub fn membership_snapshot(&self) -> ClusterMembershipSnapshot { + membership_snapshot_from_endpoint_pools(&self.endpoint_pools) + } + + pub fn pool_state_snapshot(&self) -> ClusterPoolStateSnapshot { + pool_state_snapshot_from_endpoint_pools(&self.endpoint_pools) + } + + pub fn local_node_storage_snapshot(&self) -> ClusterLocalNodeStorageSnapshot { + let membership = self.membership_snapshot(); + local_node_storage_snapshot_from_membership(&membership) + } + + pub fn peer_health_snapshot(&self) -> ClusterPeerHealthSnapshot { + let membership = self.membership_snapshot(); + peer_health_snapshot_from_membership(&membership) + } + + pub fn read_snapshot(&self) -> ClusterControlPlaneSnapshot { + let membership = self.membership_snapshot(); + + ClusterControlPlaneSnapshot { + topology: self.topology_snapshot(), + pool_state: self.pool_state_snapshot(), + local_storage: local_node_storage_snapshot_from_membership(&membership), + peer_health: peer_health_snapshot_from_membership(&membership), + membership, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterControlPlaneSnapshot { + pub topology: TopologySnapshot, + pub pool_state: ClusterPoolStateSnapshot, + pub local_storage: ClusterLocalNodeStorageSnapshot, + pub peer_health: ClusterPeerHealthSnapshot, + pub membership: ClusterMembershipSnapshot, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ClusterMembershipSnapshot { + pub nodes: Vec, + pub drives: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterNodeMembership { + pub node_id: String, + pub grid_host: String, + pub is_local: bool, + pub pools: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterDriveMembership { + pub pool_index: usize, + pub set_index: usize, + pub disk_index: usize, + pub node_id: String, + pub is_local: bool, + pub endpoint_type: ClusterEndpointType, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ClusterPoolStateSnapshot { + pub pools: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterPoolState { + pub pool_index: usize, + pub set_count: usize, + pub drives_per_set: usize, + pub endpoint_count: usize, + pub local_drive_count: usize, + pub remote_drive_count: usize, + pub legacy: bool, + pub endpoint_types: Vec, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ClusterLocalNodeStorageSnapshot { + pub nodes: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterLocalNodeStorage { + pub node_id: String, + pub pools: Vec, + pub drive_count: usize, + pub path_drive_count: usize, + pub url_drive_count: usize, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ClusterPeerHealthSnapshot { + pub peers: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterPeerHealth { + pub node_id: String, + pub is_local: bool, + pub status: CapabilityStatus, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub enum ClusterEndpointType { + Path, + Url, +} + +pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot { + topology_snapshot_from_endpoint_pools_with_capabilities( + endpoint_pools, + default_topology_capabilities(), + default_disk_capabilities(), + ) +} + +pub fn topology_snapshot_from_endpoint_pools_with_capabilities( + endpoint_pools: &EndpointServerPools, + capabilities: TopologyCapabilities, + disk_capabilities: DiskCapabilities, +) -> TopologySnapshot { + TopologySnapshot { + pools: endpoint_pools + .as_ref() + .iter() + .enumerate() + .map(|(pool_index, pool)| { + let sets = + topology_sets_from_endpoints(pool_index, pool.drives_per_set, pool.endpoints.as_ref(), &disk_capabilities); + TopologyPool { + pool_index, + pool_id: None, + labels: TopologyLabels::default(), + sets, + } + }) + .collect(), + capabilities, + } +} + +pub fn membership_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> ClusterMembershipSnapshot { + let mut nodes = BTreeMap::::new(); + let mut drives = Vec::new(); + + for (pool_index, pool) in endpoint_pools.as_ref().iter().enumerate() { + for (endpoint_index, endpoint) in pool.endpoints.as_ref().iter().enumerate() { + let (set_index, disk_index) = endpoint_indices(pool_index, endpoint_index, pool.drives_per_set, endpoint); + let node_id = endpoint_node_id(endpoint); + + match nodes.entry(node_id.clone()) { + std::collections::btree_map::Entry::Vacant(entry) => { + entry.insert(ClusterNodeMembership { + node_id: node_id.clone(), + grid_host: endpoint.grid_host(), + is_local: endpoint.is_local, + pools: vec![pool_index], + }); + } + std::collections::btree_map::Entry::Occupied(mut entry) => { + let node = entry.get_mut(); + node.is_local |= endpoint.is_local; + if !node.pools.contains(&pool_index) { + node.pools.push(pool_index); + } + } + } + + drives.push(ClusterDriveMembership { + pool_index, + set_index, + disk_index, + node_id, + is_local: endpoint.is_local, + endpoint_type: endpoint_type(endpoint), + }); + } + } + + ClusterMembershipSnapshot { + nodes: nodes.into_values().collect(), + drives, + } +} + +pub fn pool_state_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> ClusterPoolStateSnapshot { + ClusterPoolStateSnapshot { + pools: endpoint_pools + .as_ref() + .iter() + .enumerate() + .map(|(pool_index, pool)| { + let mut endpoint_types = BTreeSet::new(); + let mut local_drive_count = 0; + + for endpoint in pool.endpoints.as_ref() { + endpoint_types.insert(endpoint_type(endpoint)); + if endpoint.is_local { + local_drive_count += 1; + } + } + + let endpoint_count = pool.endpoints.as_ref().len(); + ClusterPoolState { + pool_index, + set_count: pool.set_count, + drives_per_set: pool.drives_per_set, + endpoint_count, + local_drive_count, + remote_drive_count: endpoint_count.saturating_sub(local_drive_count), + legacy: pool.legacy, + endpoint_types: endpoint_types.into_iter().collect(), + } + }) + .collect(), + } +} + +pub fn local_node_storage_snapshot_from_membership(membership: &ClusterMembershipSnapshot) -> ClusterLocalNodeStorageSnapshot { + ClusterLocalNodeStorageSnapshot { + nodes: membership + .nodes + .iter() + .filter(|node| node.is_local) + .map(|node| { + let mut path_drive_count = 0; + let mut url_drive_count = 0; + + for drive in membership.drives.iter().filter(|drive| drive.node_id == node.node_id) { + match drive.endpoint_type { + ClusterEndpointType::Path => path_drive_count += 1, + ClusterEndpointType::Url => url_drive_count += 1, + } + } + + ClusterLocalNodeStorage { + node_id: node.node_id.clone(), + pools: node.pools.clone(), + drive_count: path_drive_count + url_drive_count, + path_drive_count, + url_drive_count, + } + }) + .collect(), + } +} + +pub fn peer_health_snapshot_from_membership(membership: &ClusterMembershipSnapshot) -> ClusterPeerHealthSnapshot { + ClusterPeerHealthSnapshot { + peers: membership + .nodes + .iter() + .map(|node| ClusterPeerHealth { + node_id: node.node_id.clone(), + is_local: node.is_local, + status: CapabilityStatus::unknown().with_reason(PEER_HEALTH_NOT_REPORTED), + }) + .collect(), + } +} + +fn topology_sets_from_endpoints( + pool_index: usize, + drives_per_set: usize, + endpoints: &[Endpoint], + disk_capabilities: &DiskCapabilities, +) -> Vec { + let mut sets = BTreeMap::>::new(); + + for (endpoint_index, endpoint) in endpoints.iter().enumerate() { + let (set_index, disk_index) = endpoint_indices(pool_index, endpoint_index, drives_per_set, endpoint); + sets.entry(set_index).or_default().push(topology_disk_from_endpoint( + pool_index, + set_index, + disk_index, + endpoint, + disk_capabilities, + )); + } + + sets.into_iter() + .map(|(set_index, mut disks)| { + disks.sort_by_key(|disk| disk.disk_index); + TopologySet { + pool_index, + set_index, + set_id: None, + labels: TopologyLabels::default(), + disks, + } + }) + .collect() +} + +fn topology_disk_from_endpoint( + pool_index: usize, + set_index: usize, + disk_index: usize, + endpoint: &Endpoint, + disk_capabilities: &DiskCapabilities, +) -> TopologyDisk { + TopologyDisk { + pool_index, + set_index, + disk_index, + disk_id: endpoint_disk_id(endpoint), + labels: endpoint_labels(endpoint), + capabilities: disk_capabilities.clone(), + } +} + +fn endpoint_indices(pool_index: usize, endpoint_index: usize, drives_per_set: usize, endpoint: &Endpoint) -> (usize, usize) { + let safe_drives_per_set = drives_per_set.max(1); + let set_index = non_negative_index(endpoint.set_idx).unwrap_or(endpoint_index / safe_drives_per_set); + let disk_index = non_negative_index(endpoint.disk_idx).unwrap_or(endpoint_index % safe_drives_per_set); + + debug_assert_eq!(non_negative_index(endpoint.pool_idx).unwrap_or(pool_index), pool_index); + (set_index, disk_index) +} + +fn endpoint_disk_id(endpoint: &Endpoint) -> Option { + let host_port = endpoint.host_port(); + if host_port.is_empty() { None } else { Some(host_port) } +} + +fn endpoint_node_id(endpoint: &Endpoint) -> String { + endpoint_disk_id(endpoint).unwrap_or_else(|| LOCAL_NODE_ID.to_owned()) +} + +fn endpoint_labels(endpoint: &Endpoint) -> TopologyLabels { + let mut additional = BTreeMap::new(); + additional.insert(ENDPOINT_TYPE_LABEL.to_owned(), endpoint_type_label(endpoint).to_owned()); + additional.insert(LOCAL_ENDPOINT_LABEL.to_owned(), endpoint.is_local.to_string()); + + TopologyLabels { + node: Some(endpoint_node_id(endpoint)), + additional, + ..TopologyLabels::default() + } +} + +fn endpoint_type(endpoint: &Endpoint) -> ClusterEndpointType { + match endpoint.get_type() { + EndpointType::Path => ClusterEndpointType::Path, + EndpointType::Url => ClusterEndpointType::Url, + } +} + +fn endpoint_type_label(endpoint: &Endpoint) -> &'static str { + match endpoint_type(endpoint) { + ClusterEndpointType::Path => "path", + ClusterEndpointType::Url => "url", + } +} + +fn non_negative_index(index: i32) -> Option { + usize::try_from(index).ok() +} + +fn default_topology_capabilities() -> TopologyCapabilities { + TopologyCapabilities { + profiling: CapabilityStatus::unknown().with_reason(PROFILING_NOT_WIRED), + numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED), + failure_domain_labels: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED), + media_labels: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED), + } +} + +fn default_disk_capabilities() -> DiskCapabilities { + DiskCapabilities { + media_type: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED), + failure_domain: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED), + numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED), + profiling: CapabilityStatus::unknown().with_reason(PROFILING_NOT_WIRED), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeSet; + + use crate::endpoints::{Endpoints, PoolEndpoints}; + + #[test] + fn topology_snapshot_maps_endpoint_sets_without_local_paths() { + let endpoint_pools = sample_path_endpoint_pools(); + let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools); + + assert_eq!(snapshot.pools.len(), 1); + assert_eq!(snapshot.pools[0].sets.len(), 2); + assert_eq!(snapshot.pools[0].sets[0].disks.len(), 2); + assert_eq!(snapshot.pools[0].sets[1].disks.len(), 2); + assert_eq!(snapshot.pools[0].sets[1].disks[1].disk_index, 1); + assert_eq!( + snapshot.pools[0].sets[0].disks[0] + .labels + .additional + .get(ENDPOINT_TYPE_LABEL) + .map(String::as_str), + Some("path") + ); + assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some(LOCAL_NODE_ID)); + + let encoded = serde_json::to_string(&snapshot).expect("serialize topology snapshot"); + assert!(!encoded.contains("/tmp/rustfs-cluster-control-plane")); + } + + #[test] + fn topology_snapshot_uses_url_hosts_as_disk_ids() { + let endpoint_pools = sample_url_endpoint_pools(); + let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools); + + assert_eq!(snapshot.pools[0].sets[0].disks[0].disk_id.as_deref(), Some("node1.example:9000")); + assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some("node1.example:9000")); + assert_eq!( + snapshot.pools[0].sets[0].disks[0] + .labels + .additional + .get(ENDPOINT_TYPE_LABEL) + .map(String::as_str), + Some("url") + ); + } + + #[test] + fn membership_snapshot_groups_nodes_and_drives() { + let endpoint_pools = sample_url_endpoint_pools(); + let snapshot = membership_snapshot_from_endpoint_pools(&endpoint_pools); + + assert_eq!(snapshot.nodes.len(), 2); + assert_eq!(snapshot.drives.len(), 4); + assert_eq!(snapshot.nodes[0].node_id, "node1.example:9000"); + assert_eq!(snapshot.nodes[0].pools, vec![0]); + assert_eq!(snapshot.drives[2].node_id, "node2.example:9000"); + assert_eq!(snapshot.drives[2].set_index, 1); + assert_eq!(snapshot.drives[2].endpoint_type, ClusterEndpointType::Url); + } + + #[test] + fn pool_state_snapshot_counts_local_remote_drives_and_endpoint_types() { + let endpoint_pools = sample_mixed_endpoint_pools(); + let snapshot = pool_state_snapshot_from_endpoint_pools(&endpoint_pools); + + assert_eq!(snapshot.pools.len(), 1); + assert_eq!(snapshot.pools[0].set_count, 2); + assert_eq!(snapshot.pools[0].drives_per_set, 2); + assert_eq!(snapshot.pools[0].endpoint_count, 4); + assert_eq!(snapshot.pools[0].local_drive_count, 3); + assert_eq!(snapshot.pools[0].remote_drive_count, 1); + assert_eq!( + snapshot.pools[0].endpoint_types, + vec![ClusterEndpointType::Path, ClusterEndpointType::Url] + ); + } + + #[test] + fn local_node_storage_snapshot_keeps_only_local_drive_counts() { + let membership = membership_snapshot_from_endpoint_pools(&sample_mixed_endpoint_pools()); + let snapshot = local_node_storage_snapshot_from_membership(&membership); + + assert_eq!(snapshot.nodes.len(), 2); + assert_eq!(snapshot.nodes[0].node_id, LOCAL_NODE_ID); + assert_eq!(snapshot.nodes[0].drive_count, 2); + assert_eq!(snapshot.nodes[0].path_drive_count, 2); + assert_eq!(snapshot.nodes[0].url_drive_count, 0); + assert_eq!(snapshot.nodes[1].node_id, "node1.example:9000"); + assert_eq!(snapshot.nodes[1].drive_count, 1); + assert_eq!(snapshot.nodes[1].path_drive_count, 0); + assert_eq!(snapshot.nodes[1].url_drive_count, 1); + } + + #[test] + fn peer_health_snapshot_reports_static_unknown_status() { + let membership = membership_snapshot_from_endpoint_pools(&sample_url_endpoint_pools()); + let snapshot = peer_health_snapshot_from_membership(&membership); + + assert_eq!(snapshot.peers.len(), 2); + assert_eq!(snapshot.peers[0].node_id, "node1.example:9000"); + assert!(snapshot.peers[0].is_local); + assert_eq!(snapshot.peers[0].status.reason.as_deref(), Some(PEER_HEALTH_NOT_REPORTED)); + assert_eq!(snapshot.peers[1].node_id, "node2.example:9000"); + assert!(!snapshot.peers[1].is_local); + } + + #[test] + fn control_plane_read_snapshot_combines_topology_and_membership() { + let control_plane = ClusterControlPlane::new(sample_mixed_endpoint_pools()); + let snapshot = control_plane.read_snapshot(); + let node_ids = snapshot + .membership + .nodes + .iter() + .map(|node| node.node_id.as_str()) + .collect::>(); + + assert_eq!(snapshot.topology.pools[0].sets.len(), 2); + 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!(node_ids, BTreeSet::from([LOCAL_NODE_ID, "node1.example:9000", "node2.example:9000"])); + } + + fn sample_path_endpoint_pools() -> EndpointServerPools { + let endpoints = (0..4) + .map(|index| { + let mut endpoint = + Endpoint::try_from(format!("/tmp/rustfs-cluster-control-plane-{index}").as_str()).expect("local endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(index / 2); + endpoint.set_disk_index(index % 2); + endpoint + }) + .collect::>(); + + EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 2, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: "/tmp/rustfs-cluster-control-plane-{0...3}".to_owned(), + platform: "OS: test | Arch: test".to_owned(), + }]) + } + + fn sample_url_endpoint_pools() -> EndpointServerPools { + let endpoints = (0..4) + .map(|index| { + let host = if index < 2 { "node1.example" } else { "node2.example" }; + let mut endpoint = + Endpoint::try_from(format!("http://{host}:9000/export{index}").as_str()).expect("url endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(index / 2); + endpoint.set_disk_index(index % 2); + endpoint.is_local = index < 2; + endpoint + }) + .collect::>(); + + EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 2, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: "http://node{1...2}.example:9000/export{0...3}".to_owned(), + platform: "OS: test | Arch: test".to_owned(), + }]) + } + + fn sample_mixed_endpoint_pools() -> EndpointServerPools { + let mut endpoints = Vec::new(); + + for index in 0..2 { + let mut endpoint = + Endpoint::try_from(format!("/tmp/rustfs-cluster-control-plane-{index}").as_str()).expect("local endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(index); + endpoints.push(endpoint); + } + + for index in 2..4 { + let host = if index == 2 { "node1.example" } else { "node2.example" }; + let mut endpoint = Endpoint::try_from(format!("http://{host}:9000/export{index}").as_str()).expect("url endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(1); + endpoint.set_disk_index(index - 2); + endpoint.is_local = index == 2; + endpoints.push(endpoint); + } + + EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 2, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: "mixed-test-endpoints".to_owned(), + platform: "OS: test | Arch: test".to_owned(), + }]) + } +} diff --git a/crates/ecstore/src/cluster/mod.rs b/crates/ecstore/src/cluster/mod.rs index 3e790d476..d44510e38 100644 --- a/crates/ecstore/src/cluster/mod.rs +++ b/crates/ecstore/src/cluster/mod.rs @@ -12,624 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::collections::{BTreeMap, BTreeSet}; +mod control_plane; -use crate::storage_api_contracts::topology::{ - CapabilityStatus, DiskCapabilities, TopologyCapabilities, TopologyDisk, TopologyLabels, TopologyPool, TopologySet, - TopologySnapshot, +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, + topology_snapshot_from_endpoint_pools, topology_snapshot_from_endpoint_pools_with_capabilities, }; - -use crate::{ - endpoints::EndpointServerPools, - layout::endpoint::{Endpoint, EndpointType}, -}; - -const ENDPOINT_TYPE_LABEL: &str = "endpoint_type"; -const LOCAL_ENDPOINT_LABEL: &str = "local"; -const LOCAL_NODE_ID: &str = "local"; -const STORAGE_MEDIA_NOT_REPORTED: &str = "storage media not reported by endpoints"; -const FAILURE_DOMAIN_NOT_REPORTED: &str = "failure domain labels not reported by endpoints"; -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"; - -#[derive(Debug, Clone)] -pub struct ClusterControlPlane { - endpoint_pools: EndpointServerPools, -} - -impl ClusterControlPlane { - pub fn new(endpoint_pools: EndpointServerPools) -> Self { - Self { endpoint_pools } - } - - pub fn topology_snapshot(&self) -> TopologySnapshot { - topology_snapshot_from_endpoint_pools(&self.endpoint_pools) - } - - pub fn membership_snapshot(&self) -> ClusterMembershipSnapshot { - membership_snapshot_from_endpoint_pools(&self.endpoint_pools) - } - - pub fn pool_state_snapshot(&self) -> ClusterPoolStateSnapshot { - pool_state_snapshot_from_endpoint_pools(&self.endpoint_pools) - } - - pub fn local_node_storage_snapshot(&self) -> ClusterLocalNodeStorageSnapshot { - let membership = self.membership_snapshot(); - local_node_storage_snapshot_from_membership(&membership) - } - - pub fn peer_health_snapshot(&self) -> ClusterPeerHealthSnapshot { - let membership = self.membership_snapshot(); - peer_health_snapshot_from_membership(&membership) - } - - pub fn read_snapshot(&self) -> ClusterControlPlaneSnapshot { - let membership = self.membership_snapshot(); - - ClusterControlPlaneSnapshot { - topology: self.topology_snapshot(), - pool_state: self.pool_state_snapshot(), - local_storage: local_node_storage_snapshot_from_membership(&membership), - peer_health: peer_health_snapshot_from_membership(&membership), - membership, - } - } -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ClusterControlPlaneSnapshot { - pub topology: TopologySnapshot, - pub pool_state: ClusterPoolStateSnapshot, - pub local_storage: ClusterLocalNodeStorageSnapshot, - pub peer_health: ClusterPeerHealthSnapshot, - pub membership: ClusterMembershipSnapshot, -} - -#[derive(Debug, Clone, Default, PartialEq, Eq)] -pub struct ClusterMembershipSnapshot { - pub nodes: Vec, - pub drives: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ClusterNodeMembership { - pub node_id: String, - pub grid_host: String, - pub is_local: bool, - pub pools: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ClusterDriveMembership { - pub pool_index: usize, - pub set_index: usize, - pub disk_index: usize, - pub node_id: String, - pub is_local: bool, - pub endpoint_type: ClusterEndpointType, -} - -#[derive(Debug, Clone, Default, PartialEq, Eq)] -pub struct ClusterPoolStateSnapshot { - pub pools: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ClusterPoolState { - pub pool_index: usize, - pub set_count: usize, - pub drives_per_set: usize, - pub endpoint_count: usize, - pub local_drive_count: usize, - pub remote_drive_count: usize, - pub legacy: bool, - pub endpoint_types: Vec, -} - -#[derive(Debug, Clone, Default, PartialEq, Eq)] -pub struct ClusterLocalNodeStorageSnapshot { - pub nodes: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ClusterLocalNodeStorage { - pub node_id: String, - pub pools: Vec, - pub drive_count: usize, - pub path_drive_count: usize, - pub url_drive_count: usize, -} - -#[derive(Debug, Clone, Default, PartialEq, Eq)] -pub struct ClusterPeerHealthSnapshot { - pub peers: Vec, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct ClusterPeerHealth { - pub node_id: String, - pub is_local: bool, - pub status: CapabilityStatus, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] -pub enum ClusterEndpointType { - Path, - Url, -} - -pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> TopologySnapshot { - topology_snapshot_from_endpoint_pools_with_capabilities( - endpoint_pools, - default_topology_capabilities(), - default_disk_capabilities(), - ) -} - -pub fn topology_snapshot_from_endpoint_pools_with_capabilities( - endpoint_pools: &EndpointServerPools, - capabilities: TopologyCapabilities, - disk_capabilities: DiskCapabilities, -) -> TopologySnapshot { - TopologySnapshot { - pools: endpoint_pools - .as_ref() - .iter() - .enumerate() - .map(|(pool_index, pool)| { - let sets = - topology_sets_from_endpoints(pool_index, pool.drives_per_set, pool.endpoints.as_ref(), &disk_capabilities); - TopologyPool { - pool_index, - pool_id: None, - labels: TopologyLabels::default(), - sets, - } - }) - .collect(), - capabilities, - } -} - -pub fn membership_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> ClusterMembershipSnapshot { - let mut nodes = BTreeMap::::new(); - let mut drives = Vec::new(); - - for (pool_index, pool) in endpoint_pools.as_ref().iter().enumerate() { - for (endpoint_index, endpoint) in pool.endpoints.as_ref().iter().enumerate() { - let (set_index, disk_index) = endpoint_indices(pool_index, endpoint_index, pool.drives_per_set, endpoint); - let node_id = endpoint_node_id(endpoint); - - match nodes.entry(node_id.clone()) { - std::collections::btree_map::Entry::Vacant(entry) => { - entry.insert(ClusterNodeMembership { - node_id: node_id.clone(), - grid_host: endpoint.grid_host(), - is_local: endpoint.is_local, - pools: vec![pool_index], - }); - } - std::collections::btree_map::Entry::Occupied(mut entry) => { - let node = entry.get_mut(); - node.is_local |= endpoint.is_local; - if !node.pools.contains(&pool_index) { - node.pools.push(pool_index); - } - } - } - - drives.push(ClusterDriveMembership { - pool_index, - set_index, - disk_index, - node_id, - is_local: endpoint.is_local, - endpoint_type: endpoint_type(endpoint), - }); - } - } - - ClusterMembershipSnapshot { - nodes: nodes.into_values().collect(), - drives, - } -} - -pub fn pool_state_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> ClusterPoolStateSnapshot { - ClusterPoolStateSnapshot { - pools: endpoint_pools - .as_ref() - .iter() - .enumerate() - .map(|(pool_index, pool)| { - let mut endpoint_types = BTreeSet::new(); - let mut local_drive_count = 0; - - for endpoint in pool.endpoints.as_ref() { - endpoint_types.insert(endpoint_type(endpoint)); - if endpoint.is_local { - local_drive_count += 1; - } - } - - let endpoint_count = pool.endpoints.as_ref().len(); - ClusterPoolState { - pool_index, - set_count: pool.set_count, - drives_per_set: pool.drives_per_set, - endpoint_count, - local_drive_count, - remote_drive_count: endpoint_count.saturating_sub(local_drive_count), - legacy: pool.legacy, - endpoint_types: endpoint_types.into_iter().collect(), - } - }) - .collect(), - } -} - -pub fn local_node_storage_snapshot_from_membership(membership: &ClusterMembershipSnapshot) -> ClusterLocalNodeStorageSnapshot { - ClusterLocalNodeStorageSnapshot { - nodes: membership - .nodes - .iter() - .filter(|node| node.is_local) - .map(|node| { - let mut path_drive_count = 0; - let mut url_drive_count = 0; - - for drive in membership.drives.iter().filter(|drive| drive.node_id == node.node_id) { - match drive.endpoint_type { - ClusterEndpointType::Path => path_drive_count += 1, - ClusterEndpointType::Url => url_drive_count += 1, - } - } - - ClusterLocalNodeStorage { - node_id: node.node_id.clone(), - pools: node.pools.clone(), - drive_count: path_drive_count + url_drive_count, - path_drive_count, - url_drive_count, - } - }) - .collect(), - } -} - -pub fn peer_health_snapshot_from_membership(membership: &ClusterMembershipSnapshot) -> ClusterPeerHealthSnapshot { - ClusterPeerHealthSnapshot { - peers: membership - .nodes - .iter() - .map(|node| ClusterPeerHealth { - node_id: node.node_id.clone(), - is_local: node.is_local, - status: CapabilityStatus::unknown().with_reason(PEER_HEALTH_NOT_REPORTED), - }) - .collect(), - } -} - -fn topology_sets_from_endpoints( - pool_index: usize, - drives_per_set: usize, - endpoints: &[Endpoint], - disk_capabilities: &DiskCapabilities, -) -> Vec { - let mut sets = BTreeMap::>::new(); - - for (endpoint_index, endpoint) in endpoints.iter().enumerate() { - let (set_index, disk_index) = endpoint_indices(pool_index, endpoint_index, drives_per_set, endpoint); - sets.entry(set_index).or_default().push(topology_disk_from_endpoint( - pool_index, - set_index, - disk_index, - endpoint, - disk_capabilities, - )); - } - - sets.into_iter() - .map(|(set_index, mut disks)| { - disks.sort_by_key(|disk| disk.disk_index); - TopologySet { - pool_index, - set_index, - set_id: None, - labels: TopologyLabels::default(), - disks, - } - }) - .collect() -} - -fn topology_disk_from_endpoint( - pool_index: usize, - set_index: usize, - disk_index: usize, - endpoint: &Endpoint, - disk_capabilities: &DiskCapabilities, -) -> TopologyDisk { - TopologyDisk { - pool_index, - set_index, - disk_index, - disk_id: endpoint_disk_id(endpoint), - labels: endpoint_labels(endpoint), - capabilities: disk_capabilities.clone(), - } -} - -fn endpoint_indices(pool_index: usize, endpoint_index: usize, drives_per_set: usize, endpoint: &Endpoint) -> (usize, usize) { - let safe_drives_per_set = drives_per_set.max(1); - let set_index = non_negative_index(endpoint.set_idx).unwrap_or(endpoint_index / safe_drives_per_set); - let disk_index = non_negative_index(endpoint.disk_idx).unwrap_or(endpoint_index % safe_drives_per_set); - - debug_assert_eq!(non_negative_index(endpoint.pool_idx).unwrap_or(pool_index), pool_index); - (set_index, disk_index) -} - -fn endpoint_disk_id(endpoint: &Endpoint) -> Option { - let host_port = endpoint.host_port(); - if host_port.is_empty() { None } else { Some(host_port) } -} - -fn endpoint_node_id(endpoint: &Endpoint) -> String { - endpoint_disk_id(endpoint).unwrap_or_else(|| LOCAL_NODE_ID.to_owned()) -} - -fn endpoint_labels(endpoint: &Endpoint) -> TopologyLabels { - let mut additional = BTreeMap::new(); - additional.insert(ENDPOINT_TYPE_LABEL.to_owned(), endpoint_type_label(endpoint).to_owned()); - additional.insert(LOCAL_ENDPOINT_LABEL.to_owned(), endpoint.is_local.to_string()); - - TopologyLabels { - node: Some(endpoint_node_id(endpoint)), - additional, - ..TopologyLabels::default() - } -} - -fn endpoint_type(endpoint: &Endpoint) -> ClusterEndpointType { - match endpoint.get_type() { - EndpointType::Path => ClusterEndpointType::Path, - EndpointType::Url => ClusterEndpointType::Url, - } -} - -fn endpoint_type_label(endpoint: &Endpoint) -> &'static str { - match endpoint_type(endpoint) { - ClusterEndpointType::Path => "path", - ClusterEndpointType::Url => "url", - } -} - -fn non_negative_index(index: i32) -> Option { - usize::try_from(index).ok() -} - -fn default_topology_capabilities() -> TopologyCapabilities { - TopologyCapabilities { - profiling: CapabilityStatus::unknown().with_reason(PROFILING_NOT_WIRED), - numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED), - failure_domain_labels: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED), - media_labels: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED), - } -} - -fn default_disk_capabilities() -> DiskCapabilities { - DiskCapabilities { - media_type: CapabilityStatus::unknown().with_reason(STORAGE_MEDIA_NOT_REPORTED), - failure_domain: CapabilityStatus::unknown().with_reason(FAILURE_DOMAIN_NOT_REPORTED), - numa: CapabilityStatus::unsupported().with_reason(NUMA_NOT_WIRED), - profiling: CapabilityStatus::unknown().with_reason(PROFILING_NOT_WIRED), - } -} - -#[cfg(test)] -mod tests { - use super::*; - use std::collections::BTreeSet; - - use crate::endpoints::{Endpoints, PoolEndpoints}; - - #[test] - fn topology_snapshot_maps_endpoint_sets_without_local_paths() { - let endpoint_pools = sample_path_endpoint_pools(); - let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools); - - assert_eq!(snapshot.pools.len(), 1); - assert_eq!(snapshot.pools[0].sets.len(), 2); - assert_eq!(snapshot.pools[0].sets[0].disks.len(), 2); - assert_eq!(snapshot.pools[0].sets[1].disks.len(), 2); - assert_eq!(snapshot.pools[0].sets[1].disks[1].disk_index, 1); - assert_eq!( - snapshot.pools[0].sets[0].disks[0] - .labels - .additional - .get(ENDPOINT_TYPE_LABEL) - .map(String::as_str), - Some("path") - ); - assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some(LOCAL_NODE_ID)); - - let encoded = serde_json::to_string(&snapshot).expect("serialize topology snapshot"); - assert!(!encoded.contains("/tmp/rustfs-cluster-control-plane")); - } - - #[test] - fn topology_snapshot_uses_url_hosts_as_disk_ids() { - let endpoint_pools = sample_url_endpoint_pools(); - let snapshot = topology_snapshot_from_endpoint_pools(&endpoint_pools); - - assert_eq!(snapshot.pools[0].sets[0].disks[0].disk_id.as_deref(), Some("node1.example:9000")); - assert_eq!(snapshot.pools[0].sets[0].disks[0].labels.node.as_deref(), Some("node1.example:9000")); - assert_eq!( - snapshot.pools[0].sets[0].disks[0] - .labels - .additional - .get(ENDPOINT_TYPE_LABEL) - .map(String::as_str), - Some("url") - ); - } - - #[test] - fn membership_snapshot_groups_nodes_and_drives() { - let endpoint_pools = sample_url_endpoint_pools(); - let snapshot = membership_snapshot_from_endpoint_pools(&endpoint_pools); - - assert_eq!(snapshot.nodes.len(), 2); - assert_eq!(snapshot.drives.len(), 4); - assert_eq!(snapshot.nodes[0].node_id, "node1.example:9000"); - assert_eq!(snapshot.nodes[0].pools, vec![0]); - assert_eq!(snapshot.drives[2].node_id, "node2.example:9000"); - assert_eq!(snapshot.drives[2].set_index, 1); - assert_eq!(snapshot.drives[2].endpoint_type, ClusterEndpointType::Url); - } - - #[test] - fn pool_state_snapshot_counts_local_remote_drives_and_endpoint_types() { - let endpoint_pools = sample_mixed_endpoint_pools(); - let snapshot = pool_state_snapshot_from_endpoint_pools(&endpoint_pools); - - assert_eq!(snapshot.pools.len(), 1); - assert_eq!(snapshot.pools[0].set_count, 2); - assert_eq!(snapshot.pools[0].drives_per_set, 2); - assert_eq!(snapshot.pools[0].endpoint_count, 4); - assert_eq!(snapshot.pools[0].local_drive_count, 3); - assert_eq!(snapshot.pools[0].remote_drive_count, 1); - assert_eq!( - snapshot.pools[0].endpoint_types, - vec![ClusterEndpointType::Path, ClusterEndpointType::Url] - ); - } - - #[test] - fn local_node_storage_snapshot_keeps_only_local_drive_counts() { - let membership = membership_snapshot_from_endpoint_pools(&sample_mixed_endpoint_pools()); - let snapshot = local_node_storage_snapshot_from_membership(&membership); - - assert_eq!(snapshot.nodes.len(), 2); - assert_eq!(snapshot.nodes[0].node_id, LOCAL_NODE_ID); - assert_eq!(snapshot.nodes[0].drive_count, 2); - assert_eq!(snapshot.nodes[0].path_drive_count, 2); - assert_eq!(snapshot.nodes[0].url_drive_count, 0); - assert_eq!(snapshot.nodes[1].node_id, "node1.example:9000"); - assert_eq!(snapshot.nodes[1].drive_count, 1); - assert_eq!(snapshot.nodes[1].path_drive_count, 0); - assert_eq!(snapshot.nodes[1].url_drive_count, 1); - } - - #[test] - fn peer_health_snapshot_reports_static_unknown_status() { - let membership = membership_snapshot_from_endpoint_pools(&sample_url_endpoint_pools()); - let snapshot = peer_health_snapshot_from_membership(&membership); - - assert_eq!(snapshot.peers.len(), 2); - assert_eq!(snapshot.peers[0].node_id, "node1.example:9000"); - assert!(snapshot.peers[0].is_local); - assert_eq!(snapshot.peers[0].status.reason.as_deref(), Some(PEER_HEALTH_NOT_REPORTED)); - assert_eq!(snapshot.peers[1].node_id, "node2.example:9000"); - assert!(!snapshot.peers[1].is_local); - } - - #[test] - fn control_plane_read_snapshot_combines_topology_and_membership() { - let control_plane = ClusterControlPlane::new(sample_mixed_endpoint_pools()); - let snapshot = control_plane.read_snapshot(); - let node_ids = snapshot - .membership - .nodes - .iter() - .map(|node| node.node_id.as_str()) - .collect::>(); - - assert_eq!(snapshot.topology.pools[0].sets.len(), 2); - 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!(node_ids, BTreeSet::from([LOCAL_NODE_ID, "node1.example:9000", "node2.example:9000"])); - } - - fn sample_path_endpoint_pools() -> EndpointServerPools { - let endpoints = (0..4) - .map(|index| { - let mut endpoint = - Endpoint::try_from(format!("/tmp/rustfs-cluster-control-plane-{index}").as_str()).expect("local endpoint"); - endpoint.set_pool_index(0); - endpoint.set_set_index(index / 2); - endpoint.set_disk_index(index % 2); - endpoint - }) - .collect::>(); - - EndpointServerPools::from(vec![PoolEndpoints { - legacy: false, - set_count: 2, - drives_per_set: 2, - endpoints: Endpoints::from(endpoints), - cmd_line: "/tmp/rustfs-cluster-control-plane-{0...3}".to_owned(), - platform: "OS: test | Arch: test".to_owned(), - }]) - } - - fn sample_url_endpoint_pools() -> EndpointServerPools { - let endpoints = (0..4) - .map(|index| { - let host = if index < 2 { "node1.example" } else { "node2.example" }; - let mut endpoint = - Endpoint::try_from(format!("http://{host}:9000/export{index}").as_str()).expect("url endpoint"); - endpoint.set_pool_index(0); - endpoint.set_set_index(index / 2); - endpoint.set_disk_index(index % 2); - endpoint.is_local = index < 2; - endpoint - }) - .collect::>(); - - EndpointServerPools::from(vec![PoolEndpoints { - legacy: false, - set_count: 2, - drives_per_set: 2, - endpoints: Endpoints::from(endpoints), - cmd_line: "http://node{1...2}.example:9000/export{0...3}".to_owned(), - platform: "OS: test | Arch: test".to_owned(), - }]) - } - - fn sample_mixed_endpoint_pools() -> EndpointServerPools { - let mut endpoints = Vec::new(); - - for index in 0..2 { - let mut endpoint = - Endpoint::try_from(format!("/tmp/rustfs-cluster-control-plane-{index}").as_str()).expect("local endpoint"); - endpoint.set_pool_index(0); - endpoint.set_set_index(0); - endpoint.set_disk_index(index); - endpoints.push(endpoint); - } - - for index in 2..4 { - let host = if index == 2 { "node1.example" } else { "node2.example" }; - let mut endpoint = Endpoint::try_from(format!("http://{host}:9000/export{index}").as_str()).expect("url endpoint"); - endpoint.set_pool_index(0); - endpoint.set_set_index(1); - endpoint.set_disk_index(index - 2); - endpoint.is_local = index == 2; - endpoints.push(endpoint); - } - - EndpointServerPools::from(vec![PoolEndpoints { - legacy: false, - set_count: 2, - drives_per_set: 2, - endpoints: Endpoints::from(endpoints), - cmd_line: "mixed-test-endpoints".to_owned(), - platform: "OS: test | Arch: test".to_owned(), - }]) - } -} diff --git a/crates/ecstore/src/pools.rs b/crates/ecstore/src/core/pools.rs similarity index 100% rename from crates/ecstore/src/pools.rs rename to crates/ecstore/src/core/pools.rs diff --git a/crates/ecstore/src/pools_test.rs b/crates/ecstore/src/core/pools_test.rs similarity index 100% rename from crates/ecstore/src/pools_test.rs rename to crates/ecstore/src/core/pools_test.rs diff --git a/crates/ecstore/src/sets.rs b/crates/ecstore/src/core/sets.rs similarity index 96% rename from crates/ecstore/src/sets.rs rename to crates/ecstore/src/core/sets.rs index 98b7d1a10..ab5c68491 100644 --- a/crates/ecstore/src/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -1090,9 +1090,36 @@ async fn init_storage_disks_with_errors( #[cfg(test)] mod tests { use super::*; + use crate::endpoints::SetupType; use crate::layout::endpoint::Endpoint; use crate::storage_api_contracts::heal::HealOperations as _; use crate::storage_api_contracts::list::ListOperations as _; + use rustfs_lock::client::local::LocalClient; + use serial_test::serial; + + struct SetupTypeGuard { + previous: SetupType, + } + + impl SetupTypeGuard { + async fn switch_to(next: SetupType) -> Self { + let previous = runtime_sources::current_setup_type().await; + runtime_sources::set_setup_type(next).await; + Self { previous } + } + } + + impl Drop for SetupTypeGuard { + fn drop(&mut self) { + let previous = self.previous.clone(); + let handle = tokio::runtime::Handle::current(); + tokio::task::block_in_place(|| { + handle.block_on(async move { + runtime_sources::set_setup_type(previous).await; + }); + }); + } + } #[test] fn test_apply_delete_objects_results_preserves_original_order_for_out_of_order_batches() { @@ -1203,8 +1230,10 @@ mod tests { assert_eq!(result, (Some(3), Some(1), Some(0))); } - #[tokio::test] + #[tokio::test(flavor = "multi_thread")] + #[serial] async fn sets_list_objects_v2_lists_objects_within_the_pool() { + let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::Erasure).await; let format = FormatV3::new(1, 2); let mut endpoints = Vec::new(); let mut disks = Vec::new(); @@ -1247,7 +1276,7 @@ mod tests { 0, endpoints.clone(), format.clone(), - Vec::new(), + vec![Arc::new(LocalClient::new()), Arc::new(LocalClient::new())], ) .await; @@ -1272,23 +1301,26 @@ mod tests { exit_signal: None, }); - sets.make_bucket("bucket", &MakeBucketOptions::default()) + let bucket = format!("bucket-{}", Uuid::new_v4().simple()); + let object = format!("object-{}", Uuid::new_v4().simple()); + + sets.make_bucket(&bucket, &MakeBucketOptions::default()) .await .expect("bucket should be created"); let mut reader = PutObjReader::from_vec(b"hello".to_vec()); - sets.put_object("bucket", "object", &mut reader, &ObjectOptions::default()) + sets.put_object(&bucket, &object, &mut reader, &ObjectOptions::default()) .await .expect("object should be written"); let result = sets .clone() - .list_objects_v2("bucket", "", None, None, 1000, false, None, false) + .list_objects_v2(&bucket, "", None, None, 1000, false, None, false) .await .expect("pool-level listing should succeed"); assert_eq!(result.objects.len(), 1); - assert_eq!(result.objects[0].name, "object"); + assert_eq!(result.objects[0].name, object); } #[tokio::test] diff --git a/crates/ecstore/src/store_test.rs b/crates/ecstore/src/core/store_test.rs similarity index 100% rename from crates/ecstore/src/store_test.rs rename to crates/ecstore/src/core/store_test.rs diff --git a/crates/ecstore/src/data_movement_backpressure.rs b/crates/ecstore/src/data_movement/backpressure.rs similarity index 100% rename from crates/ecstore/src/data_movement_backpressure.rs rename to crates/ecstore/src/data_movement/backpressure.rs diff --git a/crates/ecstore/src/data_movement.rs b/crates/ecstore/src/data_movement/mod.rs similarity index 100% rename from crates/ecstore/src/data_movement.rs rename to crates/ecstore/src/data_movement/mod.rs diff --git a/crates/ecstore/src/data_usage.rs b/crates/ecstore/src/data_usage/mod.rs similarity index 100% rename from crates/ecstore/src/data_usage.rs rename to crates/ecstore/src/data_usage/mod.rs diff --git a/crates/ecstore/src/admin_server_info.rs b/crates/ecstore/src/diagnostics/admin_server_info.rs similarity index 100% rename from crates/ecstore/src/admin_server_info.rs rename to crates/ecstore/src/diagnostics/admin_server_info.rs diff --git a/crates/ecstore/src/get_diagnostics.rs b/crates/ecstore/src/diagnostics/get.rs similarity index 100% rename from crates/ecstore/src/get_diagnostics.rs rename to crates/ecstore/src/diagnostics/get.rs diff --git a/crates/ecstore/src/error.rs b/crates/ecstore/src/error/mod.rs similarity index 100% rename from crates/ecstore/src/error.rs rename to crates/ecstore/src/error/mod.rs diff --git a/crates/ecstore/src/bitrot.rs b/crates/ecstore/src/io_support/bitrot.rs similarity index 100% rename from crates/ecstore/src/bitrot.rs rename to crates/ecstore/src/io_support/bitrot.rs diff --git a/crates/ecstore/src/compress.rs b/crates/ecstore/src/io_support/compress.rs similarity index 100% rename from crates/ecstore/src/compress.rs rename to crates/ecstore/src/io_support/compress.rs diff --git a/crates/ecstore/src/rio.rs b/crates/ecstore/src/io_support/rio.rs similarity index 100% rename from crates/ecstore/src/rio.rs rename to crates/ecstore/src/io_support/rio.rs diff --git a/crates/ecstore/src/disks_layout.rs b/crates/ecstore/src/layout/disks_layout_facade.rs similarity index 100% rename from crates/ecstore/src/disks_layout.rs rename to crates/ecstore/src/layout/disks_layout_facade.rs diff --git a/crates/ecstore/src/endpoints.rs b/crates/ecstore/src/layout/endpoints_facade.rs similarity index 100% rename from crates/ecstore/src/endpoints.rs rename to crates/ecstore/src/layout/endpoints_facade.rs diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index 269d125d7..0fe686eec 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -15,36 +15,51 @@ extern crate core; +#[path = "diagnostics/admin_server_info.rs"] mod admin_server_info; pub mod api; +#[path = "services/batch_processor.rs"] mod batch_processor; +#[path = "io_support/bitrot.rs"] mod bitrot; mod bucket; mod cache_value; mod cluster; +#[path = "io_support/compress.rs"] mod compress; mod config; mod data_movement; +#[path = "data_movement/backpressure.rs"] mod data_movement_backpressure; mod data_usage; mod disk; +#[path = "layout/disks_layout_facade.rs"] mod disks_layout; +#[path = "layout/endpoints_facade.rs"] mod endpoints; mod erasure_codec; mod erasure_coding; mod error; +#[path = "diagnostics/get.rs"] mod get_diagnostics; +#[path = "runtime/global.rs"] mod global; pub(crate) mod layout; +#[path = "services/metrics_realtime.rs"] mod metrics_realtime; +#[path = "services/notification_sys.rs"] mod notification_sys; mod object_api; +#[path = "core/pools.rs"] mod pools; mod rebalance; +#[path = "io_support/rio.rs"] mod rio; mod rpc; +#[path = "runtime/sources.rs"] mod runtime_sources; mod set_disk; +#[path = "core/sets.rs"] mod sets; mod storage_api_contracts; mod store; @@ -58,10 +73,13 @@ mod store_utils; // pub mod checksum; mod client; mod event; +#[path = "services/event_notification.rs"] mod event_notification; #[cfg(test)] +#[path = "core/pools_test.rs"] mod pools_test; #[cfg(test)] +#[path = "core/store_test.rs"] mod store_test; mod tier; diff --git a/crates/ecstore/src/rebalance.rs b/crates/ecstore/src/rebalance/mod.rs similarity index 100% rename from crates/ecstore/src/rebalance.rs rename to crates/ecstore/src/rebalance/mod.rs diff --git a/crates/ecstore/src/global.rs b/crates/ecstore/src/runtime/global.rs similarity index 100% rename from crates/ecstore/src/global.rs rename to crates/ecstore/src/runtime/global.rs diff --git a/crates/ecstore/src/runtime_sources.rs b/crates/ecstore/src/runtime/sources.rs similarity index 100% rename from crates/ecstore/src/runtime_sources.rs rename to crates/ecstore/src/runtime/sources.rs diff --git a/crates/ecstore/src/batch_processor.rs b/crates/ecstore/src/services/batch_processor.rs similarity index 100% rename from crates/ecstore/src/batch_processor.rs rename to crates/ecstore/src/services/batch_processor.rs diff --git a/crates/ecstore/src/event_notification.rs b/crates/ecstore/src/services/event_notification.rs similarity index 100% rename from crates/ecstore/src/event_notification.rs rename to crates/ecstore/src/services/event_notification.rs diff --git a/crates/ecstore/src/metrics_realtime.rs b/crates/ecstore/src/services/metrics_realtime.rs similarity index 100% rename from crates/ecstore/src/metrics_realtime.rs rename to crates/ecstore/src/services/metrics_realtime.rs diff --git a/crates/ecstore/src/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs similarity index 100% rename from crates/ecstore/src/notification_sys.rs rename to crates/ecstore/src/services/notification_sys.rs diff --git a/crates/ecstore/src/storage_api_contracts.rs b/crates/ecstore/src/storage_api_contracts/mod.rs similarity index 100% rename from crates/ecstore/src/storage_api_contracts.rs rename to crates/ecstore/src/storage_api_contracts/mod.rs diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 870684b7e..a8652037d 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,20 +5,27 @@ 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-store-init-layout` +- Branch: `overtrue/arch-ecstore-cluster-control-plane-layout` - 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-019/E-STORE-003 ECStore store init support module owner layout. -- Based on: E-018 branch while PR #3929 is pending; rebase onto `origin/main` - after E-017 and E-018 merge before opening this PR. -- PR type for this branch: `pure-move` -- Runtime behavior changes: none expected for E-019; this is a pure module file - move that keeps the existing `store_init` module path. -- Rust code changes: move ECStore store init support module into the `store/` - owner directory without function-body changes. +- Current phase PR: E-028/E-CLUSTER-001 ECStore cluster control-plane owner layout. +- Based on: E-027 branch while PR #3930 is pending; rebase onto `origin/main` + after E-017 through E-027 merge before opening this PR. +- PR type for this branch: `pure-move`. +- Runtime behavior changes: none expected for E-028; production changes are a + module file move that keeps the existing `cluster` facade paths. +- Rust code changes: move ECStore cluster control-plane implementation under + the `cluster/control_plane` owner module without function-body changes. - CI/script changes: reject restoring ECStore root `store.rs`, `set_disk.rs`, - `store_list_objects.rs`, `store_utils.rs`, or `store_init.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 + `notification_sys.rs`, `data_movement.rs`, or + `data_movement_backpressure.rs`, `admin_server_info.rs`, `data_usage.rs`, + `get_diagnostics.rs`, `global.rs`, `runtime_sources.rs`, `bitrot.rs`, + `compress.rs`, `rio.rs`, `error.rs`, `rebalance.rs`, `pools.rs`, + `sets.rs`, `pools_test.rs`, or `store_test.rs`; lock completed config model ownership against restoring old `rustfs_ecstore::config` model/accessor compatibility paths, and lock ECStore public facades against re-exporting the moved server-config symbols; @@ -54,8 +61,17 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block files after API-256, and reject restoring old ECStore server-config model or global accessor compatibility paths after CFG-009, reject restoring ECStore root `store.rs` or `set_disk.rs` after E-017, reject restoring ECStore root - store support modules after E-018, and reject restoring ECStore root - `store_init.rs` after E-019. + store support modules after E-018, reject restoring ECStore root + `store_init.rs` after E-019, reject restoring ECStore root layout facade + and storage API contract support modules after E-020, and reject restoring + ECStore root service runtime modules after E-021, and reject restoring + ECStore root data movement modules after E-022, and reject restoring ECStore + root usage and diagnostics modules after E-023, and reject restoring ECStore + root runtime global modules after E-024, and reject restoring ECStore root + I/O support modules after E-025, and reject restoring ECStore root error and + rebalance facade modules after E-026, and reject restoring ECStore root core + runtime/test modules after E-027, and keep the ECStore cluster root module as + a re-export-only owner facade after E-028. ## Phase 0 Tasks @@ -3932,6 +3948,158 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block formatting, diff hygiene, Rust risk scan, branch freshness check, pre-commit quality gate, and three-expert review. +- [x] `E-020/E-LAYOUT-001` Move ECStore layout facade and contract support modules under owners. + - Do: move `crates/ecstore/src/disks_layout.rs` to + `crates/ecstore/src/layout/disks_layout_facade.rs`, + `crates/ecstore/src/endpoints.rs` to + `crates/ecstore/src/layout/endpoints_facade.rs`, and + `crates/ecstore/src/storage_api_contracts.rs` to + `crates/ecstore/src/storage_api_contracts/mod.rs` while preserving the + existing crate module names; harden the pool-level list-object test against + parallel global lock-state pollution. + - Acceptance: `crate::disks_layout`, `crate::endpoints`, and + `crate::storage_api_contracts` continue to resolve through the same module + names, and migration rules reject restoring the old root files. + - Must preserve: disk layout facade re-exports, endpoint pool facade + re-exports, storage API contract domain aliases, ECStore internal contract + imports, pool-level list-object behavior, and all runtime side effects. + - Verification: focused ECStore compile/tests, migration and layer guards, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit quality gate, and three-expert review. + +- [x] `E-021/E-SERVICES-001` Move ECStore service runtime modules under the services owner. + - Do: move `crates/ecstore/src/batch_processor.rs`, + `crates/ecstore/src/event_notification.rs`, + `crates/ecstore/src/metrics_realtime.rs`, and + `crates/ecstore/src/notification_sys.rs` under + `crates/ecstore/src/services/` while preserving the existing crate module + names through path declarations; harden the pool-level list-object test + against parallel global setup-state pollution. + - Acceptance: `crate::batch_processor`, `crate::event_notification`, + `crate::metrics_realtime`, and `crate::notification_sys` continue to + resolve through the same module names, and migration rules reject restoring + the old root files. + - Must preserve: batch job serialization/status behavior, notification event + registration/dispatch, realtime metrics snapshots, notification system + lifecycle, peer event behavior, pool-level list-object behavior, and all + runtime side effects. + - Verification: focused ECStore compile/tests, migration and layer guards, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit quality gate, and three-expert review. + +- [x] `E-022/E-DATA-MOVEMENT-001` Move ECStore data movement modules under their owner. + - Do: move `crates/ecstore/src/data_movement.rs` to + `crates/ecstore/src/data_movement/mod.rs` and + `crates/ecstore/src/data_movement_backpressure.rs` to + `crates/ecstore/src/data_movement/backpressure.rs` while preserving the + existing crate module names. + - Acceptance: `crate::data_movement` and + `crate::data_movement_backpressure` continue to resolve through the same + module names, and migration rules reject restoring the old root files. + - Must preserve: decommission/rebalance data movement copy semantics, + overwrite-resume behavior, checksum propagation, abort flags, backpressure + admission policy, metrics, cancellation behavior, and all runtime side + effects. + - Verification: focused ECStore compile/tests, migration and layer guards, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit quality gate, and three-expert review. + +- [x] `E-023/E-USAGE-DIAGNOSTICS-001` Move ECStore usage and diagnostics modules under owners. + - Do: move `crates/ecstore/src/data_usage.rs` to + `crates/ecstore/src/data_usage/mod.rs`, + `crates/ecstore/src/admin_server_info.rs` to + `crates/ecstore/src/diagnostics/admin_server_info.rs`, and + `crates/ecstore/src/get_diagnostics.rs` to + `crates/ecstore/src/diagnostics/get.rs` while preserving the existing + crate module names. + - Acceptance: `crate::data_usage`, `crate::admin_server_info`, and + `crate::get_diagnostics` continue to resolve through the same module names, + and migration rules reject restoring the old root files. + - Must preserve: usage cache paths, local usage snapshots, bucket usage + aggregation, server info assembly, GET diagnostic labels, diagnostics error + classification, and all runtime side effects. + - Verification: focused ECStore compile/tests, migration and layer guards, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit quality gate, and three-expert review. + +- [x] `E-024/E-RUNTIME-001` Move ECStore runtime global modules under the runtime owner. + - Do: move `crates/ecstore/src/global.rs` to + `crates/ecstore/src/runtime/global.rs` and + `crates/ecstore/src/runtime_sources.rs` to + `crates/ecstore/src/runtime/sources.rs` while preserving the existing + crate module names through path declarations. + - Acceptance: `crate::global` and `crate::runtime_sources` continue to + resolve through the same module names, and migration rules reject restoring + the old root files. + - Must preserve: process-global state ownership, setup-type flags, endpoint + globals, object-store publication, lock-client publication, runtime source + adapters, test reset helpers, and all runtime side effects. + - Verification: focused ECStore compile/tests, migration and layer guards, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit quality gate, and three-expert review. + +- [x] `E-025/E-IO-001` Move ECStore I/O support modules under the IO owner. + - Do: move `crates/ecstore/src/bitrot.rs`, + `crates/ecstore/src/compress.rs`, and `crates/ecstore/src/rio.rs` under + `crates/ecstore/src/io_support/` while preserving the existing crate module + names through path declarations. + - Acceptance: `crate::bitrot`, `crate::compress`, and `crate::rio` continue + to resolve through the same module names, and migration rules reject + restoring the old root files. + - Must preserve: bitrot reader construction, disk compression matching, + compression config defaults/env handling, RIO backend selection, encryption + and compression metadata behavior, reader wrappers, and all runtime side + effects. + - Verification: focused ECStore compile/tests, migration and layer guards, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit quality gate, and three-expert review. + +- [x] `E-026/E-CORE-001` Move ECStore error and rebalance facades under owners. + - Do: move `crates/ecstore/src/error.rs` to + `crates/ecstore/src/error/mod.rs` and `crates/ecstore/src/rebalance.rs` to + `crates/ecstore/src/rebalance/mod.rs` while preserving the existing crate + module names through directory module resolution. + - Acceptance: `crate::error` and `crate::rebalance` continue to resolve + through the same module names, and migration rules reject restoring the old + root files. + - Must preserve: storage error variants/conversions/classifiers, S3 error + mapping, rebalance metadata exports, rebalance orchestration constants, + rebalance tests, and all runtime side effects. + - Verification: focused ECStore compile/tests, migration and layer guards, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit quality gate, and three-expert review. + +- [x] `E-027/E-CORE-001` Move ECStore core runtime and test harness modules under owner. + - Do: move `crates/ecstore/src/pools.rs` and + `crates/ecstore/src/sets.rs` under `crates/ecstore/src/core/`, and move + the root ECStore core harness files `pools_test.rs` and `store_test.rs` + under the same owner while preserving crate module names through path + declarations. + - Acceptance: `crate::pools`, `crate::sets`, and the existing test modules + continue to resolve through the same module names, and migration rules + reject restoring the old root files. + - Must preserve: pool orchestration, set routing, ECStore object API + behavior, test harness setup/teardown, lock/setup isolation, and all + runtime side effects. + - Verification: focused ECStore compile/tests, migration and layer guards, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit quality gate, and three-expert review. + +- [x] `E-028/E-CLUSTER-001` Move ECStore cluster control-plane implementation under owner. + - Do: move `crates/ecstore/src/cluster/mod.rs` implementation into + `crates/ecstore/src/cluster/control_plane.rs` and keep + `cluster/mod.rs` as a re-export-only facade. + - Acceptance: `rustfs_ecstore::api::cluster::*` and existing + `crate::cluster::*` consumers keep the same symbols, and migration rules + reject restoring control-plane implementation declarations to the cluster + root module. + - Must preserve: topology snapshots, membership grouping, pool-state + reporting, local-node storage projection, peer-health unknown status, + read-only facade behavior, and all cluster tests. + - Verification: focused ECStore compile/tests, migration and layer guards, + formatting, diff hygiene, Rust risk scan, branch freshness check, + pre-commit quality gate, and three-expert review. + - [x] `API-129` Route RustFS internal ECStore consumers through owner boundary. - Do: expose crate-local ECStore facade module aliases from `rustfs/src/storage/mod.rs` and migrate RustFS startup, server, capacity, @@ -5786,6 +5954,33 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| +| Quality/architecture | pass | E-028 moves the ECStore cluster control-plane implementation into an owner submodule and leaves the root module as a facade. | +| Migration preservation | pass | `crate::cluster::*` and `rustfs_ecstore::api::cluster::*` stay stable through root re-exports; the moved implementation is a 100% rename. | +| Testing/verification | pass | ECStore cluster tests, ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | +| Quality/architecture | pass | E-027 moves ECStore pools, sets, and root core test harness modules under the core owner directory and extends the root-file guard. | +| Migration preservation | pass | `crate::pools`, `crate::sets`, `pools_test`, and `store_test` stay stable through path declarations; moved files are 100% renames. | +| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | +| Quality/architecture | pass | E-026 moves ECStore error and rebalance facades under owner directories and extends the root-file guard. | +| Migration preservation | pass | `crate::error` and `crate::rebalance` stay stable through directory module resolution; moved files are 100% renames. | +| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | +| Quality/architecture | pass | E-025 moves ECStore I/O support modules under the `io_support` owner directory and extends the root-file guard. | +| Migration preservation | pass | `crate::bitrot`, `crate::compress`, and `crate::rio` stay stable through path declarations; moved files are 100% renames. | +| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | +| Quality/architecture | pass | E-024 moves ECStore runtime global state and runtime-source adapters under the `runtime` owner directory and extends the root-file guard. | +| Migration preservation | pass | `crate::global` and `crate::runtime_sources` stay stable through path declarations; moved files are 100% renames. | +| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | +| Quality/architecture | pass | E-023 moves ECStore usage and diagnostics/status modules under owner directories and extends the root-file guard. | +| Migration preservation | pass | `crate::data_usage`, `crate::admin_server_info`, and `crate::get_diagnostics` stay stable through directory module resolution or path declarations; moved files are 100% renames. | +| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | +| Quality/architecture | pass | E-022 moves ECStore data movement modules under the `data_movement` owner directory and extends the root-file guard. | +| Migration preservation | pass | `crate::data_movement` and `crate::data_movement_backpressure` stay stable through directory module resolution and a path declaration; moved files are 100% renames. | +| Testing/verification | pass | ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | +| Quality/architecture | pass | E-021 moves ECStore service runtime modules under the `services` owner directory, extends the root-file guard, and keeps setup-state test hardening local to `sets` coverage. | +| Migration preservation | pass | `crate::batch_processor`, `crate::event_notification`, `crate::metrics_realtime`, and `crate::notification_sys` stay stable through path declarations; moved files are 100% renames and the `sets` test restores setup state. | +| Testing/verification | pass | ECStore target test, default ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | +| Quality/architecture | pass | E-020 moves ECStore disk-layout/endpoints facade shims and storage API contracts under owner directories, extends the root-file guard, and keeps test hardening local to `sets` coverage. | +| Migration preservation | pass | `crate::disks_layout`, `crate::endpoints`, and `crate::storage_api_contracts` stay stable through module path declarations or directory module resolution; the `sets` test now owns its lock clients and resource names. | +| Testing/verification | pass | ECStore target test, default ECStore full tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | | Quality/architecture | pass | E-019 moves ECStore store init format support under the store owner directory and extends the existing root support guard. | | Migration preservation | pass | `crate::store_init` stays stable through a path declaration, and the moved file is a 100% rename with no function-body changes. | | Testing/verification | pass | ECStore tests, architecture/layer guards, formatting, diff hygiene, diff-added Rust risk scan, and pre-commit passed. | diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index a008b5ac8..4c784d147 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -132,6 +132,15 @@ EXTERNAL_ECSTORE_API_BOUNDARY_HITS_FILE="${TMP_DIR}/external_ecstore_api_boundar LEGACY_ECSTORE_CONFIG_MODEL_HITS_FILE="${TMP_DIR}/legacy_ecstore_config_model_hits.txt" ECSTORE_ROOT_STORE_SET_DISK_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_store_set_disk_module_hits.txt" ECSTORE_ROOT_STORE_SUPPORT_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_store_support_module_hits.txt" +ECSTORE_ROOT_LAYOUT_CONTRACT_SUPPORT_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_layout_contract_support_module_hits.txt" +ECSTORE_ROOT_SERVICE_RUNTIME_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_service_runtime_module_hits.txt" +ECSTORE_ROOT_DATA_MOVEMENT_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_data_movement_module_hits.txt" +ECSTORE_ROOT_USAGE_DIAGNOSTICS_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_usage_diagnostics_module_hits.txt" +ECSTORE_ROOT_RUNTIME_GLOBAL_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_runtime_global_module_hits.txt" +ECSTORE_ROOT_IO_SUPPORT_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_io_support_module_hits.txt" +ECSTORE_ROOT_ERROR_REBALANCE_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_error_rebalance_module_hits.txt" +ECSTORE_ROOT_CORE_RUNTIME_MODULE_HITS_FILE="${TMP_DIR}/ecstore_root_core_runtime_module_hits.txt" +ECSTORE_CLUSTER_ROOT_IMPL_HITS_FILE="${TMP_DIR}/ecstore_cluster_root_impl_hits.txt" ALL_STORAGE_COMPAT_SELF_FACADE_PATH_HITS_FILE="${TMP_DIR}/all_storage_compat_self_facade_path_hits.txt" RUSTFS_LOCAL_COMPAT_OWNER_SELF_PATH_HITS_FILE="${TMP_DIR}/rustfs_local_compat_owner_self_path_hits.txt" RUSTFS_ROOT_COMPAT_RELATIVE_CONSUMER_HITS_FILE="${TMP_DIR}/rustfs_root_compat_relative_consumer_hits.txt" @@ -700,17 +709,21 @@ fi rg -n --with-filename '^use rustfs_storage_api|^pub use rustfs_storage_api|rustfs_storage_api::' \ crates/ecstore/src \ --glob '*.rs' \ - --glob '!storage_api_contracts.rs' || true + --glob '!crates/ecstore/src/storage_api_contracts/mod.rs' \ + --glob '!crates/ecstore/src/storage_api_contracts/**' \ + --glob '!storage_api_contracts.rs' \ + --glob '!storage_api_contracts/mod.rs' \ + --glob '!storage_api_contracts/**' || true ) >"$ECSTORE_DIRECT_STORAGE_API_SOURCE_HITS_FILE" if [[ -s "$ECSTORE_DIRECT_STORAGE_API_SOURCE_HITS_FILE" ]]; then - report_failure "ECStore modules must route storage-api symbols through crates/ecstore/src/storage_api_contracts.rs: $(paste -sd '; ' "$ECSTORE_DIRECT_STORAGE_API_SOURCE_HITS_FILE")" + report_failure "ECStore modules must route storage-api symbols through crates/ecstore/src/storage_api_contracts: $(paste -sd '; ' "$ECSTORE_DIRECT_STORAGE_API_SOURCE_HITS_FILE")" fi ( cd "$ROOT_DIR" rg -n --with-filename '^pub(?:\(crate\))? use rustfs_storage_api' \ - crates/ecstore/src/storage_api_contracts.rs || true + crates/ecstore/src/storage_api_contracts/mod.rs || true ) >"$ECSTORE_STORAGE_API_ROOT_REEXPORT_HITS_FILE" if [[ -s "$ECSTORE_STORAGE_API_ROOT_REEXPORT_HITS_FILE" ]]; then @@ -723,11 +736,19 @@ fi rg -n -U --with-filename 'storage_api_contracts::\{\s*[A-Z]' \ crates/ecstore/src \ --glob '*.rs' \ - --glob '!storage_api_contracts.rs' || true + --glob '!crates/ecstore/src/storage_api_contracts/mod.rs' \ + --glob '!crates/ecstore/src/storage_api_contracts/**' \ + --glob '!storage_api_contracts.rs' \ + --glob '!storage_api_contracts/mod.rs' \ + --glob '!storage_api_contracts/**' || true rg -n --with-filename 'storage_api_contracts::[A-Z]' \ crates/ecstore/src \ --glob '*.rs' \ - --glob '!storage_api_contracts.rs' || true + --glob '!crates/ecstore/src/storage_api_contracts/mod.rs' \ + --glob '!crates/ecstore/src/storage_api_contracts/**' \ + --glob '!storage_api_contracts.rs' \ + --glob '!storage_api_contracts/mod.rs' \ + --glob '!storage_api_contracts/**' || true } ) >"$ECSTORE_STORAGE_API_ROOT_CONSUMER_HITS_FILE" @@ -2805,6 +2826,125 @@ if [[ -s "$ECSTORE_ROOT_STORE_SUPPORT_MODULE_HITS_FILE" ]]; then report_failure "ECStore store support modules must stay under the store owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_STORE_SUPPORT_MODULE_HITS_FILE")" fi +( + cd "$ROOT_DIR" + { + [[ -e crates/ecstore/src/disks_layout.rs ]] && printf '%s\n' 'crates/ecstore/src/disks_layout.rs' + [[ -e crates/ecstore/src/endpoints.rs ]] && printf '%s\n' 'crates/ecstore/src/endpoints.rs' + [[ -e crates/ecstore/src/storage_api_contracts.rs ]] && printf '%s\n' 'crates/ecstore/src/storage_api_contracts.rs' + true + } +) >"$ECSTORE_ROOT_LAYOUT_CONTRACT_SUPPORT_MODULE_HITS_FILE" + +if [[ -s "$ECSTORE_ROOT_LAYOUT_CONTRACT_SUPPORT_MODULE_HITS_FILE" ]]; then + report_failure "ECStore layout facade and storage API contract modules must stay under owner directories: $(paste -sd '; ' "$ECSTORE_ROOT_LAYOUT_CONTRACT_SUPPORT_MODULE_HITS_FILE")" +fi + +( + cd "$ROOT_DIR" + { + [[ -e crates/ecstore/src/batch_processor.rs ]] && printf '%s\n' 'crates/ecstore/src/batch_processor.rs' + [[ -e crates/ecstore/src/event_notification.rs ]] && printf '%s\n' 'crates/ecstore/src/event_notification.rs' + [[ -e crates/ecstore/src/metrics_realtime.rs ]] && printf '%s\n' 'crates/ecstore/src/metrics_realtime.rs' + [[ -e crates/ecstore/src/notification_sys.rs ]] && printf '%s\n' 'crates/ecstore/src/notification_sys.rs' + true + } +) >"$ECSTORE_ROOT_SERVICE_RUNTIME_MODULE_HITS_FILE" + +if [[ -s "$ECSTORE_ROOT_SERVICE_RUNTIME_MODULE_HITS_FILE" ]]; then + report_failure "ECStore service runtime modules must stay under the services owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_SERVICE_RUNTIME_MODULE_HITS_FILE")" +fi + +( + cd "$ROOT_DIR" + { + [[ -e crates/ecstore/src/data_movement.rs ]] && printf '%s\n' 'crates/ecstore/src/data_movement.rs' + [[ -e crates/ecstore/src/data_movement_backpressure.rs ]] && printf '%s\n' 'crates/ecstore/src/data_movement_backpressure.rs' + true + } +) >"$ECSTORE_ROOT_DATA_MOVEMENT_MODULE_HITS_FILE" + +if [[ -s "$ECSTORE_ROOT_DATA_MOVEMENT_MODULE_HITS_FILE" ]]; then + report_failure "ECStore data movement modules must stay under the data_movement owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_DATA_MOVEMENT_MODULE_HITS_FILE")" +fi + +( + cd "$ROOT_DIR" + { + [[ -e crates/ecstore/src/admin_server_info.rs ]] && printf '%s\n' 'crates/ecstore/src/admin_server_info.rs' + [[ -e crates/ecstore/src/data_usage.rs ]] && printf '%s\n' 'crates/ecstore/src/data_usage.rs' + [[ -e crates/ecstore/src/get_diagnostics.rs ]] && printf '%s\n' 'crates/ecstore/src/get_diagnostics.rs' + true + } +) >"$ECSTORE_ROOT_USAGE_DIAGNOSTICS_MODULE_HITS_FILE" + +if [[ -s "$ECSTORE_ROOT_USAGE_DIAGNOSTICS_MODULE_HITS_FILE" ]]; then + report_failure "ECStore usage and diagnostics modules must stay under owner directories: $(paste -sd '; ' "$ECSTORE_ROOT_USAGE_DIAGNOSTICS_MODULE_HITS_FILE")" +fi + +( + cd "$ROOT_DIR" + { + [[ -e crates/ecstore/src/global.rs ]] && printf '%s\n' 'crates/ecstore/src/global.rs' + [[ -e crates/ecstore/src/runtime_sources.rs ]] && printf '%s\n' 'crates/ecstore/src/runtime_sources.rs' + true + } +) >"$ECSTORE_ROOT_RUNTIME_GLOBAL_MODULE_HITS_FILE" + +if [[ -s "$ECSTORE_ROOT_RUNTIME_GLOBAL_MODULE_HITS_FILE" ]]; then + report_failure "ECStore runtime global modules must stay under the runtime owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_RUNTIME_GLOBAL_MODULE_HITS_FILE")" +fi + +( + cd "$ROOT_DIR" + { + [[ -e crates/ecstore/src/bitrot.rs ]] && printf '%s\n' 'crates/ecstore/src/bitrot.rs' + [[ -e crates/ecstore/src/compress.rs ]] && printf '%s\n' 'crates/ecstore/src/compress.rs' + [[ -e crates/ecstore/src/rio.rs ]] && printf '%s\n' 'crates/ecstore/src/rio.rs' + true + } +) >"$ECSTORE_ROOT_IO_SUPPORT_MODULE_HITS_FILE" + +if [[ -s "$ECSTORE_ROOT_IO_SUPPORT_MODULE_HITS_FILE" ]]; then + report_failure "ECStore I/O support modules must stay under the io_support owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_IO_SUPPORT_MODULE_HITS_FILE")" +fi + +( + cd "$ROOT_DIR" + { + [[ -e crates/ecstore/src/error.rs ]] && printf '%s\n' 'crates/ecstore/src/error.rs' + [[ -e crates/ecstore/src/rebalance.rs ]] && printf '%s\n' 'crates/ecstore/src/rebalance.rs' + true + } +) >"$ECSTORE_ROOT_ERROR_REBALANCE_MODULE_HITS_FILE" + +if [[ -s "$ECSTORE_ROOT_ERROR_REBALANCE_MODULE_HITS_FILE" ]]; then + report_failure "ECStore error and rebalance modules must stay under owner directories: $(paste -sd '; ' "$ECSTORE_ROOT_ERROR_REBALANCE_MODULE_HITS_FILE")" +fi + +( + cd "$ROOT_DIR" + { + [[ -e crates/ecstore/src/pools.rs ]] && printf '%s\n' 'crates/ecstore/src/pools.rs' + [[ -e crates/ecstore/src/sets.rs ]] && printf '%s\n' 'crates/ecstore/src/sets.rs' + [[ -e crates/ecstore/src/pools_test.rs ]] && printf '%s\n' 'crates/ecstore/src/pools_test.rs' + [[ -e crates/ecstore/src/store_test.rs ]] && printf '%s\n' 'crates/ecstore/src/store_test.rs' + true + } +) >"$ECSTORE_ROOT_CORE_RUNTIME_MODULE_HITS_FILE" + +if [[ -s "$ECSTORE_ROOT_CORE_RUNTIME_MODULE_HITS_FILE" ]]; then + report_failure "ECStore core runtime modules must stay under the core owner directory: $(paste -sd '; ' "$ECSTORE_ROOT_CORE_RUNTIME_MODULE_HITS_FILE")" +fi + +rg -n --with-filename 'pub (struct|enum|fn) Cluster|pub fn (topology|membership|pool_state|local_node_storage|peer_health)_snapshot' \ + "${ROOT_DIR}/crates/ecstore/src/cluster/mod.rs" \ + >"$ECSTORE_CLUSTER_ROOT_IMPL_HITS_FILE" || true + +if [[ -s "$ECSTORE_CLUSTER_ROOT_IMPL_HITS_FILE" ]]; then + report_failure "ECStore cluster root module must only re-export control-plane owner symbols: $(paste -sd '; ' "$ECSTORE_CLUSTER_ROOT_IMPL_HITS_FILE")" +fi + cat >"$ECSTORE_COMPAT_PASSTHROUGH_EXPECTED_FILE" <<'EOF' EOF sort -o "$ECSTORE_COMPAT_PASSTHROUGH_EXPECTED_FILE" "$ECSTORE_COMPAT_PASSTHROUGH_EXPECTED_FILE"