diff --git a/crates/data-usage/src/data_usage.rs b/crates/data-usage/src/data_usage.rs index ebbd03261..b08281d0f 100644 --- a/crates/data-usage/src/data_usage.rs +++ b/crates/data-usage/src/data_usage.rs @@ -870,6 +870,157 @@ pub struct DataUsageCacheInfo { pub snapshot_complete: bool, } +/// Prefix-level usage over a raw entry map — the shared core behind +/// [`DataUsageCache::prefix_usage`], usable by any cache-shaped reader (the +/// scanner's writer-side cache has the same map type). +/// +/// Cache keys are cleaned literal paths (`bucket/pre/fix`), so sub-prefix +/// names come straight off the child keys — no reverse mapping exists or is +/// needed. A compacted prefix carries its aggregate but no children, which +/// the `compacted` flag reports so callers can say why the breakdown is +/// empty. `truncated` is set when the breakdown exceeded `max_entries` and +/// was cut (largest first). +pub fn prefix_usage_in_cache( + cache: &HashMap, + bucket: &str, + prefix: &str, + max_entries: usize, +) -> Option { + let prefix = prefix.trim_matches('/'); + let root = if prefix.is_empty() { + bucket.to_string() + } else { + format!("{bucket}/{prefix}") + }; + let entry = cache.get(&hash_path(&root).key())?.clone(); + + let usage = PrefixUsageSummary::from_entry(&flatten_entry(cache, &entry, 0)?); + + let child_prefix = format!("{root}/"); + let mut sub_prefixes: Vec = entry + .children + .iter() + .filter_map(|child_key| { + let child = cache.get(child_key)?; + let child_flat = flatten_entry(cache, child, 1)?; + // Child keys are literal `bucket/pre/name` paths; a trailing + // slash marks a directory object and is display-only here. + let name = child_key + .strip_prefix(child_prefix.as_str()) + .unwrap_or(child_key.as_str()) + .trim_end_matches('/') + .to_string(); + Some(PrefixUsageEntry { + prefix: name, + usage: PrefixUsageSummary::from_entry(&child_flat), + }) + }) + .collect(); + sub_prefixes.sort_by(|left, right| { + right + .usage + .size + .cmp(&left.usage.size) + .then_with(|| left.prefix.cmp(&right.prefix)) + }); + let truncated = sub_prefixes.len() > max_entries; + sub_prefixes.truncate(max_entries); + + Some(PrefixUsageQuery { + usage, + compacted: entry.compacted, + truncated, + sub_prefixes, + }) +} + +/// Maximum subtree depth [`flatten_entry`] will walk before declaring the +/// cache corrupt — the same bound the scanner's checked flatten uses. +const PREFIX_USAGE_MAX_DEPTH: usize = 1024; + +/// Flatten one entry's subtree into an aggregate: the free-function twin of +/// [`DataUsageCache::flatten`], carrying the scanner checked-flatten +/// hardening so a corrupt cache (cycles, over-deep trees, overflowing +/// counters) yields `None` instead of unbounded recursion or wrapped totals. +fn flatten_entry(cache: &HashMap, root: &DataUsageEntry, depth: usize) -> Option { + if depth > PREFIX_USAGE_MAX_DEPTH { + return None; + } + let mut flattened = DataUsageEntry::default(); + if !flattened.checked_merge(root) { + return None; + } + flattened.compacted = root.compacted; + // The root itself is not pre-seeded: it is merged above, and a corrupt + // child edge pointing back at the root's own key is still terminated by + // the visited set on first encounter. + let mut visited: HashSet<&str> = HashSet::new(); + let mut pending: Vec<(&String, usize)> = root.children.iter().map(|child| (child, depth + 1)).collect(); + while let Some((key, child_depth)) = pending.pop() { + if child_depth > PREFIX_USAGE_MAX_DEPTH || !visited.insert(key.as_str()) { + return None; + } + let entry = cache.get(key)?; + if !flattened.checked_merge(entry) { + return None; + } + pending.extend(entry.children.iter().map(|child| (child, child_depth + 1))); + } + flattened.children.clear(); + Some(flattened) +} + +/// Flattened counters of one prefix subtree, as returned by +/// [`DataUsageCache::prefix_usage`]. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct PrefixUsageSummary { + pub size: u64, + pub objects: u64, + pub versions: u64, + pub delete_markers: u64, +} + +impl PrefixUsageSummary { + fn from_entry(entry: &DataUsageEntry) -> Self { + Self { + size: entry.size as u64, + objects: entry.objects as u64, + versions: entry.versions as u64, + delete_markers: entry.delete_markers as u64, + } + } + + /// Add another set's counters into this one (entries are partitioned by + /// set, so per-set results sum). + pub fn merge(&mut self, other: &Self) { + self.size = self.size.saturating_add(other.size); + self.objects = self.objects.saturating_add(other.objects); + self.versions = self.versions.saturating_add(other.versions); + self.delete_markers = self.delete_markers.saturating_add(other.delete_markers); + } +} + +/// One first-level sub-prefix row of a [`PrefixUsageQuery`]. +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)] +pub struct PrefixUsageEntry { + pub prefix: String, + pub usage: PrefixUsageSummary, +} + +/// Result of [`DataUsageCache::prefix_usage`]. +#[derive(Clone, Debug, Default, PartialEq, Eq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct PrefixUsageQuery { + pub usage: PrefixUsageSummary, + /// The prefix entry was compacted by the scanner: its aggregate is valid + /// but no sub-prefix breakdown exists on disk. + pub compacted: bool, + /// The breakdown had more entries than `max_entries`; the largest remain. + pub truncated: bool, + pub sub_prefixes: Vec, +} + /// Read-only projection of a scanner-written `.usage-cache.bin` file. /// /// The scanner-side `DataUsageCache` (`crates/scanner/src/data_usage_define.rs`) @@ -997,6 +1148,21 @@ impl DataUsageCache { } } + /// Prefix-level usage for one bucket subtree, plus the one-level + /// breakdown below it (rustfs/backlog#1872, MinIO + /// `loadPrefixUsageFromBackend` parity and beyond: arbitrary prefixes and + /// full counters instead of first-level sizes only). + /// + /// Cache keys are cleaned literal paths (`bucket/pre/fix`), so sub-prefix + /// names come straight off the child keys — no reverse mapping exists or + /// is needed. A compacted prefix carries its aggregate but no children, + /// which the `compacted` flag reports so callers can say why the + /// breakdown is empty. `truncated` is set when the breakdown exceeded + /// `max_entries` and was cut (largest first). + pub fn prefix_usage(&self, bucket: &str, prefix: &str, max_entries: usize) -> Option { + prefix_usage_in_cache(&self.cache, bucket, prefix, max_entries) + } + pub fn force_compact(&mut self, limit: usize) { if self.cache.len() < limit { return; @@ -1898,6 +2064,126 @@ mod tests { ); } + /// Build a cache shaped like `bucket/{a,b/{c,d}},bucket/loose` with + /// distinct counters so aggregation is observable. + fn prefix_usage_fixture_cache() -> DataUsageCache { + let mut cache = DataUsageCache::default(); + let mut insert = |path: &str, parent: &str, size: usize, objects: usize, versions: usize, delete_markers: usize| { + cache.replace( + path, + parent, + DataUsageEntry { + size, + objects, + versions, + delete_markers, + ..Default::default() + }, + ); + }; + insert("bucket", "", 0, 0, 0, 0); + insert("bucket/a", "bucket", 100, 1, 1, 0); + insert("bucket/b", "bucket", 0, 0, 0, 0); + insert("bucket/b/c", "bucket/b", 200, 2, 2, 1); + insert("bucket/b/d", "bucket/b", 40, 1, 3, 0); + insert("bucket/loose", "bucket", 10, 1, 1, 1); + cache + } + + #[test] + fn prefix_usage_aggregates_bucket_root_and_one_level_below() { + let cache = prefix_usage_fixture_cache(); + + let root = cache + .prefix_usage("bucket", "", 100) + .expect("root query must find the bucket entry"); + assert_eq!(root.usage.size, 350, "root aggregate flattens the whole subtree"); + assert_eq!(root.usage.objects, 5); + assert_eq!(root.usage.versions, 7); + assert_eq!(root.usage.delete_markers, 2); + assert!(!root.compacted); + assert!(!root.truncated); + // Breakdown is one level: b (240) before a (100) before loose (10), + // each flattened to its own subtree total. + let names: Vec<(&str, u64)> = root + .sub_prefixes + .iter() + .map(|entry| (entry.prefix.as_str(), entry.usage.size)) + .collect(); + assert_eq!(names, vec![("b", 240), ("a", 100), ("loose", 10)]); + } + + #[test] + fn prefix_usage_drills_into_arbitrary_prefixes() { + let cache = prefix_usage_fixture_cache(); + + let b = cache.prefix_usage("bucket", "b", 100).expect("nested prefix must resolve"); + assert_eq!(b.usage.size, 240); + assert_eq!(b.usage.versions, 5); + let names: Vec<&str> = b.sub_prefixes.iter().map(|entry| entry.prefix.as_str()).collect(); + assert_eq!(names, vec!["c", "d"]); + + // Prefix slashes are normalized away. + let slashed = cache.prefix_usage("bucket", "/b/", 100).expect("slash-insensitive lookup"); + assert_eq!(slashed.usage.size, 240); + + assert!(cache.prefix_usage("bucket", "absent", 100).is_none(), "unknown prefix must be a miss"); + assert!(cache.prefix_usage("other", "", 100).is_none(), "unknown bucket must be a miss"); + } + + #[test] + fn prefix_usage_reports_and_respects_truncation() { + let cache = prefix_usage_fixture_cache(); + let capped = cache.prefix_usage("bucket", "", 2).expect("root query"); + assert!(capped.truncated, "three children capped to two must flag truncation"); + let names: Vec<&str> = capped.sub_prefixes.iter().map(|entry| entry.prefix.as_str()).collect(); + assert_eq!(names, vec!["b", "a"], "largest prefixes survive the cut"); + } + + #[test] + fn prefix_usage_marks_compacted_entries() { + let mut cache = DataUsageCache::default(); + cache.replace( + "bucket", + "", + DataUsageEntry { + size: 999, + objects: 9, + compacted: true, + ..Default::default() + }, + ); + + let compacted = cache.prefix_usage("bucket", "", 100).expect("compacted root resolves"); + assert!(compacted.compacted, "compaction must be visible to callers"); + assert_eq!(compacted.usage.size, 999); + assert!(compacted.sub_prefixes.is_empty(), "a compacted entry carries no children"); + } + + #[test] + fn prefix_usage_rejects_cyclic_and_dangling_caches() { + // A self-referencing child (corrupt cache) must yield a miss for the + // whole query, not unbounded recursion. + let mut cache = prefix_usage_fixture_cache(); + if let Some(entry) = cache.cache.get_mut("bucket/b") { + entry.children.insert("bucket/b".to_string()); + } + assert!(cache.prefix_usage("bucket", "b", 100).is_none(), "a cyclic subtree must be rejected"); + // The unaffected sibling still answers. + assert!(cache.prefix_usage("bucket", "a", 100).is_some()); + + // A child key with no entry (dangling link) is rejected rather than + // silently dropped: half a tree would under-report usage. + let mut dangling = prefix_usage_fixture_cache(); + if let Some(entry) = dangling.cache.get_mut("bucket/b") { + entry.children.insert("bucket/b/ghost".to_string()); + } + assert!( + dangling.prefix_usage("bucket", "b", 100).is_none(), + "a dangling child link must be rejected" + ); + } + #[test] fn hash_path_uses_portable_slash_semantics() { for (input, expected) in [ diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index f51fa6df5..8d02ef441 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -216,6 +216,16 @@ impl std::fmt::Debug for ECStore { /// These delegate to the process-global statics. No local state — the globals /// remain the single source of truth until the migration is complete. impl ECStore { + /// Every erasure set across all pools, pool-major order. + /// + /// Read-only queries that must consult each set's own copy of a + /// per-bucket object (e.g. the scanner's `.usage-cache.bin`) iterate + /// this instead of the hash-routed store path, which would always land + /// on one set (rustfs/backlog#1872). + pub fn all_set_disks(&self) -> Vec> { + self.pools.iter().flat_map(|pool| pool.disk_set.iter().cloned()).collect() + } + /// Get server configuration (delegates to global) pub fn get_server_config(&self) -> Option { runtime_sources::server_config() diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 67a199661..8ac714314 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -28,7 +28,8 @@ use rustfs_common::heal_channel::HealScanMode; 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, TierStats, hash_path, + DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, LEGACY_DATA_USAGE_OBJECT_NAME, PrefixUsageEntry, + PrefixUsageQuery, PrefixUsageSummary, TierStats, hash_path, prefix_usage_in_cache, }; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use tokio::time::{Duration, Instant, sleep, timeout}; @@ -430,6 +431,13 @@ pub(crate) enum DataUsageCachePrepareOutcome { } impl DataUsageCache { + /// Prefix-level usage query over this (writer-side) cache; see + /// [`prefix_usage_in_cache`] for the semantics + /// (rustfs/backlog#1872). + pub fn prefix_usage(&self, bucket: &str, prefix: &str, max_entries: usize) -> Option { + prefix_usage_in_cache(&self.cache, bucket, prefix, max_entries) + } + pub(crate) fn prepare_for_scan( &mut self, name: &str, diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 4a5cc7543..36cc6b817 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -53,6 +53,7 @@ use tokio_util::sync::CancellationToken; pub mod data_usage_define; pub mod error; +pub mod prefix_usage; mod remote_scanner; pub mod runtime_config; pub mod scanner; @@ -64,6 +65,7 @@ pub(crate) mod storage_api; pub use data_usage_define::*; pub use error::ScannerError; +pub use prefix_usage::{BucketPrefixUsageResponse, bucket_prefix_usage, invalidate_prefix_usage_cache}; pub use remote_scanner::{ NS_SCANNER_MAX_REQUEST_BODY_SIZE, RemoteScannerAdmission, RemoteScannerRequest, admit_remote_scanner_request, claim_remote_scanner_request, decode_remote_scanner_request, preflight_remote_scanner_request, diff --git a/crates/scanner/src/prefix_usage.rs b/crates/scanner/src/prefix_usage.rs new file mode 100644 index 000000000..9f94957de --- /dev/null +++ b/crates/scanner/src/prefix_usage.rs @@ -0,0 +1,349 @@ +// 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. + +//! Prefix-level bucket usage for admin/console consumers (rustfs/backlog#1872, +//! MinIO `loadPrefixUsageFromBackend` parity). +//! +//! The per-bucket, per-set `.usage-cache.bin` objects already hold a +//! path-keyed prefix tree; this module reads every set's copy through that +//! set's own object layer (the hash-routed store path would always land on +//! one set), aggregates the overlapping trees, and serves the result from a +//! bounded 30-second cache. Bucket writes poke the cache through the +//! dirty-usage hook so a fresh scan is visible immediately. + +use crate::data_usage_define::{DATA_USAGE_CACHE_NAME, DataUsageCache}; +use crate::error::ScannerError; +use crate::storage_api::owner::{ + EcstoreSetDisks, EcstoreStore, ecstore_is_reserved_or_invalid_bucket, ecstore_resolve_object_store_handle, +}; +use futures::future::join_all; +use rustfs_data_usage::{PrefixUsageEntry, PrefixUsageSummary}; +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, SystemTime}; +use tracing::{debug, warn}; + +const LOG_COMPONENT_SCANNER: &str = "scanner"; +const LOG_SUBSYSTEM_PREFIX_USAGE: &str = "prefix_usage"; +const EVENT_PREFIX_USAGE_CACHE_STATE: &str = "prefix_usage_cache_state"; + +/// How long a computed breakdown stays fresh. MinIO uses the same 30s for +/// its prefix-usage cache; bucket writes additionally invalidate on the spot. +const CACHE_TTL: Duration = Duration::from_secs(30); +/// Hard entry cap for the result cache; exceeded, expired entries go first +/// and the map clears rather than growing past the bound. +const CACHE_MAX_ENTRIES: usize = 128; +/// Per-set cache read budget. The underlying loader retries for up to a +/// minute per attempt on backend errors — far too long for an admin GET, so +/// a slow set degrades to "not reporting" instead of stalling the caller. +const PER_SET_LOAD_TIMEOUT: Duration = Duration::from_secs(5); + +/// Aggregated prefix-usage answer across every erasure set. +#[derive(Clone, Debug, PartialEq, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct BucketPrefixUsageResponse { + pub bucket: String, + pub prefix: String, + pub usage: PrefixUsageSummary, + /// Every reporting set's prefix entry was compacted: the aggregate is + /// valid, the sub-prefix breakdown is empty on disk. + pub compacted: bool, + /// The sub-prefix breakdown is incomplete: at least one reporting set + /// had the prefix compacted (or absent while others found it), so its + /// objects cannot be attributed to a sub-prefix. + pub sub_prefixes_partial: bool, + /// The breakdown exceeded the caller's entry limit; largest remain. + pub truncated: bool, + pub sub_prefixes: Vec, + /// Sets whose cache held this bucket and prefix. + pub sets_reporting: usize, + pub sets_total: usize, + /// Newest `last_update` across reporting sets, unix seconds. + pub last_update_unix_secs: Option, +} + +#[derive(Clone)] +struct CachedResponse { + computed_at: std::time::Instant, + response: Arc, +} + +/// Cache key: (lowercased bucket, normalized prefix, max entries). +type PrefixUsageCacheKey = (String, String, usize); +type PrefixUsageCacheMap = Option>; + +static PREFIX_USAGE_CACHE: Mutex = Mutex::new(None); + +/// Drop cached results for `bucket` (empty string clears everything). Wired +/// into the dirty-usage recording path so a write makes the next prefix +/// query recompute instead of serving up to `CACHE_TTL` seconds of stale +/// numbers. +pub fn invalidate_prefix_usage_cache(bucket: &str) { + let mut guard = PREFIX_USAGE_CACHE.lock().unwrap_or_else(|poison| poison.into_inner()); + let Some(map) = guard.as_mut() else { + return; + }; + if bucket.is_empty() { + map.clear(); + return; + } + map.retain(|(cached_bucket, ..), _| !cached_bucket.eq_ignore_ascii_case(bucket)); +} + +/// Query prefix usage for `bucket` (arbitrary `prefix`, empty = whole +/// bucket), merging every erasure set's own cache copy. `max_entries` bounds +/// the sub-prefix rows (largest first). +pub async fn bucket_prefix_usage( + bucket: &str, + prefix: &str, + max_entries: usize, +) -> Result { + if ecstore_is_reserved_or_invalid_bucket(bucket, true) { + return Err(ScannerError::Other(format!("invalid bucket name: {bucket}"))); + } + let normalized_prefix = prefix.trim_matches('/').to_string(); + let cache_key = (bucket.to_ascii_lowercase(), normalized_prefix.clone(), max_entries); + if let Some(response) = lookup_cached(&cache_key) { + return Ok((*response).clone()); + } + + let store = ecstore_resolve_object_store_handle() + .ok_or_else(|| ScannerError::Other("object store is not initialized".to_string()))?; + let response = Arc::new(compute_prefix_usage(store, bucket, &normalized_prefix, max_entries).await); + store_cached(cache_key, response.clone()); + Ok((*response).clone()) +} + +async fn compute_prefix_usage( + store: Arc, + bucket: &str, + prefix: &str, + max_entries: usize, +) -> BucketPrefixUsageResponse { + let sets: Vec> = store.all_set_disks(); + let sets_total = sets.len(); + let cache_name = format!("{bucket}/{DATA_USAGE_CACHE_NAME}"); + + let per_set = join_all(sets.into_iter().map(|set| { + let cache_name = cache_name.clone(); + async move { + let mut cache = DataUsageCache::default(); + // A set that has never scanned this bucket (or cannot be read + // within the budget) reports nothing — the remaining sets still + // produce a usable, flagged answer. + let loaded = match tokio::time::timeout(PER_SET_LOAD_TIMEOUT, cache.load(set, &cache_name)).await { + Ok(Ok(())) => cache, + Ok(Err(err)) => { + debug!( + target: "rustfs::scanner::prefix_usage", + event = EVENT_PREFIX_USAGE_CACHE_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_PREFIX_USAGE, + bucket = %bucket, + state = "set_load_failed", + error = %err, + "Prefix usage set cache load failed" + ); + return None; + } + Err(_) => { + warn!( + target: "rustfs::scanner::prefix_usage", + event = EVENT_PREFIX_USAGE_CACHE_STATE, + component = LOG_COMPONENT_SCANNER, + subsystem = LOG_SUBSYSTEM_PREFIX_USAGE, + bucket = %bucket, + state = "set_load_timeout", + "Prefix usage set cache load timed out" + ); + return None; + } + }; + if loaded.info.name != bucket { + // Empty or stale-scoped cache: this set has no data for the bucket. + return None; + } + let last_update = loaded.info.last_update; + let query = loaded.prefix_usage(bucket, prefix, max_entries); + Some((query, last_update)) + } + })) + .await; + + let mut usage = PrefixUsageSummary::default(); + let mut sub_prefix_map: HashMap = HashMap::new(); + let mut sets_reporting = 0usize; + let mut reporting_but_absent = 0usize; + let mut any_compacted = false; + let mut all_compacted = true; + let mut truncated = false; + let mut last_update: Option = None; + + for (query, set_last_update) in per_set.into_iter().flatten() { + // last_update counts every set that has scanned the bucket, even + // when the prefix itself is absent on that set. + if let Some(set_last_update) = set_last_update + && last_update.map(|current| set_last_update > current).unwrap_or(true) + { + last_update = Some(set_last_update); + } + let Some(query) = query else { + // The set knows the bucket but not this prefix: legitimate when + // the prefix's objects all hash to other sets, but it means the + // breakdown below cannot attribute that set's (zero) objects. + reporting_but_absent += 1; + continue; + }; + sets_reporting += 1; + usage.merge(&query.usage); + if query.compacted { + any_compacted = true; + } else { + all_compacted = false; + } + truncated |= query.truncated; + for entry in query.sub_prefixes { + sub_prefix_map.entry(entry.prefix).or_default().merge(&entry.usage); + } + } + + let mut sub_prefixes: Vec = sub_prefix_map + .into_iter() + .map(|(prefix, usage)| PrefixUsageEntry { prefix, usage }) + .collect(); + sub_prefixes.sort_by(|left, right| { + right + .usage + .size + .cmp(&left.usage.size) + .then_with(|| left.prefix.cmp(&right.prefix)) + }); + // Merged rows can exceed max_entries only when per-set truncation + // already flagged; enforce the caller bound on the merged view too. + if sub_prefixes.len() > max_entries { + truncated = true; + sub_prefixes.truncate(max_entries); + } + + let found = sets_reporting > 0; + BucketPrefixUsageResponse { + bucket: bucket.to_string(), + prefix: prefix.to_string(), + usage, + compacted: found && all_compacted, + sub_prefixes_partial: any_compacted || reporting_but_absent > 0, + truncated, + sub_prefixes, + sets_reporting, + sets_total, + last_update_unix_secs: last_update + .and_then(|time| time.duration_since(SystemTime::UNIX_EPOCH).ok()) + .map(|dur| dur.as_secs()), + } +} + +fn lookup_cached(key: &(String, String, usize)) -> Option> { + let mut guard = PREFIX_USAGE_CACHE.lock().unwrap_or_else(|poison| poison.into_inner()); + let map = guard.as_mut()?; + let cached = map.get(key)?; + if cached.computed_at.elapsed() > CACHE_TTL { + map.remove(key); + return None; + } + Some(cached.response.clone()) +} + +fn store_cached(key: (String, String, usize), response: Arc) { + let mut guard = PREFIX_USAGE_CACHE.lock().unwrap_or_else(|poison| poison.into_inner()); + let map = guard.get_or_insert_with(HashMap::new); + // Bound the cache: drop expired entries first, and if the cap is still + // exceeded clear wholesale — the next queries recompute in milliseconds. + if map.len() >= CACHE_MAX_ENTRIES { + map.retain(|_, cached| cached.computed_at.elapsed() <= CACHE_TTL); + if map.len() >= CACHE_MAX_ENTRIES { + map.clear(); + } + } + map.insert( + key, + CachedResponse { + computed_at: std::time::Instant::now(), + response, + }, + ); +} + +#[cfg(test)] +mod tests { + use super::{CACHE_MAX_ENTRIES, PREFIX_USAGE_CACHE, invalidate_prefix_usage_cache, store_cached}; + use rustfs_data_usage::PrefixUsageSummary; + + fn response(bucket: &str) -> super::BucketPrefixUsageResponse { + super::BucketPrefixUsageResponse { + bucket: bucket.to_string(), + prefix: String::new(), + usage: PrefixUsageSummary::default(), + compacted: false, + sub_prefixes_partial: false, + truncated: false, + sub_prefixes: Vec::new(), + sets_reporting: 1, + sets_total: 1, + last_update_unix_secs: None, + } + } + + fn seed(bucket: &str, prefix: &str) { + store_cached( + (bucket.to_ascii_lowercase(), prefix.to_string(), 10), + std::sync::Arc::new(response(bucket)), + ); + } + + fn contains(bucket: &str, prefix: &str) -> bool { + PREFIX_USAGE_CACHE + .lock() + .unwrap_or_else(|poison| poison.into_inner()) + .as_ref() + .is_some_and(|map| map.contains_key(&(bucket.to_ascii_lowercase(), prefix.to_string(), 10))) + } + + /// All cache tests run inside one test to keep the process-global map + /// free of cross-test ordering (the flake class this module avoids). + #[test] + fn invalidation_scopes_to_bucket_and_cache_stays_bounded() { + invalidate_prefix_usage_cache(""); + seed("alpha", "x"); + seed("beta", "y"); + + // Case-insensitive bucket scoping. + invalidate_prefix_usage_cache("ALPHA"); + assert!(!contains("alpha", "x")); + assert!(contains("beta", "y")); + + // Wholesale clear. + invalidate_prefix_usage_cache(""); + assert!(!contains("beta", "y")); + + // Hard cap: overflow clears rather than grows. + for index in 0..=(CACHE_MAX_ENTRIES / 2) { + let bucket = format!("cap-bucket-{index}"); + seed(&bucket, "a"); + seed(&bucket, "b"); + } + let guard = PREFIX_USAGE_CACHE.lock().unwrap_or_else(|poison| poison.into_inner()); + let map = guard.as_ref().expect("seeded"); + assert!(map.len() <= CACHE_MAX_ENTRIES, "cache must stay bounded, got {}", map.len()); + } +} diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 15ee9cca0..f722ff186 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -231,6 +231,10 @@ pub fn record_dirty_usage_bucket(bucket: &str) { dirty_buckets.len() }; global_metrics().record_scanner_dirty_usage_pending(usize_to_u64_saturated(pending_buckets)); + // A write invalidates this bucket's prefix-usage answers on the spot so + // admin/console consumers never ride the full TTL after a change + // (rustfs/backlog#1872). + crate::prefix_usage::invalidate_prefix_usage_cache(bucket); DIRTY_USAGE_BUCKET_NOTIFY.notify_one(); } diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index 8c837eb05..f0a32f402 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -64,6 +64,7 @@ mod target_descriptor; pub mod tier; pub mod tls_debug; pub mod trace; +pub mod usage_prefix; pub mod user; pub mod user_iam; pub mod user_lifecycle; diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index c7ad93b54..0909568e5 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -1158,10 +1158,10 @@ impl Operation for RuntimeCapabilitiesHandler { } } -/// Authorization gate for GET datausageinfo: any-of the dedicated admin action -/// OR the bucket listing action. Pinned by a unit test so the gate cannot -/// silently narrow or widen (rustfs/backlog#1306). -fn data_usage_info_gate_actions() -> Vec { +/// Authorization gate for GET datausageinfo (and prefix usage): any-of the +/// dedicated admin action OR the bucket listing action. Pinned by a unit test +/// so the gate cannot silently narrow or widen (rustfs/backlog#1306). +pub(crate) fn data_usage_info_gate_actions() -> Vec { vec![ Action::AdminAction(AdminAction::DataUsageInfoAdminAction), Action::S3Action(S3Action::ListBucketAction), diff --git a/rustfs/src/admin/handlers/usage_prefix.rs b/rustfs/src/admin/handlers/usage_prefix.rs new file mode 100644 index 000000000..10cce50d5 --- /dev/null +++ b/rustfs/src/admin/handlers/usage_prefix.rs @@ -0,0 +1,142 @@ +// 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. + +//! Prefix-level bucket usage admin handler (rustfs/backlog#1872). +//! +//! `GET /rustfs/admin/v3/usage/{bucket}?prefix=&max-entries=` answers +//! "what does this bucket / this prefix hold" from the scanner's per-set +//! usage caches, with a one-level sub-prefix breakdown — the data console +//! buckets view MinIO serves from `loadPrefixUsageFromBackend`. + +use crate::admin::auth::validate_admin_request; +use crate::admin::handlers::system::data_usage_info_gate_actions; +use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::auth::{check_key_valid, get_session_token}; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use http::{HeaderMap, HeaderValue, StatusCode}; +use hyper::Method; +use matchit::Params; +use s3s::header::CONTENT_TYPE; +use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; + +const JSON_CONTENT_TYPE: &str = "application/json"; +const DEFAULT_MAX_ENTRIES: usize = 1000; +const MAX_ENTRIES_LIMIT: usize = 10_000; + +pub struct BucketPrefixUsageHandler {} + +pub fn register_usage_prefix_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v3/usage/{bucket}").as_str(), + AdminOperation(&BucketPrefixUsageHandler {}), + )?; + Ok(()) +} + +/// Parse `prefix` and `max-entries` from the query string. Unknown keys are +/// rejected so a typo'd parameter cannot silently change the answer's shape. +fn parse_usage_prefix_query(query: Option<&str>) -> S3Result<(String, usize)> { + let mut prefix: Option = None; + let mut max_entries: Option = None; + for (key, value) in url::form_urlencoded::parse(query.unwrap_or_default().as_bytes()) { + match key.as_ref() { + "prefix" => prefix = Some(value.into_owned()), + "max-entries" => { + max_entries = Some( + value + .parse::() + .map_err(|_| s3_error!(InvalidArgument, "max-entries must be a positive integer"))?, + ); + } + other => return Err(s3_error!(InvalidArgument, "unknown query parameter: {other}")), + } + } + let max_entries = max_entries.unwrap_or(DEFAULT_MAX_ENTRIES).clamp(1, MAX_ENTRIES_LIMIT); + Ok((prefix.unwrap_or_default(), max_entries)) +} + +#[async_trait::async_trait] +impl Operation for BucketPrefixUsageHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let Some(input_cred) = req.credentials else { + return Err(s3_error!(InvalidRequest, "get cred failed")); + }; + + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + + let remote_addr = req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)); + validate_admin_request(&req.headers, &cred, owner, false, data_usage_info_gate_actions(), remote_addr).await?; + + let bucket = params.get("bucket").unwrap_or_default().to_string(); + if bucket.is_empty() { + return Err(s3_error!(InvalidRequest, "bucket path parameter is required")); + } + let (prefix, max_entries) = parse_usage_prefix_query(req.uri.query())?; + + // Authorization is bucket-scoped by the same any-of gate as the + // datausageinfo route; the bucket name itself is validated by the + // scanner layer, which rejects reserved/invalid names. + let response = rustfs_scanner::bucket_prefix_usage(&bucket, &prefix, max_entries) + .await + .map_err(|err| s3_error!(InvalidArgument, "{}", err))?; + + let data = serde_json::to_vec(&response) + .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "parse prefix usage failed"))?; + let mut header = HeaderMap::new(); + header.insert(CONTENT_TYPE, HeaderValue::from_static(JSON_CONTENT_TYPE)); + + Ok(S3Response::with_headers((StatusCode::OK, Body::from(data)), header)) + } +} + +#[cfg(test)] +mod tests { + use super::{DEFAULT_MAX_ENTRIES, MAX_ENTRIES_LIMIT, parse_usage_prefix_query}; + use s3s::S3Error; + + fn query(raw: &str) -> Result<(String, usize), S3Error> { + parse_usage_prefix_query(Some(raw)) + } + + #[test] + fn defaults_apply_when_no_query_is_given() { + assert_eq!(parse_usage_prefix_query(None).unwrap(), (String::new(), DEFAULT_MAX_ENTRIES)); + assert_eq!(query("").unwrap(), (String::new(), DEFAULT_MAX_ENTRIES)); + } + + #[test] + fn prefix_round_trips_url_encoded_characters() { + let (prefix, _) = query("prefix=pre%2Ffix%20name").unwrap(); + assert_eq!(prefix, "pre/fix name"); + } + + #[test] + fn max_entries_parses_and_clamps_to_documented_bounds() { + assert_eq!(query("max-entries=5").unwrap().1, 5); + assert_eq!(query("max-entries=0").unwrap().1, 1, "zero must clamp up, not mean unlimited"); + assert_eq!(query("max-entries=99999999").unwrap().1, MAX_ENTRIES_LIMIT); + assert!(query("max-entries=-3").is_err()); + assert!(query("max-entries=abc").is_err()); + } + + #[test] + fn unknown_parameters_are_rejected_not_ignored() { + assert!( + query("prefixes=x").is_err(), + "a typo'd parameter must fail the request, not widen the query" + ); + } +} diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index d61abf932..fe07d4056 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -40,7 +40,8 @@ use handlers::{ audit, batch_job, bucket_meta, cluster_snapshot, config_admin, diagnostics, durability as durability_handler, extensions, heal, health, idp_compat, ilm_transition, inspect_archive, kms, module_switch, object_data_cache, object_zip_download, oidc, plugins_catalog, plugins_instances, pools, profile_admin, quota as quota_handler, rebalance, - replication as replication_handler, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, user, + replication as replication_handler, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, usage_prefix, + user, }; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -80,6 +81,7 @@ fn register_admin_routes(r: &mut S3Router) -> std::io::Result<() bucket_meta::register_bucket_meta_route(r)?; config_admin::register_config_route(r)?; scanner::register_scanner_route(r)?; + usage_prefix::register_usage_prefix_route(r)?; ilm_transition::register_ilm_transition_route(r)?; object_data_cache::register_object_data_cache_route(r)?; audit::register_audit_target_route(r)?; diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 474d3ad0f..2c1ad5e1b 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -1558,6 +1558,11 @@ pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[ "/rustfs/admin/v3/datausageinfo", DeferredRoutePolicyReason::MultipleActions, ), + deferred( + HttpMethod::Get, + "/rustfs/admin/v3/usage/{bucket}", + DeferredRoutePolicyReason::MultipleActions, + ), deferred( HttpMethod::Post, "/rustfs/admin/v3/object-zip-downloads", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index f829b8c94..e81f13d17 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -172,6 +172,7 @@ fn expected_admin_route_matrix() -> Vec { admin_route(Method::POST, "/v4/inspect/archive"), admin_route(Method::GET, "/v3/storageinfo"), admin_route(Method::GET, "/v3/datausageinfo"), + admin_route_sample(Method::GET, "/v3/usage/{bucket}", "/v3/usage/test-bucket"), admin_route(Method::GET, "/v3/metrics"), admin_route(Method::GET, "/v3/object-data-cache/stats"), admin_route(Method::POST, "/v3/object-data-cache/flush"),