diff --git a/Cargo.lock b/Cargo.lock index f4884dc5a..c949e65fe 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9955,8 +9955,11 @@ dependencies = [ name = "rustfs-test-utils" version = "1.0.0-beta.10" dependencies = [ + "rustfs-data-usage", "rustfs-ecstore", "rustfs-storage-api", + "serde_json", + "tempfile", "tokio", "tokio-util", "tracing-subscriber", diff --git a/crates/ecstore/src/config/com.rs b/crates/ecstore/src/config/com.rs index 5733b26f3..a32662f6a 100644 --- a/crates/ecstore/src/config/com.rs +++ b/crates/ecstore/src/config/com.rs @@ -279,6 +279,17 @@ where Ok(data) } +/// Read an existing config object without treating an empty payload as absent. +/// Callers that validate their own payload format need to distinguish corruption +/// from `ConfigNotFound`. +pub(crate) async fn read_config_preserve_empty(api: Arc, file: &str) -> Result> +where + S: EcstoreObjectIO, +{ + let (data, _obj) = read_config_with_metadata_inner(api, file, &ObjectOptions::default(), true).await?; + Ok(data) +} + pub async fn read_config_no_lock(api: Arc, file: &str) -> Result> where S: ObjectIO< @@ -304,6 +315,26 @@ where } pub async fn read_config_with_metadata(api: Arc, file: &str, opts: &ObjectOptions) -> Result<(Vec, ObjectInfo)> +where + S: ObjectIO< + Error = Error, + RangeSpec = HTTPRangeSpec, + HeaderMap = HeaderMap, + ObjectOptions = ObjectOptions, + ObjectInfo = ObjectInfo, + GetObjectReader = GetObjectReader, + PutObjectReader = PutObjReader, + >, +{ + read_config_with_metadata_inner(api, file, opts, false).await +} + +async fn read_config_with_metadata_inner( + api: Arc, + file: &str, + opts: &ObjectOptions, + preserve_empty: bool, +) -> Result<(Vec, ObjectInfo)> where S: ObjectIO< Error = Error, @@ -330,7 +361,7 @@ where let data = rd.read_all().await?; - if data.is_empty() { + if data.is_empty() && !preserve_empty { return Err(Error::ConfigNotFound); } @@ -1596,7 +1627,7 @@ where mod tests { use super::{ configs_semantically_equal, decode_server_config_blob, encode_server_config_blob, is_standard_object_server_config, - read_config_with_metadata, storage_class_kvs_mut, + read_config, read_config_preserve_empty, read_config_with_metadata, storage_class_kvs_mut, }; use crate::config::{audit, notify, oidc}; use crate::disk::endpoint::Endpoint; @@ -3059,6 +3090,21 @@ mod tests { } } + #[tokio::test] + async fn read_config_preserve_empty_distinguishes_empty_object() { + let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(Vec::new()), None)); + + let err = read_config(store.clone(), "config/empty.json") + .await + .expect_err("the existing config contract treats empty objects as missing"); + assert!(matches!(err, Error::ConfigNotFound)); + + let data = read_config_preserve_empty(store, "config/empty.json") + .await + .expect("payload-validating callers must observe the empty object"); + assert!(data.is_empty()); + } + #[async_trait::async_trait] impl StorageAdminApi for RecoveryMockStore { type BackendInfo = rustfs_madmin::BackendInfo; diff --git a/crates/ecstore/src/data_usage/mod.rs b/crates/ecstore/src/data_usage/mod.rs index a67e69d57..6b41c994e 100644 --- a/crates/ecstore/src/data_usage/mod.rs +++ b/crates/ecstore/src/data_usage/mod.rs @@ -24,7 +24,7 @@ use crate::storage_api_contracts::{ }; use crate::{ bucket::{metadata_sys::get_replication_config, versioning::VersioningApi as _, versioning_sys::BucketVersioningSys}, - config::com::read_config, + config::com::{read_config, read_config_preserve_empty}, disk::DiskAPI, error::{Error, classify_system_path_failure_reason}, object_api::ObjectInfo, @@ -83,8 +83,47 @@ static LIVE_BUCKET_USAGE_CACHE: OnceLock = OnceLock::new() /// endpoints for up to `DATA_USAGE_CACHE_TTL_SECS` between backend reads. #[derive(Debug, Clone)] struct CachedDataUsageSnapshot { - info: DataUsageInfo, - loaded_at: SystemTime, + info: Option, + loaded_at: tokio::time::Instant, +} + +impl CachedDataUsageSnapshot { + fn result(&self) -> Result { + self.info + .clone() + .ok_or_else(|| Error::other("data usage snapshot load recently failed")) + } +} + +fn fresh_cached_data_usage_snapshot( + cache: &Option, + now: tokio::time::Instant, + ttl: Duration, +) -> Option> { + cache + .as_ref() + .filter(|cached| now.duration_since(cached.loaded_at) < ttl) + .map(CachedDataUsageSnapshot::result) +} + +fn cache_data_usage_snapshot_result( + cache: &mut Option, + result: Result, + loaded_at: tokio::time::Instant, +) -> Result { + match result { + Ok(info) => { + *cache = Some(CachedDataUsageSnapshot { + info: Some(info.clone()), + loaded_at, + }); + Ok(info) + } + Err(e) => { + *cache = Some(CachedDataUsageSnapshot { info: None, loaded_at }); + Err(e) + } + } } type DataUsageSnapshotCache = Arc>>; @@ -164,6 +203,7 @@ lazy_static::lazy_static! { SLASH_SEPARATOR, DATA_USAGE_OBJ_NAME ); + static ref DATA_USAGE_OBJ_BACKUP_PATH: String = format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()); pub static ref DATA_USAGE_BLOOM_NAME_PATH: String = format!("{}{}{}", crate::disk::BUCKET_META_PREFIX, SLASH_SEPARATOR, @@ -195,13 +235,33 @@ fn stale_data_usage_persist_reason(incoming: &DataUsageInfo, existing: &DataUsag } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum UsageSnapshotSource { + Primary, + Backup, + Missing, +} + +fn stale_data_usage_persist_reason_for_source( + incoming: &DataUsageInfo, + existing: &DataUsageInfo, + source: UsageSnapshotSource, + now: SystemTime, +) -> Option<&'static str> { + let reason = stale_data_usage_persist_reason(incoming, existing, now); + if source == UsageSnapshotSource::Backup && incoming.last_update == existing.last_update { + None + } else { + reason + } +} + /// Store data usage info to backend storage #[instrument(skip(store))] pub async fn store_data_usage_in_backend(data_usage_info: DataUsageInfo, store: Arc) -> Result<(), Error> { // Prevent older data from overwriting newer persisted stats - if let Ok(buf) = read_config(store.clone(), &DATA_USAGE_OBJ_NAME_PATH).await - && let Ok(existing) = serde_json::from_slice::(&buf) - && let Some(reason) = stale_data_usage_persist_reason(&data_usage_info, &existing, SystemTime::now()) + if let Ok((existing, source)) = load_data_usage_snapshot(store.clone()).await + && let Some(reason) = stale_data_usage_persist_reason_for_source(&data_usage_info, &existing, source, SystemTime::now()) { info!( "Skip persisting data usage ({reason}): incoming last_update {:?} <= existing {:?}", @@ -277,8 +337,8 @@ pub async fn remove_bucket_usage_from_backend(store: Arc, bucket: &str) live_bucket_usage_cache().invalidate(bucket).await; clear_bucket_usage_memory(bucket).await; - let data_usage_info = load_data_usage_from_backend(store.clone()).await?; - let existing = load_data_usage_from_backend(store.clone()).await.ok(); + let data_usage_info = load_primary_data_usage_from_backend(store.clone()).await?; + let existing = load_primary_data_usage_from_backend(store.clone()).await.ok(); if let Some(data_usage_info) = merge_bucket_usage_removal(data_usage_info, existing, bucket) { save_data_usage_in_backend(data_usage_info, store).await?; @@ -287,47 +347,123 @@ pub async fn remove_bucket_usage_from_backend(store: Arc, bucket: &str) Ok(()) } +fn record_usage_snapshot_failure(operation: &'static str, object: &str, e: &Error) { + let reason = classify_system_path_failure_reason(e); + record_system_path_failure("data_usage", operation, reason); + error!( + event = "data_usage_snapshot_load_failed", + component = "ecstore", + subsystem = "data_usage", + state = "read_failed", + operation, + reason, + object = %object, + error = %e, + "data usage snapshot load failed" + ); +} + +fn record_usage_snapshot_decode_failure(operation: &'static str, object: &str, e: &Error) { + const REASON: &str = "decode_error"; + record_system_path_failure("data_usage", operation, REASON); + error!( + event = "data_usage_snapshot_load_failed", + component = "ecstore", + subsystem = "data_usage", + state = "decode_failed", + operation, + reason = REASON, + object = %object, + error = %e, + "data usage snapshot load failed" + ); +} + +fn parse_usage_snapshot(buf: &[u8]) -> Result { + serde_json::from_slice(buf).map_err(|e| { + Error::other(format!( + "Failed to deserialize data usage info: {:?} at line {} column {}", + e.classify(), + e.line(), + e.column() + )) + }) +} + +/// Decide which usage snapshot to serve: the primary `.usage.json` or its +/// `.bkp` backup. The backup future is polled only when the primary is unusable, +/// so the healthy path performs a single read. +/// +/// `Ok(DataUsageInfo::default())` is returned only when BOTH the primary and +/// the backup are `ConfigNotFound` — a genuine "no snapshot yet". A real +/// primary failure with a missing backup propagates as an error instead of +/// being rendered as confirmed all-zero stats. +async fn resolve_loaded_snapshot_with_source( + primary: Result, Error>, + backup: impl Future, Error>>, +) -> Result<(DataUsageInfo, UsageSnapshotSource), Error> { + let primary_failure = match primary { + Ok(buf) => match parse_usage_snapshot(&buf) { + Ok(info) => return Ok((info, UsageSnapshotSource::Primary)), + Err(e) => { + record_usage_snapshot_decode_failure("parse_primary", DATA_USAGE_OBJ_NAME_PATH.as_str(), &e); + e + } + }, + Err(e) => { + if e != Error::ConfigNotFound { + record_usage_snapshot_failure("read_primary", DATA_USAGE_OBJ_NAME_PATH.as_str(), &e); + } + e + } + }; + match backup.await { + Ok(buf) => match parse_usage_snapshot(&buf) { + Ok(info) => Ok((info, UsageSnapshotSource::Backup)), + Err(e) => { + record_usage_snapshot_decode_failure("parse_backup", &DATA_USAGE_OBJ_BACKUP_PATH, &e); + Err(e) + } + }, + Err(Error::ConfigNotFound) if primary_failure == Error::ConfigNotFound => { + Ok((DataUsageInfo::default(), UsageSnapshotSource::Missing)) + } + Err(Error::ConfigNotFound) => Err(primary_failure), + Err(e) => { + record_usage_snapshot_failure("read_backup", &DATA_USAGE_OBJ_BACKUP_PATH, &e); + Err(e) + } + } +} + +async fn resolve_loaded_snapshot( + primary: Result, Error>, + backup: impl Future, Error>>, +) -> Result { + Ok(resolve_loaded_snapshot_with_source(primary, backup).await?.0) +} + +async fn load_data_usage_snapshot(store: Arc) -> Result<(DataUsageInfo, UsageSnapshotSource), Error> { + let primary = read_config_preserve_empty(store.clone(), &DATA_USAGE_OBJ_NAME_PATH).await; + resolve_loaded_snapshot_with_source( + primary, + async move { read_config_preserve_empty(store, &DATA_USAGE_OBJ_BACKUP_PATH).await }, + ) + .await +} + +async fn load_primary_data_usage_from_backend(store: Arc) -> Result { + let buf = read_config_preserve_empty(store, &DATA_USAGE_OBJ_NAME_PATH).await?; + Ok(normalize_loaded_data_usage(parse_usage_snapshot(&buf)?).await) +} + /// Load data usage info from backend storage #[instrument(skip(store))] pub async fn load_data_usage_from_backend(store: Arc) -> Result { - let buf: Vec = match read_config(store.clone(), &DATA_USAGE_OBJ_NAME_PATH).await { - Ok(data) => data, - Err(e) => { - let reason = classify_system_path_failure_reason(&e); - record_system_path_failure("data_usage", "read_primary", reason); - error!( - path_kind = "data_usage", - operation = "read_primary", - reason, - object = %DATA_USAGE_OBJ_NAME_PATH.as_str(), - error = %e, - "system path read failed" - ); - - match read_config(store.clone(), format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()).as_str()).await { - Ok(data) => data, - Err(e) => { - if e == Error::ConfigNotFound { - return Ok(DataUsageInfo::default()); - } - let reason = classify_system_path_failure_reason(&e); - record_system_path_failure("data_usage", "read_backup", reason); - error!( - path_kind = "data_usage", - operation = "read_backup", - reason, - object = %format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()), - error = %e, - "system path read failed" - ); - return Err(Error::other(e)); - } - } - } - }; - let mut data_usage_info: DataUsageInfo = - serde_json::from_slice(&buf).map_err(|e| Error::other(format!("Failed to deserialize data usage info: {e}")))?; + Ok(normalize_loaded_data_usage(load_data_usage_snapshot(store).await?.0).await) +} +async fn normalize_loaded_data_usage(mut data_usage_info: DataUsageInfo) -> DataUsageInfo { info!("Loaded data usage info from backend with {} buckets", data_usage_info.buckets_count); // Handle backward compatibility @@ -378,7 +514,7 @@ pub async fn load_data_usage_from_backend(store: Arc) -> Result) -> Result< { let cache = data_usage_snapshot_cache().read().await; - if let Some(cached) = cache.as_ref() - && SystemTime::now().duration_since(cached.loaded_at).unwrap_or_default() < ttl - { - return Ok(cached.info.clone()); + if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl) { + return result; } } // Re-check under the write lock so concurrent expirations trigger a single // backend read instead of a stampede. let mut cache = data_usage_snapshot_cache().write().await; - if let Some(cached) = cache.as_ref() - && SystemTime::now().duration_since(cached.loaded_at).unwrap_or_default() < ttl - { - return Ok(cached.info.clone()); + if let Some(result) = fresh_cached_data_usage_snapshot(&cache, tokio::time::Instant::now(), ttl) { + return result; } - let info = load_data_usage_from_backend(store).await?; - *cache = Some(CachedDataUsageSnapshot { - info: info.clone(), - loaded_at: SystemTime::now(), - }); - Ok(info) + let result = load_data_usage_from_backend(store).await; + cache_data_usage_snapshot_result(&mut cache, result, tokio::time::Instant::now()) } /// Aggregate usage information from local disk snapshots. @@ -1012,7 +1140,7 @@ async fn update_usage_cache_if_needed() { let updating_clone = (*cache_updating()).clone(); tokio::spawn(async move { if let Some(store) = runtime_sources::object_store_handle() - && let Ok(data_usage_info) = load_data_usage_from_backend(store.clone()).await + && let Ok(data_usage_info) = load_data_usage_from_backend(store).await { replace_bucket_usage_memory_from_info(&data_usage_info).await; } @@ -1036,7 +1164,7 @@ async fn update_usage_cache_if_needed() { drop(updating); if let Some(store) = runtime_sources::object_store_handle() - && let Ok(data_usage_info) = load_data_usage_from_backend(store.clone()).await + && let Ok(data_usage_info) = load_data_usage_from_backend(store).await { replace_bucket_usage_memory_from_info(&data_usage_info).await; } @@ -1536,6 +1664,175 @@ mod tests { assert_eq!(stale_data_usage_persist_reason(&both_none, &usage_with_last_update(None), now), None); } + #[test] + fn backup_stale_guard_allows_equal_snapshot_to_repair_primary() { + let now = SystemTime::UNIX_EPOCH + Duration::from_secs(1_000_000); + let existing = usage_with_last_update(Some(now - Duration::from_secs(60))); + let equal = usage_with_last_update(existing.last_update); + let older = usage_with_last_update(Some(now - Duration::from_secs(120))); + + assert_eq!( + stale_data_usage_persist_reason_for_source(&equal, &existing, UsageSnapshotSource::Backup, now), + None + ); + assert_eq!( + stale_data_usage_persist_reason_for_source(&older, &existing, UsageSnapshotSource::Backup, now), + Some("older_or_equal_last_update") + ); + } + + fn snapshot_bytes(bucket: &str) -> Vec { + let info = data_usage_info_for_test(bucket, 3, 42, SystemTime::UNIX_EPOCH); + serde_json::to_vec(&info).expect("test snapshot must serialize") + } + + #[test] + fn data_usage_backup_path_appends_suffix_to_primary() { + assert_eq!(DATA_USAGE_OBJ_BACKUP_PATH.as_str(), format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str())); + } + + fn assert_snapshot_bucket(info: &DataUsageInfo, bucket: &str) { + assert!( + info.buckets_usage.contains_key(bucket), + "expected snapshot for bucket {bucket}, got buckets {:?}", + info.buckets_usage.keys().collect::>() + ); + } + + #[test] + fn parse_snapshot_error_does_not_include_payload_value() { + let err = + parse_usage_snapshot(br#"{"buckets_count":"secret-marker"}"#).expect_err("an invalid field type must fail decoding"); + assert!(!err.to_string().contains("secret-marker")); + } + + #[test] + fn cached_snapshot_failure_is_reused_until_ttl_expires() { + let loaded_at = tokio::time::Instant::now(); + let mut cache = None; + + let first = cache_data_usage_snapshot_result(&mut cache, Err(Error::ErasureReadQuorum), loaded_at); + assert!(matches!(first, Err(Error::ErasureReadQuorum))); + + let cached = fresh_cached_data_usage_snapshot(&cache, loaded_at + Duration::from_secs(1), Duration::from_secs(30)) + .expect("the failed load must remain cached within the TTL"); + assert!(cached.is_err()); + + assert!(fresh_cached_data_usage_snapshot(&cache, loaded_at + Duration::from_secs(30), Duration::from_secs(30)).is_none()); + } + + #[test] + fn cached_snapshot_success_is_reused_until_ttl_expires() { + let loaded_at = tokio::time::Instant::now(); + let expected = data_usage_info_for_test("bucket", 3, 42, SystemTime::UNIX_EPOCH); + let mut cache = None; + + let first = + cache_data_usage_snapshot_result(&mut cache, Ok(expected), loaded_at).expect("successful load must be returned"); + assert_snapshot_bucket(&first, "bucket"); + + let cached = fresh_cached_data_usage_snapshot(&cache, loaded_at + Duration::from_secs(1), Duration::from_secs(30)) + .expect("successful load must remain cached within the TTL") + .expect("cached success must remain successful"); + assert_snapshot_bucket(&cached, "bucket"); + } + + #[tokio::test] + async fn resolve_snapshot_primary_ok_is_used_without_backup_read() { + let backup_read = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let backup_read_probe = Arc::clone(&backup_read); + + let info = resolve_loaded_snapshot(Ok(snapshot_bytes("primary-bucket")), async move { + backup_read_probe.store(true, std::sync::atomic::Ordering::SeqCst); + Ok(snapshot_bytes("backup-bucket")) + }) + .await + .expect("healthy primary snapshot must load"); + + assert_snapshot_bucket(&info, "primary-bucket"); + assert!( + !backup_read.load(std::sync::atomic::Ordering::SeqCst), + "backup must not be read when the primary parses" + ); + } + + #[tokio::test] + async fn resolve_snapshot_primary_corrupt_falls_back_to_backup() { + let (info, source) = + resolve_loaded_snapshot_with_source(Ok(b"{not json".to_vec()), async { Ok(snapshot_bytes("backup-bucket")) }) + .await + .expect("backup snapshot must be served when the primary is corrupt"); + + assert_snapshot_bucket(&info, "backup-bucket"); + assert_eq!(source, UsageSnapshotSource::Backup); + } + + #[tokio::test] + async fn resolve_snapshot_primary_corrupt_backup_corrupt_is_error() { + let err = resolve_loaded_snapshot(Ok(b"{not json".to_vec()), async { Ok(b"also not json".to_vec()) }) + .await + .expect_err("two corrupt snapshots must not produce stats"); + assert!(err.to_string().contains("deserialize"), "unexpected error: {err}"); + } + + #[tokio::test] + async fn resolve_snapshot_primary_corrupt_backup_missing_is_error() { + let err = resolve_loaded_snapshot(Ok(b"{not json".to_vec()), async { Err(Error::ConfigNotFound) }) + .await + .expect_err("a corrupt primary without a backup must not produce stats"); + assert!(err.to_string().contains("deserialize"), "unexpected error: {err}"); + } + + #[tokio::test] + async fn resolve_snapshot_primary_empty_backup_missing_is_error() { + let err = resolve_loaded_snapshot(Ok(Vec::new()), async { Err(Error::ConfigNotFound) }) + .await + .expect_err("an empty primary object is corrupt, not an absent snapshot"); + assert!(err.to_string().contains("deserialize"), "unexpected error: {err}"); + } + + #[tokio::test] + async fn resolve_snapshot_primary_read_error_falls_back_to_backup() { + let info = resolve_loaded_snapshot(Err(Error::ErasureReadQuorum), async { Ok(snapshot_bytes("backup-bucket")) }) + .await + .expect("backup snapshot must be served when the primary read fails"); + + assert_snapshot_bucket(&info, "backup-bucket"); + } + + #[tokio::test] + async fn resolve_snapshot_primary_real_error_backup_missing_is_error_not_zeros() { + let err = resolve_loaded_snapshot(Err(Error::ErasureReadQuorum), async { Err(Error::ConfigNotFound) }) + .await + .expect_err("primary corruption must not be rendered as all-zero stats"); + assert!(matches!(err, Error::ErasureReadQuorum), "unexpected error: {err}"); + } + + #[tokio::test] + async fn resolve_snapshot_primary_missing_uses_backup() { + let info = resolve_loaded_snapshot(Err(Error::ConfigNotFound), async { Ok(snapshot_bytes("backup-bucket")) }) + .await + .expect("an existing backup must be used when the primary is missing"); + assert_snapshot_bucket(&info, "backup-bucket"); + } + + #[tokio::test] + async fn resolve_snapshot_backup_read_error_is_preserved() { + let err = resolve_loaded_snapshot(Err(Error::ConfigNotFound), async { Err(Error::ErasureReadQuorum) }) + .await + .expect_err("a backup read failure must be returned"); + assert!(matches!(err, Error::ErasureReadQuorum), "unexpected error: {err}"); + } + + #[tokio::test] + async fn resolve_snapshot_both_missing_is_default() { + let info = resolve_loaded_snapshot(Err(Error::ConfigNotFound), async { Err(Error::ConfigNotFound) }) + .await + .expect("no snapshot yet must resolve to empty stats"); + assert_eq!(info.buckets_count, 0); + assert!(info.buckets_usage.is_empty()); + } + #[tokio::test] async fn compute_usage_preserves_same_object_across_1000_entry_page_boundary() { let first_page = UsageVersionPage { diff --git a/crates/ecstore/src/diagnostics/admin_server_info.rs b/crates/ecstore/src/diagnostics/admin_server_info.rs index 31cc656c9..91e76c5e8 100644 --- a/crates/ecstore/src/diagnostics/admin_server_info.rs +++ b/crates/ecstore/src/diagnostics/admin_server_info.rs @@ -13,7 +13,7 @@ // limitations under the License. use crate::cluster::rpc::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client}; -use crate::data_usage::{DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, load_data_usage_from_backend}; +use crate::data_usage::{DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT, load_data_usage_from_backend_cached}; use crate::error::{Error, Result}; use crate::{disk::endpoint::Endpoint, runtime::sources as runtime_sources}; @@ -42,6 +42,33 @@ use shadow_rs::shadow; shadow!(build); const SERVER_PING_TIMEOUT: Duration = Duration::from_secs(1); +const DATA_USAGE_UNAVAILABLE_ERROR: &str = "data usage snapshot unavailable"; + +fn apply_data_usage_result( + result: Result, + buckets: &mut rustfs_madmin::Buckets, + objects: &mut rustfs_madmin::Objects, + versions: &mut rustfs_madmin::Versions, + delete_markers: &mut rustfs_madmin::DeleteMarkers, + usage: &mut rustfs_madmin::Usage, +) { + match result { + Ok(info) => { + buckets.count = info.buckets_count; + objects.count = info.objects_total_count; + versions.count = info.versions_total_count; + delete_markers.count = info.delete_markers_total_count; + usage.size = info.objects_total_size; + } + Err(_) => { + buckets.error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()); + objects.error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()); + versions.error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()); + delete_markers.error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()); + usage.error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()); + } + } +} // pub const ITEM_OFFLINE: &str = "offline"; // pub const ITEM_INITIALIZING: &str = "initializing"; @@ -246,22 +273,14 @@ pub async fn get_server_info(get_pools: bool) -> InfoMessage { if let Some(store) = runtime_sources::object_store_handle() { mode = ITEM_ONLINE; - match load_data_usage_from_backend(store.clone()).await { - Ok(res) => { - buckets.count = res.buckets_count; - objects.count = res.objects_total_count; - versions.count = res.versions_total_count; - delete_markers.count = res.delete_markers_total_count; - usage.size = res.objects_total_size; - } - Err(err) => { - buckets.error = Some(err.to_string()); - objects.error = Some(err.to_string()); - versions.error = Some(err.to_string()); - delete_markers.error = Some(err.to_string()); - usage.error = Some(err.to_string()); - } - } + apply_data_usage_result( + load_data_usage_from_backend_cached(store.clone()).await, + &mut buckets, + &mut objects, + &mut versions, + &mut delete_markers, + &mut usage, + ); let after3 = OffsetDateTime::now_utc(); @@ -433,7 +452,10 @@ mod tests { use crate::runtime::sources as runtime_sources; use rustfs_madmin::{Disk, ITEM_OFFLINE, ITEM_UNKNOWN}; - use super::{get_local_server_property, get_online_offline_disks_stats, get_server_info}; + use super::{ + DATA_USAGE_UNAVAILABLE_ERROR, apply_data_usage_result, get_local_server_property, get_online_offline_disks_stats, + get_server_info, + }; fn disk_with_state(endpoint: &str, state: &str) -> Disk { Disk { @@ -471,6 +493,55 @@ mod tests { ); } + #[test] + fn data_usage_errors_are_sanitized_in_server_info() { + let mut buckets = rustfs_madmin::Buckets::default(); + let mut objects = rustfs_madmin::Objects::default(); + let mut versions = rustfs_madmin::Versions::default(); + let mut delete_markers = rustfs_madmin::DeleteMarkers::default(); + let mut usage = rustfs_madmin::Usage::default(); + + apply_data_usage_result( + Err(crate::error::Error::other("sensitive disk path")), + &mut buckets, + &mut objects, + &mut versions, + &mut delete_markers, + &mut usage, + ); + + assert_eq!(buckets.error.as_deref(), Some(DATA_USAGE_UNAVAILABLE_ERROR)); + assert_eq!(objects.error.as_deref(), Some(DATA_USAGE_UNAVAILABLE_ERROR)); + assert_eq!(versions.error.as_deref(), Some(DATA_USAGE_UNAVAILABLE_ERROR)); + assert_eq!(delete_markers.error.as_deref(), Some(DATA_USAGE_UNAVAILABLE_ERROR)); + assert_eq!(usage.error.as_deref(), Some(DATA_USAGE_UNAVAILABLE_ERROR)); + } + + #[test] + fn data_usage_counts_are_mapped_into_server_info() { + let mut buckets = rustfs_madmin::Buckets::default(); + let mut objects = rustfs_madmin::Objects::default(); + let mut versions = rustfs_madmin::Versions::default(); + let mut delete_markers = rustfs_madmin::DeleteMarkers::default(); + let mut usage = rustfs_madmin::Usage::default(); + let info = rustfs_data_usage::DataUsageInfo { + buckets_count: 2, + objects_total_count: 3, + versions_total_count: 4, + delete_markers_total_count: 5, + objects_total_size: 6, + ..Default::default() + }; + + apply_data_usage_result(Ok(info), &mut buckets, &mut objects, &mut versions, &mut delete_markers, &mut usage); + + assert_eq!(buckets.count, 2); + assert_eq!(objects.count, 3); + assert_eq!(versions.count, 4); + assert_eq!(delete_markers.count, 5); + assert_eq!(usage.size, 6); + } + #[serial] #[tokio::test] async fn server_info_includes_global_deployment_id() { diff --git a/crates/test-utils/Cargo.toml b/crates/test-utils/Cargo.toml index 3fa408370..81d2a53c0 100644 --- a/crates/test-utils/Cargo.toml +++ b/crates/test-utils/Cargo.toml @@ -33,6 +33,11 @@ tokio-util = { workspace = true, features = ["io", "compat"] } tracing-subscriber = { workspace = true, features = ["env-filter", "time"] } uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] } +[dev-dependencies] +rustfs-data-usage = { workspace = true } +serde_json = { workspace = true } +tempfile = { workspace = true } + [lints] workspace = true diff --git a/crates/test-utils/src/data_usage_snapshot_tests.rs b/crates/test-utils/src/data_usage_snapshot_tests.rs new file mode 100644 index 000000000..94ed7694a --- /dev/null +++ b/crates/test-utils/src/data_usage_snapshot_tests.rs @@ -0,0 +1,132 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use rustfs_data_usage::{BucketUsageInfo, DataUsageInfo}; +use std::time::{Duration, SystemTime}; + +use crate::TestECStoreEnv; +use crate::ecstore_test_compat::fixture::{ + BUCKET_META_PREFIX, delete_config, load_data_usage_from_backend, load_data_usage_from_backend_cached, read_config, + remove_bucket_usage_from_backend, save_config, store_data_usage_in_backend, +}; + +fn snapshot(bucket: &str, last_update: SystemTime) -> DataUsageInfo { + let mut info = DataUsageInfo { + last_update: Some(last_update), + ..Default::default() + }; + info.buckets_usage.insert( + bucket.to_string(), + BucketUsageInfo { + objects_count: 3, + size: 42, + ..Default::default() + }, + ); + info.bucket_sizes.insert(bucket.to_string(), 42); + info.buckets_count = 1; + info.calculate_totals(); + info +} + +#[tokio::test] +async fn data_usage_snapshot_recovery_wires_primary_backup_and_negative_cache() { + let temp_root = tempfile::tempdir().expect("create test temp root"); + let env = TestECStoreEnv::builder() + .base_dir(temp_root.path()) + .init_bucket_metadata(false) + .build() + .await; + let primary = format!("{BUCKET_META_PREFIX}/.usage.json"); + let backup = format!("{primary}.bkp"); + let last_update = SystemTime::UNIX_EPOCH + Duration::from_secs(100); + let expected = snapshot("bucket-a", last_update); + + store_data_usage_in_backend(expected.clone(), env.ecstore.clone()) + .await + .expect("seed primary snapshot"); + let valid = read_config(env.ecstore.clone(), &primary).await.expect("read seeded primary"); + save_config(env.ecstore.clone(), &backup, valid.clone()) + .await + .expect("seed backup snapshot"); + + let corrupt = b"{not json".to_vec(); + save_config(env.ecstore.clone(), &primary, corrupt.clone()) + .await + .expect("corrupt primary snapshot"); + let recovered = load_data_usage_from_backend(env.ecstore.clone()) + .await + .expect("public loader must fall back to backup"); + assert!(recovered.buckets_usage.contains_key("bucket-a")); + + store_data_usage_in_backend(expected, env.ecstore.clone()) + .await + .expect("equal backup timestamp must repair primary"); + let repaired = read_config(env.ecstore.clone(), &primary) + .await + .expect("read repaired primary"); + serde_json::from_slice::(&repaired).expect("primary must contain a valid snapshot after repair"); + + save_config(env.ecstore.clone(), &primary, corrupt.clone()) + .await + .expect("corrupt primary before destructive update"); + remove_bucket_usage_from_backend(env.ecstore.clone(), "bucket-a") + .await + .expect_err("destructive updates must not promote a backup snapshot"); + assert_eq!( + read_config(env.ecstore.clone(), &primary) + .await + .expect("read corrupt primary"), + corrupt + ); + + delete_config(env.ecstore.clone(), &backup) + .await + .expect("remove backup before empty-primary check"); + save_config(env.ecstore.clone(), &primary, Vec::new()) + .await + .expect("write empty primary snapshot"); + load_data_usage_from_backend(env.ecstore.clone()) + .await + .expect_err("an empty primary with no backup must not look missing"); + + delete_config(env.ecstore.clone(), &primary) + .await + .expect("remove primary before empty-backup check"); + save_config(env.ecstore.clone(), &backup, Vec::new()) + .await + .expect("write empty backup snapshot"); + load_data_usage_from_backend(env.ecstore.clone()) + .await + .expect_err("an empty backup must not look missing"); + + delete_config(env.ecstore.clone(), &backup) + .await + .expect("remove backup before negative-cache check"); + save_config(env.ecstore.clone(), &primary, corrupt.clone()) + .await + .expect("restore corrupt primary before negative-cache check"); + load_data_usage_from_backend_cached(env.ecstore.clone()) + .await + .expect_err("corrupt primary without backup must fail"); + save_config(env.ecstore.clone(), &primary, valid) + .await + .expect("repair primary without invalidating the in-process cache"); + load_data_usage_from_backend(env.ecstore.clone()) + .await + .expect("direct load must observe repaired primary"); + load_data_usage_from_backend_cached(env.ecstore.clone()) + .await + .expect_err("cached loader must reuse the recent failure until its TTL expires"); +} diff --git a/crates/test-utils/src/ecstore_test_compat.rs b/crates/test-utils/src/ecstore_test_compat.rs index cfb208537..e21d534d8 100644 --- a/crates/test-utils/src/ecstore_test_compat.rs +++ b/crates/test-utils/src/ecstore_test_compat.rs @@ -28,4 +28,14 @@ pub(crate) mod fixture { pub(crate) use rustfs_ecstore::api::layout::{EndpointServerPools, Endpoints, PoolEndpoints}; pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks}; pub(crate) use rustfs_storage_api::{BucketOperations, BucketOptions, MakeBucketOptions}; + + #[cfg(test)] + pub(crate) use rustfs_ecstore::api::config::com::{delete_config, read_config, save_config}; + #[cfg(test)] + pub(crate) use rustfs_ecstore::api::data_usage::{ + load_data_usage_from_backend, load_data_usage_from_backend_cached, remove_bucket_usage_from_backend, + store_data_usage_in_backend, + }; + #[cfg(test)] + pub(crate) use rustfs_ecstore::api::disk::BUCKET_META_PREFIX; } diff --git a/crates/test-utils/src/lib.rs b/crates/test-utils/src/lib.rs index 977112225..1511f2aa2 100644 --- a/crates/test-utils/src/lib.rs +++ b/crates/test-utils/src/lib.rs @@ -23,6 +23,8 @@ //! Single-process integration scope only — multi-node / chaos harnesses are //! out of scope (backlog#1100). +#[cfg(test)] +mod data_usage_snapshot_tests; mod ecstore_test_compat; use std::path::PathBuf; diff --git a/rustfs/src/admin/handlers/account_info.rs b/rustfs/src/admin/handlers/account_info.rs index 829c2ee3f..575384980 100644 --- a/rustfs/src/admin/handlers/account_info.rs +++ b/rustfs/src/admin/handlers/account_info.rs @@ -25,7 +25,7 @@ use crate::server::{ADMIN_PREFIX, RemoteAddr}; use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; use matchit::Params; -use rustfs_data_usage::BucketUsageInfo; +use rustfs_data_usage::{BucketUsageInfo, DataUsageInfo}; use rustfs_policy::policy::BucketPolicy; use rustfs_policy::policy::default::DEFAULT_POLICIES; use rustfs_policy::policy::{Args, action::Action, action::S3Action}; @@ -36,6 +36,12 @@ use serde::Serialize; use std::collections::HashMap; use std::sync::Arc; +const DATA_USAGE_LOAD_ERROR_MESSAGE: &str = "failed to load data usage"; + +fn map_data_usage_result(result: Result) -> S3Result { + result.map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, DATA_USAGE_LOAD_ERROR_MESSAGE)) +} + #[allow(dead_code)] #[derive(Debug, Serialize, Default)] #[serde(rename_all = "PascalCase", default)] @@ -256,9 +262,7 @@ impl Operation for AccountInfoHandler { // Serve the last persisted scanner snapshot plus the in-memory overlay. // This request path must never trigger a live full-version listing // (rustfs/backlog#1306); freshness is owned by the scanner. - let mut data_usage_info = load_data_usage_from_backend_cached(store.clone()) - .await - .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?; + let mut data_usage_info = map_data_usage_result(load_data_usage_from_backend_cached(store.clone()).await)?; apply_bucket_usage_memory_overlay(&mut data_usage_info).await; for bucket in buckets.iter() { @@ -371,6 +375,14 @@ mod tests { assert_eq!(bucket_info.object_versions_histogram, None); } + #[test] + fn accountinfo_data_usage_error_message_is_generic() { + let err = map_data_usage_result::<&str>(Err("sensitive disk path")).expect_err("load failure must reach the response"); + + assert_eq!(err.code(), &S3ErrorCode::InternalError); + assert_eq!(err.message(), Some("failed to load data usage")); + } + #[test] fn accountinfo_bucket_details_marks_object_lock_enabled() { let enabled = ObjectLockConfiguration { diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 678aea387..e3901a793 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -209,6 +209,10 @@ impl DefaultAdminUsecase { Self::app_error(code, message) } + fn map_data_usage_load_result(result: Result) -> AdminUsecaseResult { + result.map_err(|_| Self::app_error(S3ErrorCode::InternalError, "load_data_usage_from_backend failed")) + } + async fn refresh_rebalance_status_snapshot(store: &ECStore) -> AdminUsecaseResult<()> { store.refresh_rebalance_status_meta().await.map_err(|err| { error!("refresh rebalance metadata for pool status failed: {:?}", err); @@ -248,10 +252,7 @@ impl DefaultAdminUsecase { /// This request path must never trigger a live full-version listing /// (rustfs/backlog#1306); freshness is owned by the scanner. pub(crate) async fn query_data_usage_info_with_store(store: Arc) -> AdminUsecaseResult { - let mut info = load_data_usage_from_backend_cached(store.clone()).await.map_err(|e| { - error!("load_data_usage_from_backend failed {:?}", e); - Self::app_error(S3ErrorCode::InternalError, "load_data_usage_from_backend failed") - })?; + let mut info = Self::map_data_usage_load_result(load_data_usage_from_backend_cached(store.clone()).await)?; apply_bucket_usage_memory_overlay(&mut info).await; let storage_info = StorageAdminApi::storage_info(store.as_ref()).await; @@ -617,6 +618,30 @@ mod tests { use super::super::storage_api::admin_usecase::capacity::{PoolDecommissionInfo, PoolStatus}; use super::*; use time::OffsetDateTime; + use tracing_subscriber::{Layer, Registry, layer::Context, prelude::*}; + + struct RejectEvents; + + impl Layer for RejectEvents + where + S: tracing::Subscriber, + { + fn on_event(&self, _event: &tracing::Event<'_>, _ctx: Context<'_, S>) { + panic!("data usage error mapping must not emit a duplicate event"); + } + } + + #[test] + fn data_usage_load_error_is_generic_and_not_logged_again() { + let subscriber = Registry::default().with(RejectEvents); + + tracing::subscriber::with_default(subscriber, || { + let err = DefaultAdminUsecase::map_data_usage_load_result::<&str>(Err("sensitive disk path")) + .expect_err("load failure must reach the response"); + assert_eq!(err.code, S3ErrorCode::InternalError); + assert_eq!(err.message, "load_data_usage_from_backend failed"); + }); + } #[tokio::test] async fn execute_query_storage_info_returns_internal_error_when_store_uninitialized() {