gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-20 03:22:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: centralize scanner runtime source helpers (#3827)

Browse Source 
* refactor: batch lifecycle runtime source handles

* refactor: centralize scanner runtime source helpers
This commit is contained in:
Zhengchao An
2026-06-24 22:05:03 +08:00
committed by GitHub
parent 5cc8bd7952
commit a935854b32
19 changed files with 191 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+8
View File
@@ -123,6 +123,14 @@ lazy_static! {
pub static ref GLOBAL_TransitionState: Arc<TransitionState> = TransitionState::new();
}
pub fn get_global_expiry_state() -> Arc<RwLock<ExpiryState>> {
GLOBAL_ExpiryState.clone()
}
pub fn get_global_transition_state() -> Arc<TransitionState> {
GLOBAL_TransitionState.clone()
}
fn resolve_transition_worker_count() -> (i64, i64, i64) {
let fallback = std::cmp::min(num_cpus::get() as i64, DEFAULT_TRANSITION_WORKERS_CAP);
let configured = env::var(ENV_TRANSITION_WORKERS)
crates/scanner/src/lib.rs
+16 -11
View File
@@ -21,11 +21,12 @@
)]
use http::HeaderMap;
use rustfs_config::server_config::{Config as ServerConfig, get_global_server_config as config_get_global_server_config};
use rustfs_ecstore::api::bucket::bucket_target_sys::BucketTargetSys as EcstoreBucketTargetSys;
use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc as EcstoreLcEventSrc;
use rustfs_ecstore::api::bucket::lifecycle::bucket_lifecycle_ops::{
GLOBAL_ExpiryState as ECSTORE_GLOBAL_EXPIRY_STATE, apply_expiry_rule as ecstore_apply_expiry_rule,
apply_transition_rule as ecstore_apply_transition_rule,
apply_expiry_rule as ecstore_apply_expiry_rule, apply_transition_rule as ecstore_apply_transition_rule,
get_global_expiry_state as ecstore_get_global_expiry_state,
};
use rustfs_ecstore::api::bucket::lifecycle::evaluator::Evaluator as EcstoreEvaluator;
use rustfs_ecstore::api::bucket::lifecycle::lifecycle::{
@@ -67,7 +68,7 @@ use rustfs_ecstore::api::disk::{
use rustfs_ecstore::api::disk::{DiskOption as EcstoreDiskOption, DiskStore as EcstoreDiskStore, new_disk as ecstore_new_disk};
use rustfs_ecstore::api::error::{Error as EcstoreErrorType, Result as EcstoreResultType, StorageError as EcstoreStorageError};
use rustfs_ecstore::api::global::{
GLOBAL_TierConfigMgr as ECSTORE_GLOBAL_TIER_CONFIG_MGR, is_erasure as ecstore_is_erasure,
get_global_tier_config_mgr as ecstore_get_global_tier_config_mgr, is_erasure as ecstore_is_erasure,
is_erasure_sd as ecstore_is_erasure_sd, resolve_object_store_handle as ecstore_resolve_object_store_handle,
};
use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks;
@@ -271,21 +272,25 @@ pub(crate) async fn apply_expiry_rule(event: &Event, src: &LcEventSrc, oi: &Scan
ecstore_apply_expiry_rule(event, src, oi).await
}
pub(crate) async fn list_global_tiers() -> Vec<EcstoreTierConfig> {
ECSTORE_GLOBAL_TIER_CONFIG_MGR.read().await.list_tiers()
pub(crate) fn resolve_scanner_server_config() -> Option<ServerConfig> {
config_get_global_server_config()
}
pub(crate) async fn enqueue_global_free_version(oi: ScannerObjectInfo) {
ECSTORE_GLOBAL_EXPIRY_STATE.write().await.enqueue_free_version(oi).await;
pub(crate) async fn list_runtime_tiers() -> Vec<EcstoreTierConfig> {
ecstore_get_global_tier_config_mgr().read().await.list_tiers()
}
pub(crate) async fn enqueue_global_newer_noncurrent(
pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo) {
ecstore_get_global_expiry_state().write().await.enqueue_free_version(oi).await;
}
pub(crate) async fn enqueue_runtime_newer_noncurrent(
bucket: &str,
to_delete_objs: Vec<ObjectToDelete>,
event: Event,
src: &LcEventSrc,
) -> bool {
ECSTORE_GLOBAL_EXPIRY_STATE
ecstore_get_global_expiry_state()
.write()
.await
.enqueue_by_newer_noncurrent(bucket, to_delete_objs, event, src)
@@ -317,11 +322,11 @@ pub(crate) fn path2_bucket_object_with_base_path(base_path: &str, path: &str) ->
ecstore_path2_bucket_object_with_base_path(base_path, path)
}
pub(crate) async fn is_erasure() -> bool {
pub(crate) async fn scanner_is_erasure() -> bool {
ecstore_is_erasure().await
}
pub(crate) async fn is_erasure_sd() -> bool {
pub(crate) async fn scanner_is_erasure_sd() -> bool {
ecstore_is_erasure_sd().await
}
crates/scanner/src/runtime_config.rs
+2 -1
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::resolve_scanner_server_config;
use crate::scanner_budget::ScannerCycleBudgetConfig;
use crate::sleeper::{SCANNER_SLEEPER, scanner_default_speed};
use rustfs_config::{
@@ -674,7 +675,7 @@ fn apply_resolved_runtime_config(config: ScannerRuntimeConfig) {
}
fn current_server_config() -> Option<ServerConfig> {
rustfs_config::server_config::get_global_server_config()
resolve_scanner_server_config()
}
pub(crate) fn resolve_scanner_runtime_config_from_global() -> ScannerRuntimeConfig {
crates/scanner/src/scanner.rs
+5 -4
View File
@@ -47,7 +47,8 @@ use tracing::{debug, error, info, instrument, warn};
use crate::{
ECStore, EcstoreError, RUSTFS_META_BUCKET, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _,
get_lifecycle_config, get_replication_config, is_erasure_sd, read_config, replace_bucket_usage_memory_from_info, save_config,
get_lifecycle_config, get_replication_config, read_config, replace_bucket_usage_memory_from_info, save_config,
scanner_is_erasure_sd,
};
const LOG_COMPONENT_SCANNER: &str = "scanner";
@@ -419,7 +420,7 @@ async fn detect_scanner_maintenance_features(storeapi: &Arc<ECStore>) -> Scanner
}
async fn configure_scanner_defaults(storeapi: &Arc<ECStore>) -> ScannerMaintenanceFeatures {
if is_erasure_sd().await {
if scanner_is_erasure_sd().await {
let features = detect_scanner_maintenance_features(storeapi).await;
let default_cycle_secs = single_disk_default_cycle_secs(features);
set_scanner_default_speed(single_disk_default_speed());
@@ -568,7 +569,7 @@ pub struct BackgroundHealInfo {
/// Read background healing information from storage
pub async fn read_background_heal_info(storeapi: Arc<ECStore>) -> BackgroundHealInfo {
// Skip for ErasureSD setup
if is_erasure_sd().await {
if scanner_is_erasure_sd().await {
return BackgroundHealInfo::default();
}
@@ -610,7 +611,7 @@ pub async fn read_background_heal_info(storeapi: Arc<ECStore>) -> BackgroundHeal
#[instrument(skip(storeapi))]
pub async fn save_background_heal_info(storeapi: Arc<ECStore>, info: BackgroundHealInfo) {
// Skip for ErasureSD setup
if is_erasure_sd().await {
if scanner_is_erasure_sd().await {
return;
}
crates/scanner/src/scanner_folder.rs
+5 -5
View File
@@ -54,8 +54,8 @@ use crate::{
BucketVersioningSys, Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts,
ReplicationConfig, ReplicationQueueAdmission, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule,
enqueue_global_newer_noncurrent, is_erasure, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
path2_bucket_object_with_base_path, queue_replication_heal_internal,
enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object,
path2_bucket_object_with_base_path, queue_replication_heal_internal, scanner_is_erasure,
};
use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete};
@@ -910,7 +910,7 @@ impl ScannerItem {
let action = event.action;
let count = u64::try_from(to_delete_objs.len()).unwrap_or(u64::MAX);
let done_ilm = Metrics::time_ilm(action);
let queued = enqueue_global_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner).await;
let queued = enqueue_runtime_newer_noncurrent(&self.bucket, to_delete_objs, event, &LcEventSrc::Scanner).await;
if record_scanner_ilm_action_if_queued(global_metrics(), action, count, queued) {
done_ilm(count)();
remaining_versions = remaining_versions.saturating_sub(noncurrent_accounting.len());
@@ -1711,7 +1711,7 @@ impl FolderScanner {
return Err(ScannerError::Other("Operation cancelled".to_string()));
}
if found_objects && is_erasure().await {
if found_objects && scanner_is_erasure().await {
// If we found an object in erasure mode, we skip subdirs (only datadirs)...
debug!(
target: "rustfs::scanner::folder",
@@ -2310,7 +2310,7 @@ pub async fn scan_data_folder(
let (update_current_path, close_disk) = current_path_updater(&base_path, &cache.info.name);
// Create skip_heal flag
let is_erasure_mode = is_erasure().await;
let is_erasure_mode = scanner_is_erasure().await;
let skip_heal = Arc::new(std::sync::atomic::AtomicBool::new(!is_erasure_mode || cache.info.skip_healing));
// Create heal_object_select flag
crates/scanner/src/scanner_io.rs
+5 -5
View File
@@ -46,9 +46,9 @@ use crate::ScannerObjectInfo as ObjectInfo;
use crate::{
BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result,
ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_global_free_version,
get_lifecycle_config, get_object_lock_config, get_replication_config, list_global_tiers, resolve_scanner_object_store_handle,
storageclass,
ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_runtime_free_version,
get_lifecycle_config, get_object_lock_config, get_replication_config, list_runtime_tiers,
resolve_scanner_object_store_handle, storageclass,
};
pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file";
@@ -1350,7 +1350,7 @@ impl ScannerIODisk for Disk {
let mut size_summary = SizeSummary::default();
let tiers = list_global_tiers().await;
let tiers = list_runtime_tiers().await;
for tier in tiers.iter() {
size_summary.tier_stats.insert(tier.name.clone(), TierStats::default());
@@ -1373,7 +1373,7 @@ impl ScannerIODisk for Disk {
if !free_version_infos.is_empty() {
for oi in free_version_infos {
enqueue_global_free_version(oi).await;
enqueue_runtime_free_version(oi).await;
}
}
crates/scanner/tests/lifecycle_integration_test.rs
+5 -3
View File
@@ -26,7 +26,7 @@ use rustfs_ecstore::api::bucket::versioning_sys::BucketVersioningSys;
use rustfs_ecstore::api::capacity::path2_bucket_object_with_base_path;
use rustfs_ecstore::api::client::transition_api::{ReadCloser, ReaderImpl};
use rustfs_ecstore::api::disk::{DiskAPI as _, DiskOption, STORAGE_FORMAT_FILE, endpoint::Endpoint, new_disk};
use rustfs_ecstore::api::global::GLOBAL_TierConfigMgr;
use rustfs_ecstore::api::global::get_global_tier_config_mgr;
use rustfs_ecstore::api::layout::{EndpointServerPools, Endpoints, PoolEndpoints};
use rustfs_ecstore::api::storage::{ECStore, init_local_disks};
use rustfs_ecstore::api::tier::tier_config::{TierConfig, TierMinIO, TierType};
@@ -423,7 +423,8 @@ async fn create_test_tier(server: u32) {
})
},
};
let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await;
let tier_config_mgr_handle = get_global_tier_config_mgr();
let mut tier_config_mgr = tier_config_mgr_handle.write().await;
if let Err(err) = tier_config_mgr.add(args, false).await {
println!("tier_config_mgr add failed, e: {err:?}");
panic!("tier add failed. {err}");
@@ -761,7 +762,8 @@ impl ScannerWarmBackend for MockWarmBackend {
async fn register_mock_tier(tier_name: &str) -> MockWarmBackend {
let backend = MockWarmBackend::default();
let mut tier_config_mgr = GLOBAL_TierConfigMgr.write().await;
let tier_config_mgr_handle = get_global_tier_config_mgr();
let mut tier_config_mgr = tier_config_mgr_handle.write().await;
tier_config_mgr.tiers.insert(
tier_name.to_string(),
TierConfig {