From 62a31e4ec4968a0dd0f008db2fceefc9e94c963b Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 8 Jul 2026 00:15:07 +0800 Subject: [PATCH] fix(storage): address pending metadata and health gaps (#4380) --- Cargo.lock | 1 + crates/ecstore/src/bucket/metadata_sys.rs | 202 +++++++++++-- .../src/diagnostics/admin_server_info.rs | 46 ++- crates/ecstore/src/disk/local.rs | 126 +++++++- crates/ecstore/src/set_disk/metadata.rs | 129 +++++++- crates/madmin/Cargo.toml | 1 + crates/madmin/src/health.rs | 132 +++++++- .../protocols/src/swift/expiration_worker.rs | 283 ++++++++++++++++-- 8 files changed, 841 insertions(+), 79 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ee1a51633..35a200d3a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9387,6 +9387,7 @@ dependencies = [ "rmp-serde", "serde", "serde_json", + "sysinfo", "time", ] diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index aba5028f3..2edc43f9f 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -39,19 +39,26 @@ use std::{collections::HashMap, sync::Arc}; use time::OffsetDateTime; use tokio::sync::RwLock; use tokio::time::sleep; -use tracing::error; +use tokio_util::sync::CancellationToken; +use tracing::{error, warn}; lazy_static! { pub static ref GLOBAL_BUCKET_METADATA_SYS: OnceLock>> = OnceLock::new(); } +const BUCKET_METADATA_REFRESH_INTERVAL: Duration = Duration::from_secs(15 * 60); + pub async fn init_bucket_metadata_sys(api: Arc, buckets: Vec) { let mut sys = BucketMetadataSys::new(api); sys.init(buckets).await; let sys = Arc::new(RwLock::new(sys)); - GLOBAL_BUCKET_METADATA_SYS.set(sys).unwrap(); + GLOBAL_BUCKET_METADATA_SYS.set(sys.clone()).unwrap(); + + if runtime_sources::setup_is_dist_erasure().await { + start_refresh_buckets_metadata_loop(sys); + } } pub fn get_global_bucket_metadata_sys() -> Option>> { @@ -85,6 +92,67 @@ pub async fn remove_bucket_metadata(bucket: &str) -> Result { Ok(lock.remove(bucket).await) } +fn start_refresh_buckets_metadata_loop(sys: Arc>) { + let Some(cancel_token) = runtime_sources::background_services_cancel_token().cloned() else { + warn!("bucket metadata refresh loop skipped because background cancellation token is not initialized"); + return; + }; + + tokio::spawn(async move { + refresh_buckets_metadata_loop(sys, cancel_token).await; + }); +} + +async fn refresh_buckets_metadata_loop(sys: Arc>, cancel_token: CancellationToken) { + loop { + if !wait_refresh_interval_or_cancel(&cancel_token, BUCKET_METADATA_REFRESH_INTERVAL).await { + break; + } + refresh_buckets_metadata_once(sys.clone()).await; + } +} + +async fn wait_refresh_interval_or_cancel(cancel_token: &CancellationToken, interval: Duration) -> bool { + tokio::select! { + _ = cancel_token.cancelled() => false, + _ = sleep(interval) => true, + } +} + +async fn refresh_buckets_metadata_once(sys: Arc>) { + let buckets = { + let sys = sys.read().await; + sys.bucket_names().await + }; + if buckets.is_empty() { + return; + } + + let count = runtime_sources::endpoint_erasure_set_count() + .map(|count| count * 10) + .unwrap_or(10) + .max(1); + let mut failed_buckets = HashSet::new(); + + for chunk in buckets.chunks(count) { + let sys = sys.read().await; + sys.concurrent_load(chunk, &mut failed_buckets).await; + } + + if !failed_buckets.is_empty() { + warn!( + failed_bucket_count = failed_buckets.len(), + "bucket metadata refresh loop left buckets queued for retry" + ); + } +} + +async fn sync_bucket_target_sys(bucket: &str, bm: &BucketMetadata) { + BucketTargetSys::get() + .update_all_targets(bucket, bm.bucket_target_config.as_ref()) + .await; +} + pub async fn get(bucket: &str) -> Result> { let sys = get_bucket_metadata_sys()?; let lock = sys.read().await; @@ -295,10 +363,6 @@ impl BucketMetadataSys { let mut initialized = self.initialized.write().await; *initialized = true; - if runtime_sources::setup_is_dist_erasure().await { - // TODO: refresh_buckets_metadata_loop - } - Ok(()) } @@ -325,18 +389,11 @@ impl BucketMetadataSys { let results = join_all(futures).await; - let mut idx = 0; - - let mut mp = self.metadata_map.write().await; - - // TODO: EventNotifier - for res in results { + for (idx, res) in results.into_iter().enumerate() { match res { Ok(res) => { if let Some(bucket) = buckets.get(idx) { - let x = Arc::new(res); - mp.insert(bucket.clone(), x.clone()); - BucketTargetSys::get().set(bucket, &x).await; + self.set(bucket.clone(), Arc::new(res)).await; } } Err(e) => { @@ -346,8 +403,6 @@ impl BucketMetadataSys { } } } - - idx += 1; } } @@ -367,7 +422,9 @@ impl BucketMetadataSys { pub async fn set(&self, bucket: String, bm: Arc) { if !is_meta_bucketname(&bucket) { let mut map = self.metadata_map.write().await; - map.insert(bucket, bm); + map.insert(bucket.clone(), bm.clone()); + drop(map); + sync_bucket_target_sys(&bucket, &bm).await; } } @@ -379,7 +436,12 @@ impl BucketMetadataSys { return false; } let mut map = self.metadata_map.write().await; - map.remove(bucket).is_some() + let removed = map.remove(bucket).is_some(); + drop(map); + if removed { + BucketTargetSys::get().delete(bucket).await; + } + removed } async fn _reset(&mut self) { @@ -466,6 +528,10 @@ impl BucketMetadataSys { Ok(()) } + async fn bucket_names(&self) -> Vec { + self.metadata_map.read().await.keys().cloned().collect() + } + pub async fn get_config_from_disk(&self, bucket: &str) -> Result { if is_meta_bucketname(bucket) { return Err(Error::other("errInvalidArgument")); @@ -498,6 +564,8 @@ impl BucketMetadataSys { let bm = Arc::new(bm); map.insert(bucket.to_string(), bm.clone()); + drop(map); + sync_bucket_target_sys(bucket, &bm).await; Ok((bm, true)) } @@ -697,13 +765,9 @@ impl BucketMetadataSys { } pub async fn get_replication_config(&self, bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> { - let (bm, reload) = self.get_config(bucket).await?; + let (bm, _) = self.get_config(bucket).await?; if let Some(config) = &bm.replication_config { - if reload { - // TODO: globalBucketTargetSys - } - //println!("549 {:?}", config.clone()); Ok((config.clone(), bm.replication_config_updated_at)) } else { Err(Error::ConfigNotFound) @@ -711,17 +775,95 @@ impl BucketMetadataSys { } pub async fn get_bucket_targets_config(&self, bucket: &str) -> Result { - let (bm, reload) = self.get_config(bucket).await?; + let (bm, _) = self.get_config(bucket).await?; if let Some(config) = &bm.bucket_target_config { - if reload { - // TODO: globalBucketTargetSys - //config. - } - Ok(config.clone()) } else { Err(Error::ConfigNotFound) } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::bucket::target::{BucketTarget, BucketTargetType, Credentials}; + use serial_test::serial; + use tokio::time::timeout; + + fn target(bucket: &str, id: &str) -> BucketTarget { + BucketTarget { + source_bucket: bucket.to_string(), + endpoint: format!("{id}.example.com:9000"), + credentials: Some(Credentials { + access_key: "access".to_string(), + secret_key: "secret".to_string(), + ..Default::default() + }), + target_bucket: format!("{bucket}-{id}"), + arn: format!("arn:rustfs:replication:us-east-1:{bucket}:{id}"), + target_type: BucketTargetType::ReplicationService, + region: "us-east-1".to_string(), + ..Default::default() + } + } + + #[tokio::test] + #[serial] + async fn metadata_reload_syncs_bucket_target_sys() { + let bucket = "metadata-reload-targets"; + let target_sys = BucketTargetSys::get(); + target_sys.delete(bucket).await; + + let mut bm = BucketMetadata::new(bucket); + bm.bucket_target_config = Some(BucketTargets { + targets: vec![target(bucket, "fresh")], + }); + + sync_bucket_target_sys(bucket, &bm).await; + + let targets = target_sys + .list_bucket_targets(bucket) + .await + .expect("target sync should publish bucket targets"); + assert_eq!(targets.targets.len(), 1); + assert_eq!(targets.targets[0].arn, format!("arn:rustfs:replication:us-east-1:{bucket}:fresh")); + + target_sys.delete(bucket).await; + } + + #[tokio::test] + #[serial] + async fn metadata_reload_clears_stale_bucket_targets_when_config_is_removed() { + let bucket = "metadata-clear-targets"; + let target_sys = BucketTargetSys::get(); + target_sys.delete(bucket).await; + target_sys + .targets_map + .write() + .await + .insert(bucket.to_string(), vec![target(bucket, "stale")]); + + let bm = BucketMetadata::new(bucket); + sync_bucket_target_sys(bucket, &bm).await; + + assert!(target_sys.list_bucket_targets(bucket).await.is_err()); + target_sys.delete(bucket).await; + } + + #[tokio::test] + async fn refresh_wait_exits_when_cancelled() { + let cancel_token = CancellationToken::new(); + cancel_token.cancel(); + + let should_refresh = timeout( + Duration::from_millis(100), + wait_refresh_interval_or_cancel(&cancel_token, Duration::from_secs(60)), + ) + .await + .expect("cancelled refresh wait should not sleep until the interval"); + + assert!(!should_refresh); + } +} diff --git a/crates/ecstore/src/diagnostics/admin_server_info.rs b/crates/ecstore/src/diagnostics/admin_server_info.rs index 50d9a4231..76fc5b318 100644 --- a/crates/ecstore/src/diagnostics/admin_server_info.rs +++ b/crates/ecstore/src/diagnostics/admin_server_info.rs @@ -21,7 +21,7 @@ use crate::data_usage::load_data_usage_cache; use crate::storage_api_contracts::admin::StorageAdminApi; use rustfs_common::heal_channel::DriveState; use rustfs_madmin::{ - BackendDisks, Disk, ErasureSetInfo, ITEM_INITIALIZING, ITEM_OFFLINE, ITEM_ONLINE, InfoMessage, ServerProperties, + BackendDisks, Disk, ErasureSetInfo, ITEM_INITIALIZING, ITEM_OFFLINE, ITEM_ONLINE, InfoMessage, MemStats, ServerProperties, }; use rustfs_protos::{ models::{PingBody, PingBodyBuilder}, @@ -128,10 +128,22 @@ pub async fn get_local_server_property() -> ServerProperties { let addr = runtime_sources::local_node_name().await; let mut pool_numbers = HashSet::new(); let mut network = HashMap::new(); + let (mem_stats, max_procs, num_cpu) = collect_runtime_server_stats(); let endpoints = match runtime_sources::endpoint_pools() { Some(eps) => eps, - None => return ServerProperties::default(), + None => { + return ServerProperties { + state: ITEM_INITIALIZING.to_string(), + endpoint: addr, + uptime: runtime_sources::boot_uptime_secs(), + version: get_commit_id(), + mem_stats, + max_procs, + num_cpu, + ..Default::default() + }; + } }; for ep in endpoints.as_ref().iter() { for endpoint in ep.endpoints.as_ref().iter() { @@ -154,14 +166,14 @@ pub async fn get_local_server_property() -> ServerProperties { } } - // todo: mem collect - // let mem_stats = - let mut props = ServerProperties { endpoint: addr, uptime: runtime_sources::boot_uptime_secs(), network, version: get_commit_id(), + mem_stats, + max_procs, + num_cpu, ..Default::default() }; @@ -189,6 +201,15 @@ pub async fn get_local_server_property() -> ServerProperties { props } +fn collect_runtime_server_stats() -> (MemStats, u64, u64) { + let num_cpu = u64::try_from(num_cpus::get()).unwrap_or(u64::MAX); + let max_procs = std::thread::available_parallelism() + .map(|parallelism| u64::try_from(parallelism.get()).unwrap_or(u64::MAX)) + .unwrap_or(num_cpu.max(1)); + + (rustfs_madmin::health::collect_mem_stats(), max_procs, num_cpu) +} + pub async fn get_server_info(get_pools: bool) -> InfoMessage { let nowt: OffsetDateTime = OffsetDateTime::now_utc(); @@ -393,7 +414,7 @@ mod tests { use crate::runtime::sources as runtime_sources; - use super::get_server_info; + use super::{get_local_server_property, get_server_info}; #[serial] #[tokio::test] @@ -403,4 +424,17 @@ mod tests { assert_eq!(info.deployment_id, expected_deployment_id); } + + #[serial] + #[tokio::test] + async fn local_server_property_includes_runtime_stats_without_endpoint_pools() { + let props = get_local_server_property().await; + + assert!(props.num_cpu > 0); + assert!(props.max_procs > 0); + assert!( + props.mem_stats.alloc > 0 || props.mem_stats.total_alloc > 0 || props.mem_stats.heap_alloc > 0, + "memory stats should not remain fixed placeholders" + ); + } } diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index c2aac3244..774ab8779 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -36,7 +36,7 @@ use crate::erasure::coding::bitrot_verify; use crate::runtime::sources as runtime_sources; use bytes::Bytes; use metrics::counter; -use parking_lot::RwLock as ParkingLotRwLock; +use parking_lot::{Mutex as ParkingLotMutex, RwLock as ParkingLotRwLock}; use rustfs_filemeta::{ Cache, FileInfo, FileInfoOpts, FileMeta, MetaCacheEntry, MetacacheWriter, ObjectPartInfo, Opts, RawFileInfo, UpdateFn, get_file_info, read_xl_meta_no_data, @@ -1461,6 +1461,7 @@ pub struct LocalDisk { pub major: u64, pub minor: u64, pub nrrequests: u64, + scan_locks: Arc>>, // Performance optimization fields path_cache: Arc>>, current_dir: Arc>, @@ -1698,6 +1699,7 @@ impl LocalDisk { minor: Default::default(), major: Default::default(), nrrequests: Default::default(), + scan_locks: Arc::new(ParkingLotMutex::new(HashSet::new())), // // format_legacy, // format_file_info: Mutex::new(format_meta), // format_data: Mutex::new(format_data), @@ -2060,6 +2062,20 @@ impl LocalDisk { reject_local_disk_symlink_components(&self.root, path) } + fn try_acquire_scan_lock(&self, opts: &WalkDirOptions) -> Result { + let key = local_disk_scan_lock_key(&opts.bucket, &opts.base_dir, opts.filter_prefix.as_deref()); + let mut scan_locks = self.scan_locks.lock(); + + if !scan_locks.insert(key.clone()) { + return Err(DiskError::DiskOngoingReq); + } + + Ok(LocalScanLockGuard { + scan_locks: Arc::clone(&self.scan_locks), + key, + }) + } + // Batch path generation with single lock acquisition fn get_object_paths_batch(&self, requests: &[(String, String)]) -> Result> { let mut results = Vec::with_capacity(requests.len()); @@ -3015,6 +3031,34 @@ impl Drop for ScanGuard { } } +struct LocalScanLockGuard { + scan_locks: Arc>>, + key: (String, String), +} + +impl Drop for LocalScanLockGuard { + fn drop(&mut self) { + self.scan_locks.lock().remove(&self.key); + } +} + +fn local_disk_scan_lock_key(bucket: &str, base_dir: &str, filter_prefix: Option<&str>) -> (String, String) { + let mut prefix = base_dir.trim_matches('/').to_owned(); + if let Some(filter_prefix) = filter_prefix + .map(|prefix| prefix.trim_matches('/')) + .filter(|prefix| !prefix.is_empty()) + { + if prefix.is_empty() { + prefix.push_str(filter_prefix); + } else { + prefix.push_str(SLASH_SEPARATOR); + prefix.push_str(filter_prefix); + } + } + + (bucket.to_owned(), prefix) +} + fn is_root_path(path: impl AsRef) -> bool { path.as_ref().components().count() == 1 && path.as_ref().has_root() } @@ -3850,6 +3894,8 @@ impl DiskAPI for LocalDisk { return Err(to_access_error(e, DiskError::VolumeAccessDenied).into()); } + let _scan_lock = self.try_acquire_scan_lock(&opts)?; + let mut wr = wr; let mut out = MetacacheWriter::new(&mut wr); @@ -4828,7 +4874,7 @@ mod test { use std::io; use std::pin::Pin; use std::task::{Context, Poll}; - use tokio::io::{AsyncReadExt, ReadBuf}; + use tokio::io::{AsyncReadExt, AsyncWrite, ReadBuf}; fn test_file_info(name: &str, version_id: Uuid, data_dir: Option, data: Option) -> FileInfo { let size = data @@ -4859,6 +4905,82 @@ mod test { } } + struct BlockingScanWriter { + entered_tx: Option>, + } + + impl AsyncWrite for BlockingScanWriter { + fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll> { + if let Some(tx) = self.entered_tx.take() { + let _ = tx.send(()); + } + Poll::Pending + } + + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Pending + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Pending + } + } + + #[tokio::test] + async fn test_local_disk_scan_rejects_concurrent_same_prefix_and_releases_on_cancel() { + use tempfile::tempdir; + + let dir = tempdir().expect("temp dir should be created"); + let bucket = "test-bucket"; + let object_dir = dir.path().join(bucket).join("prefix/object"); + fs::create_dir_all(&object_dir).await.expect("object dir should be created"); + fs::write(object_dir.join(STORAGE_FORMAT_FILE), b"meta") + .await + .expect("object metadata should be written"); + + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let opts = WalkDirOptions { + bucket: bucket.to_string(), + base_dir: "prefix/".to_string(), + recursive: true, + ..Default::default() + }; + + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let first_disk = Arc::clone(&disk); + let first_opts = opts.clone(); + let mut blocking_writer = BlockingScanWriter { + entered_tx: Some(entered_tx), + }; + let first_scan = tokio::spawn(async move { first_disk.walk_dir(first_opts, &mut blocking_writer).await }); + + entered_rx.await.expect("first scan should enter write path"); + + let mut second_writer = tokio::io::sink(); + let second_scan = disk.walk_dir(opts.clone(), &mut second_writer).await; + assert!( + matches!(second_scan, Err(DiskError::DiskOngoingReq)), + "concurrent scan of same bucket and prefix must be rejected, got {second_scan:?}" + ); + + first_scan.abort(); + assert!( + first_scan + .await + .expect_err("first scan task should be cancelled") + .is_cancelled(), + "aborting the blocked scan should cancel the task" + ); + + let mut after_cancel_writer = tokio::io::sink(); + let after_cancel = disk.walk_dir(opts, &mut after_cancel_writer).await; + assert!( + after_cancel.is_ok(), + "cancelled scan must release the bucket/prefix lock, got {after_cancel:?}" + ); + } + #[tokio::test] async fn test_rename_data_writes_old_metadata_backup_before_non_inline_undo() { use tempfile::tempdir; diff --git a/crates/ecstore/src/set_disk/metadata.rs b/crates/ecstore/src/set_disk/metadata.rs index 4feab05ca..c27eece70 100644 --- a/crates/ecstore/src/set_disk/metadata.rs +++ b/crates/ecstore/src/set_disk/metadata.rs @@ -235,7 +235,8 @@ impl SetDisks { pub(super) fn list_object_parities(parts_metadata: &[FileInfo], errs: &[Option]) -> Vec { let total_shards = parts_metadata.len(); - let half = total_shards as i32 / 2; + let total_shards_i32 = i32::try_from(total_shards).unwrap_or(i32::MAX); + let half = total_shards_i32 / 2; let mut parities: Vec = vec![-1; total_shards]; for (index, metadata) in parts_metadata.iter().enumerate() { @@ -251,11 +252,12 @@ impl SetDisks { if metadata.deleted || metadata.size == 0 { parities[index] = half; - // } else if metadata.transition_status == "TransitionComplete" { - // TODO: metadata.transition_status - // parities[index] = total_shards - (total_shards / 2 + 1); + } else if metadata.transition_status == TRANSITION_COMPLETE { + let majority_metadata_parity = total_shards_i32 - (half + 1); + let erasure_parity = i32::try_from(metadata.erasure.parity_blocks).unwrap_or(i32::MAX); + parities[index] = majority_metadata_parity.max(erasure_parity); } else { - parities[index] = metadata.erasure.parity_blocks as i32; + parities[index] = i32::try_from(metadata.erasure.parity_blocks).unwrap_or(i32::MAX); } } parities @@ -502,6 +504,12 @@ impl SetDisks { } } + fn file_info_has_encryption_metadata(meta: &FileInfo) -> bool { + meta.metadata + .keys() + .any(|name| http::is_encryption_metadata_key(name) || http::is_sse_header(name)) + } + fn starts_with_ignore_ascii_case(value: &str, prefix: &str) -> bool { value .get(..prefix.len()) @@ -560,6 +568,11 @@ impl SetDisks { hasher.update(meta.size.to_le_bytes()); hasher.update([u8::from(meta.deleted), u8::from(meta.mark_deleted)]); hasher.update([u8::from(meta.expire_restored)]); + hasher.update([ + u8::from(meta.is_remote()), + u8::from(Self::file_info_has_encryption_metadata(meta)), + u8::from(meta.is_compressed()), + ]); Self::update_hash_optional_time(hasher, meta.mod_time); Self::update_hash_str(hasher, &meta.transition_status); Self::update_hash_str(hasher, &meta.transition_tier); @@ -747,14 +760,6 @@ impl SetDisks { let mod_valid = mod_time == &meta.mod_time; if etag_only || mod_valid { - if meta.is_remote() { - // TODO: - } - - // TODO: IsEncrypted - - // TODO: IsCompressed - meta_hashes[i] = Some(Self::file_info_quorum_hash(meta)); } else { debug!( @@ -1003,6 +1008,104 @@ impl SetDisks { mod tests { use super::*; + fn metadata_quorum_test_fileinfo(mod_time: OffsetDateTime, erasure_index: usize) -> FileInfo { + let mut fi = FileInfo::new("bucket/object", 2, 2); + fi.name = "bucket/object".to_string(); + fi.size = 8 * 1024 * 1024; + fi.mod_time = Some(mod_time); + fi.data_dir = Some(Uuid::new_v4()); + fi.metadata.insert("etag".to_string(), "object-etag".to_string()); + fi.add_object_part(1, "part-etag".to_string(), 8 * 1024 * 1024, Some(mod_time), 8 * 1024 * 1024, None, None); + fi.erasure.index = erasure_index; + fi + } + + fn transition_metadata_quorum_fileinfo(erasure_index: usize) -> FileInfo { + let mut fi = FileInfo::new("bucket/object", 5, 1); + fi.name = "bucket/object".to_string(); + fi.size = 8 * 1024 * 1024; + fi.mod_time = Some(OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp")); + fi.metadata.insert("etag".to_string(), "object-etag".to_string()); + fi.transition_status = TRANSITION_COMPLETE.to_string(); + fi.transition_tier = "WARM".to_string(); + fi.transitioned_objname = "remote/object".to_string(); + fi.transition_version_id = Some(Uuid::new_v4()); + fi.erasure.index = erasure_index; + fi + } + + fn expect_metadata_quorum_error(metas: Vec, mod_time: OffsetDateTime, message: &str) { + let err = SetDisks::find_file_info_in_quorum(&metas, &Some(mod_time), &None, 3).expect_err(message); + assert_eq!(err, DiskError::ErasureReadQuorum); + } + + #[test] + fn metadata_quorum_uses_simple_majority_for_transitioned_objects() { + let parts_metadata = (1..=6).map(transition_metadata_quorum_fileinfo).collect::>(); + let errs = vec![None; parts_metadata.len()]; + + let parities = SetDisks::list_object_parities(&parts_metadata, &errs); + let quorum = SetDisks::object_quorum_from_meta(&parts_metadata, &errs, 1) + .expect("transitioned metadata should resolve object quorum"); + + assert_eq!(parities, vec![2; 6]); + assert_eq!(quorum, (4, 4)); + } + + #[test] + fn find_file_info_in_quorum_rejects_encrypted_plain_metadata_split() { + let mod_time = OffsetDateTime::now_utc(); + let mut encrypted_a = metadata_quorum_test_fileinfo(mod_time, 1); + let mut encrypted_b = metadata_quorum_test_fileinfo(mod_time, 2); + let plain = metadata_quorum_test_fileinfo(mod_time, 3); + encrypted_a + .metadata + .insert("x-rustfs-encryption-key".to_string(), "encrypted-key".to_string()); + encrypted_b + .metadata + .insert("x-rustfs-encryption-key".to_string(), "encrypted-key".to_string()); + + expect_metadata_quorum_error( + vec![encrypted_a, encrypted_b, plain], + mod_time, + "mixed encrypted and plain metadata must not reach quorum", + ); + } + + #[test] + fn find_file_info_in_quorum_rejects_compressed_plain_metadata_split() { + let mod_time = OffsetDateTime::now_utc(); + let mut compressed_a = metadata_quorum_test_fileinfo(mod_time, 1); + let mut compressed_b = metadata_quorum_test_fileinfo(mod_time, 2); + let plain = metadata_quorum_test_fileinfo(mod_time, 3); + http::insert_str(&mut compressed_a.metadata, http::SUFFIX_COMPRESSION, "lz4".to_string()); + http::insert_str(&mut compressed_b.metadata, http::SUFFIX_COMPRESSION, "lz4".to_string()); + + expect_metadata_quorum_error( + vec![compressed_a, compressed_b, plain], + mod_time, + "mixed compressed and plain metadata must not reach quorum", + ); + } + + #[test] + fn find_file_info_in_quorum_rejects_remote_local_metadata_split() { + let mod_time = OffsetDateTime::now_utc(); + let mut remote = metadata_quorum_test_fileinfo(mod_time, 1); + let local_a = metadata_quorum_test_fileinfo(mod_time, 2); + let local_b = metadata_quorum_test_fileinfo(mod_time, 3); + remote.transition_status = TRANSITION_COMPLETE.to_string(); + remote.transition_tier = "WARM".to_string(); + remote.transitioned_objname = "remote/object".to_string(); + remote.transition_version_id = Some(Uuid::new_v4()); + + expect_metadata_quorum_error( + vec![remote, local_a, local_b], + mod_time, + "mixed remote and local metadata must not reach quorum", + ); + } + async fn shuffle_test_disks(tempdir: &tempfile::TempDir, count: usize) -> Vec> { let endpoint = Endpoint::try_from(tempdir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse"); diff --git a/crates/madmin/Cargo.toml b/crates/madmin/Cargo.toml index 192656800..375d223e8 100644 --- a/crates/madmin/Cargo.toml +++ b/crates/madmin/Cargo.toml @@ -34,6 +34,7 @@ humantime.workspace = true hyper.workspace = true serde.workspace = true serde_json.workspace = true +sysinfo.workspace = true time.workspace = true [lib] diff --git a/crates/madmin/src/health.rs b/crates/madmin/src/health.rs index 7b4fecd2b..d7d5b2bff 100644 --- a/crates/madmin/src/health.rs +++ b/crates/madmin/src/health.rs @@ -15,6 +15,7 @@ use std::collections::HashMap; use serde::{Deserialize, Serialize}; +use sysinfo::{Pid, System}; #[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct NodeCommon { @@ -63,8 +64,63 @@ pub struct Cpus { } pub fn get_cpus() -> Cpus { - // todo - Cpus::default() + let mut sys = System::new_all(); + sys.refresh_cpu_all(); + let cores = available_parallelism_count(); + + let mut cpus: Vec = sys + .cpus() + .iter() + .enumerate() + .map(|(idx, cpu)| Cpu { + vendor_id: cpu.vendor_id().to_string(), + physical_id: idx.to_string(), + model_name: cpu.brand().to_string(), + mhz: frequency_to_mhz(cpu.frequency()), + cores, + ..Default::default() + }) + .collect(); + + if cpus.is_empty() { + cpus.push(Cpu { + model_name: "unknown".to_string(), + cores: available_parallelism_count(), + ..Default::default() + }); + } + + let mut cpu_freq_stats: Vec = sys + .cpus() + .iter() + .enumerate() + .map(|(idx, cpu)| { + let frequency = cpu.frequency(); + CpuFreqStats { + name: if cpu.name().is_empty() { + format!("cpu{idx}") + } else { + cpu.name().to_string() + }, + cpuinfo_current_frequency: Some(frequency), + scaling_current_frequency: Some(frequency), + ..Default::default() + } + }) + .collect(); + + if cpu_freq_stats.is_empty() { + cpu_freq_stats.push(CpuFreqStats { + name: "cpu0".to_string(), + ..Default::default() + }); + } + + Cpus { + cpus, + cpu_freq_stats, + ..Default::default() + } } #[derive(Debug, Default, Serialize, Deserialize)] @@ -197,8 +253,62 @@ pub struct MemInfo { limit: Option, } -pub fn get_mem_info(_addr: &str) -> MemInfo { - MemInfo::default() +pub fn get_mem_info(addr: &str) -> MemInfo { + let mut sys = System::new_all(); + sys.refresh_memory(); + + let total = sys.total_memory(); + let used = sys.used_memory(); + let total_swap = sys.total_swap(); + let used_swap = sys.used_swap(); + + MemInfo { + node_common: NodeCommon { + addr: addr.to_string(), + error: None, + }, + total: Some(total), + used: Some(used), + free: Some(total.saturating_sub(used)), + available: Some(sys.available_memory()), + swap_space_total: Some(total_swap), + swap_space_free: Some(total_swap.saturating_sub(used_swap)), + limit: Some(total), + ..Default::default() + } +} + +pub fn collect_mem_stats() -> crate::MemStats { + let mut sys = System::new_all(); + sys.refresh_memory(); + + let process_memory = sys + .process(Pid::from_u32(std::process::id())) + .map(|process| process.memory()) + .filter(|memory| *memory > 0) + .unwrap_or_else(|| sys.used_memory()); + + crate::MemStats { + alloc: process_memory, + total_alloc: sys.used_memory(), + mallocs: 0, + frees: 0, + heap_alloc: process_memory, + } +} + +fn available_parallelism_count() -> u64 { + std::thread::available_parallelism() + .map(|parallelism| usize_to_u64_saturated(parallelism.get())) + .unwrap_or(1) +} + +fn usize_to_u64_saturated(value: usize) -> u64 { + u64::try_from(value).unwrap_or(u64::MAX) +} + +fn frequency_to_mhz(frequency: u64) -> f64 { + f64::from(u32::try_from(frequency).unwrap_or(u32::MAX)) } #[cfg(test)] @@ -332,8 +442,10 @@ mod tests { fn test_get_cpus_function() { let cpus = get_cpus(); assert!(cpus.node_common.addr.is_empty()); - assert!(cpus.cpus.is_empty()); - assert!(cpus.cpu_freq_stats.is_empty()); + assert!(!cpus.cpus.is_empty()); + assert!(cpus.cpus.iter().any(|cpu| cpu.cores > 0)); + assert!(cpus.cpus.iter().any(|cpu| !cpu.model_name.is_empty())); + assert!(!cpus.cpu_freq_stats.is_empty()); } #[test] @@ -666,9 +778,11 @@ mod tests { #[test] fn test_get_mem_info_function() { let mem_info = get_mem_info("memory-server"); - assert!(mem_info.node_common.addr.is_empty()); - assert!(mem_info.total.is_none()); - assert!(mem_info.used.is_none()); + assert_eq!(mem_info.node_common.addr, "memory-server"); + let total = mem_info.total.expect("total memory should be collected"); + assert!(total > 0); + assert!(mem_info.used.is_some()); + assert!(mem_info.available.is_some()); } #[test] diff --git a/crates/protocols/src/swift/expiration_worker.rs b/crates/protocols/src/swift/expiration_worker.rs index f4c609b6e..473dccd8d 100644 --- a/crates/protocols/src/swift/expiration_worker.rs +++ b/crates/protocols/src/swift/expiration_worker.rs @@ -48,9 +48,12 @@ //! - Objects are assigned to workers based on hash(account + container + object) % max_workers //! - This prevents duplicate deletions and distributes load -use super::SwiftResult; +use super::container::ContainerMapper; +use super::object::ObjectKeyMapper; +use super::storage_api::object::ObjectOperations as _; +use super::{SwiftError, SwiftObjectOptions, SwiftResult, resolve_swift_object_store_handle}; use std::cmp::Reverse; -use std::collections::BinaryHeap; +use std::collections::{BinaryHeap, HashMap}; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; use tokio::sync::RwLock; @@ -64,6 +67,87 @@ const EVENT_SWIFT_EXPIRATION_OBJECT_TRACKING: &str = "swift_expiration_object_tr const EVENT_SWIFT_EXPIRATION_ITERATION_SUMMARY: &str = "swift_expiration_iteration_summary"; const EVENT_SWIFT_EXPIRATION_DELETE_STATE: &str = "swift_expiration_delete_state"; const EVENT_SWIFT_EXPIRATION_SCAN_STATE: &str = "swift_expiration_scan_state"; +const SWIFT_DELETE_AT_METADATA: &str = "x-delete-at"; + +#[async_trait::async_trait] +trait ExpirationObjectBackend: Send + Sync { + async fn object_metadata(&self, account: &str, container: &str, object: &str) + -> SwiftResult>>; + + async fn delete_object(&self, account: &str, container: &str, object: &str) -> SwiftResult<()>; +} + +#[derive(Debug, Default)] +struct SwiftStorageExpirationBackend; + +#[async_trait::async_trait] +impl ExpirationObjectBackend for SwiftStorageExpirationBackend { + async fn object_metadata( + &self, + account: &str, + container: &str, + object: &str, + ) -> SwiftResult>> { + let (bucket, object_key) = swift_storage_location(account, container, object)?; + let Some(store) = resolve_swift_object_store_handle() else { + return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string())); + }; + + let opts = SwiftObjectOptions::default(); + match store.get_object_info(&bucket, &object_key, &opts).await { + Ok(info) if info.delete_marker => Ok(None), + Ok(info) => Ok(Some(info.user_defined.as_ref().clone())), + Err(err) => { + let err_msg = err.to_string(); + if storage_error_is_not_found(&err_msg) { + Ok(None) + } else { + Err(storage_error("Object expiration metadata retrieval", err_msg)) + } + } + } + } + + async fn delete_object(&self, account: &str, container: &str, object: &str) -> SwiftResult<()> { + let (bucket, object_key) = swift_storage_location(account, container, object)?; + let Some(store) = resolve_swift_object_store_handle() else { + return Err(SwiftError::InternalServerError("Storage layer not initialized".to_string())); + }; + + store + .delete_object(&bucket, &object_key, SwiftObjectOptions::default()) + .await + .map(|_| ()) + .map_err(|err| storage_error("Object expiration deletion", err)) + } +} + +fn swift_storage_location(account: &str, container: &str, object: &str) -> SwiftResult<(String, String)> { + let project_id = account + .strip_prefix("AUTH_") + .ok_or_else(|| SwiftError::Unauthorized("Invalid account format".to_string()))?; + let object_key = ObjectKeyMapper::swift_to_s3_key(object)?; + let bucket = ContainerMapper::default().swift_to_s3_bucket(container, project_id); + Ok((bucket, object_key)) +} + +fn storage_error_is_not_found(message: &str) -> bool { + let normalized = message.to_ascii_lowercase(); + normalized.contains("does not exist") || normalized.contains("not found") +} + +fn storage_error(operation: &str, error: E) -> SwiftError { + error!( + event = EVENT_SWIFT_EXPIRATION_DELETE_STATE, + component = LOG_COMPONENT_PROTOCOLS, + subsystem = LOG_SUBSYSTEM_SWIFT_EXPIRATION, + operation = %operation, + error = %error, + result = "failed", + "swift expiration delete state changed" + ); + SwiftError::InternalServerError(format!("{operation} operation failed")) +} /// Configuration for expiration worker #[derive(Debug, Clone)] @@ -489,14 +573,20 @@ impl ExpirationWorker { /// - Ok(false) if object doesn't exist or expiration was removed /// - Err if deletion failed async fn delete_expired_object(account: &str, container: &str, object: &str, expected_expires_at: u64) -> SwiftResult { - // Note: This is a placeholder implementation - // In a real system, this would: - // 1. HEAD the object to verify it still exists and has X-Delete-At metadata - // 2. Check that X-Delete-At matches expected_expires_at (not modified) - // 3. DELETE the object - // 4. Handle errors (NotFound = Ok(false), others = Err) + Self::delete_expired_object_with_backend(&SwiftStorageExpirationBackend, account, container, object, expected_expires_at) + .await + } - // For now, we'll log the deletion + async fn delete_expired_object_with_backend( + backend: &B, + account: &str, + container: &str, + object: &str, + expected_expires_at: u64, + ) -> SwiftResult + where + B: ExpirationObjectBackend + ?Sized, + { debug!( event = EVENT_SWIFT_EXPIRATION_DELETE_STATE, component = LOG_COMPONENT_PROTOCOLS, @@ -509,17 +599,23 @@ impl ExpirationWorker { "swift expiration delete state changed" ); - // TODO: Integrate with actual object storage - // let info = object::head_object(account, container, object, &None).await?; - // if let Some(delete_at_str) = info.metadata.get("x-delete-at") { - // let delete_at = delete_at_str.parse::().unwrap_or(0); - // if delete_at == expected_expires_at && expiration::is_expired(delete_at) { - // object::delete_object(account, container, object, &None).await?; - // return Ok(true); - // } - // } + let Some(metadata) = backend.object_metadata(account, container, object).await? else { + return Ok(false); + }; - Ok(false) // Placeholder: object doesn't exist or expiration removed + let Some(delete_at) = metadata + .get(SWIFT_DELETE_AT_METADATA) + .and_then(|value| value.parse::().ok()) + else { + return Ok(false); + }; + + if delete_at != expected_expires_at || !super::expiration::is_expired(delete_at) { + return Ok(false); + } + + backend.delete_object(account, container, object).await?; + Ok(true) } /// Scan all objects and add those with expiration to tracking @@ -557,6 +653,79 @@ impl ExpirationWorker { #[cfg(test)] mod tests { use super::*; + use std::collections::VecDeque; + use std::sync::Mutex; + + #[derive(Default)] + struct MockExpirationObjectBackend { + metadata_results: Mutex>>>>, + delete_results: Mutex>>, + deleted_objects: Mutex>, + } + + impl MockExpirationObjectBackend { + fn with_metadata_result(result: SwiftResult>>) -> Self { + Self { + metadata_results: Mutex::new(VecDeque::from([result])), + ..Default::default() + } + } + + fn with_delete_result(self, result: SwiftResult<()>) -> Self { + self.delete_results + .lock() + .expect("delete results mutex should not be poisoned") + .push_back(result); + self + } + + fn deleted_objects(&self) -> Vec<(String, String, String)> { + self.deleted_objects + .lock() + .expect("deleted objects mutex should not be poisoned") + .clone() + } + } + + #[async_trait::async_trait] + impl ExpirationObjectBackend for MockExpirationObjectBackend { + async fn object_metadata( + &self, + _account: &str, + _container: &str, + _object: &str, + ) -> SwiftResult>> { + self.metadata_results + .lock() + .expect("metadata results mutex should not be poisoned") + .pop_front() + .expect("metadata result should be queued") + } + + async fn delete_object(&self, account: &str, container: &str, object: &str) -> SwiftResult<()> { + self.deleted_objects + .lock() + .expect("deleted objects mutex should not be poisoned") + .push((account.to_string(), container.to_string(), object.to_string())); + + self.delete_results + .lock() + .expect("delete results mutex should not be poisoned") + .pop_front() + .unwrap_or(Ok(())) + } + } + + fn now_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time should be after unix epoch") + .as_secs() + } + + fn metadata_with_delete_at(delete_at: u64) -> HashMap { + HashMap::from([(SWIFT_DELETE_AT_METADATA.to_string(), delete_at.to_string())]) + } #[test] fn test_expiration_entry_ordering() { @@ -714,4 +883,80 @@ mod tests { let metrics = worker.get_metrics().await; assert_eq!(metrics.queue_size, 2); } + + #[tokio::test] + async fn test_delete_expired_object_deletes_when_expired_metadata_matches() { + let expires_at = now_secs().saturating_sub(1); + let backend = MockExpirationObjectBackend::with_metadata_result(Ok(Some(metadata_with_delete_at(expires_at)))); + + let deleted = + ExpirationWorker::delete_expired_object_with_backend(&backend, "AUTH_test", "container", "object", expires_at) + .await + .expect("matching expired object should delete successfully"); + + assert!(deleted); + assert_eq!( + backend.deleted_objects(), + vec![("AUTH_test".to_string(), "container".to_string(), "object".to_string())] + ); + } + + #[tokio::test] + async fn test_delete_expired_object_skips_when_metadata_changed() { + let expires_at = now_secs().saturating_sub(1); + let backend = + MockExpirationObjectBackend::with_metadata_result(Ok(Some(metadata_with_delete_at(expires_at.saturating_sub(1))))); + + let deleted = + ExpirationWorker::delete_expired_object_with_backend(&backend, "AUTH_test", "container", "object", expires_at) + .await + .expect("changed expiration metadata should skip deletion"); + + assert!(!deleted); + assert!(backend.deleted_objects().is_empty()); + } + + #[tokio::test] + async fn test_delete_expired_object_skips_when_not_expired() { + let expires_at = now_secs() + 3600; + let backend = MockExpirationObjectBackend::with_metadata_result(Ok(Some(metadata_with_delete_at(expires_at)))); + + let deleted = + ExpirationWorker::delete_expired_object_with_backend(&backend, "AUTH_test", "container", "object", expires_at) + .await + .expect("future expiration should skip deletion"); + + assert!(!deleted); + assert!(backend.deleted_objects().is_empty()); + } + + #[tokio::test] + async fn test_delete_expired_object_skips_when_object_missing() { + let expires_at = now_secs().saturating_sub(1); + let backend = MockExpirationObjectBackend::with_metadata_result(Ok(None)); + + let deleted = + ExpirationWorker::delete_expired_object_with_backend(&backend, "AUTH_test", "container", "object", expires_at) + .await + .expect("missing object should skip deletion"); + + assert!(!deleted); + assert!(backend.deleted_objects().is_empty()); + } + + #[tokio::test] + async fn test_delete_expired_object_returns_error_when_delete_fails() { + let expires_at = now_secs().saturating_sub(1); + let backend = MockExpirationObjectBackend::with_metadata_result(Ok(Some(metadata_with_delete_at(expires_at)))) + .with_delete_result(Err(SwiftError::InternalServerError("delete failed".to_string()))); + + let result = + ExpirationWorker::delete_expired_object_with_backend(&backend, "AUTH_test", "container", "object", expires_at).await; + + assert!(matches!(result, Err(SwiftError::InternalServerError(_)))); + assert_eq!( + backend.deleted_objects(), + vec![("AUTH_test".to_string(), "container".to_string(), "object".to_string())] + ); + } }