diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 9dca16c22..911a0b8c6 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::config::{audit, notify, oidc, set_global_storage_class, storageclass}; +use crate::config::{audit, notify, oidc, storageclass}; use crate::disk::{MIGRATING_META_BUCKET, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; @@ -1288,7 +1288,7 @@ where match storageclass::lookup_config(&kvs, *count) { Ok(res) => { if i == 0 { - set_global_storage_class(res); + runtime_sources::set_storage_class_config(res); } } Err(err) => { @@ -1312,8 +1312,8 @@ mod tests { use crate::disk::endpoint::Endpoint; use crate::endpoints::SetupType; use crate::error::{Error, Result}; - use crate::global::{is_dist_erasure, is_erasure, is_erasure_sd, update_erasure_type}; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; + use crate::runtime_sources; use crate::set_disk::SetDisks; use http::HeaderMap; use rustfs_config::audit::{AUDIT_AMQP_SUB_SYS, AUDIT_KAFKA_SUB_SYS, AUDIT_MQTT_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS}; @@ -1413,7 +1413,7 @@ mod tests { impl SetupTypeGuard { async fn switch_to(next: SetupType) -> Self { let previous = current_setup_type().await; - update_erasure_type(next).await; + runtime_sources::set_setup_type(next).await; Self { previous } } } @@ -1424,22 +1424,14 @@ mod tests { let handle = tokio::runtime::Handle::current(); tokio::task::block_in_place(|| { handle.block_on(async move { - update_erasure_type(previous).await; + runtime_sources::set_setup_type(previous).await; }); }); } } async fn current_setup_type() -> SetupType { - if is_dist_erasure().await { - SetupType::DistErasure - } else if is_erasure_sd().await { - SetupType::ErasureSD - } else if is_erasure().await { - SetupType::Erasure - } else { - SetupType::Unknown - } + runtime_sources::current_setup_type().await } impl LockingConfigStorage { diff --git a/crates/ecstore/src/runtime_sources.rs b/crates/ecstore/src/runtime_sources.rs index 05657bbbd..a76e87544 100644 --- a/crates/ecstore/src/runtime_sources.rs +++ b/crates/ecstore/src/runtime_sources.rs @@ -17,12 +17,14 @@ use std::{collections::HashMap, sync::Arc, time::SystemTime}; use crate::bucket::bandwidth::monitor::Monitor; use crate::disk::endpoint::Endpoint; use crate::{ + batch_processor::{GlobalBatchProcessors, get_global_processors}, bucket::lifecycle::bucket_lifecycle_ops::{ExpiryState, GLOBAL_ExpiryState, GLOBAL_TransitionState, TransitionState}, bucket::metadata_sys::{BucketMetadataSys, get_global_bucket_metadata_sys}, bucket::replication::{DynReplicationPool, GLOBAL_REPLICATION_POOL, GLOBAL_REPLICATION_STATS, ReplicationStats}, config::{get_global_storage_class, set_global_storage_class, storageclass}, disk::{DiskAPI, DiskOption, DiskStore, new_disk}, endpoints::EndpointServerPools, + endpoints::SetupType, error::Result, event_notification::EventNotifier, global::{ @@ -31,7 +33,7 @@ use crate::{ TypeLocalDiskSetDrives, get_global_bucket_monitor, get_global_deployment_id, get_global_endpoints, get_global_endpoints_opt, get_global_lock_clients, get_global_region, get_global_tier_config_mgr, global_rustfs_port, init_global_bucket_monitor, is_dist_erasure, is_erasure, is_first_cluster_node_local, resolve_object_store_handle, - set_global_deployment_id, set_global_lock_client, set_global_lock_clients, set_object_layer, + set_global_deployment_id, set_global_lock_client, set_global_lock_clients, set_object_layer, update_erasure_type, }, notification_sys::{NotificationSys, get_global_notification_sys}, store::ECStore, @@ -95,6 +97,22 @@ pub(crate) async fn setup_is_erasure_sd() -> bool { *GLOBAL_IsErasureSD.read().await } +pub(crate) async fn current_setup_type() -> SetupType { + if setup_is_dist_erasure().await { + SetupType::DistErasure + } else if setup_is_erasure_sd().await { + SetupType::ErasureSD + } else if setup_is_erasure().await { + SetupType::Erasure + } else { + SetupType::Unknown + } +} + +pub(crate) async fn set_setup_type(setup_type: SetupType) { + update_erasure_type(setup_type).await; +} + pub(crate) async fn local_node_name() -> String { GLOBAL_LOCAL_NODE_NAME.read().await.clone() } @@ -244,6 +262,10 @@ pub(crate) fn set_storage_class_config(config: storageclass::Config) { set_global_storage_class(config); } +pub(crate) fn batch_processors() -> &'static GlobalBatchProcessors { + get_global_processors() +} + pub(crate) fn global_tier_config_mgr() -> Arc> { get_global_tier_config_mgr() } diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 143e5914a..359e97778 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -15,7 +15,7 @@ #![allow(unused_imports)] #![allow(unused_variables)] -use crate::batch_processor::{AsyncBatchProcessor, get_global_processors}; +use crate::batch_processor::AsyncBatchProcessor; use crate::bitrot::{create_bitrot_reader, create_bitrot_writer}; use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE; use crate::bucket::metadata_sys; @@ -5306,7 +5306,6 @@ mod tests { use crate::disk::error::DiskError; use crate::disk::health_state::RuntimeDriveHealthState; use crate::endpoints::SetupType; - use crate::global::{is_dist_erasure, is_erasure, is_erasure_sd, update_erasure_type}; use crate::object_api::ObjectInfo; use crate::store_init::save_format_file; use crate::store_list_objects::ListPathOptions; @@ -5447,7 +5446,7 @@ mod tests { impl SetupTypeGuard { async fn switch_to(next: SetupType) -> Self { let previous = current_setup_type().await; - update_erasure_type(next).await; + runtime_sources::set_setup_type(next).await; Self { previous } } } @@ -5458,22 +5457,14 @@ mod tests { let handle = tokio::runtime::Handle::current(); tokio::task::block_in_place(|| { handle.block_on(async move { - update_erasure_type(previous).await; + runtime_sources::set_setup_type(previous).await; }); }); } } async fn current_setup_type() -> SetupType { - if is_dist_erasure().await { - SetupType::DistErasure - } else if is_erasure_sd().await { - SetupType::ErasureSD - } else if is_erasure().await { - SetupType::Erasure - } else { - SetupType::Unknown - } + runtime_sources::current_setup_type().await } async fn make_formatted_local_disk_for_info_test(disk_idx: usize, format: &FormatV3) -> (TempDir, Endpoint, DiskStore) { diff --git a/crates/ecstore/src/set_disk/lock.rs b/crates/ecstore/src/set_disk/lock.rs index e6200b9fa..6b3bc2621 100644 --- a/crates/ecstore/src/set_disk/lock.rs +++ b/crates/ecstore/src/set_disk/lock.rs @@ -166,7 +166,7 @@ impl SetDisks { }); } - let processor = get_global_processors().metadata_processor(); + let processor = runtime_sources::batch_processors().metadata_processor(); let results = processor.execute_batch(futures).await; for (submitted_idx, result) in results.into_iter().enumerate() { diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index bb0585520..e8fbe4bc7 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -307,7 +307,7 @@ impl SetDisks { let version_id = version_id.to_string(); let opts = opts.clone(); - let processor = get_global_processors().read_processor(); + let processor = runtime_sources::batch_processors().read_processor(); let tasks: Vec<_> = disks .iter() .take(required_reads + 2) // Read a few extra for reliability diff --git a/crates/ecstore/src/set_disk/write.rs b/crates/ecstore/src/set_disk/write.rs index 9da1b97fd..2704f140b 100644 --- a/crates/ecstore/src/set_disk/write.rs +++ b/crates/ecstore/src/set_disk/write.rs @@ -270,7 +270,7 @@ impl SetDisks { let mut errs = Vec::with_capacity(disks.len()); // Use improved simple batch processor instead of join_all for better performance - let processor = get_global_processors().write_processor(); + let processor = runtime_sources::batch_processors().write_processor(); let tasks: Vec<_> = disks .iter() diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index a2046db8b..562b69622 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,9 +5,9 @@ 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-accessor-runtime-sources-batch` -- Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124/API-125/API-126/API-127/API-128/API-129/API-130/API-131/API-132/API-133/API-134/API-135/API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153/API-154/API-155/API-156/API-157/API-158/API-159/API-160/API-161/API-162/API-163/API-164/API-165/API-166/API-167/API-168/API-169/API-170/API-171/API-172/API-173/API-174/API-175/API-176/API-177/API-178/API-179/API-180/API-181/API-182/API-183/API-184/API-185/API-186/API-187/API-188/API-189/API-190/API-191/API-192/API-193/API-194/API-195/API-196`. -- Based on: stacked on API-195 PR branch while PR #3812 is pending. +- Branch: `overtrue/arch-ecstore-batch-config-runtime-sources` +- 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`. +- Based on: API-196 merged main. - PR type for this branch: `consumer-migration` - Runtime behavior changes: none. - Rust code changes: route replication pool, outbound TLS generation, runtime @@ -32,7 +32,8 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block client paths, plus ECStore facade server-config, storage-class, notification, bucket-metadata, endpoint, region, tier-config, server-address, object-store publication, lock-client publication, and local-node publication - paths, + paths, plus ECStore batch processor, dynamic storage-class publication, and + RustFS cluster snapshot facade alias paths, through AppContext-first or owner-crate resolver boundaries. - CI/script changes: lock completed owner and test/fuzz boundaries against bare/glob imports, scattered raw ECStore facade subpaths, and startup @@ -42,7 +43,7 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block and storage owner thin bridge regressions, plus app context and notify event-bridge thin module regressions; accept the reviewed AppContext resolver reverse dependencies in the layer baseline. -- Docs changes: record the API-136 through API-196 owner facade cleanup. +- Docs changes: record the API-136 through API-197 owner facade cleanup. ## Phase 0 Tasks @@ -4865,6 +4866,20 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block formatting, diff hygiene, residual runtime-source scan, migration/layer guards, PR-before-push quality gate, and three-expert review. +- [x] `API-197` Centralize ECStore batch/config runtime sources and cluster aliases. + - Do: route set-disk batch processor selection and dynamic storage-class + publication through the ECStore-owned runtime-source module, and keep + RustFS cluster snapshot ECStore facade imports behind local ecstore aliases. + - Acceptance: set-disk read/write/metadata paths, dynamic config + application, and cluster snapshot consumers no longer import or call those + ECStore process-global or facade paths directly outside owner boundaries. + - Must preserve: read/write/metadata batch concurrency levels, dynamic + storage-class lookup by drive count, first-set publication behavior, and + config error handling. + - Verification: ECStore compile coverage, focused set-disk/config checks, + formatting, diff hygiene, residual runtime-source scan, migration/layer + guards, PR-before-push quality gate, and three-expert review. + ## Next PRs 1. `consumer-migration`: continue reducing direct global reads behind AppContext resolver boundaries. @@ -4873,6 +4888,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| +| Quality/architecture | pass | API-197 keeps set-disk batch processor lookup, dynamic storage-class publication, and RustFS cluster snapshot facade imports behind owner boundaries. | +| Migration preservation | pass | Batch concurrency defaults, storage-class lookup by drive count, first-set publication, config error handling, and cluster snapshot response shape preserve existing behavior. | +| Testing/verification | pass | ECStore compile, focused set-disk/config checks, formatting, migration/layer guards, residual scan, fast PR gate, and full PR gate are planned before PR. | | Quality/architecture | pass | API-196 keeps ECStore facade, object-layer, lock-client, local-node, and server-address runtime access behind the runtime-source boundary. | | Migration preservation | pass | Facade signatures, endpoint fallback, object-store publication, lock-client duplicate handling, and local-node fallback behavior preserve existing semantics. | | Testing/verification | pass | ECStore compile and focused store-init/accessor tests have passed; formatting, migration/layer guards, residual scan, and PR gate are planned before PR. | diff --git a/rustfs/src/admin/handlers/cluster_snapshot.rs b/rustfs/src/admin/handlers/cluster_snapshot.rs index 953269941..cd6974934 100644 --- a/rustfs/src/admin/handlers/cluster_snapshot.rs +++ b/rustfs/src/admin/handlers/cluster_snapshot.rs @@ -14,6 +14,11 @@ use crate::admin::{ auth::validate_admin_request, + ecstore_cluster::{ + ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot, + ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState, + ClusterPoolStateSnapshot, + }, router::{AdminOperation, Operation, S3Router}, system, }; @@ -28,11 +33,6 @@ use hyper::Method; use matchit::Params; use rustfs_concurrency::AdmissionState as WorkloadAdmissionState; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadClass}; -use rustfs_ecstore::api::cluster::{ - ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot, - ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState, - ClusterPoolStateSnapshot, -}; use rustfs_policy::policy::action::{Action, AdminAction}; use rustfs_storage_api::{CapabilityState, CapabilityStatus, ObservabilitySnapshot, TopologySnapshot}; use s3s::header::CONTENT_TYPE; @@ -590,14 +590,14 @@ fn summarize_named_capability_statuses( #[cfg(test)] mod tests { use super::{ClusterSnapshotResponse, ClusterSnapshotSummary, ClusterSnapshotView}; - use crate::cluster_snapshot::{ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot}; - use crate::server::{DependencyReadiness, ReadinessDegradedReason}; - use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadClass}; - use rustfs_ecstore::api::cluster::{ + use crate::admin::ecstore_cluster::{ ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot, }; + use crate::cluster_snapshot::{ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot}; + use crate::server::{DependencyReadiness, ReadinessDegradedReason}; + use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadClass}; use rustfs_storage_api::CapabilityState; use rustfs_storage_api::{CapabilityStatus, ObservabilitySnapshot, TopologySnapshot}; diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index bb95f4501..291457a7e 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -119,6 +119,14 @@ mod ecstore_client { pub(crate) use crate::storage::ecstore_client::admin_handler_utils; } +mod ecstore_cluster { + pub(crate) use crate::storage::ecstore_cluster::{ + ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot, + ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState, + ClusterPoolStateSnapshot, + }; +} + mod ecstore_config { pub(crate) use crate::storage::ecstore_config::{com, init, storageclass}; } diff --git a/rustfs/src/cluster_snapshot.rs b/rustfs/src/cluster_snapshot.rs index e2abd1a90..9eb9dfa6f 100644 --- a/rustfs/src/cluster_snapshot.rs +++ b/rustfs/src/cluster_snapshot.rs @@ -17,12 +17,12 @@ use crate::server::{ DependencyReadiness, DependencyReadinessReport, ReadinessDegradedReason, snapshot_dependency_readiness_report, }; use crate::storage::EndpointServerPools; -use crate::workload_admission::workload_admission_registry_snapshot; -use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot}; -use rustfs_ecstore::api::cluster::{ +use crate::storage::ecstore_cluster::{ ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot, ClusterPeerHealthSnapshot, ClusterPoolStateSnapshot, }; +use crate::workload_admission::workload_admission_registry_snapshot; +use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot}; use rustfs_storage_api::{ObservabilitySnapshot, TopologySnapshot}; #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index 2ea723e01..81adfff77 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -84,7 +84,12 @@ pub(crate) mod ecstore_compression { } pub(crate) mod ecstore_cluster { - pub(crate) use rustfs_ecstore::api::cluster::topology_snapshot_from_endpoint_pools_with_capabilities; + pub(crate) use rustfs_ecstore::api::cluster::{ + ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, + ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, + ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot, + topology_snapshot_from_endpoint_pools_with_capabilities, + }; } pub(crate) mod ecstore_config {