refactor: route ecstore internals through object resolver (#3441)

This commit is contained in:
安正超
2026-06-14 19:55:21 +08:00
committed by GitHub
parent bf3a3a6189
commit 323302255c
11 changed files with 66 additions and 50 deletions
+4 -4
View File
@@ -18,8 +18,8 @@ use crate::rpc::{TonicInterceptor, gen_tonic_signature_interceptor, node_service
use crate::{ use crate::{
disk::endpoint::Endpoint, disk::endpoint::Endpoint,
global::{GLOBAL_BOOT_TIME, GLOBAL_Endpoints, get_global_deployment_id}, global::{GLOBAL_BOOT_TIME, GLOBAL_Endpoints, get_global_deployment_id},
new_object_layer_fn,
notification_sys::get_global_notification_sys, notification_sys::get_global_notification_sys,
resolve_object_store_handle,
}; };
use crate::data_usage::load_data_usage_cache; use crate::data_usage::load_data_usage_cache;
@@ -185,7 +185,7 @@ pub async fn get_local_server_property() -> ServerProperties {
// let mut sensitive = HashSet::new(); // let mut sensitive = HashSet::new();
// sensitive.insert(rustfs_config::ENV_RUSTFS_ACCESS_KEY.to_string()); // sensitive.insert(rustfs_config::ENV_RUSTFS_ACCESS_KEY.to_string());
// sensitive.insert(rustfs_config::ENV_RUSTFS_SECRET_KEY.to_string()); // sensitive.insert(rustfs_config::ENV_RUSTFS_SECRET_KEY.to_string());
if let Some(store) = new_object_layer_fn() { if let Some(store) = resolve_object_store_handle() {
let storage_info = StorageAdminApi::local_storage_info(store.as_ref()).await; let storage_info = StorageAdminApi::local_storage_info(store.as_ref()).await;
props.state = ITEM_ONLINE.to_string(); props.state = ITEM_ONLINE.to_string();
props.disks = storage_info.disks; props.disks = storage_info.disks;
@@ -229,7 +229,7 @@ pub async fn get_server_info(get_pools: bool) -> InfoMessage {
let mut backend = rustfs_madmin::ErasureBackend::default(); let mut backend = rustfs_madmin::ErasureBackend::default();
let mut pools: HashMap<i32, HashMap<i32, ErasureSetInfo>> = HashMap::new(); let mut pools: HashMap<i32, HashMap<i32, ErasureSetInfo>> = HashMap::new();
if let Some(store) = new_object_layer_fn() { if let Some(store) = resolve_object_store_handle() {
mode = ITEM_ONLINE; mode = ITEM_ONLINE;
match load_data_usage_from_backend(store.clone()).await { match load_data_usage_from_backend(store.clone()).await {
Ok(res) => { Ok(res) => {
@@ -347,7 +347,7 @@ fn get_online_offline_disks_stats(disks_info: &[Disk]) -> (BackendDisks, Backend
} }
async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32, ErasureSetInfo>>> { async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32, ErasureSetInfo>>> {
let Some(store) = new_object_layer_fn() else { let Some(store) = resolve_object_store_handle() else {
return Err(Error::other("ServerNotInitialized")); return Err(Error::other("ServerNotInitialized"));
}; };
@@ -735,7 +735,7 @@ impl TransitionState {
tokio::spawn(async move { tokio::spawn(async move {
Self::inc_counter(&state.compensation_running_tasks); Self::inc_counter(&state.compensation_running_tasks);
state.record_scanner_transition_state(); state.record_scanner_transition_state();
let Some(api) = crate::new_object_layer_fn() else { let Some(api) = crate::resolve_object_store_handle() else {
scheduled.lock().unwrap().remove(&bucket); scheduled.lock().unwrap().remove(&bucket);
Self::add_counter(&state.compensation_running_tasks, -1); Self::add_counter(&state.compensation_running_tasks, -1);
state.record_scanner_transition_state(); state.record_scanner_transition_state();
@@ -1646,7 +1646,7 @@ pub async fn enqueue_immediate_expiry(oi: &ObjectInfo, src: LcEventSrc) {
let Some(lifecycle) = GLOBAL_LifecycleSys.get(&oi.bucket).await else { let Some(lifecycle) = GLOBAL_LifecycleSys.get(&oi.bucket).await else {
return; return;
}; };
let Some(api) = crate::new_object_layer_fn() else { let Some(api) = crate::resolve_object_store_handle() else {
return; return;
}; };
+2 -2
View File
@@ -20,7 +20,7 @@ use crate::bucket::utils::deserialize;
use crate::config::com::{read_config, save_config}; use crate::config::com::{read_config, save_config};
use crate::disk::BUCKET_META_PREFIX; use crate::disk::BUCKET_META_PREFIX;
use crate::error::{Error, Result}; use crate::error::{Error, Result};
use crate::new_object_layer_fn; use crate::resolve_object_store_handle;
use crate::store::ECStore; use crate::store::ECStore;
use byteorder::{BigEndian, ByteOrder, LittleEndian}; use byteorder::{BigEndian, ByteOrder, LittleEndian};
use rustfs_policy::policy::BucketPolicy; use rustfs_policy::policy::BucketPolicy;
@@ -765,7 +765,7 @@ impl BucketMetadata {
} }
pub async fn save(&mut self) -> Result<()> { pub async fn save(&mut self) -> Result<()> {
let Some(store) = new_object_layer_fn() else { let Some(store) = resolve_object_store_handle() else {
return Err(Error::other("errServerNotInitialized")); return Err(Error::other("errServerNotInitialized"));
}; };
+2 -2
View File
@@ -19,7 +19,7 @@ use crate::bucket::bucket_target_sys::BucketTargetSys;
use crate::bucket::metadata::{BUCKET_LIFECYCLE_CONFIG, load_bucket_metadata_parse}; use crate::bucket::metadata::{BUCKET_LIFECYCLE_CONFIG, load_bucket_metadata_parse};
use crate::bucket::utils::{deserialize, is_meta_bucketname}; use crate::bucket::utils::{deserialize, is_meta_bucketname};
use crate::error::{Error, Result, is_err_bucket_not_found}; use crate::error::{Error, Result, is_err_bucket_not_found};
use crate::global::{GLOBAL_Endpoints, is_dist_erasure, is_erasure, new_object_layer_fn}; use crate::global::{GLOBAL_Endpoints, is_dist_erasure, is_erasure, resolve_object_store_handle};
use crate::store::ECStore; use crate::store::ECStore;
use crate::store_api::HealOperations as _; use crate::store_api::HealOperations as _;
use futures::future::join_all; use futures::future::join_all;
@@ -406,7 +406,7 @@ impl BucketMetadataSys {
} }
async fn update_and_parse(&mut self, bucket: &str, config_file: &str, data: Vec<u8>, parse: bool) -> Result<OffsetDateTime> { async fn update_and_parse(&mut self, bucket: &str, config_file: &str, data: Vec<u8>, parse: bool) -> Result<OffsetDateTime> {
let Some(store) = new_object_layer_fn() else { let Some(store) = resolve_object_store_handle() else {
return Err(Error::other("errServerNotInitialized")); return Err(Error::other("errServerNotInitialized"));
}; };
@@ -35,7 +35,7 @@ use crate::set_disk::get_lock_acquire_timeout;
use crate::store_api::{ use crate::store_api::{
DeletedObject, HTTPRangeSpec, NamespaceLocking, ObjectIO, ObjectInfo, ObjectOptions, ObjectToDelete, WalkOptions, DeletedObject, HTTPRangeSpec, NamespaceLocking, ObjectIO, ObjectInfo, ObjectOptions, ObjectToDelete, WalkOptions,
}; };
use crate::{StorageAPI, new_object_layer_fn}; use crate::{StorageAPI, resolve_object_store_handle};
use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError}; use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput}; use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput};
use aws_sdk_s3::primitives::ByteStream; use aws_sdk_s3::primitives::ByteStream;
@@ -1580,7 +1580,7 @@ impl ObjectInfoExt for ObjectInfo {
} }
pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision { pub async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision {
if new_object_layer_fn().is_none() { if resolve_object_store_handle().is_none() {
return ReplicateDecision::default(); return ReplicateDecision::default();
} }
+2 -2
View File
@@ -793,10 +793,10 @@ pub async fn load_data_usage_cache(store: &crate::set_disk::SetDisks, name: &str
pub async fn save_data_usage_cache(cache: &DataUsageCache, name: &str) -> crate::error::Result<()> { pub async fn save_data_usage_cache(cache: &DataUsageCache, name: &str) -> crate::error::Result<()> {
use crate::config::com::save_config; use crate::config::com::save_config;
use crate::disk::BUCKET_META_PREFIX; use crate::disk::BUCKET_META_PREFIX;
use crate::new_object_layer_fn; use crate::resolve_object_store_handle;
use std::path::Path; use std::path::Path;
let Some(store) = new_object_layer_fn() else { let Some(store) = resolve_object_store_handle() else {
return Err(Error::other("errServerNotInitialized")); return Err(Error::other("errServerNotInitialized"));
}; };
let buf = cache.marshal_msg().map_err(Error::other)?; let buf = cache.marshal_msg().map_err(Error::other)?;
+2 -2
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and // See the License for the specific language governing permissions and
// limitations under the License. // limitations under the License.
use crate::{admin_server_info::get_local_server_property, new_object_layer_fn}; use crate::{admin_server_info::get_local_server_property, resolve_object_store_handle};
use chrono::Utc; use chrono::Utc;
use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::DriveState, metrics::global_metrics}; use rustfs_common::{GLOBAL_LOCAL_NODE_NAME, GLOBAL_RUSTFS_ADDR, heal_channel::DriveState, metrics::global_metrics};
use rustfs_io_metrics::internode_metrics::global_internode_metrics; use rustfs_io_metrics::internode_metrics::global_internode_metrics;
@@ -264,7 +264,7 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts)
} }
async fn collect_local_disks_metrics(disks: &HashSet<String>) -> HashMap<String, DiskMetric> { async fn collect_local_disks_metrics(disks: &HashSet<String>) -> HashMap<String, DiskMetric> {
let store = match new_object_layer_fn() { let store = match resolve_object_store_handle() {
Some(store) => store, Some(store) => store,
None => return HashMap::new(), None => return HashMap::new(),
}; };
+2 -2
View File
@@ -18,7 +18,7 @@ use crate::global::{GLOBAL_BOOT_TIME, get_global_endpoints};
use crate::metrics_realtime::{CollectMetricsOpts, MetricType}; use crate::metrics_realtime::{CollectMetricsOpts, MetricType};
use crate::rebalance::RebalSaveOpt; use crate::rebalance::RebalSaveOpt;
use crate::rpc::PeerRestClient; use crate::rpc::PeerRestClient;
use crate::{endpoints::EndpointServerPools, new_object_layer_fn}; use crate::{endpoints::EndpointServerPools, resolve_object_store_handle};
use futures::future::join_all; use futures::future::join_all;
use lazy_static::lazy_static; use lazy_static::lazy_static;
use rustfs_madmin::health::{Cpus, MemInfo, OsInfo, Partitions, ProcInfo, SysConfig, SysErrors, SysServices}; use rustfs_madmin::health::{Cpus, MemInfo, OsInfo, Partitions, ProcInfo, SysConfig, SysErrors, SysServices};
@@ -520,7 +520,7 @@ impl NotificationSys {
pub async fn stop_rebalance(&self) -> Result<()> { pub async fn stop_rebalance(&self) -> Result<()> {
warn!("notification stop_rebalance start"); warn!("notification stop_rebalance start");
let Some(store) = new_object_layer_fn() else { let Some(store) = resolve_object_store_handle() else {
error!("stop_rebalance: not init"); error!("stop_rebalance: not init");
return Err(Error::other("stop_rebalance: object layer not initialized")); return Err(Error::other("stop_rebalance: object layer not initialized"));
}; };
+2 -2
View File
@@ -35,8 +35,8 @@ use crate::error::{
StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_data_movement_overwrite, is_err_object_not_found, StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_data_movement_overwrite, is_err_object_not_found,
is_err_operation_canceled, is_err_version_not_found, is_err_operation_canceled, is_err_version_not_found,
}; };
use crate::new_object_layer_fn;
use crate::notification_sys::get_global_notification_sys; use crate::notification_sys::get_global_notification_sys;
use crate::resolve_object_store_handle;
use crate::set_disk::SetDisks; use crate::set_disk::SetDisks;
use crate::store_api::{ use crate::store_api::{
BucketOperations, BucketOptions, GetObjectReader, HealOperations, MakeBucketOptions, ObjectIO, ObjectOperations, BucketOperations, BucketOptions, GetObjectReader, HealOperations, MakeBucketOptions, ObjectIO, ObjectOperations,
@@ -1384,7 +1384,7 @@ impl ECStore {
ensure_decommission_not_rebalancing(self.is_rebalance_conflicting_with_decommission().await)?; ensure_decommission_not_rebalancing(self.is_rebalance_conflicting_with_decommission().await)?;
let store = require_decommission_store(new_object_layer_fn(), "start decommission")?; let store = require_decommission_store(resolve_object_store_handle(), "start decommission")?;
self.start_decommission(indices.clone()).await?; self.start_decommission(indices.clone()).await?;
if let Err(err) = self.spawn_decommission_routines(store, rx, indices.clone()).await { if let Err(err) = self.spawn_decommission_routines(store, rx, indices.clone()).await {
+2 -2
View File
@@ -38,7 +38,7 @@ use tracing::{debug, error, info, warn};
use crate::client::admin_handler_utils::AdminError; use crate::client::admin_handler_utils::AdminError;
use crate::error::{Error, Result, StorageError}; use crate::error::{Error, Result, StorageError};
use crate::new_object_layer_fn; use crate::resolve_object_store_handle;
use crate::tier::{ use crate::tier::{
tier_admin::TierCreds, tier_admin::TierCreds,
tier_config::{TierConfig, TierType}, tier_config::{TierConfig, TierType},
@@ -1024,7 +1024,7 @@ impl TierConfigMgr {
#[tracing::instrument(level = "debug", name = "tier_save", skip(self))] #[tracing::instrument(level = "debug", name = "tier_save", skip(self))]
pub async fn save(&self) -> std::result::Result<(), std::io::Error> { pub async fn save(&self) -> std::result::Result<(), std::io::Error> {
let Some(api) = new_object_layer_fn() else { let Some(api) = resolve_object_store_handle() else {
return Err(tier_config_not_initialized_error("save tiering config")); return Err(tier_config_not_initialized_error("save tiering config"));
}; };
//let (pr, opts) = GLOBAL_TierConfigMgr.write().config_reader()?; //let (pr, opts) = GLOBAL_TierConfigMgr.write().config_reader()?;
+44 -28
View File
@@ -5,18 +5,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
## Current Context ## Current Context
- Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
- Branch: `overtrue/arch-server-storage-object-store-resolver` - Branch: `overtrue/arch-ecstore-internal-object-store-resolver`
- Baseline: `upstream/main` at `9372ee70329d08ea7aafae8df84f6b1ecb5bd686` - Baseline: `upstream/main` at `8f6b1d47b5af689e0b3576ab4448cc4274786df1`
- PR type for this branch: `consumer-migration` - PR type for this branch: `consumer-migration`
- Runtime behavior changes: no external behavior change expected; server and - Runtime behavior changes: no external behavior change expected; ECStore
storage infra object-store lookups prefer the AppContext-owned object store internal/background object-store lookups prefer the AppContext-owned object
through the ECStore-owned resolver and keep the existing global fallback. store through the ECStore-owned resolver and keep the existing global
- Rust code changes: migrate server readiness/module-switch and storage access, fallback.
ecfs extension, and node RPC object-store lookups to the ECStore-owned - Rust code changes: migrate ECStore internal metrics, notification, tier,
resolver without touching app compatibility fallbacks or ECStore internal decommission, admin info, bucket metadata, replication, lifecycle, and data
background consumers. usage object-store lookups to the ECStore-owned resolver without touching app
compatibility fallbacks or the global fallback definition.
- CI/script changes: none. - CI/script changes: none.
- Docs changes: record `CTX-009` consumer migration scope and verification. - Docs changes: record `CTX-010` consumer migration scope and verification.
## Phase 0 Tasks ## Phase 0 Tasks
@@ -217,6 +218,20 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
error paths. error paths.
- Verification: formatting, compile checks, migration guards, diff hygiene, - Verification: formatting, compile checks, migration guards, diff hygiene,
Rust risk scan, and full `make pre-commit`. Rust risk scan, and full `make pre-commit`.
- [x] `CTX-010` Migrate ECStore internal object-store consumers.
- Do: migrate ECStore internal/background object-store lookups to the
ECStore-owned resolver.
- Acceptance: ECStore metrics realtime, notification, tier config save,
decommission, admin server info, bucket metadata, replication decision,
lifecycle compensation/expiry, and data-usage cache consumers prefer the
AppContext-owned object store after context initialization and preserve the
existing global object-layer fallback.
- Must preserve: metrics collection, notification rebalance stop behavior,
tier config persistence, decommission startup, admin server info reporting,
bucket metadata persistence, replication decisions, lifecycle queueing, data
usage cache persistence, and existing storage error paths.
- Verification: formatting, compile checks, migration guards, diff hygiene,
Rust risk scan, and full `make pre-commit`.
## Phase 1 Security Governance Tasks ## Phase 1 Security Governance Tasks
@@ -523,9 +538,8 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
## Next PRs ## Next PRs
1. `consumer-migration`: review app compatibility fallback call sites and 1. `consumer-migration`: review app compatibility fallback call sites before
ECStore internal background consumers before the final global-accessor the final global-accessor cleanup phase.
cleanup phase.
2. `pure-move`: start `R-009` boot wrapper with the IAM degraded readiness 2. `pure-move`: start `R-009` boot wrapper with the IAM degraded readiness
contract covered. contract covered.
@@ -533,40 +547,42 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
| Expert | Status | Notes | | Expert | Status | Notes |
|---|---|---| |---|---|---|
| Quality/architecture | passed | Single consumer-migration slice; server/storage infra consumers use the ECStore-owned resolver without widening application-layer dependencies or changing ECStore internals. | | Quality/architecture | passed | Single consumer-migration slice; ECStore internal/background consumers use the ECStore-owned resolver without adding application-layer dependencies or changing the fallback definition. |
| Migration preservation | passed | Readiness, module-switch, storage access, ecfs extension, and node RPC call sites keep the same error paths while preferring the AppContext-owned store when installed. | | Migration preservation | passed | Metrics, notification, tier, decommission, admin info, bucket metadata, replication, lifecycle, and data-usage call sites keep the same missing-store/error behavior while preferring the AppContext-owned store when installed. |
| Testing/verification | passed | Formatting, compile check, diff hygiene, migration/layer guards, Rust risk scan, branch freshness check, and full `make pre-commit` passed. | | Testing/verification | passed | Formatting, compile checks, diff hygiene, migration/layer guards, Rust risk scan, branch freshness check, and full `make pre-commit` passed. |
## Verification Notes ## Verification Notes
Passed on `9372ee70329d08ea7aafae8df84f6b1ecb5bd686`: Passed on `8f6b1d47b5af689e0b3576ab4448cc4274786df1`:
- `cargo fmt --all --check`. - `cargo fmt --all --check`.
- `cargo check -p rustfs-ecstore`.
- `cargo check -p rustfs --lib`. - `cargo check -p rustfs --lib`.
- `git diff --check`. - `git diff --check`.
- `./scripts/check_architecture_migration_rules.sh`. - `./scripts/check_architecture_migration_rules.sh`.
- `./scripts/check_layer_dependencies.sh`. - `./scripts/check_layer_dependencies.sh`.
- `git rev-list --left-right --count HEAD...origin/main` returned `0 0`. - `git rev-list --left-right --count HEAD...origin/main` returned `1 0`
after rebase.
- Rust risk scan for changed Rust files; full-file matches were existing tests, - Rust risk scan for changed Rust files; full-file matches were existing tests,
existing relaxed counters, existing numeric casts, and existing string error existing relaxed counters, existing numeric casts, existing print/debug
signatures, and the added-line scan returned no `unwrap`/`expect`, numeric helpers, and existing unwrap/expect paths, and the added-line scan returned
cast, string error, boxed error, print macro, or relaxed-ordering match. no `unwrap`/`expect`, numeric cast, string error, boxed error, print macro, or
relaxed-ordering match.
- `make pre-commit`; all checks passed, including nextest 5954 passed and 111 - `make pre-commit`; all checks passed, including nextest 5954 passed and 111
skipped, plus doctests. skipped, plus doctests.
Notes: Notes:
- This slice migrates server readiness/module-switch and storage access, ecfs - This slice migrates ECStore internal/background object-store lookups to the
extension, and node RPC object-store lookups to the ECStore-owned resolver. ECStore-owned resolver.
- App compatibility fallback call sites remain on `new_object_layer_fn` as an - App compatibility fallback call sites remain on `new_object_layer_fn` as an
explicit fallback path. explicit fallback path.
- ECStore internal background consumers remain on the old global accessor for a - The old global accessor remains as the resolver fallback and public
separate review. compatibility re-export for a later cleanup slice.
## Handoff Notes ## Handoff Notes
- CTX-009 is complete. - CTX-010 is complete.
- App compatibility fallback call sites remain on `new_object_layer_fn` as an - App compatibility fallback call sites remain on `new_object_layer_fn` as an
explicit fallback path. explicit fallback path.
- ECStore internal background consumers should be reviewed separately before - The global fallback definition and re-export remain for a later cleanup slice.
changing their global accessor behavior.