From 15cdeee775a67bbed1ef829b72b218afa6681d6e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Sun, 21 Jun 2026 10:45:36 +0800 Subject: [PATCH] refactor: add ecstore cluster read models (#3681) --- crates/ecstore/src/api/mod.rs | 8 + crates/ecstore/src/cluster/mod.rs | 409 ++++++++++++++++++ crates/ecstore/src/lib.rs | 1 + docs/architecture/crate-boundaries.md | 4 + docs/architecture/migration-progress.md | 74 +++- .../storage-control-data-plane.md | 5 + rustfs/src/runtime_capabilities.rs | 93 +--- rustfs/src/storage_compat.rs | 1 + scripts/check_architecture_migration_rules.sh | 10 +- 9 files changed, 509 insertions(+), 96 deletions(-) create mode 100644 crates/ecstore/src/cluster/mod.rs diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index d4465ccd0..6948f5e5e 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -45,6 +45,14 @@ pub mod client { pub use crate::client::{admin_handler_utils, object_api_utils, transition_api}; } +pub mod cluster { + pub use crate::cluster::{ + ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterDriveMembership, ClusterEndpointType, ClusterMembershipSnapshot, + ClusterNodeMembership, membership_snapshot_from_endpoint_pools, topology_snapshot_from_endpoint_pools, + topology_snapshot_from_endpoint_pools_with_capabilities, + }; +} + pub mod compression { pub use crate::compress::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible}; } diff --git a/crates/ecstore/src/cluster/mod.rs b/crates/ecstore/src/cluster/mod.rs new file mode 100644 index 000000000..f56d6d7ef --- /dev/null +++ b/crates/ecstore/src/cluster/mod.rs @@ -0,0 +1,409 @@ +// 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; + +use rustfs_storage_api::{ + 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"; + +#[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 read_snapshot(&self) -> ClusterControlPlaneSnapshot { + ClusterControlPlaneSnapshot { + topology: self.topology_snapshot(), + membership: self.membership_snapshot(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterControlPlaneSnapshot { + pub topology: TopologySnapshot, + 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, Copy, PartialEq, Eq)] +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, + } +} + +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 { + 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") + ); + + 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 + .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 control_plane_read_snapshot_combines_topology_and_membership() { + let control_plane = ClusterControlPlane::new(sample_path_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!(node_ids, BTreeSet::from([LOCAL_NODE_ID])); + } + + 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(), + }]) + } +} diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index 644d1cae9..3e8fa724f 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -21,6 +21,7 @@ mod batch_processor; mod bitrot; mod bucket; mod cache_value; +mod cluster; mod compress; mod config; mod data_movement; diff --git a/docs/architecture/crate-boundaries.md b/docs/architecture/crate-boundaries.md index 2f695de61..419dbd759 100644 --- a/docs/architecture/crate-boundaries.md +++ b/docs/architecture/crate-boundaries.md @@ -143,6 +143,10 @@ batch processor root modules once their facade groups exist. ECStore root `global` re-exports must also stay removed once consumers use `rustfs_ecstore::api::global` or crate-internal `crate::global` paths. +ECStore ClusterControlPlane read models must stay owned by the crate-private +`cluster` module. Public access goes through `rustfs_ecstore::api::cluster` so +outer crates cannot depend on ECStore root control-plane internals. + RustFS startup internals must stay crate-private after the startup owner split. Only `startup_entrypoint` remains a public startup module for the binary entrypoint; IAM bootstrap, optional runtime, and profiling startup shims must diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 70df41426..1f9d7cc99 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,15 +5,15 @@ 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-app-context-bootstrap-result` -- Baseline: completed `R-050/R-051`. -- Stacked on: embedded startup config identity slice. -- PR type for this branch: `pure-move` +- Branch: `overtrue/arch-cluster-control-plane-read-models` +- Baseline: completed `API-078`. +- Stacked on: ECStore root global facade prune. +- PR type for this branch: `api-extraction` - Runtime behavior changes: none. -- Rust code changes: make IAM AppContext bootstrap outcome explicit and reuse - it in inline and deferred IAM readiness paths. -- CI/script changes: none. -- Docs changes: record the AppContext bootstrap result slices. +- Rust code changes: add ECStore ClusterControlPlane read models and route + RustFS runtime endpoint topology snapshots through the ECStore owner facade. +- CI/script changes: extend architecture rules for the cluster facade boundary. +- Docs changes: record the C-001/C-002/C-003 read-model slice. ## Phase 0 Tasks @@ -498,6 +498,47 @@ 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. +## Phase 5 Cluster Control Plane Tasks + +- [x] `C-001` Add topology model. + - Completed slice: move endpoint-pool topology mapping behind ECStore's + crate-private `cluster` owner module and expose it through + `rustfs_ecstore::api::cluster`. + - Acceptance: pool, set, and disk topology snapshots are built from existing + endpoint assignments without exposing local disk paths or changing + placement, readiness, or endpoint construction. + - Must preserve: endpoint pool/set/disk indexes, local path privacy, + storage-api topology contract shape, runtime capability reasons, and + existing RustFS topology provider behavior. + - Verification: ECStore topology tests, RustFS runtime topology tests, + migration guard, compile coverage, formatting, diff hygiene, risk scan, + pre-commit quality gate, and three-expert review. + +- [x] `C-002` Add membership model. + - Completed slice: add a static membership snapshot that groups endpoint + drives by node identity and records drive placement without dynamic + membership, health checks, control RPC, or hot-path changes. + - Acceptance: URL endpoints group by host:port, path endpoints group under a + local node identity, and drive membership carries pool/set/disk placement + plus endpoint type/local flags. + - Must preserve: no Raft, no Kubernetes watcher, no peer-health behavior, no + dynamic membership, and no object I/O or lock-quorum behavior changes. + - Verification: ECStore membership tests, compile coverage, migration guard, + formatting, diff hygiene, risk scan, pre-commit quality gate, and + three-expert review. + +- [x] `C-003` Add read-only control plane facade. + - Completed slice: add `ClusterControlPlane` as a read-only facade that + combines topology and membership snapshots from existing endpoint pools. + - Acceptance: outer crates use `rustfs_ecstore::api::cluster` for the facade, + while ECStore root `cluster` remains crate-private and migration rules + reject restoring it as a public root module. + - Must preserve: no worker start/stop, health impact, lock registry mutation, + pool state mutation, endpoint publication, or readiness behavior changes. + - Verification: control-plane read-snapshot test, migration guard, compile + coverage, formatting, diff hygiene, risk scan, pre-commit 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 @@ -3026,14 +3067,25 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | API-078 removes remaining ECStore global root re-exports without runtime logic changes. | -| Migration preservation | passed | Outer access remains through `rustfs_ecstore::api::global`; internal ECStore users route directly to `crate::global`. | -| Testing/verification | passed | ECStore all-target and RustFS lib/bin compile coverage, migration guards, formatting, diff hygiene, added-line risk scan, and pre-commit gate passed. | +| Quality/architecture | passed | C-001/C-002/C-003 keeps ClusterControlPlane read models in ECStore with public access only through `api::cluster`. | +| Migration preservation | passed | Topology and static membership are read-only snapshots from existing endpoint pools; no placement, readiness, health, or lock behavior changes. | +| Testing/verification | passed | ECStore cluster tests, RustFS runtime capability tests, compile coverage, migration guard, formatting, diff hygiene, risk scan, and pre-commit gate passed. | ## Verification Notes Passed before push: +- Issue #660 C-001/C-002/C-003 current slice: + - `cargo check -p rustfs-ecstore --all-targets`: passed. + - `cargo check -p rustfs --lib --bins`: passed. + - `cargo test -p rustfs-ecstore cluster -- --nocapture`: passed; 4 tests. + - `cargo test -p rustfs --lib runtime_capabilities -- --nocapture`: passed; 3 tests. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - Rust added-line risk scan on changed Rust files and guard script: passed. + - `make pre-commit`: passed. + - Issue #660 API-078 current slice: - `cargo check -p rustfs-ecstore --all-targets`: passed. - `cargo check -p rustfs --lib --bins`: passed. diff --git a/docs/architecture/storage-control-data-plane.md b/docs/architecture/storage-control-data-plane.md index cff608c81..71de6152d 100644 --- a/docs/architecture/storage-control-data-plane.md +++ b/docs/architecture/storage-control-data-plane.md @@ -37,6 +37,11 @@ Initial scope: - Peer health snapshot. - Pool state snapshot. +The first read-only implementation lives behind `rustfs_ecstore::api::cluster`. +It maps existing endpoint pools into the shared storage-api topology contract and +an ECStore-owned static membership snapshot. It must not expose local disk paths, +start health checks, mutate endpoint ownership, or change placement/readiness. + Risk controls: - Distributed lock quorum remains per set. diff --git a/rustfs/src/runtime_capabilities.rs b/rustfs/src/runtime_capabilities.rs index 7b34bae55..1ad90a5d0 100644 --- a/rustfs/src/runtime_capabilities.rs +++ b/rustfs/src/runtime_capabilities.rs @@ -12,15 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::collections::BTreeMap; - use rustfs_storage_api::{ CapabilitySnapshotError, CapabilityStatus, DiskCapabilities, MemorySamplingState, ObservabilitySnapshot, - ObservabilitySnapshotProvider, PlatformSupport, TopologyCapabilities, TopologyDisk, TopologyLabels, TopologyPool, - TopologySet, TopologySnapshot, TopologySnapshotProvider, UserspaceProfilingCapability, + ObservabilitySnapshotProvider, PlatformSupport, TopologyCapabilities, TopologySnapshot, TopologySnapshotProvider, + UserspaceProfilingCapability, }; -use crate::storage_compat::{Endpoint, EndpointServerPools}; +use crate::storage_compat::EndpointServerPools; const NOT_WIRED_INTO_RUNTIME: &str = "not wired into runtime"; const STORAGE_MEDIA_NOT_REPORTED: &str = "storage media not reported by endpoints"; @@ -82,94 +80,21 @@ impl TopologySnapshotProvider for EndpointTopologySnapshotProvider { } pub fn topology_snapshot_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> 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()); - TopologyPool { - pool_index, - pool_id: None, - labels: TopologyLabels::default(), - sets, - } - }) - .collect(), - capabilities: TopologyCapabilities { + crate::storage_compat::topology_snapshot_from_endpoint_pools_with_capabilities( + endpoint_pools, + TopologyCapabilities { profiling: cpu_profiling_status(), 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 topology_sets_from_endpoints(pool_index: usize, drives_per_set: usize, endpoints: &[Endpoint]) -> Vec { - let safe_drives_per_set = drives_per_set.max(1); - let mut sets = BTreeMap::>::new(); - - for (endpoint_index, endpoint) in endpoints.iter().enumerate() { - 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); - - sets.entry(set_index) - .or_default() - .push(topology_disk_from_endpoint(pool_index, set_index, disk_index, endpoint)); - } - - 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) -> TopologyDisk { - TopologyDisk { - pool_index, - set_index, - disk_index, - disk_id: endpoint_disk_id(endpoint), - labels: endpoint_labels(endpoint), - 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: cpu_profiling_status(), }, - } -} - -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_labels(endpoint: &Endpoint) -> TopologyLabels { - let mut additional = BTreeMap::new(); - additional.insert( - "endpoint_type".to_owned(), - if endpoint.url.scheme() == "file" { "path" } else { "url" }.to_owned(), - ); - additional.insert("local".to_owned(), endpoint.is_local.to_string()); - - TopologyLabels { - additional, - ..TopologyLabels::default() - } -} - -fn non_negative_index(index: i32) -> Option { - usize::try_from(index).ok() + ) } fn runtime_telemetry_status() -> CapabilityStatus { @@ -207,7 +132,7 @@ fn cgroup_memory_status() -> CapabilityStatus { #[cfg(test)] mod tests { use super::*; - use crate::storage_compat::{Endpoints, PoolEndpoints}; + use crate::storage_compat::{Endpoint, Endpoints, PoolEndpoints}; use rustfs_storage_api::{CapabilityState, ObservabilitySnapshotProvider, TopologySnapshotProvider}; #[tokio::test] diff --git a/rustfs/src/storage_compat.rs b/rustfs/src/storage_compat.rs index 2c7f6c6b2..92002095a 100644 --- a/rustfs/src/storage_compat.rs +++ b/rustfs/src/storage_compat.rs @@ -18,6 +18,7 @@ pub(crate) use rustfs_ecstore::api::bucket::replication::{ GLOBAL_REPLICATION_STATS, get_global_replication_pool, init_background_replication, }; pub(crate) use rustfs_ecstore::api::bucket::{metadata, metadata_sys, quota}; +pub(crate) use rustfs_ecstore::api::cluster::topology_snapshot_from_endpoint_pools_with_capabilities; pub(crate) use rustfs_ecstore::api::config::{com, init, init_global_config_sys, try_migrate_server_config}; pub(crate) use rustfs_ecstore::api::disk::{DiskAPI, RUSTFS_META_BUCKET, endpoint::Endpoint}; pub(crate) use rustfs_ecstore::api::error::{Error as EcstoreError, Result as EcstoreResult, StorageError}; diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 3c8c69fab..db219301f 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -233,6 +233,14 @@ require_source_line \ "crates/ecstore/src/lib.rs" \ "mod endpoints;" \ "ECStore legacy endpoint compatibility module crate-private visibility" +require_source_line \ + "crates/ecstore/src/lib.rs" \ + "mod cluster;" \ + "ECStore cluster control-plane owner module crate-private visibility" +require_source_line \ + "crates/ecstore/src/api/mod.rs" \ + "pub mod cluster {" \ + "ECStore cluster control-plane public facade" for ecstore_private_module in \ admin_server_info \ bucket \ @@ -798,7 +806,7 @@ if rg -n --no-heading '^\s*pub\s+mod\s+store_api\s*;' "$ROOT_DIR/crates/ecstore/ report_failure "ECStore store_api must remain private; expose ECStore-owned object DTO and reader aliases through rustfs_ecstore::object_api" fi -if rg -n --no-heading '^\s*pub\s+mod\s+(batch_processor|bitrot|erasure_coding|event|object_api|store_list_objects)\s*;' "$ROOT_DIR/crates/ecstore/src/lib.rs" >"$ECSTORE_PUBLIC_ROOT_MODULE_HITS_FILE"; then +if rg -n --no-heading '^\s*pub\s+mod\s+(batch_processor|bitrot|cluster|erasure_coding|event|object_api|store_list_objects)\s*;' "$ROOT_DIR/crates/ecstore/src/lib.rs" >"$ECSTORE_PUBLIC_ROOT_MODULE_HITS_FILE"; then report_failure "facade-covered ECStore root modules must remain private; expose compatibility through rustfs_ecstore::api: $(paste -sd '; ' "$ECSTORE_PUBLIC_ROOT_MODULE_HITS_FILE")" fi