diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 36cc6b817..0e347a8c1 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -26,6 +26,8 @@ use http::HeaderMap; use rustfs_config::server_config::{Config as ServerConfig, get_global_server_config as config_get_global_server_config}; use std::path::PathBuf; use std::sync::Arc; +use std::sync::RwLock; +use std::time::{Duration, Instant}; use storage_api::owner::{ ECSTORE_BUCKET_META_PREFIX, ECSTORE_RUSTFS_META_BUCKET, ECSTORE_STORAGE_FORMAT_FILE, ECSTORE_STORAGECLASS_RRS, ECSTORE_STORAGECLASS_STANDARD, ECSTORE_TRANSITION_COMPLETE, EcstoreBucketTargetSys, EcstoreBucketVersioningSys, EcstoreDisk, @@ -33,7 +35,7 @@ use storage_api::owner::{ EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts, EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, EcstoreSetDisks, EcstoreStorageError, EcstoreStore, - EcstoreTierConfig, EcstoreVersioningApi, HTTPPreconditions, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete, + EcstoreVersioningApi, HTTPPreconditions, HTTPRangeSpec, ObjectIO, ObjectOperations, ObjectToDelete, ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache, @@ -363,8 +365,46 @@ pub(crate) fn resolve_scanner_server_config() -> Option { config_get_global_server_config() } -pub(crate) async fn list_runtime_tiers() -> Vec { - ecstore_get_global_tier_config_mgr().read().await.list_tiers() +/// How long the scanner caches the runtime tier-name list before re-reading +/// the tier configuration manager. +const TIER_NAME_CACHE_TTL: Duration = Duration::from_secs(30); + +/// Process-wide TTL cache of runtime tier names. +/// +/// The scan hot path only needs tier *names* to seed `SizeSummary::tier_stats` +/// per object, but every `list_tiers()` call clones each full `TierConfig` +/// (endpoints, credentials, prefixes) from the global manager. Caching just +/// the names keeps the per-object cost at an `Arc` clone. +/// +/// Staleness bounds: a newly added tier starts showing up in scans at most +/// `TIER_NAME_CACHE_TTL` later; a removed tier can leave an all-zero +/// `TierStats` seed behind for one cache generation, which merges harmlessly +/// by key in per-object accounting and disappears on the next refresh. +static TIER_NAME_CACHE: RwLock)>> = RwLock::new(None); + +/// Tier names currently registered in the tier configuration, cached for +/// `TIER_NAME_CACHE_TTL`. +pub(crate) async fn runtime_tier_names() -> Arc<[String]> { + { + let cached = TIER_NAME_CACHE.read().unwrap_or_else(|err| err.into_inner()).clone(); + if let Some((refreshed_at, names)) = cached + && refreshed_at.elapsed() < TIER_NAME_CACHE_TTL + { + return names; + } + } + + let tiers = ecstore_get_global_tier_config_mgr().read().await.list_tiers(); + let names: Arc<[String]> = tiers.iter().map(|tier| tier.name.clone()).collect::>().into(); + *TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((Instant::now(), Arc::clone(&names))); + names +} + +/// Test-only cache reset; the production cache has no invalidation hook +/// because the TTL is its only refresh path. +#[cfg(test)] +fn reset_tier_name_cache_for_test() { + *TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = None; } pub(crate) async fn enqueue_runtime_free_version(oi: ScannerObjectInfo) { @@ -561,6 +601,20 @@ mod tests { use super::*; use serial_test::serial; + #[tokio::test] + #[serial] + async fn runtime_tier_names_serves_cached_arc_within_ttl() { + reset_tier_name_cache_for_test(); + // The tier config manager is unconfigured in unit tests, so the + // first call populates the cache from an empty tier list... + let first = runtime_tier_names().await; + assert!(first.is_empty()); + // ...and a second call within the TTL must return the cached Arc + // (pointer-equal) without re-reading the manager. + let second = runtime_tier_names().await; + assert!(Arc::ptr_eq(&first, &second)); + } + #[test] #[serial] fn foreground_read_guard_tracks_stream_lifetime() { diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index c15ed6c41..4b279eea2 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -45,7 +45,7 @@ use rustfs_common::metrics::{ use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count}; use rustfs_filemeta::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams}; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; -use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration}; +use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, VersioningConfiguration}; use time::OffsetDateTime; use tokio::select; use tokio::sync::mpsc; @@ -53,10 +53,10 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, warn}; use crate::{ - BucketVersioningSys, Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, - ReplicationConfig, ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE, - ScannerDiskExt as _, ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, - apply_transition_rule, enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object, + Disk, DiskError, DiskInfoOptions, Evaluator, Event, LcEventSrc, ListPathRawOptions, ObjectOpts, ReplicationConfig, + ReplicationHealObject, ReplicationQueueAdmission, ReplicationStatusType, STORAGE_FORMAT_FILE, ScannerDiskExt as _, + ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, apply_expiry_rule, apply_transition_rule, + enqueue_runtime_newer_noncurrent, is_reserved_or_invalid_bucket, list_path_raw, path2_bucket_object, path2_bucket_object_with_base_path, queue_replication_heal, scanner_is_erasure, scanner_replication_config_for_lifecycle_eval, }; @@ -934,6 +934,7 @@ impl ScannerItem { &mut self, object_infos: Vec, lock_retention: Option>, + versioning_config: VersioningConfiguration, size_summary: &mut SizeSummary, ) { if object_infos.is_empty() { @@ -958,21 +959,8 @@ impl ScannerItem { "Scanner lifecycle evaluation started" ); - let versioning_config = match BucketVersioningSys::get(&self.bucket).await { - Ok(versioning_config) => versioning_config, - Err(_) => { - warn!( - target: "rustfs::scanner::folder", - event = EVENT_SCANNER_LIFECYCLE_ACTION, - component = LOG_COMPONENT_SCANNER, - subsystem = LOG_SUBSYSTEM_LIFECYCLE, - bucket = %self.bucket, - state = "versioning_lookup_failed_defaulting", - "Scanner lifecycle action falling back to default bucket versioning" - ); - Default::default() - } - }; + // `versioning_config` is resolved once per object by the caller + // (`get_size`) and handed in; only `prefix_enabled` is consulted here. let Some(lifecycle) = self.lifecycle.as_ref() else { let mut cumulative_size = 0; @@ -1402,6 +1390,11 @@ impl ScannerItem { fn alert_excessive_versions(&self, remaining_versions: usize, cumulative_size: i64) { ensure_scanner_alert_metrics_registered(); let (too_many_versions, too_large_versions) = should_alert_excessive_versions(remaining_versions, cumulative_size); + // Threshold check first so healthy objects never pay for the + // object-path allocation below. + if !too_many_versions && !too_large_versions { + return; + } let object_path = self.object_path(); if too_many_versions { global_metrics().record_scanner_source_executed(ScannerWorkSource::Alerts, 1); diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index f722ff186..52ec53deb 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -32,7 +32,9 @@ use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo}; use rustfs_filemeta::FileMeta; use rustfs_lock::{LockError, NamespaceLockGuard}; use rustfs_utils::path::path_join_buf; -use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration}; +use s3s::dto::{ + BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration, VersioningConfiguration, +}; use sha2::{Digest as _, Sha256}; use std::collections::{HashMap, HashSet}; use std::future::Future; @@ -55,7 +57,7 @@ use crate::{ BucketTargetSys, BucketVersioningSys, Disk, DiskError, ECStore, EcstoreError as Error, EcstoreResult as Result, RUSTFS_META_BUCKET, ReplicationConfig, STORAGE_FORMAT_FILE, ScannerDiskExt as _, ScannerLifecycleConfigExt as _, ScannerReplicationConfigExt as _, ScannerVersioningConfigExt as _, SetDisks, StorageError, enqueue_runtime_free_version, - get_lifecycle_config, get_object_lock_config, get_replication_config, list_runtime_tiers, storageclass, + get_lifecycle_config, get_object_lock_config, get_replication_config, runtime_tier_names, storageclass, }; pub(crate) const SCANNER_SKIP_FILE_ERROR: &str = "skip file"; @@ -63,6 +65,11 @@ pub(crate) const SCANNER_METADATA_CORRUPT_ERROR: &str = "scanner metadata corrup pub(crate) const SCANNER_METADATA_TRANSIENT_ERROR: &str = "scanner metadata transient"; const LOG_COMPONENT_SCANNER: &str = "scanner"; const LOG_SUBSYSTEM_IO: &str = "io"; +// Mirrors `scanner_folder.rs` so the versioning-lookup fallback warn keeps its +// historical `rustfs::scanner::folder` lifecycle event identity after the +// lookup moved into `get_size`. +const LOG_SUBSYSTEM_LIFECYCLE: &str = "lifecycle"; +const EVENT_SCANNER_LIFECYCLE_ACTION: &str = "scanner_lifecycle_action"; const EVENT_SCANNER_DISK_BUCKET_STATE: &str = "scanner_disk_bucket_state"; const EVENT_SCANNER_DATA_USAGE_STREAM: &str = "scanner_data_usage_stream"; const EVENT_SCANNER_CACHE_PERSIST_STATE: &str = "scanner_cache_persist_state"; @@ -3822,6 +3829,24 @@ impl ScannerIOCache for SetDisks { } } +/// Seed [`SizeSummary::tier_stats`] from the cached tier-name list. +/// +/// Preserves the original seeding semantics: with no tiers configured the map +/// stays completely empty (STANDARD/RRS are not seeded either); otherwise the +/// standard storage classes are seeded alongside every configured tier so +/// per-object accounting always finds its tier key. +fn tier_stats_template(tier_names: &[String]) -> HashMap { + let mut tier_stats = HashMap::with_capacity(tier_names.len() + 2); + for tier_name in tier_names { + tier_stats.insert(tier_name.clone(), TierStats::default()); + } + if !tier_stats.is_empty() { + tier_stats.insert(storageclass::STANDARD.to_string(), TierStats::default()); + tier_stats.insert(storageclass::RRS.to_string(), TierStats::default()); + } + tier_stats +} + #[async_trait::async_trait] impl ScannerIODisk for Disk { async fn get_size(&self, mut item: ScannerItem) -> Result { @@ -3861,10 +3886,26 @@ impl ScannerIODisk for Disk { } }; - let versioned = BucketVersioningSys::get(&item.bucket) - .await - .map(|v| v.versioned(&item.object_path())) - .unwrap_or(false); + // Single versioning lookup per object, shared with `apply_actions` + // (which used to query it a second time). On failure keep the + // historical fallback: default configuration (versioned = false) plus + // the warn that `apply_actions` used to emit. + let versioning_config = match BucketVersioningSys::get(&item.bucket).await { + Ok(versioning_config) => versioning_config, + Err(_) => { + warn!( + target: "rustfs::scanner::folder", + event = EVENT_SCANNER_LIFECYCLE_ACTION, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + bucket = %item.bucket, + state = "versioning_lookup_failed_defaulting", + "Scanner lifecycle action falling back to default bucket versioning" + ); + VersioningConfiguration::default() + } + }; + let versioned = versioning_config.versioned(&item.object_path()); let object_infos = fivs .versions @@ -3879,19 +3920,10 @@ impl ScannerIODisk for Disk { let mut size_summary = SizeSummary::default(); - let tiers = list_runtime_tiers().await; - - for tier in tiers.iter() { - size_summary.tier_stats.insert(tier.name.clone(), TierStats::default()); - } - if !size_summary.tier_stats.is_empty() { - size_summary - .tier_stats - .insert(storageclass::STANDARD.to_string(), TierStats::default()); - size_summary - .tier_stats - .insert(storageclass::RRS.to_string(), TierStats::default()); - } + // Tier names come from the process-wide TTL cache; seeding from them + // replaces the per-object clone of every full TierConfig. + let tier_names = runtime_tier_names().await; + size_summary.tier_stats = tier_stats_template(&tier_names); let lock_config = object_lock_config_for_scanner_item(&item).await; @@ -3901,7 +3933,8 @@ impl ScannerIODisk for Disk { // `object_infos`. global_metrics().record_scanner_versions_scanned(object_infos.len() as u64); - item.apply_actions(object_infos, lock_config, &mut size_summary).await; + item.apply_actions(object_infos, lock_config, versioning_config, &mut size_summary) + .await; if !free_version_infos.is_empty() { for oi in free_version_infos { @@ -4968,6 +5001,23 @@ mod tests { assert!(is_xl_meta_path("/data/bucket/object/xl.meta")); } + #[test] + fn tier_stats_template_seeds_tiers_and_standard_classes() { + let template = tier_stats_template(&["WARM".to_string(), "COLD".to_string()]); + + assert_eq!(template.len(), 4); + for tier in ["WARM", "COLD", storageclass::STANDARD, storageclass::RRS] { + assert_eq!(template.get(tier), Some(&TierStats::default()), "missing seed for tier {tier}"); + } + } + + #[test] + fn tier_stats_template_stays_empty_without_tiers() { + let template = tier_stats_template(&[]); + + assert!(template.is_empty()); + } + #[tokio::test] async fn get_size_treats_missing_metadata_as_skip_file() { let temp_dir = std::env::temp_dir().join(format!("rustfs-scanner-missing-meta-{}", Uuid::new_v4())); diff --git a/crates/scanner/src/storage_api.rs b/crates/scanner/src/storage_api.rs index cbd7c6486..d8779a0aa 100644 --- a/crates/scanner/src/storage_api.rs +++ b/crates/scanner/src/storage_api.rs @@ -99,7 +99,6 @@ pub(crate) use rustfs_ecstore::api::set_disk::SetDisks as EcstoreSetDisks; pub(crate) use rustfs_ecstore::api::storage::ECStore as EcstoreStore; #[cfg(test)] pub(crate) use rustfs_ecstore::api::storage::init_local_disks_with_instance_ctx as ecstore_init_local_disks_with_instance_ctx; -pub(crate) use rustfs_ecstore::api::tier::tier_config::TierConfig as EcstoreTierConfig; use rustfs_storage_api as storage_contracts; pub(crate) mod owner { @@ -114,15 +113,15 @@ pub(crate) mod owner { EcstoreDiskLocation, EcstoreDiskResult, EcstoreErrorType, EcstoreEvaluator, EcstoreEvent, EcstoreEventArgs, EcstoreLcEventSrc, EcstoreLifecycle, EcstoreListPathRawOptions, EcstoreNsScannerOpenRequest, EcstoreObjectOpts, EcstoreReplicationConfigurationExt, EcstoreReplicationScannerBridge, EcstoreResultType, EcstoreScanGuard, - EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreTierConfig, EcstoreVersioningApi, - ScannerReplicationHealObject, ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, - ecstore_apply_transition_rule, ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, - ecstore_get_lifecycle_config, ecstore_get_object_lock_config, ecstore_get_replication_config, - ecstore_invalidate_admin_data_usage_snapshot_cache, ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, - ecstore_is_erasure_sd, ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, - ecstore_object_opts_from_object_info, ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, - ecstore_read_config, ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, - ecstore_save_config, ecstore_send_event, scanner_replication_config_for_lifecycle_eval, + EcstoreSetDisks, EcstoreStorageError, EcstoreStore, EcstoreVersioningApi, ScannerReplicationHealObject, + ScannerReplicationHealResult, ScannerReplicationQueueAdmission, ecstore_apply_expiry_rule, ecstore_apply_transition_rule, + ecstore_expiry_state_handle, ecstore_get_global_tier_config_mgr, ecstore_get_lifecycle_config, + ecstore_get_object_lock_config, ecstore_get_replication_config, ecstore_invalidate_admin_data_usage_snapshot_cache, + ecstore_invalidate_data_usage_snapshot_cache, ecstore_is_erasure, ecstore_is_erasure_sd, + ecstore_is_reserved_or_invalid_bucket, ecstore_list_path_raw, ecstore_object_opts_from_object_info, + ecstore_path2_bucket_object, ecstore_path2_bucket_object_with_base_path, ecstore_read_config, + ecstore_replace_bucket_usage_memory_from_info, ecstore_resolve_object_store_handle, ecstore_save_config, + ecstore_send_event, scanner_replication_config_for_lifecycle_eval, }; #[cfg(test)]