mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-17 18:27:49 +00:00
Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 28040dd1fa | |||
| c842d00074 | |||
| 59b7d13095 | |||
| e0b87b0e7e |
@@ -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<String, DataUsageEntry>,
|
||||
bucket: &str,
|
||||
prefix: &str,
|
||||
max_entries: usize,
|
||||
) -> Option<PrefixUsageQuery> {
|
||||
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<PrefixUsageEntry> = 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<String, DataUsageEntry>, root: &DataUsageEntry, depth: usize) -> Option<DataUsageEntry> {
|
||||
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<PrefixUsageEntry>,
|
||||
}
|
||||
|
||||
/// 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<PrefixUsageQuery> {
|
||||
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 [
|
||||
|
||||
@@ -1549,8 +1549,8 @@ impl Default for PutObjectOptions {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
impl PutObjectOptions {
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn set_match_etag(&mut self, etag: &str) {
|
||||
if etag == "*" {
|
||||
self.custom_header
|
||||
@@ -1561,6 +1561,7 @@ impl PutObjectOptions {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn set_match_etag_except(&mut self, etag: &str) {
|
||||
if etag == "*" {
|
||||
self.custom_header
|
||||
@@ -1696,6 +1697,7 @@ impl PutObjectOptions {
|
||||
header
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn validate(&self, _c: Arc<TargetClient>) -> Result<(), std::io::Error> {
|
||||
//if self.checksum.is_set() {
|
||||
/*if !self.trailing_header_support {
|
||||
|
||||
@@ -456,16 +456,23 @@ impl<'a> LifecycleExpiryTrace<'a> {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
impl ExpiryStats {
|
||||
pub fn missed_tasks(&self) -> i64 {
|
||||
self.missed_expiry_tasks.load(Ordering::SeqCst)
|
||||
}
|
||||
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
|
||||
)]
|
||||
fn missed_free_vers_tasks(&self) -> i64 {
|
||||
self.missed_freevers_tasks.load(Ordering::SeqCst)
|
||||
}
|
||||
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
|
||||
)]
|
||||
fn missed_tier_journal_tasks(&self) -> i64 {
|
||||
self.missed_tier_journal_tasks.load(Ordering::SeqCst)
|
||||
}
|
||||
|
||||
@@ -80,7 +80,10 @@ impl LastDayTierStats {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "asserted by this file's tests; the lib target cannot see test-only consumers (backlog#1823)"
|
||||
)]
|
||||
fn merge(&self, m: LastDayTierStats) -> LastDayTierStats {
|
||||
let mut cl = self.clone();
|
||||
let mut cm = m;
|
||||
|
||||
@@ -177,9 +177,10 @@ fn should_record_remote_delete_failure(err: &std::io::Error) -> bool {
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
#[allow(dead_code)]
|
||||
struct ObjSweeper {
|
||||
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
|
||||
object: String,
|
||||
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
|
||||
bucket: String,
|
||||
version_id: Option<Uuid>,
|
||||
versioned: bool,
|
||||
@@ -191,9 +192,9 @@ struct ObjSweeper {
|
||||
remote_object: String,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
impl ObjSweeper {
|
||||
#[allow(clippy::new_ret_no_self)]
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
pub async fn new(bucket: &str, object: &str) -> Result<Self, std::io::Error> {
|
||||
Ok(Self {
|
||||
object: object.into(),
|
||||
@@ -202,17 +203,20 @@ impl ObjSweeper {
|
||||
})
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
pub fn with_version(&mut self, vid: Option<Uuid>) -> &Self {
|
||||
self.version_id = vid.clone();
|
||||
self
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
pub fn with_versioning(&mut self, versioned: bool, suspended: bool) -> &Self {
|
||||
self.versioned = versioned;
|
||||
self.suspended = suspended;
|
||||
self
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
pub fn get_opts(&self) -> lifecycle::ObjectOpts {
|
||||
let mut opts = ObjectOpts {
|
||||
version_id: self.version_id.clone(),
|
||||
@@ -226,6 +230,7 @@ impl ObjSweeper {
|
||||
opts
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
pub fn set_transition_state(&mut self, info: TransitionedObject) {
|
||||
self.transition_tier = info.tier;
|
||||
self.transition_status = info.status;
|
||||
@@ -266,6 +271,7 @@ impl ObjSweeper {
|
||||
None
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
pub async fn sweep(&self, api: Arc<ECStore>) {
|
||||
let Some(je) = self.should_remove_remote_object() else {
|
||||
return;
|
||||
|
||||
@@ -312,9 +312,7 @@ mod tests {
|
||||
}
|
||||
#[derive(Deserialize)]
|
||||
struct LegacyBucketQuota {
|
||||
#[allow(dead_code)]
|
||||
quota: Option<u64>,
|
||||
#[allow(dead_code)]
|
||||
quota_type: LegacyQuotaType,
|
||||
}
|
||||
let legacy = serde_json::from_slice::<LegacyBucketQuota>(&json)
|
||||
|
||||
@@ -95,7 +95,6 @@ impl TransitionClient {
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
#[allow(dead_code)]
|
||||
pub struct GetRequest {
|
||||
pub buffer: Vec<u8>,
|
||||
pub offset: i64,
|
||||
@@ -107,11 +106,12 @@ pub struct GetRequest {
|
||||
pub setting_object_info: bool,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub struct GetResponse {
|
||||
pub size: i64,
|
||||
//pub error: error,
|
||||
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
|
||||
pub did_read: bool,
|
||||
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
|
||||
pub object_info: ObjectInfo,
|
||||
}
|
||||
|
||||
|
||||
@@ -27,7 +27,6 @@ use tracing::warn;
|
||||
use crate::client::api_error_response::err_invalid_argument;
|
||||
|
||||
#[derive(Default)]
|
||||
#[allow(dead_code)]
|
||||
pub struct AdvancedGetOptions {
|
||||
pub replication_delete_marker: bool,
|
||||
pub is_replication_ready_for_delete_marker: bool,
|
||||
|
||||
@@ -360,7 +360,6 @@ impl TransitionClient {
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
#[allow(dead_code)]
|
||||
pub struct ListObjectsOptions {
|
||||
reverse_versions: bool,
|
||||
with_versions: bool,
|
||||
|
||||
@@ -137,8 +137,8 @@ impl Default for PutObjectOptions {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
impl PutObjectOptions {
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn set_match_etag(&mut self, etag: &str) {
|
||||
if etag == "*" {
|
||||
self.custom_header.insert("If-Match", HeaderValue::from_static("*"));
|
||||
@@ -149,6 +149,7 @@ impl PutObjectOptions {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn set_match_etag_except(&mut self, etag: &str) {
|
||||
if etag == "*" {
|
||||
self.custom_header.insert("If-None-Match", HeaderValue::from_static("*"));
|
||||
@@ -259,6 +260,7 @@ impl PutObjectOptions {
|
||||
header
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn validate(&self, c: TransitionClient) -> Result<(), std::io::Error> {
|
||||
//if self.checksum.is_set() {
|
||||
/*if !self.trailing_header_support {
|
||||
|
||||
@@ -55,7 +55,6 @@ pub struct RemoveBucketOptions {
|
||||
const DELETE_RESPONSE_PREVIEW_LEN: usize = 1024;
|
||||
|
||||
#[derive(Debug)]
|
||||
#[allow(dead_code)]
|
||||
pub struct AdvancedRemoveOptions {
|
||||
pub replication_delete_marker: bool,
|
||||
pub replication_status: ReplicationStatus,
|
||||
@@ -465,10 +464,10 @@ impl TransitionClient {
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
#[allow(dead_code)]
|
||||
pub struct RemoveObjectError {
|
||||
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
|
||||
object_name: String,
|
||||
#[allow(dead_code)]
|
||||
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
|
||||
version_id: String,
|
||||
err: Option<std::io::Error>,
|
||||
}
|
||||
|
||||
@@ -372,8 +372,8 @@ pub struct Checksum {
|
||||
computed: bool,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
impl Checksum {
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn new(t: ChecksumMode, b: &[u8]) -> Checksum {
|
||||
if t.is_set() && b.len() == t.raw_byte_len() {
|
||||
return Checksum {
|
||||
@@ -385,7 +385,7 @@ impl Checksum {
|
||||
Checksum::default()
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn new_checksum_string(t: ChecksumMode, s: &str) -> Result<Checksum, std::io::Error> {
|
||||
let b = match base64_decode(s.as_bytes()) {
|
||||
Ok(b) => b,
|
||||
@@ -412,7 +412,7 @@ impl Checksum {
|
||||
base64_encode(&self.r)
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn raw(&self) -> Option<Vec<u8>> {
|
||||
if !self.is_set() {
|
||||
return None;
|
||||
|
||||
@@ -37,16 +37,17 @@ pub struct PutObjReader {
|
||||
//pub sealMD5Fn: SealMD5CurrFn,
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
impl PutObjReader {
|
||||
pub fn new(reader: HashReader) -> Self {
|
||||
Self { reader }
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn md5_current_hex_string(&self) -> String {
|
||||
self.reader.checksum().map(|v| v.encoded).unwrap_or_default()
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn with_encryption(&mut self, enc_reader: HashReader) -> Result<(), std::io::Error> {
|
||||
self.reader = enc_reader;
|
||||
|
||||
|
||||
@@ -39,7 +39,6 @@ use rustfs_config::{
|
||||
};
|
||||
use std::sync::LazyLock;
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[allow(clippy::declare_interior_mutable_const)]
|
||||
/// Default KVS for audit webhook settings.
|
||||
pub static DEFAULT_AUDIT_WEBHOOK_KVS: LazyLock<KVS> = LazyLock::new(|| {
|
||||
@@ -117,7 +116,6 @@ pub static DEFAULT_AUDIT_WEBHOOK_KVS: LazyLock<KVS> = LazyLock::new(|| {
|
||||
])
|
||||
});
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[allow(clippy::declare_interior_mutable_const)]
|
||||
/// Default KVS for audit MQTT settings.
|
||||
pub static DEFAULT_AUDIT_MQTT_KVS: LazyLock<KVS> = LazyLock::new(|| {
|
||||
@@ -375,7 +373,6 @@ pub static DEFAULT_AUDIT_NATS_KVS: LazyLock<KVS> = LazyLock::new(|| {
|
||||
])
|
||||
});
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub static DEFAULT_AUDIT_PULSAR_KVS: LazyLock<KVS> = LazyLock::new(|| {
|
||||
KVS(vec![
|
||||
KV {
|
||||
|
||||
@@ -12,12 +12,9 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::error::{Error, Result};
|
||||
use rustfs_config::server_config::{KV, KVS};
|
||||
use rustfs_config::{DEFAULT_HEAL_BITROT_CYCLE_SECS, HEAL_BITROT_CYCLE};
|
||||
use rustfs_utils::string::parse_bool;
|
||||
use std::sync::LazyLock;
|
||||
use std::time::Duration;
|
||||
|
||||
pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
|
||||
KVS(vec![KV {
|
||||
@@ -26,59 +23,3 @@ pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
|
||||
hidden_if_empty: false,
|
||||
}])
|
||||
});
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
pub struct Config {
|
||||
pub bitrot: String,
|
||||
pub sleep: Duration,
|
||||
pub io_count: usize,
|
||||
pub drive_workers: usize,
|
||||
pub cache: Duration,
|
||||
}
|
||||
|
||||
impl Config {
|
||||
pub fn bitrot_scan_cycle(&self) -> Duration {
|
||||
self.cache
|
||||
}
|
||||
|
||||
pub fn get_workers(&self) -> usize {
|
||||
self.drive_workers
|
||||
}
|
||||
|
||||
pub fn update(&mut self, nopts: &Config) {
|
||||
self.bitrot = nopts.bitrot.clone();
|
||||
self.io_count = nopts.io_count;
|
||||
self.sleep = nopts.sleep;
|
||||
self.drive_workers = nopts.drive_workers;
|
||||
}
|
||||
}
|
||||
|
||||
const RUSTFS_BITROT_CYCLE_IN_MONTHS: u64 = 1;
|
||||
|
||||
fn parse_bitrot_config(s: &str) -> Result<Duration> {
|
||||
match parse_bool(s) {
|
||||
Ok(enabled) => {
|
||||
if enabled {
|
||||
Ok(Duration::from_secs_f64(0.0))
|
||||
} else {
|
||||
Ok(Duration::from_secs_f64(-1.0))
|
||||
}
|
||||
}
|
||||
Err(_) => {
|
||||
if !s.ends_with("m") {
|
||||
return Err(Error::other("unknown format"));
|
||||
}
|
||||
|
||||
match s.trim_end_matches('m').parse::<u64>() {
|
||||
Ok(months) => {
|
||||
if months < RUSTFS_BITROT_CYCLE_IN_MONTHS {
|
||||
return Err(Error::other(format!("minimum bitrot cycle is {RUSTFS_BITROT_CYCLE_IN_MONTHS} month(s)")));
|
||||
}
|
||||
|
||||
Ok(Duration::from_secs(months * 30 * 24 * 60))
|
||||
}
|
||||
Err(err) => Err(Error::other(err)),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -16,7 +16,6 @@
|
||||
|
||||
mod audit;
|
||||
pub mod com;
|
||||
#[allow(dead_code)]
|
||||
pub mod heal;
|
||||
mod notify;
|
||||
mod oidc;
|
||||
|
||||
@@ -1996,11 +1996,11 @@ impl PoolMeta {
|
||||
Ok(false)
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub fn validate(&self, pools: Vec<Arc<Sets>>) -> Result<bool> {
|
||||
struct PoolInfo {
|
||||
position: usize,
|
||||
completed: bool,
|
||||
#[allow(dead_code, reason = "written but never read back (backlog#1823)")]
|
||||
decom_started: bool,
|
||||
}
|
||||
|
||||
@@ -4958,13 +4958,19 @@ fn is_disk_online_state(state: &str) -> bool {
|
||||
}
|
||||
|
||||
#[deprecated(since = "0.1.0", note = "Use fallback_total_capacity_dedup instead")]
|
||||
#[allow(dead_code)]
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "superseded by the replacement named in the comment at pools.rs:5071 (backlog#1823)"
|
||||
)]
|
||||
fn fallback_total_capacity(disks: &[rustfs_madmin::Disk]) -> usize {
|
||||
fallback_total_capacity_dedup(disks)
|
||||
}
|
||||
|
||||
#[deprecated(since = "0.1.0", note = "Use fallback_free_capacity_dedup instead")]
|
||||
#[allow(dead_code)]
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "superseded by the replacement named in the comment at pools.rs:5071 (backlog#1823)"
|
||||
)]
|
||||
fn fallback_free_capacity(disks: &[rustfs_madmin::Disk]) -> usize {
|
||||
fallback_free_capacity_dedup(disks)
|
||||
}
|
||||
|
||||
@@ -6562,7 +6562,7 @@ impl LocalDisk {
|
||||
Ok(f)
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn get_metrics(&self) -> DiskMetrics {
|
||||
DiskMetrics::default()
|
||||
}
|
||||
|
||||
@@ -132,7 +132,6 @@ impl RebalanceStopPropagationRecord {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct DiskStat {
|
||||
pub total_space: u64,
|
||||
|
||||
@@ -16,8 +16,16 @@ use serde::{Deserialize, Serialize};
|
||||
use std::{fmt::Display, io};
|
||||
use tracing::info;
|
||||
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "tier config wire version stamped by the parity constructors below (backlog#1823)"
|
||||
)]
|
||||
const C_TIER_CONFIG_VER: &str = "v1";
|
||||
|
||||
#[allow(
|
||||
dead_code,
|
||||
reason = "tier-name validation message reached only from the parity constructors below (backlog#1823)"
|
||||
)]
|
||||
const ERR_TIER_NAME_EMPTY: &str = "remote tier name empty";
|
||||
const WASABI_US_EAST_ENDPOINT: &str = "https://s3.wasabisys.com";
|
||||
const WASABI_ALTERNATIVE_ENDPOINTS: &[(&str, &str)] = &[
|
||||
@@ -264,7 +272,6 @@ impl Clone for TierConfig {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
impl TierConfig {
|
||||
pub(crate) fn clone_with_credentials(&self) -> Self {
|
||||
Self {
|
||||
@@ -284,6 +291,7 @@ impl TierConfig {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn endpoint(&self) -> String {
|
||||
match self.tier_type {
|
||||
TierType::S3 => self.s3.as_ref().map(|s| s.endpoint.clone()).unwrap_or_default(),
|
||||
@@ -303,6 +311,7 @@ impl TierConfig {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn bucket(&self) -> String {
|
||||
match self.tier_type {
|
||||
TierType::S3 => self.s3.as_ref().map(|s| s.bucket.clone()).unwrap_or_default(),
|
||||
@@ -322,6 +331,7 @@ impl TierConfig {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn prefix(&self) -> String {
|
||||
match self.tier_type {
|
||||
TierType::S3 => self.s3.as_ref().map(|s| s.prefix.clone()).unwrap_or_default(),
|
||||
@@ -341,6 +351,7 @@ impl TierConfig {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn region(&self) -> String {
|
||||
match self.tier_type {
|
||||
TierType::S3 => self.s3.as_ref().map(|s| s.region.clone()).unwrap_or_default(),
|
||||
@@ -457,7 +468,7 @@ impl TierWasabi {
|
||||
}
|
||||
|
||||
impl TierS3 {
|
||||
#[allow(dead_code)]
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn create<F>(
|
||||
name: &str,
|
||||
access_key: &str,
|
||||
@@ -528,7 +539,7 @@ pub struct TierMinIO {
|
||||
}
|
||||
|
||||
impl TierMinIO {
|
||||
#[allow(dead_code)]
|
||||
#[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
|
||||
fn create<F>(
|
||||
name: &str,
|
||||
endpoint: &str,
|
||||
|
||||
@@ -14,7 +14,6 @@
|
||||
|
||||
use crate::services::tier::tier::TierConfigMgr;
|
||||
|
||||
#[allow(dead_code)]
|
||||
impl TierConfigMgr {
|
||||
pub fn msg_size(&self) -> usize {
|
||||
100
|
||||
|
||||
@@ -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<Arc<crate::set_disk::SetDisks>> {
|
||||
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<Config> {
|
||||
runtime_sources::server_config()
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -6023,7 +6023,10 @@ fn edit_generation_wall_clock() -> u64 {
|
||||
/// node's clock behind the clock that fed the previous lifetime) mints
|
||||
/// below the stale mark and the origin stays fenced — but only until real
|
||||
/// time passes the previous lifetime's last allocation, because every later
|
||||
/// allocation takes the wall-clock floor again. Bounded by the skew,
|
||||
/// allocation takes the wall-clock floor again (and never longer than
|
||||
/// [`PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS`]: a regression past the window
|
||||
/// leaves the mark implausibly distant and the origin runs unfenced
|
||||
/// immediately). Bounded by the skew,
|
||||
/// self-healing, and no rollback window beyond the plain counter's: a
|
||||
/// delivery applies only at or above the receiver's mark, so the one
|
||||
/// cross-lifetime interleaving that can apply stale content — a
|
||||
@@ -6063,6 +6066,52 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
|
||||
Some((origin.clone(), generation))
|
||||
}
|
||||
|
||||
/// How far below the recorded high-water mark a delivery may sit and still
|
||||
/// be fenced as stale. The distance a GENUINE superseded delivery can trail
|
||||
/// its origin's mark is small: retransmissions re-run the sender flow and
|
||||
/// mint a fresh generation (the retry queue keys on the bare path and never
|
||||
/// replays a fenced URL), so only an in-flight straggler of the losing
|
||||
/// fan-out race trails the mark, by delivery latency — minutes at the
|
||||
/// outside. A mark further above than this window cannot be explained by
|
||||
/// any genuine race, only by a forged fence (the shared service account
|
||||
/// lets any peer stamp any origin) or by a persisted clock excursion the
|
||||
/// origin has since left behind — and fencing on it would silently drop the
|
||||
/// origin's real edits, so the stale check ignores it instead.
|
||||
const PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS: u64 = 24 * 60 * 60 * 1_000_000_000;
|
||||
|
||||
/// Whether an incoming fence may be honoured, as far as this site can vouch
|
||||
/// for it. The sender's identity is unverifiable (shared service account),
|
||||
/// so the check runs over what the receiving state knows: the claimed origin
|
||||
/// must be a site this state currently replicates with — the same membership
|
||||
/// rule the load-time mark pruning applies, so every mark recorded behind
|
||||
/// this check is one a reload would keep — and not this site itself, which
|
||||
/// never delivers edits to itself. The caller IGNORES an inadmissible fence
|
||||
/// rather than failing the request: the delivery applies exactly as an
|
||||
/// unstamped (pre-fence) delivery would, no high-water mark is read or
|
||||
/// written, and the worst a forged fence achieves is forfeiting an ordering
|
||||
/// guarantee its sender was never owed. The generation itself is NOT
|
||||
/// bounded here: a genuine origin whose hybrid clock persisted a wall-clock
|
||||
/// excursion allocates arbitrarily far in the future, and refusing to
|
||||
/// record its marks would strip the ordering fence from exactly the
|
||||
/// deliveries that still race — the staleness window on the read side is
|
||||
/// what defuses forged marks instead.
|
||||
fn peer_edit_fence_is_admissible(state: &SiteReplicationState, local_deployment_id: &str, fence: &(String, u64)) -> bool {
|
||||
let (origin, generation) = fence;
|
||||
if origin != local_deployment_id && state.peers.contains_key(origin) {
|
||||
return true;
|
||||
}
|
||||
warn!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
result = "fence_origin_not_a_remote_peer",
|
||||
origin = %origin,
|
||||
generation = *generation,
|
||||
"ignoring inadmissible peer-edit fence"
|
||||
);
|
||||
false
|
||||
}
|
||||
|
||||
/// True when a strictly newer edit from the same origin site already landed
|
||||
/// here. No lock on the sending side can order deliveries issued by two
|
||||
/// nodes of that site, so ordering is decided here, on the generation the
|
||||
@@ -6070,11 +6119,42 @@ fn peer_edit_fence(queries: &HashMap<String, String>) -> Option<(String, u64)> {
|
||||
/// stale: one edit legitimately fans out several deliveries under a single
|
||||
/// generation (the ILM-expiry edit sends every peer's record), and a replay of
|
||||
/// an applied delivery re-applies the same edit idempotently.
|
||||
///
|
||||
/// A mark more than [`PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS`] above the
|
||||
/// delivery is implausible and does NOT fence: the shared service account
|
||||
/// means any peer can stamp any origin, so a forged `u64::MAX`-scale mark
|
||||
/// would otherwise silently swallow the origin's genuine edits for good.
|
||||
/// Bounding the fence by distance instead of by an absolute ceiling keeps
|
||||
/// ordering intact wherever the origin's clock actually operates — two
|
||||
/// racing deliveries trail each other by seconds whether the hybrid clock
|
||||
/// tracks wall time or persists a long-gone excursion far ahead of it —
|
||||
/// while a mark no genuine race can explain merely downgrades the origin to
|
||||
/// unfenced (pre-fence) delivery instead of dropping its edits. (One genuine
|
||||
/// shape does land out here: a plain-counter straggler arriving after its
|
||||
/// origin's first hybrid-clock edit. It gets the same downgrade — applied
|
||||
/// unfenced — once, at upgrade time; fencing it instead would silence the
|
||||
/// mirror case, a hybrid-clock origin downgraded back to the plain counter.)
|
||||
fn peer_edit_delivery_is_stale(state: &SiteReplicationState, origin: &str, generation: u64) -> bool {
|
||||
state
|
||||
.applied_edit_generations
|
||||
.get(origin)
|
||||
.is_some_and(|applied| *applied > generation)
|
||||
let Some(applied) = state.applied_edit_generations.get(origin) else {
|
||||
return false;
|
||||
};
|
||||
if *applied <= generation {
|
||||
return false;
|
||||
}
|
||||
if *applied - generation > PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS {
|
||||
warn!(
|
||||
event = EVENT_ADMIN_SITE_REPLICATION_STATE,
|
||||
component = LOG_COMPONENT_ADMIN,
|
||||
subsystem = LOG_SUBSYSTEM_SITE_REPLICATION,
|
||||
result = "fence_mark_beyond_staleness_window",
|
||||
origin,
|
||||
generation,
|
||||
applied_mark = *applied,
|
||||
"ignoring implausibly distant peer-edit high-water mark"
|
||||
);
|
||||
return false;
|
||||
}
|
||||
true
|
||||
}
|
||||
|
||||
fn record_applied_peer_edit_generation(state: &mut SiteReplicationState, origin: &str, generation: u64) {
|
||||
@@ -10698,6 +10778,11 @@ impl Operation for SRPeerEditHandler {
|
||||
let outcome = update_site_replication_state_when_changed(move |state| {
|
||||
let mut incoming = incoming;
|
||||
let local_peer = local_peer_at_endpoint(commit_endpoint, state);
|
||||
// The fence is self-reported — the shared service account means
|
||||
// the sender cannot be identified — so it is honoured only after
|
||||
// the admissibility check, against the same state it will gate.
|
||||
let commit_fence =
|
||||
commit_fence.filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence));
|
||||
// Ordering fence: the sending site allocates the generation under
|
||||
// its state-object lock, so a delivery that lost the race carries
|
||||
// a generation this site has already passed. Applying it would
|
||||
@@ -13393,6 +13478,15 @@ mod tests {
|
||||
handler_block.contains("record_applied_peer_edit_generation(state, origin, *generation);"),
|
||||
"SRPeerEditHandler must record the applied generation so later stale deliveries are recognised"
|
||||
);
|
||||
// Fence hardening: origin and generation are self-reported by a
|
||||
// caller the shared service account cannot identify, so the handler
|
||||
// must pass the fence through the admissibility check — against the
|
||||
// same state the fence gates, i.e. inside the transaction — before
|
||||
// reading or raising any high-water mark.
|
||||
assert!(
|
||||
handler_block.contains(".filter(|fence| peer_edit_fence_is_admissible(state, &local_peer.deployment_id, fence))"),
|
||||
"SRPeerEditHandler must admit a fence only through peer_edit_fence_is_admissible inside the state transaction"
|
||||
);
|
||||
// P1-15 PR2: both halves of the fence and the edit they fence share
|
||||
// ONE transaction. Checking the fence against a state read outside the
|
||||
// lock would let the check pass on one snapshot and the write land on
|
||||
@@ -14769,6 +14863,121 @@ mod tests {
|
||||
assert!(peer_edit_delivery_is_stale(&state, origin, generation - 1));
|
||||
}
|
||||
|
||||
/// A fence is self-reported: every site authenticates peer traffic with
|
||||
/// the same site-replicator credential, so a compromised peer can stamp
|
||||
/// ANY origin with ANY generation. An origin the receiver does not
|
||||
/// replicate with — or the receiver itself — is ignored and plants no
|
||||
/// mark; a mark a compromised peer plants for a CURRENT origin cannot
|
||||
/// silence that origin, because the staleness window refuses to fence on
|
||||
/// a mark implausibly far above the genuine deliveries.
|
||||
#[test]
|
||||
fn forged_peer_edit_fences_cannot_poison_the_high_water_marks() {
|
||||
let mut state = SiteReplicationState {
|
||||
peers: BTreeMap::from([
|
||||
(
|
||||
"site-local".to_string(),
|
||||
PeerInfo {
|
||||
deployment_id: "site-local".to_string(),
|
||||
..peer("local", "https://local.example:9000")
|
||||
},
|
||||
),
|
||||
(
|
||||
"site-victim".to_string(),
|
||||
PeerInfo {
|
||||
deployment_id: "site-victim".to_string(),
|
||||
..peer("victim", "https://victim.example:9000")
|
||||
},
|
||||
),
|
||||
]),
|
||||
..Default::default()
|
||||
};
|
||||
// An origin outside the current membership is refused outright...
|
||||
let unknown = ("site-unknown".to_string(), 4u64);
|
||||
assert!(!peer_edit_fence_is_admissible(&state, "site-local", &unknown));
|
||||
|
||||
// No site delivers edits to itself: a fence claiming the receiver as
|
||||
// its origin is forged by construction, current peer or not.
|
||||
let own = ("site-local".to_string(), 4u64);
|
||||
assert!(!peer_edit_fence_is_admissible(&state, "site-local", &own));
|
||||
|
||||
// A current remote peer's fence is admitted and works end to end.
|
||||
let genuine = ("site-victim".to_string(), 1u64);
|
||||
assert!(peer_edit_fence_is_admissible(&state, "site-local", &genuine));
|
||||
assert!(!peer_edit_delivery_is_stale(&state, &genuine.0, genuine.1));
|
||||
record_applied_peer_edit_generation(&mut state, &genuine.0, genuine.1);
|
||||
assert_eq!(state.applied_edit_generations.get("site-victim"), Some(&1));
|
||||
|
||||
// A forged u64::MAX-scale mark CAN be recorded — the shared service
|
||||
// account means the receiver cannot tell the stamp was forged — but
|
||||
// it is inert: the victim's genuine hybrid-clock deliveries sit far
|
||||
// more than the staleness window below it, so they keep applying
|
||||
// instead of being silently acked-and-dropped.
|
||||
record_applied_peer_edit_generation(&mut state, "site-victim", u64::MAX);
|
||||
assert!(!peer_edit_delivery_is_stale(&state, "site-victim", edit_generation_wall_clock()));
|
||||
}
|
||||
|
||||
/// The staleness window bounds the fence by DISTANCE from the mark, not
|
||||
/// by an absolute clock ceiling, so ordering must hold wherever the
|
||||
/// origin's hybrid clock actually operates. The regression that matters:
|
||||
/// a temporary wall-clock excursion far in the future is persisted by
|
||||
/// `next_peer_edit_generation` (`max(now, prev + 1)` never comes back
|
||||
/// down), and two later edits g+1 then g can arrive in reverse order —
|
||||
/// g must still be fenced, even though both generations dwarf the
|
||||
/// receiver's clock. Conversely a mark further above a delivery than any
|
||||
/// genuine race can explain must not fence it.
|
||||
#[test]
|
||||
fn peer_edit_fence_orders_a_persisted_future_clock_and_defuses_distant_marks() {
|
||||
let mut state = SiteReplicationState {
|
||||
peers: BTreeMap::from([(
|
||||
"site-origin".to_string(),
|
||||
PeerInfo {
|
||||
deployment_id: "site-origin".to_string(),
|
||||
..peer("origin", "https://origin.example:9000")
|
||||
},
|
||||
)]),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
// The origin's clock once jumped ten years ahead; the hybrid clock
|
||||
// keeps allocating from there long after the clock was corrected.
|
||||
let excursion = edit_generation_wall_clock() + 10 * 365 * 24 * 60 * 60 * 1_000_000_000;
|
||||
let fence = ("site-origin".to_string(), excursion + 1);
|
||||
assert!(peer_edit_fence_is_admissible(&state, "site-local", &fence));
|
||||
record_applied_peer_edit_generation(&mut state, &fence.0, fence.1);
|
||||
|
||||
// The reverse delivery of the race: g arrives after g+1 landed.
|
||||
// Without the fence it would commit last and roll g+1 back.
|
||||
assert!(peer_edit_delivery_is_stale(&state, "site-origin", excursion));
|
||||
// Equal generation (same edit's fan-out or a replay) still applies,
|
||||
// as does the next edit.
|
||||
assert!(!peer_edit_delivery_is_stale(&state, "site-origin", excursion + 1));
|
||||
assert!(!peer_edit_delivery_is_stale(&state, "site-origin", excursion + 2));
|
||||
|
||||
// The window's exact boundary: a delivery trailing the mark by the
|
||||
// full window is still fenced; one nanosecond further is not — that
|
||||
// distance is no longer explicable by a genuine race, only by a
|
||||
// forged mark or an excursion the origin has left behind.
|
||||
let mark = fence.1;
|
||||
// A straggler trailing by a concrete hour must still be fenced —
|
||||
// pins the window's real magnitude, not just its symbolic boundary.
|
||||
assert!(peer_edit_delivery_is_stale(&state, "site-origin", mark - 60 * 60 * 1_000_000_000));
|
||||
assert!(peer_edit_delivery_is_stale(
|
||||
&state,
|
||||
"site-origin",
|
||||
mark - PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS
|
||||
));
|
||||
assert!(!peer_edit_delivery_is_stale(
|
||||
&state,
|
||||
"site-origin",
|
||||
mark - PEER_EDIT_FENCE_STALENESS_WINDOW_NANOS - 1
|
||||
));
|
||||
|
||||
// A pre-hybrid plain-counter origin trails such a mark by eons: it
|
||||
// is not fenced (the rc.2-era downgrade case), it just runs
|
||||
// unfenced until its counter regime catches up.
|
||||
assert!(!peer_edit_delivery_is_stale(&state, "site-origin", 3));
|
||||
}
|
||||
|
||||
/// P1-15 review follow-up: a site that leaves the mesh drops below two
|
||||
/// peers, which clears its state object and restarts its generation
|
||||
/// counter at zero. A mark left over from its previous membership would
|
||||
|
||||
@@ -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<Action> {
|
||||
/// 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<Action> {
|
||||
vec![
|
||||
Action::AdminAction(AdminAction::DataUsageInfoAdminAction),
|
||||
Action::S3Action(S3Action::ListBucketAction),
|
||||
|
||||
@@ -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<AdminOperation>) -> 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<String> = None;
|
||||
let mut max_entries: Option<usize> = 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::<usize>()
|
||||
.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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
|
||||
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::<Option<RemoteAddr>>().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"
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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<AdminOperation>) -> 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)?;
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -172,6 +172,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
|
||||
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"),
|
||||
|
||||
Reference in New Issue
Block a user