gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-04 03:05:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: centralize ecstore runtime owner sources (#3798)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-24 06:34:06 +08:00
committed by GitHub
parent 1735dcde9c
commit 7e60432588
18 changed files with 499 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/store/bucket.rs
+2 -4
View File
@@ -17,7 +17,7 @@ use crate::bucket::{
metadata::{BUCKET_TABLE_RESERVED_PREFIX, table_bucket_catalog_metadata_prefix},
utils::is_meta_bucketname,
};
use crate::global::get_global_bucket_monitor;
use crate::runtime_sources;
use crate::set_disk::get_lock_acquire_timeout;
use rustfs_storage_api::NamespaceLocking as _;
@@ -258,9 +258,7 @@ impl ECStore {
for prefix in bucket_delete_metadata_cleanup_prefixes(bucket) {
self.delete_all(RUSTFS_META_BUCKET, prefix.as_str()).await?;
}
if let Some(monitor) = get_global_bucket_monitor() {
monitor.delete_bucket(bucket);
}
runtime_sources::delete_bucket_monitor_entry(bucket);
Ok(())
}
}
crates/ecstore/src/store/init.rs
+13 -22
View File
@@ -14,11 +14,9 @@
use super::*;
use crate::error::is_err_decommission_running;
use crate::global::{
GLOBAL_EventNotifier, GLOBAL_LOCAL_DISK_ID_MAP, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, GLOBAL_TierConfigMgr,
get_global_bucket_monitor, is_dist_erasure, is_first_cluster_node_local,
};
use crate::global::{is_dist_erasure, is_first_cluster_node_local};
use crate::pools::local_decommission_queue_prefix;
use crate::runtime_sources;
use tracing::{debug, error, info, warn};
const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -302,11 +300,7 @@ impl ECStore {
// Replace the local disk
if !is_dist_erasure().await {
let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await;
for disk in local_disks {
let path = disk.endpoint().to_string();
global_local_disk_map.insert(path, Some(disk.clone()));
}
runtime_sources::record_local_disks(local_disks).await;
}
let peer_sys = S3PeerSys::new(&endpoint_pools);
@@ -325,19 +319,17 @@ impl ECStore {
start_gate: tokio::sync::Mutex::new(()),
pool_meta_save_gate: tokio::sync::Mutex::new(()),
local_disk_map: GLOBAL_LOCAL_DISK_MAP.clone(),
local_disk_id_map: GLOBAL_LOCAL_DISK_ID_MAP.clone(),
local_disk_set_drives: GLOBAL_LOCAL_DISK_SET_DRIVES.clone(),
tier_config_mgr: GLOBAL_TierConfigMgr.clone(),
event_notifier: GLOBAL_EventNotifier.clone(),
local_disk_map: runtime_sources::local_disk_map_handle(),
local_disk_id_map: runtime_sources::local_disk_id_map_handle(),
local_disk_set_drives: runtime_sources::local_disk_set_drives_handle(),
tier_config_mgr: runtime_sources::tier_config_mgr_handle(),
event_notifier: runtime_sources::event_notifier_handle(),
bucket_monitor: OnceLock::new(),
});
// Only set it when the global deployment ID is not yet configured
if let Some(dep_id) = deployment_id
&& get_global_deployment_id().is_none()
{
set_global_deployment_id(dep_id);
if let Some(dep_id) = deployment_id {
runtime_sources::ensure_deployment_id(dep_id);
}
let wait_sec = 5;
@@ -362,7 +354,7 @@ impl ECStore {
set_object_layer(ec.clone()).await;
if let Some(monitor) = get_global_bucket_monitor() {
if let Some(monitor) = runtime_sources::bucket_monitor() {
let _ = ec.bucket_monitor.set(monitor);
}
@@ -449,8 +441,7 @@ impl ECStore {
});
}
let num_nodes = get_global_endpoints().get_nodes().len() as u64;
init_global_bucket_monitor(num_nodes);
runtime_sources::init_bucket_monitor_for_current_endpoints();
init_background_expiry(self.clone()).await;
crate::bucket::lifecycle::bucket_lifecycle_ops::init_background_stale_multipart_upload_cleanup(self.clone());
@@ -458,7 +449,7 @@ impl ECStore {
TransitionState::init(self.clone()).await;
crate::tier::tier::try_migrate_tiering_config(self.clone()).await;
if let Err(err) = GLOBAL_TierConfigMgr.write().await.init(self.clone()).await {
if let Err(err) = runtime_sources::init_tier_config_mgr(self.clone()).await {
info!("TierConfigMgr init error: {}", err);
}
crates/ecstore/src/store/peer.rs
+10 -67
View File
@@ -13,7 +13,7 @@
// limitations under the License.
use super::*;
use crate::global::GLOBAL_LOCAL_DISK_ID_MAP;
use crate::runtime_sources;
use tracing::{debug, error};
const LOG_COMPONENT_ECSTORE: &str = "ecstore";
@@ -23,25 +23,16 @@ const EVENT_LOCK_CLIENT_INITIALIZATION_FAILED: &str = "lock_client_initializatio
async fn remember_local_disk_id(disk: &DiskStore) -> Option<Uuid> {
let disk_id = disk.get_disk_id().await.ok().flatten()?;
GLOBAL_LOCAL_DISK_ID_MAP
.write()
.await
.insert(disk_id, disk.endpoint().to_string());
runtime_sources::record_local_disk_id(disk_id, disk.endpoint().to_string()).await;
Some(disk_id)
}
pub async fn find_local_disk(disk_path: &String) -> Option<DiskStore> {
let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await;
if let Some(disk) = disk_map.get(disk_path) {
disk.as_ref().cloned()
} else {
None
}
pub async fn find_local_disk(disk_path: &str) -> Option<DiskStore> {
runtime_sources::local_disk_by_path(disk_path).await
}
pub async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore> {
if let Some(disk) = find_local_disk(&disk_ref.to_string()).await {
if let Some(disk) = find_local_disk(disk_ref).await {
let _ = remember_local_disk_id(&disk).await;
return Some(disk);
}
@@ -50,7 +41,7 @@ pub async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore> {
return None;
};
if let Some(disk_path) = GLOBAL_LOCAL_DISK_ID_MAP.read().await.get(&disk_id).cloned()
if let Some(disk_path) = runtime_sources::local_disk_path_by_id(&disk_id).await
&& let Some(disk) = find_local_disk(&disk_path).await
{
return Some(disk);
@@ -66,35 +57,15 @@ pub async fn find_local_disk_by_ref(disk_ref: &str) -> Option<DiskStore> {
}
pub async fn get_disk_via_endpoint(endpoint: &Endpoint) -> Option<DiskStore> {
let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await;
if global_set_drives.is_empty() {
return GLOBAL_LOCAL_DISK_MAP
.read()
.await
.get(&endpoint.to_string())
.cloned()
.unwrap_or(None);
}
global_set_drives
.get(endpoint.pool_idx as usize)
.and_then(|sets| sets.get(endpoint.set_idx as usize))
.and_then(|disks| disks.get(endpoint.disk_idx as usize))
.cloned()
.unwrap_or(None)
runtime_sources::local_disk_for_endpoint(endpoint).await
}
pub async fn all_local_disk_path() -> Vec<String> {
let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await;
disk_map.keys().cloned().collect()
runtime_sources::local_disk_paths().await
}
pub async fn all_local_disk() -> Vec<DiskStore> {
let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await;
disk_map
.values()
.filter(|v| v.is_some())
.map(|v| v.as_ref().unwrap().clone())
.collect()
runtime_sources::local_disks().await
}
pub async fn prewarm_local_disk_id_map() {
@@ -121,35 +92,7 @@ pub async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<()>
health_check: true,
};
let mut global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.write().await;
for pool_eps in endpoint_pools.as_ref().iter() {
let mut set_count_drives = Vec::with_capacity(pool_eps.set_count);
for _ in 0..pool_eps.set_count {
set_count_drives.push(vec![None; pool_eps.drives_per_set]);
}
global_set_drives.push(set_count_drives);
}
let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await;
for pool_eps in endpoint_pools.as_ref().iter() {
for ep in pool_eps.endpoints.as_ref().iter() {
if !ep.is_local {
continue;
}
let disk = new_disk(ep, opt).await?;
let path = disk.endpoint().to_string();
global_local_disk_map.insert(path, Some(disk.clone()));
global_set_drives[ep.pool_idx as usize][ep.set_idx as usize][ep.disk_idx as usize] = Some(disk.clone());
}
}
Ok(())
runtime_sources::initialize_local_disk_maps(endpoint_pools, opt).await
}
pub fn init_lock_clients(endpoint_pools: EndpointServerPools) {
crates/ecstore/src/store/rebalance.rs
+4 -15
View File
@@ -13,8 +13,8 @@
// limitations under the License.
use super::*;
use crate::config::get_global_storage_class;
use crate::layout::pool_space::{ServerPoolsAvailableSpace, build_server_pools_available_space};
use crate::runtime_sources;
use rustfs_storage_api::{NamespaceLocking as _, ObjectOperations as _, StorageAdminApi};
pub(in crate::store) mod support;
use support::{
@@ -509,19 +509,8 @@ impl ECStore {
#[instrument(skip(self))]
pub(super) async fn handle_backend_info(&self) -> rustfs_madmin::BackendInfo {
let (standard_sc_parity, rr_sc_parity) = {
if let Some(sc) = get_global_storage_class() {
let sc_parity = sc
.get_parity_for_sc(storageclass::CLASS_STANDARD)
.or(Some(self.pools[0].default_parity_count));
let rrs_sc_parity = sc.get_parity_for_sc(storageclass::RRS);
(sc_parity, rrs_sc_parity)
} else {
(Some(self.pools[0].default_parity_count), None)
}
};
let (standard_sc_parity, rr_sc_parity) =
runtime_sources::backend_storage_class_parities(self.pools[0].default_parity_count);
let mut standard_sc_data = Vec::new();
let mut rr_sc_data = Vec::new();
@@ -555,7 +544,7 @@ impl ECStore {
#[instrument(skip(self))]
pub(super) async fn handle_storage_info(&self) -> rustfs_madmin::StorageInfo {
let Some(notification_sy) = get_global_notification_sys() else {
let Some(notification_sy) = runtime_sources::notification_sys() else {
return rustfs_madmin::StorageInfo::default();
};