feat(scanner): expose prefix-level bucket usage via admin API (HS-08) (#6171)

feat(scanner): expose prefix-level bucket usage via admin API

The scanner's per-bucket, per-set usage caches already hold a path-keyed
prefix tree, but dui() flattened it only to bucket names — consoles and
operators had no way to ask "what does this prefix hold" without an S3
listing sweep (rustfs/backlog#1872, MinIO loadPrefixUsageFromBackend
parity).

Add:

- data-usage: prefix_usage_in_cache — a shared aggregation over the
  entry map (arbitrary prefix, full counters, one-level sub-prefix
  breakdown with names recovered from the literal-path cache keys),
  hardened like the scanner's checked flatten: cycles, dangling child
  links, over-deep trees, and overflowing counters yield None rather
  than unbounded recursion or wrapped totals.
- ecstore: ECStore::all_set_disks — iterate every erasure set so a
  query can read each set's own cache copy; the hash-routed store path
  would always land on one set.
- scanner: bucket_prefix_usage — per-set loads (5s budget each, a slow
  set degrades to not-reporting instead of stalling the caller),
  merged across sets with partial/compacted/truncated flags, served
  from a bounded 30s cache (128 entries, hard-capped) that bucket
  writes invalidate through the dirty-usage hook.
- admin: GET /rustfs/admin/v3/usage/{bucket}?prefix=&max-entries=
  behind the same any-of gate as datausageinfo (DataUsageInfoAdminAction
  OR ListBucketAction), rejecting unknown query parameters and
  clamping max-entries to 1..=10000. Route registered in the policy
  table (deferred MultipleActions, matching datausageinfo) and the
  route matrix test.

Closes rustfs/backlog#1872.

Co-authored-by: heihutu <heihutu@gmail.com>

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<PrefixUsageQuery> {
+        prefix_usage_in_cache(&self.cache, bucket, prefix, max_entries)
+    }
+
     pub(crate) fn prepare_for_scan(
         &mut self,
         name: &str,

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,

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<PrefixUsageEntry>,
+    /// 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<u64>,
+}
+
+#[derive(Clone)]
+struct CachedResponse {
+    computed_at: std::time::Instant,
+    response: Arc<BucketPrefixUsageResponse>,
+}
+
+/// Cache key: (lowercased bucket, normalized prefix, max entries).
+type PrefixUsageCacheKey = (String, String, usize);
+type PrefixUsageCacheMap = Option<HashMap<PrefixUsageCacheKey, CachedResponse>>;
+
+static PREFIX_USAGE_CACHE: Mutex<PrefixUsageCacheMap> = 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<BucketPrefixUsageResponse, ScannerError> {
+    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<EcstoreStore>,
+    bucket: &str,
+    prefix: &str,
+    max_entries: usize,
+) -> BucketPrefixUsageResponse {
+    let sets: Vec<Arc<EcstoreSetDisks>> = 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<String, PrefixUsageSummary> = 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<SystemTime> = 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<PrefixUsageEntry> = 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<Arc<BucketPrefixUsageResponse>> {
+    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<BucketPrefixUsageResponse>) {
+    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());
+    }
+}

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();
 }