mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 20:36:38 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2ab23980f9 | |||
| 0b0c7ca4d7 |
@@ -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(
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
};
|
||||
|
||||
@@ -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)]
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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}");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user