Compare commits

...

2 Commits

Author SHA1 Message Date
马登山 2ab23980f9 docs(scanner): clarify unknown tier stats 2026-08-22 23:42:58 +08:00
马登山 0b0c7ca4d7 fix(scanner): bound retired tier accounting 2026-08-22 23:41:31 +08:00
11 changed files with 636 additions and 41 deletions
+232 -2
View File
@@ -48,6 +48,11 @@ pub const DATA_USAGE_OBSERVED_OBJECT_NAME: &str = ".usage.observed.json";
// RUSTFS_COMPAT_TODO(scanner-usage-v2): keep .usage.json readable and removable during rolling upgrades from pre-v2 scanners. Remove after supported direct-upgrade sources all write .usage.v2.json.
pub const LEGACY_DATA_USAGE_OBJECT_NAME: &str = ".usage.json";
/// Fixed bucket for objects whose storage class is not in the scanner's
/// cycle-local tier registry. Keeping this key fixed prevents untrusted or
/// stale tier names from growing persisted per-tier maps without bound.
pub const UNKNOWN_TIER: &str = "UNKNOWN_TIER";
/// Returns true when `existing_last_update` is ahead of `now` by more than
/// [`USAGE_LAST_UPDATE_FUTURE_TOLERANCE`], i.e. the persisted timestamp cannot be
/// trusted for staleness comparisons and a fresh snapshot save must be allowed.
@@ -78,12 +83,146 @@ impl TierStats {
&& self.num_objects.checked_add(u.num_objects).is_some()
}
/// Add tier counters without allowing a counter to wrap.
pub fn checked_add(&self, u: &TierStats) -> Option<TierStats> {
Some(TierStats {
total_size: self.total_size.checked_add(u.total_size)?,
num_versions: self.num_versions.checked_add(u.num_versions)?,
num_objects: self.num_objects.checked_add(u.num_objects)?,
})
}
/// True when this tier contributed nothing, i.e. merging it is a no-op.
pub fn is_empty(&self) -> bool {
self.total_size == 0 && self.num_versions == 0 && self.num_objects == 0
}
}
/// Bounded diagnostics for objects whose tier is absent from the cycle
/// registry. Counters are authoritative; diagnostics are only a small,
/// redacted reconciliation aid and may be dropped at the configured caps.
pub const UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP: usize = 64;
pub const UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP: usize = 4096;
pub const UNKNOWN_TIER_DIAGNOSTIC_TTL: Duration = Duration::from_secs(60 * 60);
const UNKNOWN_TIER_DIAGNOSTIC_KEY_BYTES: usize = 256;
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct UnknownTierStats {
/// Logical bytes retained in the scanner's normal usage total.
pub unknown_bytes: u64,
/// Physical bytes recorded in the per-tier accounting dimension.
///
/// Older writers only had `unknown_bytes`; decoding those snapshots keeps
/// this field at zero and the scanner fills it for new observations.
#[serde(default)]
pub unknown_physical_bytes: u64,
pub unknown_objects: u64,
pub unknown_versions: u64,
pub diagnostics_dropped: u64,
#[serde(default)]
pub diagnostics: Vec<String>,
#[serde(default)]
pub diagnostics_at: Option<SystemTime>,
}
impl UnknownTierStats {
/// Record one observation where the logical and physical dimensions are
/// the same. Kept as a small compatibility helper for callers that only
/// have one size value.
pub fn record(&mut self, tier: &str, bytes: u64, versions: u64, objects: u64) {
self.record_dimensions(tier, bytes, bytes, versions, objects);
}
/// Record one observation without conflating logical usage with physical
/// tier bytes. Both counters are saturating so malformed metadata cannot
/// wrap an aggregate.
pub fn record_dimensions(&mut self, tier: &str, logical_bytes: u64, physical_bytes: u64, versions: u64, objects: u64) {
self.unknown_bytes = self.unknown_bytes.saturating_add(logical_bytes);
self.unknown_physical_bytes = self.unknown_physical_bytes.saturating_add(physical_bytes);
self.unknown_objects = self.unknown_objects.saturating_add(objects);
self.unknown_versions = self.unknown_versions.saturating_add(versions);
let digest = {
let mut hasher = DefaultHasher::new();
// Bound hashing work for hostile metadata while retaining enough
// length/prefix entropy to reconcile repeated observations.
hasher.write_usize(tier.len());
hasher.write(&tier.as_bytes()[..tier.len().min(UNKNOWN_TIER_DIAGNOSTIC_KEY_BYTES)]);
hasher.write_u8(u8::from(tier.bytes().any(|byte| byte.is_ascii_control())));
format!("tier-hash:{:016x}", hasher.finish())
};
let now = SystemTime::now();
if self
.diagnostics_at
.is_some_and(|at| now.duration_since(at).unwrap_or_default() > UNKNOWN_TIER_DIAGNOSTIC_TTL)
{
self.diagnostics.clear();
}
self.diagnostics_at = Some(now);
if self.diagnostics.iter().any(|entry| entry == &digest) {
return;
}
let current_bytes: usize = self.diagnostics.iter().map(String::len).sum();
if self.diagnostics.len() >= UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP
|| current_bytes.saturating_add(digest.len()) > UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP
{
self.diagnostics_dropped = self.diagnostics_dropped.saturating_add(1);
return;
}
self.diagnostics.push(digest);
}
pub fn merge(&mut self, other: &Self) {
self.unknown_bytes = self.unknown_bytes.saturating_add(other.unknown_bytes);
self.unknown_physical_bytes = self.unknown_physical_bytes.saturating_add(other.unknown_physical_bytes);
self.unknown_objects = self.unknown_objects.saturating_add(other.unknown_objects);
self.unknown_versions = self.unknown_versions.saturating_add(other.unknown_versions);
self.diagnostics_dropped = self.diagnostics_dropped.saturating_add(other.diagnostics_dropped);
let now = SystemTime::now();
if self
.diagnostics_at
.zip(other.diagnostics_at)
.is_some_and(|(left, right)| now.duration_since(left.max(right)).unwrap_or_default() > UNKNOWN_TIER_DIAGNOSTIC_TTL)
{
self.diagnostics.clear();
}
self.diagnostics_at = Some(now);
for diagnostic in &other.diagnostics {
if self.diagnostics.iter().any(|entry| entry == diagnostic) {
continue;
}
let current_bytes: usize = self.diagnostics.iter().map(String::len).sum();
if self.diagnostics.len() >= UNKNOWN_TIER_DIAGNOSTIC_ENTRY_CAP
|| current_bytes.saturating_add(diagnostic.len()) > UNKNOWN_TIER_DIAGNOSTIC_BYTE_CAP
{
self.diagnostics_dropped = self.diagnostics_dropped.saturating_add(1);
continue;
}
self.diagnostics.push(diagnostic.clone());
}
}
pub fn fits_add(&self, other: &Self) -> bool {
self.unknown_bytes.checked_add(other.unknown_bytes).is_some()
&& self
.unknown_physical_bytes
.checked_add(other.unknown_physical_bytes)
.is_some()
&& self.unknown_objects.checked_add(other.unknown_objects).is_some()
&& self.unknown_versions.checked_add(other.unknown_versions).is_some()
&& self.diagnostics_dropped.checked_add(other.diagnostics_dropped).is_some()
}
pub fn is_empty(&self) -> bool {
self.unknown_bytes == 0
&& self.unknown_physical_bytes == 0
&& self.unknown_objects == 0
&& self.unknown_versions == 0
&& self.diagnostics_dropped == 0
&& self.diagnostics.is_empty()
}
}
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct AllTierStats {
pub tiers: HashMap<String, TierStats>,
@@ -124,6 +263,31 @@ impl AllTierStats {
.iter()
.all(|(tier, right)| self.tiers.get(tier).is_none_or(|left| left.fits_add(right)))
}
/// Fold keys from an older cache that are no longer present in the
/// current registry into the fixed unknown bucket. Built-in storage
/// classes remain known even when no remote tier is configured.
pub fn fold_unknown_tiers<'a, I>(&mut self, known_tiers: I)
where
I: IntoIterator<Item = &'a str>,
{
let known: HashSet<&str> = known_tiers.into_iter().collect();
let mut unknown = self.tiers.remove(UNKNOWN_TIER).unwrap_or_default();
let retired: Vec<String> = self
.tiers
.keys()
.filter(|tier| tier.as_str() != "STANDARD" && tier.as_str() != "REDUCED_REDUNDANCY" && !known.contains(tier.as_str()))
.cloned()
.collect();
for tier in retired {
if let Some(stats) = self.tiers.remove(&tier) {
unknown = unknown.add(&stats);
}
}
if !unknown.is_empty() {
self.tiers.insert(UNKNOWN_TIER.to_string(), unknown);
}
}
}
/// Bucket target usage info provides replication statistics
@@ -211,6 +375,10 @@ pub struct DataUsageInfo {
/// tier exists, so an absent value means "not accounted", never "zero".
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tier_stats: Option<AllTierStats>,
/// Bounded diagnostics and separate logical/physical counters for objects
/// classified into [`UNKNOWN_TIER`].
#[serde(default, skip_serializing_if = "Option::is_none")]
pub unknown_tier_stats: Option<UnknownTierStats>,
/// Total number of buckets in this cluster
pub buckets_count: u64,
@@ -336,6 +504,8 @@ pub struct SizeSummary {
pub repl_target_stats: HashMap<String, ReplTargetSizeSummary>,
/// Per-tier accounting, keyed by storage class or remote tier name
pub tier_stats: HashMap<String, TierStats>,
/// Counters and bounded diagnostics for unknown tiers in this summary.
pub unknown_tier_stats: UnknownTierStats,
}
/// Replication target size summary
@@ -681,6 +851,9 @@ pub struct DataUsageEntry {
/// observed tier-classified objects.
#[serde(default)]
pub all_tier_stats: Option<AllTierStats>,
/// Bounded unknown-tier reconciliation state for this cache entry.
#[serde(default)]
pub unknown_tier_stats: Option<UnknownTierStats>,
}
impl Serialize for DataUsageEntry {
@@ -691,7 +864,7 @@ impl Serialize for DataUsageEntry {
// Keep entries map-encoded so older readers can ignore fields appended
// by newer scanner versions during rolling upgrades. The derived
// (array) encoding made any appended field a decode error for them.
let mut state = serializer.serialize_map(Some(11))?;
let mut state = serializer.serialize_map(Some(12))?;
state.serialize_entry("children", &self.children)?;
state.serialize_entry("size", &self.size)?;
state.serialize_entry("objects", &self.objects)?;
@@ -703,6 +876,7 @@ impl Serialize for DataUsageEntry {
state.serialize_entry("compacted", &self.compacted)?;
state.serialize_entry("failed_objects", &self.failed_objects)?;
state.serialize_entry("all_tier_stats", &self.all_tier_stats)?;
state.serialize_entry("unknown_tier_stats", &self.unknown_tier_stats)?;
state.end()
}
}
@@ -744,6 +918,11 @@ impl DataUsageEntry {
if let Some(o_tiers) = other.all_tier_stats.as_ref().filter(|tiers| !tiers.is_empty()) {
self.all_tier_stats.get_or_insert_with(AllTierStats::new).merge(o_tiers);
}
if let Some(other_unknown) = other.unknown_tier_stats.as_ref() {
self.unknown_tier_stats
.get_or_insert_with(UnknownTierStats::default)
.merge(other_unknown);
}
self.obj_sizes.merge_from(&other.obj_sizes);
self.obj_versions.merge_from(&other.obj_versions);
@@ -757,6 +936,15 @@ impl DataUsageEntry {
self.all_tier_stats.get_or_insert_with(AllTierStats::new).add_sizes(tiers);
}
pub fn add_unknown_tier_stats(&mut self, stats: &UnknownTierStats) {
if stats.is_empty() {
return;
}
self.unknown_tier_stats
.get_or_insert_with(UnknownTierStats::default)
.merge(stats);
}
pub fn checked_merge(&mut self, other: &DataUsageEntry) -> bool {
let scalar_counts_fit = self.objects.checked_add(other.objects).is_some()
&& self.versions.checked_add(other.versions).is_some()
@@ -820,8 +1008,12 @@ impl DataUsageEntry {
(_, None) | (None, Some(_)) => true,
(Some(left), Some(right)) => left.fits_merge(right),
};
let unknown_tier_stats_fit = match (&self.unknown_tier_stats, &other.unknown_tier_stats) {
(_, None) | (None, Some(_)) => true,
(Some(left), Some(right)) => left.fits_add(right),
};
if !scalar_counts_fit || !histograms_fit || !replication_fits || !tier_stats_fit {
if !scalar_counts_fit || !histograms_fit || !replication_fits || !tier_stats_fit || !unknown_tier_stats_fit {
return false;
}
self.merge(other);
@@ -1339,6 +1531,7 @@ impl DataUsageCache {
delete_markers_total_count: flat.delete_markers as u64,
objects_total_size: flat.size as u64,
tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
unknown_tier_stats: flat.unknown_tier_stats,
buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
buckets_usage,
usage_snapshot_complete: self.info.snapshot_complete,
@@ -1766,6 +1959,16 @@ impl SizeSummary {
self.replica_count = self.replica_count.saturating_add(other.replica_count);
self.pending_count = self.pending_count.saturating_add(other.pending_count);
self.failed_count = self.failed_count.saturating_add(other.failed_count);
self.unknown_tier_stats.merge(&other.unknown_tier_stats);
// A disk/bucket aggregate is assembled from many object summaries.
// Keep the per-tier dimension in lockstep with the scalar counters;
// dropping this map here would recreate the original silent-loss bug
// at the cross-disk merge boundary.
for (tier, stats) in &other.tier_stats {
let entry = self.tier_stats.entry(tier.clone()).or_default();
*entry = entry.add(stats);
}
// Merge replication target stats
for (target, stats) in &other.repl_target_stats {
@@ -1878,6 +2081,33 @@ mod tests {
);
}
#[test]
fn retired_tier_stats_fold_into_fixed_unknown_bucket() {
let mut stats = AllTierStats::default();
stats.tiers.insert(
"RETIRED".to_string(),
TierStats {
total_size: 9,
num_versions: 2,
num_objects: 1,
},
);
stats.tiers.insert(
"WARM".to_string(),
TierStats {
total_size: 4,
num_versions: 1,
num_objects: 1,
},
);
stats.fold_unknown_tiers(["WARM"]);
assert!(!stats.tiers.contains_key("RETIRED"));
assert_eq!(stats.tiers.get("WARM").map(|v| v.total_size), Some(4));
assert_eq!(stats.tiers.get(UNKNOWN_TIER).map(|v| v.total_size), Some(9));
}
#[test]
fn checked_merge_rejects_overflowing_tier_totals() {
let mut left = tier_entry(
+50 -9
View File
@@ -29,7 +29,8 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS;
pub use rustfs_data_usage::{
AllTierStats, BucketTargetUsageInfo, BucketUsageInfo, DATA_USAGE_OBJECT_NAME, DATA_USAGE_OBSERVED_OBJECT_NAME,
DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry,
PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, hash_path, prefix_usage_in_cache,
PrefixUsageQuery, PrefixUsageSummary, ReplTargetSizeSummary, SizeSummary, TierStats, UNKNOWN_TIER, UnknownTierStats,
hash_path, prefix_usage_in_cache,
};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use tokio::time::{Duration, Instant, sleep, timeout};
@@ -205,7 +206,8 @@ impl ScannerSizeSummaryExt for SizeSummary {
self.versions = self.versions.saturating_add(1);
}
let size = usize::try_from(size.max(0)).unwrap_or(usize::MAX);
let logical_size = size.max(0);
let size = usize::try_from(logical_size).unwrap_or(usize::MAX);
self.total_size = self.total_size.saturating_add(size);
if oi.transitioned_object.free_version {
@@ -217,12 +219,34 @@ impl ScannerSizeSummaryExt for SizeSummary {
tier = oi.transitioned_object.tier.clone();
}
if let Some(tier_stats) = self.tier_stats.get_mut(&tier) {
*tier_stats = tier_stats.add(&TierStats {
total_size: u64::try_from(oi.size).unwrap_or(0),
num_versions: 1,
num_objects: u64::from(oi.is_latest),
});
let builtin_tier = tier == storageclass::STANDARD || tier == storageclass::RRS;
let tier_registry_is_empty =
self.tier_stats.is_empty() || (self.tier_stats.len() == 1 && self.tier_stats.contains_key(UNKNOWN_TIER));
let known_tier = tier != UNKNOWN_TIER && (builtin_tier || self.tier_stats.contains_key(&tier));
// With no configured tier, retain the historical empty-map shape for
// ordinary STANDARD/RRS objects. A non-built-in key is still an
// observable unknown and must create only the fixed bucket.
if tier_registry_is_empty && known_tier {
return;
}
let accounting_key = if known_tier { tier.clone() } else { UNKNOWN_TIER.to_string() };
let tier_stats = self.tier_stats.entry(accounting_key).or_default();
let physical_size = u64::try_from(oi.size.max(0)).unwrap_or(0);
*tier_stats = tier_stats.add(&TierStats {
total_size: physical_size,
num_versions: 1,
num_objects: u64::from(oi.is_latest),
});
if !known_tier {
self.unknown_tier_stats.record_dimensions(
&tier,
u64::try_from(logical_size).unwrap_or(u64::MAX),
physical_size,
1,
u64::from(oi.is_latest),
);
}
}
}
@@ -344,6 +368,10 @@ pub struct DataUsageCacheInfo {
pub scan_plan_digest: Option<DataUsageScanPlanDigest>,
#[serde(default)]
pub cache_key_format: u16,
/// Registry generation used for the completed/partial scan. This is
/// process-local audit data; older cache writers omit it.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub tier_registry_generation: Option<u64>,
}
impl Serialize for DataUsageCacheInfo {
@@ -353,7 +381,7 @@ impl Serialize for DataUsageCacheInfo {
{
// Keep this metadata map-encoded so older readers can ignore fields
// appended by newer scanner versions during rolling upgrades.
let mut state = serializer.serialize_map(Some(16))?;
let mut state = serializer.serialize_map(Some(17))?;
state.serialize_entry("name", &self.name)?;
state.serialize_entry("next_cycle", &self.next_cycle)?;
state.serialize_entry("leader_epoch", &self.leader_epoch)?;
@@ -370,6 +398,7 @@ impl Serialize for DataUsageCacheInfo {
state.serialize_entry("snapshot_complete", &self.snapshot_complete)?;
state.serialize_entry("scan_plan_digest", &self.scan_plan_digest)?;
state.serialize_entry("cache_key_format", &self.cache_key_format)?;
state.serialize_entry("tier_registry_generation", &self.tier_registry_generation)?;
state.end()
}
}
@@ -390,6 +419,17 @@ pub(crate) enum DataUsageCachePrepareOutcome {
}
impl DataUsageCache {
/// Reconcile tier keys loaded from an older cache against the registry
/// frozen for this scan. New metadata is already routed through
/// `UNKNOWN_TIER`; this pass handles retired keys that predate that rule.
pub(crate) fn fold_retired_tiers(&mut self, tier_names: &[String]) {
for entry in self.cache.values_mut() {
if let Some(tiers) = entry.all_tier_stats.as_mut() {
tiers.fold_unknown_tiers(tier_names.iter().map(String::as_str));
}
}
}
/// Prefix-level usage query over this (writer-side) cache; see
/// [`prefix_usage_in_cache`] for the semantics
/// (rustfs/backlog#1872).
@@ -875,6 +915,7 @@ impl DataUsageCache {
delete_markers_total_count: flat.delete_markers as u64,
objects_total_size: flat.size as u64,
tier_stats: flat.all_tier_stats.filter(|tiers| !tiers.is_empty()),
unknown_tier_stats: flat.unknown_tier_stats.filter(|stats| !stats.is_empty()),
buckets_count: u64::try_from(buckets.len()).unwrap_or(u64::MAX),
buckets_usage,
..Default::default()
+179 -3
View File
@@ -573,7 +573,6 @@ fn size_summary_add_saturates_all_usage_counters() {
failed_count: usize::MAX,
},
);
let mut increment = SizeSummary {
total_size: 1,
versions: 1,
@@ -588,6 +587,24 @@ fn size_summary_add_saturates_all_usage_counters() {
failed_count: 1,
..Default::default()
};
summary.tier_stats.insert(
UNKNOWN_TIER.to_string(),
TierStats {
total_size: u64::MAX,
num_versions: u64::MAX,
num_objects: u64::MAX,
},
);
increment.tier_stats.insert(
UNKNOWN_TIER.to_string(),
TierStats {
total_size: 1,
num_versions: 1,
num_objects: 1,
},
);
increment.unknown_tier_stats.unknown_bytes = 1;
increment.unknown_tier_stats.unknown_physical_bytes = 1;
increment.repl_target_stats.insert(
target.clone(),
ReplTargetSizeSummary {
@@ -624,6 +641,8 @@ fn size_summary_add_saturates_all_usage_counters() {
assert_eq!(target_summary.failed_size, i64::MAX);
assert_eq!(target_summary.pending_count, usize::MAX);
assert_eq!(target_summary.failed_count, usize::MAX);
assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, u64::MAX);
assert_eq!(summary.unknown_tier_stats.unknown_bytes, 1);
}
#[test]
@@ -673,6 +692,162 @@ fn size_summary_actions_accounting_accumulates_tier_stats() {
);
}
#[test]
fn unknown_tier_is_bounded_and_accounted() {
let mut summary = SizeSummary::new();
summary.tier_stats.insert("WARM".to_string(), TierStats::default());
let object = ObjectInfo {
storage_class: Some("retired-tier".to_string()),
size: 11,
is_latest: true,
..Default::default()
};
summary.actions_accounting(&object, 11, 11);
assert_eq!(summary.tier_stats.len(), 2);
assert_eq!(summary.tier_stats.get(UNKNOWN_TIER).map(|stats| stats.total_size), Some(11));
assert_eq!(summary.unknown_tier_stats.unknown_bytes, 11);
assert_eq!(summary.unknown_tier_stats.unknown_physical_bytes, 11);
assert_eq!(summary.unknown_tier_stats.unknown_objects, 1);
assert!(
summary
.unknown_tier_stats
.diagnostics
.iter()
.all(|entry| !entry.contains("retired"))
);
}
#[test]
fn unknown_tier_is_accounted_when_no_remote_tier_is_configured() {
let mut summary = SizeSummary::new();
let object = ObjectInfo {
storage_class: Some("retired-tier".to_string()),
size: 3,
is_latest: true,
..Default::default()
};
summary.actions_accounting(&object, 9, 9);
assert_eq!(summary.tier_stats.len(), 1);
assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, 3);
assert_eq!(summary.unknown_tier_stats.unknown_bytes, 9);
assert_eq!(summary.unknown_tier_stats.unknown_physical_bytes, 3);
let standard = ObjectInfo {
storage_class: Some(storageclass::STANDARD.to_string()),
size: 4,
is_latest: true,
..Default::default()
};
summary.actions_accounting(&standard, 4, 4);
assert_eq!(summary.tier_stats.len(), 1, "built-ins preserve the no-tier map shape");
}
#[test]
fn million_unique_tier_keys_do_not_grow_stats_map() {
let mut summary = SizeSummary::new();
summary.tier_stats.insert("WARM".to_string(), TierStats::default());
for index in 0..1_000_000_u64 {
let object = ObjectInfo {
storage_class: Some(format!("untrusted-tier-{index}")),
size: 1,
..Default::default()
};
summary.actions_accounting(&object, 1, 1);
}
assert_eq!(summary.tier_stats.len(), 2);
assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, 1_000_000);
assert_eq!(summary.unknown_tier_stats.unknown_bytes, 1_000_000);
}
#[test]
fn unknown_tier_never_triggers_transition() {
let mut summary = SizeSummary::new();
summary.tier_stats.insert("WARM".to_string(), TierStats::default());
let mut object = ObjectInfo {
storage_class: Some("removed-tier".to_string()),
size: 7,
..Default::default()
};
object.transitioned_object.status = TRANSITION_COMPLETE.to_string();
object.transitioned_object.tier = "removed-tier".to_string();
summary.actions_accounting(&object, 7, 7);
assert_eq!(summary.tier_stats.get("removed-tier"), None);
assert_eq!(summary.tier_stats[UNKNOWN_TIER].total_size, 7);
}
#[test]
fn removed_tier_survives_restart_as_unknown() {
let mut summary = SizeSummary::new();
summary.tier_stats.insert("COLD".to_string(), TierStats::default());
let object = ObjectInfo {
storage_class: Some("retired-tier".to_string()),
size: 5,
..Default::default()
};
summary.actions_accounting(&object, 5, 5);
let mut entry = DataUsageEntry::default();
entry.add_tier_sizes(&summary.tier_stats);
entry.add_unknown_tier_stats(&summary.unknown_tier_stats);
let encoded = rmp_serde::to_vec(&entry).expect("entry should encode");
let restored: DataUsageEntry = rmp_serde::from_slice(&encoded).expect("entry should decode");
assert_eq!(restored.all_tier_stats.expect("tier stats persisted").tiers[UNKNOWN_TIER].total_size, 5);
assert_eq!(restored.unknown_tier_stats.expect("unknown stats persisted").unknown_bytes, 5);
}
#[test]
fn tier_registry_refresh_does_not_mix_cycle_generations() {
let first = crate::TierRegistrySnapshot {
generation: 1,
names: Arc::from(["WARM".to_string()]),
refresh_failed: false,
};
let second = crate::TierRegistrySnapshot {
generation: 2,
names: Arc::from(["COLD".to_string()]),
refresh_failed: false,
};
assert_ne!(first.generation, second.generation);
assert_eq!(first.names.as_ref(), ["WARM".to_string()]);
assert_eq!(second.names.as_ref(), ["COLD".to_string()]);
assert!(first.refreshed(Err(())).refresh_failed);
assert!(!second.refreshed(Ok(Arc::from(["HOT".to_string()]))).refresh_failed);
assert_eq!(first.refreshed(Err(())).generation, first.generation);
assert_eq!(first.refreshed(Err(())).names, first.names);
}
#[test]
fn unknown_tier_counter_uses_checked_arithmetic() {
let max = TierStats {
total_size: u64::MAX,
num_versions: u64::MAX,
num_objects: u64::MAX,
};
assert!(max.checked_add(&TierStats::default()).is_some());
assert!(
max.checked_add(&TierStats {
total_size: 1,
..Default::default()
})
.is_none()
);
let mut unknown = UnknownTierStats {
unknown_bytes: u64::MAX,
..Default::default()
};
unknown.record("overflow", 1, 1, 1);
assert_eq!(unknown.unknown_bytes, u64::MAX);
assert_eq!(unknown.unknown_objects, 1);
}
#[test]
fn test_data_usage_entry_merge_sums_failed_objects() {
let mut left = DataUsageEntry {
@@ -926,7 +1101,7 @@ const USAGE_CACHE_WIRE_FIXTURE: &[u8] = &[
0xdc, 0x00, 0x20, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03,
0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0x03, 0xb0, 0x63, 0x61, 0x63, 0x68, 0x65, 0x5f,
0x6b, 0x65, 0x79, 0x5f, 0x66, 0x6f, 0x72, 0x6d, 0x61, 0x74, 0x01, 0x81, 0xab, 0x77, 0x69, 0x72, 0x65, 0x2d, 0x62, 0x75, 0x63,
0x6b, 0x65, 0x74, 0x8b, 0xa8, 0x63, 0x68, 0x69, 0x6c, 0x64, 0x72, 0x65, 0x6e, 0x90, 0xa4, 0x73, 0x69, 0x7a, 0x65, 0xcd, 0x10,
0x6b, 0x65, 0x74, 0x8c, 0xa8, 0x63, 0x68, 0x69, 0x6c, 0x64, 0x72, 0x65, 0x6e, 0x90, 0xa4, 0x73, 0x69, 0x7a, 0x65, 0xcd, 0x10,
0x00, 0xa7, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x73, 0x03, 0xa8, 0x76, 0x65, 0x72, 0x73, 0x69, 0x6f, 0x6e, 0x73, 0x05, 0xae,
0x64, 0x65, 0x6c, 0x65, 0x74, 0x65, 0x5f, 0x6d, 0x61, 0x72, 0x6b, 0x65, 0x72, 0x73, 0x01, 0xa9, 0x6f, 0x62, 0x6a, 0x5f, 0x73,
0x69, 0x7a, 0x65, 0x73, 0x9b, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xac, 0x6f, 0x62, 0x6a, 0x5f,
@@ -934,7 +1109,8 @@ const USAGE_CACHE_WIRE_FIXTURE: &[u8] = &[
0x69, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x73, 0xc0, 0xa9, 0x63, 0x6f, 0x6d, 0x70, 0x61, 0x63,
0x74, 0x65, 0x64, 0xc3, 0xae, 0x66, 0x61, 0x69, 0x6c, 0x65, 0x64, 0x5f, 0x6f, 0x62, 0x6a, 0x65, 0x63, 0x74, 0x73, 0x02, 0xae,
0x61, 0x6c, 0x6c, 0x5f, 0x74, 0x69, 0x65, 0x72, 0x5f, 0x73, 0x74, 0x61, 0x74, 0x73, 0x91, 0x81, 0xa4, 0x57, 0x41, 0x52, 0x4d,
0x93, 0xcd, 0x08, 0x00, 0x02, 0x01,
0x93, 0xcd, 0x08, 0x00, 0x02, 0x01, 0xb2, 0x75, 0x6e, 0x6b, 0x6e, 0x6f, 0x77, 0x6e, 0x5f, 0x74, 0x69, 0x65, 0x72, 0x5f, 0x73,
0x74, 0x61, 0x74, 0x73, 0xc0,
];
#[test]
+71 -8
View File
@@ -94,6 +94,45 @@ static SCANNER_RUNTIME_INSTANCES: AtomicU64 = AtomicU64::new(0);
static SCANNER_FOREGROUND_READ_ACTIVITY: AtomicU64 = AtomicU64::new(0);
static SCANNER_FOREGROUND_STREAM_READS: AtomicU64 = AtomicU64::new(0);
/// Immutable tier registry captured at the beginning of a folder scan.
/// Generation makes it possible to prove that a result was classified against
/// one registry even when the process-wide TTL cache refreshes concurrently.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct TierRegistrySnapshot {
pub(crate) generation: u64,
pub(crate) names: Arc<[String]>,
/// True when the last refresh attempt failed and `names` is therefore a
/// retained last-good snapshot rather than a newly read registry.
pub(crate) refresh_failed: bool,
}
impl TierRegistrySnapshot {
/// Apply a refresh only when the registry read succeeds. A failed refresh
/// retains the prior generation, preventing a transient config failure
/// from classifying the remainder of a scan against an empty registry.
pub(crate) fn refreshed(&self, names: Result<Arc<[String]>, ()>) -> Self {
match names {
Ok(names) => Self {
generation: self.generation.saturating_add(1),
names,
refresh_failed: false,
},
Err(()) => Self {
refresh_failed: true,
..self.clone()
},
}
}
pub(crate) fn initial(names: Arc<[String]>) -> Self {
Self {
generation: 1,
names,
refresh_failed: false,
}
}
}
pub fn current_scanner_activity() -> u64 {
SCANNER_ACTIVE_WORK_UNITS.load(Ordering::Relaxed)
}
@@ -383,24 +422,48 @@ const TIER_NAME_CACHE_TTL: Duration = Duration::from_secs(30);
/// `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<Option<(Instant, Arc<[String]>)>> = RwLock::new(None);
static TIER_NAME_CACHE: RwLock<Option<(Instant, TierRegistrySnapshot)>> = RwLock::new(None);
static TIER_REGISTRY_GENERATION: AtomicU64 = AtomicU64::new(0);
/// Tier names currently registered in the tier configuration, cached for
/// `TIER_NAME_CACHE_TTL`.
pub(crate) async fn runtime_tier_names() -> Arc<[String]> {
/// Return one immutable registry snapshot for a scanner unit of work.
pub(crate) async fn runtime_tier_registry() -> TierRegistrySnapshot {
{
let cached = TIER_NAME_CACHE.read().unwrap_or_else(|err| err.into_inner()).clone();
if let Some((refreshed_at, names)) = cached
if let Some((refreshed_at, snapshot)) = cached
&& refreshed_at.elapsed() < TIER_NAME_CACHE_TTL
{
return names;
return snapshot;
}
}
let tiers = ecstore_get_global_tier_config_mgr().read().await.list_tiers();
let names: Arc<[String]> = tiers.iter().map(|tier| tier.name.clone()).collect::<Vec<_>>().into();
*TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((Instant::now(), Arc::clone(&names)));
names
let previous = TIER_NAME_CACHE
.read()
.unwrap_or_else(|err| err.into_inner())
.as_ref()
.map(|(_, snapshot)| snapshot.clone());
let snapshot = previous
.as_ref()
.map(|snapshot| {
let refreshed = snapshot.refreshed(Ok(Arc::clone(&names)));
TierRegistrySnapshot {
generation: TIER_REGISTRY_GENERATION.fetch_add(1, Ordering::Relaxed).saturating_add(1),
..refreshed
}
})
.unwrap_or_else(|| TierRegistrySnapshot {
generation: TIER_REGISTRY_GENERATION.fetch_add(1, Ordering::Relaxed).saturating_add(1),
..TierRegistrySnapshot::initial(names)
});
*TIER_NAME_CACHE.write().unwrap_or_else(|err| err.into_inner()) = Some((Instant::now(), snapshot.clone()));
snapshot
}
/// Tier names currently registered in the tier configuration, cached for
/// `TIER_NAME_CACHE_TTL`.
pub(crate) async fn runtime_tier_names() -> Arc<[String]> {
runtime_tier_registry().await.names
}
/// Test-only cache reset; the production cache has no invalidation hook
+17 -4
View File
@@ -55,9 +55,9 @@ use tracing::{debug, error, warn};
use crate::{
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,
ScannerLifecycleConfigExt as _, ScannerVersioningConfigExt as _, StorageError, TierRegistrySnapshot, 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, runtime_tier_registry, scanner_is_erasure,
scanner_replication_config_for_lifecycle_eval,
};
use crate::{ScannerObjectInfo as ObjectInfo, ScannerObjectToDelete as ObjectToDelete};
@@ -626,6 +626,7 @@ fn apply_scanner_size_summary(into: &mut DataUsageEntry, summary: &SizeSummary)
}
into.add_tier_sizes(&summary.tier_stats);
into.add_unknown_tier_stats(&summary.unknown_tier_stats);
}
fn data_usage_root_has_progress(root: &DataUsageEntry) -> bool {
@@ -670,6 +671,9 @@ pub struct FolderScanner {
budget: Arc<ScannerCycleBudget>,
skip_heal: Arc<std::sync::atomic::AtomicBool>,
local_disk: Arc<Disk>,
/// Tier registry frozen for this folder scan. A refresh applies to the
/// next scan and cannot mix generations in one aggregate.
tier_registry: TierRegistrySnapshot,
pending_heals_changed: bool,
#[cfg(test)]
list_path_raw_options_observer: Option<mpsc::UnboundedSender<ListPathRawTimeoutSnapshot>>,
@@ -1329,7 +1333,11 @@ impl FolderScanner {
continue;
}
let sz = match self.local_disk.get_size(item.clone()).await {
let sz = match self
.local_disk
.get_size_with_tier_names(item.clone(), &self.tier_registry.names)
.await
{
Ok(sz) => sz,
Err(e) => {
let failure_action = classify_get_size_failure(&item, &e);
@@ -2161,6 +2169,10 @@ pub async fn scan_data_folder(
let failed_object_ttl = rustfs_utils::get_env_u32(ENV_FAILED_OBJECT_TTL_SECS, DEFAULT_FAILED_OBJECT_TTL_SECS) as u64;
let failed_objects_max = rustfs_utils::get_env_u32(ENV_FAILED_OBJECTS_MAX, DEFAULT_FAILED_OBJECTS_MAX) as usize;
let tier_registry = runtime_tier_registry().await;
let mut cache = cache;
cache.fold_retired_tiers(&tier_registry.names);
cache.info.tier_registry_generation = Some(tier_registry.generation);
// Create folder scanner
let mut scanner = FolderScanner {
@@ -2189,6 +2201,7 @@ pub async fn scan_data_folder(
budget: budget.clone(),
skip_heal,
local_disk,
tier_registry,
pending_heals_changed: false,
#[cfg(test)]
list_path_raw_options_observer: None,
@@ -289,11 +289,42 @@ impl ScannerItem {
item.object_path()
}
fn effective_tier(oi: &ObjectInfo) -> &str {
if oi.transitioned_object.status == crate::TRANSITION_COMPLETE {
oi.transitioned_object.tier.as_str()
} else {
oi.storage_class.as_deref().unwrap_or(crate::storageclass::STANDARD)
}
}
fn tier_is_known(oi: &ObjectInfo, tier_names: &[String]) -> bool {
let tier = Self::effective_tier(oi);
if tier == crate::data_usage_define::UNKNOWN_TIER {
return false;
}
tier == crate::storageclass::STANDARD || tier == crate::storageclass::RRS || tier_names.iter().any(|name| name == tier)
}
fn action_requires_known_tier(action: IlmAction) -> bool {
matches!(
action,
IlmAction::TransitionAction
| IlmAction::TransitionVersionAction
| IlmAction::DeleteAction
| IlmAction::DeleteVersionAction
| IlmAction::DeleteRestoredAction
| IlmAction::DeleteRestoredVersionAction
| IlmAction::DeleteAllVersionsAction
| IlmAction::DelMarkerDeleteAllVersionsAction
)
}
pub async fn apply_actions(
&mut self,
object_infos: Vec<ObjectInfo>,
lock_retention: Option<Arc<ObjectLockConfiguration>>,
versioning_config: VersioningConfiguration,
tier_names: &[String],
size_summary: &mut SizeSummary,
) {
let object_path = self.object_path();
@@ -425,6 +456,18 @@ impl ScannerItem {
let mut size = actual_size;
let mut account_now = true;
// A retired/unknown source tier may point at a remote object
// that cannot be safely deleted or transitioned. Lifecycle
// evaluation is still useful for accounting, but all
// side-effecting tier actions fail closed until the registry
// recognizes the source again.
if !Self::tier_is_known(oi, tier_names) && Self::action_requires_known_tier(event.action) {
size = self.heal_actions(oi, actual_size, size_summary).await;
size_summary.actions_accounting(oi, size, actual_size);
cumulative_size += size;
continue;
}
match event.action {
IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => {
debug!(
@@ -929,4 +972,15 @@ mod tests {
assert_eq!(item.object_name, "object");
assert_eq!(item.object_path(), "object");
}
#[test]
fn unknown_tier_never_triggers_transition() {
let object = ObjectInfo {
storage_class: Some("retired-tier".to_string()),
..Default::default()
};
assert!(!ScannerItem::tier_is_known(&object, &["WARM".to_string()]));
assert!(ScannerItem::action_requires_known_tier(IlmAction::TransitionAction));
assert!(ScannerItem::action_requires_known_tier(IlmAction::DeleteVersionAction));
}
}
@@ -325,6 +325,11 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
budget: ScannerCycleBudget::new(&CancellationToken::new(), Default::default()),
skip_heal: Arc::new(AtomicBool::new(false)),
local_disk: disk,
tier_registry: crate::TierRegistrySnapshot {
generation: 0,
names: Arc::new([]),
refresh_failed: false,
},
pending_heals_changed: false,
list_path_raw_options_observer: None,
};
+3
View File
@@ -498,6 +498,9 @@ pub trait ScannerIODisk: Send + Sync + Debug + 'static {
) -> Result<ScannerDiskScanOutcome>;
async fn get_size(&self, item: ScannerItem) -> Result<SizeSummary>;
/// Read one object using a registry snapshot captured at scan start.
async fn get_size_with_tier_names(&self, item: ScannerItem, tier_names: &[String]) -> Result<SizeSummary>;
}
#[derive(Debug)]
+1
View File
@@ -216,6 +216,7 @@ pub(super) fn completed_data_usage_info(
delete_markers_total_count: u64::try_from(total.delete_markers).ok()?,
objects_total_size: u64::try_from(total.size).ok()?,
tier_stats: total.all_tier_stats.filter(|tiers| !tiers.is_empty()),
unknown_tier_stats: total.unknown_tier_stats.filter(|stats| !stats.is_empty()),
buckets_count: u64::try_from(all_buckets.len()).ok()?,
bucket_sizes,
buckets_usage,
+21 -12
View File
@@ -13,29 +13,38 @@
// limitations under the License.
/// ScannerIODisk implementation for Disk: get_size and the per-disk bucket scan.
use super::*;
use crate::UNKNOWN_TIER;
///
/// 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.
/// Preserves the original no-tier shape: with no tiers configured the map
/// stays completely empty (STANDARD/RRS/UNKNOWN are not seeded either).
/// Otherwise the standard storage classes and one fixed unknown bucket are
/// seeded alongside every configured tier so per-object accounting never
/// inserts an untrusted metadata key.
pub(super) fn tier_stats_template(tier_names: &[String]) -> HashMap<String, TierStats> {
let mut tier_stats = HashMap::with_capacity(tier_names.len() + 2);
let mut tier_stats = HashMap::with_capacity(tier_names.len() + 3);
for tier_name in tier_names {
tier_stats.insert(tier_name.clone(), TierStats::default());
if tier_name != UNKNOWN_TIER {
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.insert(UNKNOWN_TIER.to_string(), TierStats::default());
}
tier_stats
}
#[async_trait::async_trait]
impl ScannerIODisk for Disk {
async fn get_size(&self, mut item: ScannerItem) -> Result<SizeSummary> {
async fn get_size(&self, item: ScannerItem) -> Result<SizeSummary> {
self.get_size_with_tier_names(item, &runtime_tier_names().await).await
}
async fn get_size_with_tier_names(&self, mut item: ScannerItem, tier_names: &[String]) -> Result<SizeSummary> {
let done_object = Metrics::time(Metric::ScanObject);
if !is_xl_meta_path(&item.path) {
@@ -107,10 +116,10 @@ impl ScannerIODisk for Disk {
let mut size_summary = SizeSummary::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);
// The caller supplies one registry snapshot for the whole folder scan;
// seeding from it prevents a TTL refresh from mixing generations in a
// single result.
size_summary.tier_stats = tier_stats_template(tier_names);
let lock_config = object_lock_config_for_scanner_item(&item).await;
@@ -120,7 +129,7 @@ impl ScannerIODisk for Disk {
// `object_infos`.
global_metrics().record_scanner_versions_scanned(object_infos.len() as u64);
item.apply_actions(object_infos, lock_config, versioning_config, &mut size_summary)
item.apply_actions(object_infos, lock_config, versioning_config, tier_names, &mut size_summary)
.await;
if !free_version_infos.is_empty() {
+3 -3
View File
@@ -23,7 +23,7 @@ use crate::storage_api::owner::{
use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _};
use crate::{
DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerObjectOptions,
ScannerPutObjReader, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
ScannerPutObjReader, UNKNOWN_TIER, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
init_local_disks_with_instance_ctx, new_disk, path2_bucket_object_with_base_path,
};
use rustfs_filemeta::FileInfo;
@@ -986,8 +986,8 @@ fn is_xl_meta_path_accepts_forward_separator() {
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.len(), 5);
for tier in ["WARM", "COLD", storageclass::STANDARD, storageclass::RRS, UNKNOWN_TIER] {
assert_eq!(template.get(tier), Some(&TierStats::default()), "missing seed for tier {tier}");
}
}