gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(scanner): preserve complete usage snapshots (#3654)

Browse Source
This commit is contained in:
GatewayJ
2026-06-21 08:21:20 +08:00
committed by GitHub
parent 796f293b80
commit c47dea0009
1 changed files with 186 additions and 328 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/scanner_io.rs
+186 -328
View File
@@ -30,7 +30,7 @@ use rustfs_filemeta::FileMeta;
use rustfs_storage_api::{BucketInfo, BucketOperations, BucketOptions, DiskSetSelector, StorageAdminApi};
use rustfs_utils::path::path_join_buf;
use s3s::dto::{BucketLifecycleConfiguration, ReplicationConfiguration};
use std::collections::{HashMap, HashSet};
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{LazyLock, Mutex as StdMutex, MutexGuard};
@@ -406,20 +406,87 @@ fn cache_root_entry_info(cache: &DataUsageCache) -> DataUsageEntryInfo {
}
}
fn apply_bucket_result_to_cache(
cache: &mut DataUsageCache,
result: DataUsageEntryInfo,
update_time: SystemTime,
publish_immediately: bool,
) -> bool {
fn apply_bucket_result_to_cache(cache: &mut DataUsageCache, result: DataUsageEntryInfo, update_time: SystemTime) {
cache.replace(&result.name, &result.parent, result.entry);
cache.info.last_update = Some(update_time);
publish_immediately
}
fn bucket_result_should_publish_immediately(published_buckets: &mut HashSet<String>, bucket_name: &str) -> bool {
published_buckets.insert(bucket_name.to_string())
fn should_publish_completed_snapshot(completed_count: usize, total_count: usize, budget_elapsed: bool, cancelled: bool) -> bool {
total_count > 0 && completed_count == total_count && !budget_elapsed && !cancelled
}
fn completed_data_usage_info(
results: &[DataUsageCache],
all_buckets: &[String],
budget_elapsed: bool,
cancelled: bool,
) -> Option<(DataUsageInfo, SystemTime)> {
let completed_set_count = results.iter().filter(|result| result.info.last_update.is_some()).count();
if !should_publish_completed_snapshot(completed_set_count, results.len(), budget_elapsed, cancelled) {
return None;
}
let mut all_merged = DataUsageCache::default();
for result in results.iter() {
all_merged.merge(result);
}
let merged_last_update = all_merged.info.last_update.unwrap_or(SystemTime::UNIX_EPOCH);
all_merged.root()?;
Some((all_merged.dui(&all_merged.info.name, all_buckets), merged_last_update))
}
#[cfg(test)]
mod publish_gate_tests {
use super::*;
#[test]
fn should_publish_completed_snapshot_requires_full_clean_cycle() {
assert!(should_publish_completed_snapshot(3, 3, false, false));
assert!(!should_publish_completed_snapshot(2, 3, false, false));
assert!(!should_publish_completed_snapshot(3, 3, true, false));
assert!(!should_publish_completed_snapshot(3, 3, false, true));
assert!(!should_publish_completed_snapshot(0, 0, false, false));
}
fn completed_root_cache(bucket: &str, objects: usize, update_secs: u64) -> DataUsageCache {
let mut cache = DataUsageCache {
info: DataUsageCacheInfo {
name: DATA_USAGE_ROOT.to_string(),
last_update: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(update_secs)),
..Default::default()
},
..Default::default()
};
cache.replace(
bucket,
DATA_USAGE_ROOT,
DataUsageEntry {
objects,
size: objects * 10,
..Default::default()
},
);
cache
}
#[test]
fn completed_data_usage_info_requires_every_set_before_publish() {
let all_buckets = vec!["bucket-a".to_string(), "bucket-b".to_string()];
let first_set = completed_root_cache("bucket-a", 1, 10);
let second_set = completed_root_cache("bucket-b", 2, 20);
assert!(completed_data_usage_info(&[first_set.clone(), DataUsageCache::default()], &all_buckets, false, false).is_none());
assert!(completed_data_usage_info(&[first_set.clone(), second_set.clone()], &all_buckets, true, false).is_none());
assert!(completed_data_usage_info(&[first_set.clone(), second_set.clone()], &all_buckets, false, true).is_none());
let (data_usage_info, last_update) = completed_data_usage_info(&[first_set, second_set], &all_buckets, false, false)
.expect("all completed sets should produce a publishable data usage snapshot");
assert_eq!(last_update, SystemTime::UNIX_EPOCH + Duration::from_secs(20));
assert_eq!(data_usage_info.objects_total_count, 3);
assert_eq!(data_usage_info.buckets_usage.len(), 2);
}
}
async fn send_cache_root_entry_info(
@@ -467,6 +534,20 @@ async fn persist_and_publish_cache_snapshot<S: ScannerObjectIO>(
last_update
}
async fn send_merged_data_usage_update(updates: &mpsc::Sender<DataUsageInfo>, data_usage_info: DataUsageInfo) {
if let Err(e) = updates.send(data_usage_info).await {
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
state = "send_merged_failed",
error = %e,
"Scanner merged data usage publish failed"
);
}
}
#[async_trait::async_trait]
pub trait ScannerIO: Send + Sync + Debug + 'static {
async fn nsscanner(
@@ -583,14 +664,13 @@ impl ScannerIO for ECStore {
let results = vec![DataUsageCache::default(); total_results];
let results_mutex: Arc<Mutex<Vec<DataUsageCache>>> = Arc::new(Mutex::new(results));
let first_err_mutex: Arc<Mutex<Option<Error>>> = Arc::new(Mutex::new(None));
let mut results_index: i32 = -1_i32;
let mut results_index = 0usize;
let mut wait_futs = Vec::new();
for pool in self.pools.iter() {
for set in pool.disk_set.iter() {
let results_index_clone = results_index;
results_index += 1;
let results_index_clone = results_index as usize;
// Clone the Arc to move it into the spawned task
let set_clone: Arc<SetDisks> = Arc::clone(set);
let pool_label = set.pool_index.to_string();
@@ -690,6 +770,8 @@ impl ScannerIO for ECStore {
let all_buckets_clone = all_buckets.iter().map(|b| b.name.clone()).collect::<Vec<String>>();
let results_mutex_for_updates = results_mutex.clone();
let budget_for_updates = budget.clone();
let child_token_for_updates = child_token.clone();
tokio::spawn(async move {
let mut last_update = SystemTime::UNIX_EPOCH;
let mut has_sent_once = false;
@@ -697,7 +779,7 @@ impl ScannerIO for ECStore {
let mut ticker = tokio::time::interval(Duration::from_secs(30));
loop {
tokio::select! {
_ = child_token.cancelled() => {
_ = child_token_for_updates.cancelled() => {
break;
}
res = &mut update_rx => {
@@ -705,56 +787,38 @@ impl ScannerIO for ECStore {
break;
}
let results = results_mutex_for_updates.lock().await;
let mut all_merged = DataUsageCache::default();
for result in results.iter() {
if result.info.last_update.is_none() {
continue;
}
all_merged.merge(result);
}
let data_usage_update = {
let results = results_mutex_for_updates.lock().await;
completed_data_usage_info(
&results,
&all_buckets_clone,
budget_for_updates.budget_elapsed(),
child_token_for_updates.is_cancelled(),
)
};
let merged_last_update = all_merged.info.last_update.unwrap_or(SystemTime::UNIX_EPOCH);
if all_merged.root().is_some() && (!has_sent_once || merged_last_update > last_update) {
let dui = all_merged.dui(&all_merged.info.name, &all_buckets_clone);
if let Err(e) = updates.send(dui).await {
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
state = "send_merged_failed",
error = %e,
"Scanner merged data usage publish failed"
);
}
if let Some((data_usage_info, merged_last_update)) = data_usage_update
&& (!has_sent_once || merged_last_update > last_update)
{
send_merged_data_usage_update(&updates, data_usage_info).await;
}
break;
}
_ = ticker.tick() => {
let results = results_mutex_for_updates.lock().await;
let mut all_merged = DataUsageCache::default();
for result in results.iter() {
if result.info.last_update.is_none() {
continue;
}
all_merged.merge(result);
}
let data_usage_update = {
let results = results_mutex_for_updates.lock().await;
completed_data_usage_info(
&results,
&all_buckets_clone,
budget_for_updates.budget_elapsed(),
child_token_for_updates.is_cancelled(),
)
};
let merged_last_update = all_merged.info.last_update.unwrap_or(SystemTime::UNIX_EPOCH);
if all_merged.root().is_some() && (!has_sent_once || merged_last_update > last_update) {
let dui = all_merged.dui(&all_merged.info.name, &all_buckets_clone);
if let Err(e) = updates.send(dui).await {
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
state = "send_merged_failed",
error = %e,
"Scanner merged data usage publish failed"
);
}
if let Some((data_usage_info, merged_last_update)) = data_usage_update
&& (!has_sent_once || merged_last_update > last_update)
{
send_merged_data_usage_update(&updates, data_usage_info).await;
has_sent_once = true;
last_update = merged_last_update;
}
@@ -858,14 +922,10 @@ impl ScannerIOCache for SetDisks {
let mut permutes = buckets.clone();
permutes.shuffle(&mut rand::rng());
let scan_order = bucket_usage_scan_order(&permutes, &old_cache, &dirty_usage_buckets);
let mut preloaded_published_buckets = HashSet::new();
for bucket in scan_order.iter() {
if let Some(c) = old_cache.find(&bucket.name) {
cache.replace(&bucket.name, DATA_USAGE_ROOT, c.clone());
if old_cache.info.last_update.is_some() {
preloaded_published_buckets.insert(bucket.name.clone());
}
}
if let Err(e) = bucket_tx.send(bucket.clone()).await {
@@ -889,12 +949,10 @@ impl ScannerIOCache for SetDisks {
let (bucket_result_tx, mut bucket_result_rx) = mpsc::channel::<DataUsageEntryInfo>(disks.len());
let cache_mutex_clone = cache_mutex.clone();
let store_clone = self.clone();
let ctx_clone = ctx.clone();
let send_update_fut = tokio::spawn(async move {
let mut ticker = tokio::time::interval(Duration::from_secs(3 + rand::random::<u64>() % 10));
let mut last_update = None;
let completed_bucket_count = Arc::new(AtomicUsize::new(0));
let completed_bucket_count_clone = completed_bucket_count.clone();
let collect_bucket_results_fut = tokio::spawn(async move {
let mut cancelled = false;
loop {
@@ -902,50 +960,14 @@ impl ScannerIOCache for SetDisks {
_ = ctx_clone.cancelled(), if !cancelled => {
cancelled = true;
}
_ = ticker.tick(), if !cancelled => {
let cache_snapshot = {
let cache = cache_mutex_clone.lock().await;
if cache.info.last_update == last_update {
None
} else {
Some(cache.clone())
}
};
let Some(cache_snapshot) = cache_snapshot else {
continue;
};
last_update =
persist_and_publish_cache_snapshot(store_clone.clone(), &updates, cache_snapshot).await;
}
res = bucket_result_rx.recv() => {
if let Some(result) = res {
let cache_snapshot = {
let mut cache = cache_mutex_clone.lock().await;
let publish_immediately =
bucket_result_should_publish_immediately(&mut preloaded_published_buckets, &result.name);
if apply_bucket_result_to_cache(&mut cache, result, SystemTime::now(), publish_immediately) {
Some(cache.clone())
} else {
None
}
};
if let Some(cache_snapshot) = cache_snapshot {
last_update =
persist_and_publish_cache_snapshot(store_clone.clone(), &updates, cache_snapshot).await;
}
} else {
let cache_snapshot = {
let mut cache = cache_mutex_clone.lock().await;
cache.info.next_cycle = want_cycle;
cache.info.last_update = Some(SystemTime::now());
cache.clone()
};
let _ = persist_and_publish_cache_snapshot(store_clone.clone(), &updates, cache_snapshot).await;
result = bucket_result_rx.recv() => {
let Some(result) = result else {
return;
}
};
let mut cache = cache_mutex_clone.lock().await;
apply_bucket_result_to_cache(&mut cache, result, SystemTime::now());
completed_bucket_count_clone.fetch_add(1, Ordering::Relaxed);
}
}
}
@@ -1066,45 +1088,10 @@ impl ScannerIOCache for SetDisks {
"Scanner disk bucket cache ready"
);
let (updates_tx, mut updates_rx) = mpsc::channel::<DataUsageEntry>(1);
let ctx_clone_clone = ctx_clone.clone();
let bucket_name_clone = bucket.name.clone();
let bucket_result_tx_clone_clone_clone = bucket_result_tx_clone_clone.clone();
let update_fut = tokio::spawn(async move {
while let Some(result) = updates_rx.recv().await {
if ctx_clone_clone.is_cancelled() {
break;
}
if let Err(e) = bucket_result_tx_clone_clone_clone
.lock()
.await
.send(DataUsageEntryInfo {
name: bucket_name_clone.clone(),
parent: DATA_USAGE_ROOT.to_string(),
entry: result,
})
.await
{
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket_name_clone,
state = "send_failed",
error = %e,
"Scanner data usage stream failed"
);
}
}
});
let before = cache.info.last_update;
let scan_outcome = match disk_clone
.nsscanner_disk(ctx_clone.clone(), budget_clone.clone(), cache.clone(), Some(updates_tx), scan_mode)
.nsscanner_disk(ctx_clone.clone(), budget_clone.clone(), cache.clone(), None, scan_mode)
.await
{
Ok(scan_outcome) => scan_outcome,
@@ -1153,18 +1140,6 @@ impl ScannerIOCache for SetDisks {
done_save();
}
if let Err(e) = update_fut.await {
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "update_join_failed",
error = %e,
"Scanner bucket update task failed"
);
}
continue;
}
};
@@ -1173,46 +1148,37 @@ impl ScannerIOCache for SetDisks {
ScannerDiskScanOutcome::Complete(cache) => cache,
ScannerDiskScanOutcome::Partial(cache) => {
let done_save = Metrics::time(Metric::SaveUsage);
if let Err(e) = cache.save(store_clone_clone.clone(), cache_name.as_str()).await {
error!(
let partial_saved = match cache.save(store_clone_clone.clone(), cache_name.as_str()).await {
Ok(()) => true,
Err(e) => {
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
cache_name = %cache_name,
state = "partial_save_failed",
error = %e,
"Scanner partial bucket cache save failed"
);
false
}
};
done_save();
if partial_saved {
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
cache_name = %cache_name,
state = "partial_save_failed",
error = %e,
"Scanner partial bucket cache save failed"
);
}
done_save();
if let Err(e) = update_fut.await {
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "partial_update_join_failed",
error = %e,
"Scanner partial bucket update task failed"
state = "partial_saved_not_published",
"Scanner partial bucket cache saved without publishing usage aggregate"
);
}
if let Err(e) = send_cache_root_entry_info(&bucket_result_tx_clone_clone, &cache).await {
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "send_partial_root_failed",
error = %e,
"Scanner partial root entry publish failed"
);
}
continue;
}
};
@@ -1228,19 +1194,6 @@ impl ScannerIOCache for SetDisks {
"Scanner disk bucket scan completed"
);
if let Err(e) = update_fut.await {
error!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_DATA_USAGE_STREAM,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
bucket = %bucket.name,
state = "update_join_failed",
error = %e,
"Scanner bucket update task failed"
);
}
if ctx_clone.is_cancelled() {
break;
}
@@ -1295,7 +1248,31 @@ impl ScannerIOCache for SetDisks {
drop(bucket_result_tx_clone);
send_update_fut.await?;
collect_bucket_results_fut.await?;
let completed_count = completed_bucket_count.load(Ordering::Relaxed);
if should_publish_completed_snapshot(completed_count, buckets.len(), budget.budget_elapsed(), ctx.is_cancelled()) {
let cache_snapshot = {
let mut cache = cache_mutex.lock().await;
cache.info.next_cycle = want_cycle;
cache.info.last_update.get_or_insert_with(SystemTime::now);
cache.clone()
};
let _ = persist_and_publish_cache_snapshot(self.clone(), &updates, cache_snapshot).await;
} else {
debug!(
target: "rustfs::scanner::io",
event = EVENT_SCANNER_CACHE_PERSIST_STATE,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_IO,
completed_buckets = completed_count,
total_buckets = buckets.len(),
budget_elapsed = budget.budget_elapsed(),
cancelled = ctx.is_cancelled(),
state = "set_cache_publish_skipped",
"Scanner set cache publish skipped because cycle did not complete cleanly"
);
}
debug!(
target: "rustfs::scanner::io",
@@ -1799,118 +1776,8 @@ mod tests {
assert!(info.entry.children.is_empty());
}
#[tokio::test]
async fn send_cache_root_entry_info_sends_after_budget_cancellation() {
let ctx = CancellationToken::new();
ctx.cancel();
assert!(ctx.is_cancelled());
let mut cache = DataUsageCache {
info: DataUsageCacheInfo {
name: "bucket".to_string(),
..Default::default()
},
..Default::default()
};
cache.replace(
"bucket",
DATA_USAGE_ROOT,
DataUsageEntry {
size: 10,
objects: 1,
..Default::default()
},
);
let (tx, mut rx) = mpsc::channel(1);
let tx = Arc::new(Mutex::new(tx));
send_cache_root_entry_info(&tx, &cache)
.await
.expect("partial cache should be sent even after budget cancellation");
let info = rx.recv().await.expect("partial cache entry should be received");
assert_eq!(info.name, "bucket");
assert_eq!(info.parent, DATA_USAGE_ROOT);
assert_eq!(info.entry.size, 10);
assert_eq!(info.entry.objects, 1);
}
#[test]
fn apply_bucket_result_requests_immediate_publish_for_missing_bucket() {
let mut cache = DataUsageCache {
info: DataUsageCacheInfo {
name: DATA_USAGE_ROOT.to_string(),
..Default::default()
},
..Default::default()
};
let should_publish = apply_bucket_result_to_cache(
&mut cache,
DataUsageEntryInfo {
name: "bucket".to_string(),
parent: DATA_USAGE_ROOT.to_string(),
entry: DataUsageEntry {
size: 10,
objects: 1,
..Default::default()
},
},
SystemTime::now(),
true,
);
assert!(should_publish);
assert!(cache.info.last_update.is_some());
let entry = cache.find("bucket").expect("bucket entry should be inserted");
assert_eq!(entry.size, 10);
assert_eq!(entry.objects, 1);
}
#[test]
fn apply_bucket_result_defers_publish_for_existing_published_bucket() {
let mut cache = DataUsageCache {
info: DataUsageCacheInfo {
name: DATA_USAGE_ROOT.to_string(),
last_update: Some(SystemTime::now()),
..Default::default()
},
..Default::default()
};
cache.replace(
"bucket",
DATA_USAGE_ROOT,
DataUsageEntry {
size: 5,
objects: 1,
..Default::default()
},
);
let should_publish = apply_bucket_result_to_cache(
&mut cache,
DataUsageEntryInfo {
name: "bucket".to_string(),
parent: DATA_USAGE_ROOT.to_string(),
entry: DataUsageEntry {
size: 10,
objects: 2,
..Default::default()
},
},
SystemTime::now(),
false,
);
assert!(!should_publish);
let entry = cache.find("bucket").expect("bucket entry should remain present");
assert_eq!(entry.size, 10);
assert_eq!(entry.objects, 2);
}
#[test]
fn apply_bucket_result_defers_publish_for_preloaded_published_bucket() {
fn apply_bucket_result_to_cache_updates_bucket_entry() {
let mut cache = DataUsageCache {
info: DataUsageCacheInfo {
name: DATA_USAGE_ROOT.to_string(),
@@ -1928,7 +1795,8 @@ mod tests {
},
);
let should_publish = apply_bucket_result_to_cache(
let update_time = SystemTime::now();
apply_bucket_result_to_cache(
&mut cache,
DataUsageEntryInfo {
name: "bucket".to_string(),
@@ -1939,22 +1807,12 @@ mod tests {
..Default::default()
},
},
SystemTime::now(),
false,
update_time,
);
assert!(!should_publish);
assert_eq!(cache.info.last_update, Some(update_time));
let entry = cache.find("bucket").expect("bucket entry should remain present");
assert_eq!(entry.size, 10);
assert_eq!(entry.objects, 2);
}
#[test]
fn bucket_result_immediate_publish_tracks_preloaded_and_current_results() {
let mut published_buckets = HashSet::from(["existing".to_string()]);
assert!(!bucket_result_should_publish_immediately(&mut published_buckets, "existing"));
assert!(bucket_result_should_publish_immediately(&mut published_buckets, "missing"));
assert!(!bucket_result_should_publish_immediately(&mut published_buckets, "missing"));
}
}