mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 06:18:43 +00:00
feat(scanner): share node admission with background heal
This commit is contained in:
overtrue
@@ -48,6 +48,10 @@ pub enum ScannerError {
|
||||
#[error("Remote namespace scanner disk is already active")]
|
||||
RemoteDiskBusy,
|
||||
|
||||
/// Node resource pressure deferred work before its first storage poll.
|
||||
#[error("Scanner node resource admission deferred")]
|
||||
ResourceDeferred,
|
||||
|
||||
/// Partial data usage cache produced before the scanner stopped.
|
||||
#[error("Scanner stopped with partial data usage cache")]
|
||||
PartialCache(Box<DataUsageCache>),
|
||||
|
||||
@@ -123,7 +123,7 @@ pub use storage_api::scan::{
|
||||
SCANNER_DIRTY_USAGE_SNAPSHOT_PROTOCOL_VERSION, SCANNER_DIRTY_USAGE_SNAPSHOT_RPC_MAX_MESSAGE_SIZE,
|
||||
SCANNER_SCOPED_DIRTY_USAGE_ACK_MAX_ENTRIES,
|
||||
};
|
||||
pub use workload_admission::set_scanner_workload_admission_snapshot_provider;
|
||||
pub use workload_admission::{set_scanner_node_admission, set_scanner_workload_admission_snapshot_provider};
|
||||
|
||||
static SCANNER_ACTIVE_WORK_UNITS: AtomicU64 = AtomicU64::new(0);
|
||||
static SCANNER_RUNTIME_INSTANCES: AtomicU64 = AtomicU64::new(0);
|
||||
|
||||
@@ -42,6 +42,7 @@ use crate::storage_api::owner::{
|
||||
};
|
||||
#[cfg(test)]
|
||||
use crate::storage_api::owner::{EcstoreExpirationStatus as ExpirationStatus, EcstoreLifecycleRule as LifecycleRule};
|
||||
use crate::workload_admission::run_scanner_node_operation;
|
||||
use metrics::{counter, describe_counter};
|
||||
use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit, trace_subscriber_count};
|
||||
use rustfs_filemeta::{
|
||||
@@ -769,6 +770,7 @@ pub struct FolderScanner {
|
||||
resolved_failed_object_paths: HashSet<String>,
|
||||
|
||||
sleeper: DynamicSleeper,
|
||||
node_admission: Option<Arc<rustfs_concurrency::adaptive::NodeAdmission>>,
|
||||
// should_heal: Arc<dyn Fn() -> bool + Send + Sync>,
|
||||
disks: Vec<Arc<Disk>>,
|
||||
disks_quorum: usize,
|
||||
@@ -977,12 +979,13 @@ async fn next_scanner_dir_entry(
|
||||
ctx: &CancellationToken,
|
||||
sleeper: &DynamicSleeper,
|
||||
entries: &mut ScannerDirectory,
|
||||
admission: Option<&Arc<rustfs_concurrency::adaptive::NodeAdmission>>,
|
||||
) -> Result<Option<tokio::fs::DirEntry>, ScannerError> {
|
||||
if entries.batch_entries == SCANNER_DIRECTORY_BATCH_ENTRIES {
|
||||
wait_for_scanner_io(ctx, tokio::task::yield_now()).await?;
|
||||
entries.batch_entries = 0;
|
||||
}
|
||||
let entry = wait_for_scanner_io(ctx, entries.next_entry()).await??;
|
||||
let entry = wait_for_scanner_io(ctx, run_scanner_node_operation(admission, ctx, entries.next_entry())).await???;
|
||||
if entry.is_some() {
|
||||
// Tokio prefetches bounded chunks. Pacing consumption also paces the
|
||||
// next refill, without charging an extra entry for EOF.
|
||||
@@ -1861,7 +1864,12 @@ impl FolderScanner {
|
||||
// directory that vanished during iteration instead of failing the
|
||||
// whole parent subtree. The bucket root (no parent) still propagates
|
||||
// NotFound so namespace removal is detected by the caller.
|
||||
let mut dir_reader = match wait_for_scanner_io(&ctx, ScannerDirectory::open(&dir_path)).await? {
|
||||
let mut dir_reader = match wait_for_scanner_io(
|
||||
&ctx,
|
||||
run_scanner_node_operation(self.node_admission.as_ref(), &ctx, ScannerDirectory::open(&dir_path)),
|
||||
)
|
||||
.await??
|
||||
{
|
||||
Ok(dir_reader) => Some(dir_reader),
|
||||
Err(e)
|
||||
if folder.parent.is_some()
|
||||
@@ -1899,7 +1907,7 @@ impl FolderScanner {
|
||||
// Hold at most one metadata entry until a candidate reaches
|
||||
// EOF: a later part must not leave object accounting or ILM work.
|
||||
let (entry, replayed_metadata) = if let Some(reader) = dir_reader.as_mut() {
|
||||
let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader).await {
|
||||
let entry = match next_scanner_dir_entry(&ctx, &self.sleeper, reader, self.node_admission.as_ref()).await {
|
||||
Ok(Some(entry)) => entry,
|
||||
Ok(None) => {
|
||||
raw_enumeration_complete = true;
|
||||
@@ -2010,7 +2018,12 @@ impl FolderScanner {
|
||||
|
||||
// Ignore entries that disappeared during traversal or hit symlink
|
||||
// loops, but propagate other walk errors.
|
||||
let mut entry_type = match wait_for_scanner_io(&ctx, entry.file_type()).await? {
|
||||
let mut entry_type = match wait_for_scanner_io(
|
||||
&ctx,
|
||||
run_scanner_node_operation(self.node_admission.as_ref(), &ctx, entry.file_type()),
|
||||
)
|
||||
.await??
|
||||
{
|
||||
Ok(entry_type) => entry_type,
|
||||
Err(e) if e.kind() == ErrorKind::NotFound => {
|
||||
debug!(
|
||||
@@ -2061,7 +2074,12 @@ impl FolderScanner {
|
||||
let mut entry_metadata_len = None;
|
||||
if entry_type.is_symlink() {
|
||||
wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
|
||||
let metadata = match wait_for_scanner_io(&ctx, tokio::fs::metadata(&file_path)).await? {
|
||||
let metadata = match wait_for_scanner_io(
|
||||
&ctx,
|
||||
run_scanner_node_operation(self.node_admission.as_ref(), &ctx, tokio::fs::metadata(&file_path)),
|
||||
)
|
||||
.await??
|
||||
{
|
||||
Ok(metadata) => metadata,
|
||||
Err(e) if e.kind() == ErrorKind::NotFound => {
|
||||
debug!(
|
||||
@@ -2203,21 +2221,26 @@ impl FolderScanner {
|
||||
len
|
||||
} else {
|
||||
wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
|
||||
wait_for_scanner_io(&ctx, tokio::fs::metadata(&item.path))
|
||||
.await?
|
||||
.map(|metadata| metadata.len())
|
||||
.unwrap_or(0)
|
||||
wait_for_scanner_io(
|
||||
&ctx,
|
||||
run_scanner_node_operation(self.node_admission.as_ref(), &ctx, tokio::fs::metadata(&item.path)),
|
||||
)
|
||||
.await??
|
||||
.map(|metadata| metadata.len())
|
||||
.unwrap_or(0)
|
||||
}
|
||||
} else {
|
||||
0
|
||||
};
|
||||
wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, budgeted_bytes)).await?;
|
||||
|
||||
self.object_metadata_visits.increment(1);
|
||||
let sz = match self
|
||||
.local_disk
|
||||
.get_size_with_tier_names(item.clone(), &self.tier_registry.names)
|
||||
.await
|
||||
let sz = match run_scanner_node_operation(self.node_admission.as_ref(), &ctx, async {
|
||||
self.object_metadata_visits.increment(1);
|
||||
self.local_disk
|
||||
.get_size_with_tier_names(item.clone(), &self.tier_registry.names)
|
||||
.await
|
||||
})
|
||||
.await?
|
||||
{
|
||||
Ok(sz) => sz,
|
||||
Err(e) => {
|
||||
@@ -2548,7 +2571,7 @@ impl FolderScanner {
|
||||
{
|
||||
Ok(outcome) => outcome,
|
||||
Err(e) => {
|
||||
if ctx.is_cancelled() {
|
||||
if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
|
||||
self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
|
||||
.await;
|
||||
return Err(e);
|
||||
@@ -2972,7 +2995,10 @@ impl FolderScanner {
|
||||
if into.compacted {
|
||||
// In compacted mode child totals are accumulated directly into the parent entry.
|
||||
let fut = Box::pin(self.scan_folder(ctx.clone(), folder_item.clone(), into));
|
||||
fut.await.map_err(|e| ScannerError::Other(e.to_string()))?;
|
||||
fut.await.map_err(|error| match error {
|
||||
ScannerError::ResourceDeferred => ScannerError::ResourceDeferred,
|
||||
other => ScannerError::Other(other.to_string()),
|
||||
})?;
|
||||
self.send_update_for_entry(&this_hash, &folder.parent, into).await;
|
||||
tokio::task::yield_now().await;
|
||||
} else {
|
||||
@@ -2991,7 +3017,7 @@ impl FolderScanner {
|
||||
)
|
||||
.await
|
||||
{
|
||||
if ctx.is_cancelled() {
|
||||
if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
|
||||
self.preserve_partial_child_progress(&folder_item.parent, &h, into, &dst)
|
||||
.await;
|
||||
return Err(e);
|
||||
@@ -3198,6 +3224,7 @@ pub(crate) async fn scan_data_folder_scoped(
|
||||
failed_object_paths_seen: HashSet::new(),
|
||||
resolved_failed_object_paths: HashSet::new(),
|
||||
sleeper,
|
||||
node_admission: crate::workload_admission::scanner_node_admission(),
|
||||
disks,
|
||||
disks_quorum,
|
||||
updates,
|
||||
@@ -3320,7 +3347,7 @@ pub(crate) async fn scan_data_folder_scoped(
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
if ctx.is_cancelled() {
|
||||
if ctx.is_cancelled() || matches!(&e, ScannerError::ResourceDeferred) {
|
||||
let root_hash = hash_path(&bucket_name);
|
||||
let root_has_progress = data_usage_root_has_progress(&root);
|
||||
let pending_heals_changed = scanner.pending_heals_changed;
|
||||
@@ -3395,6 +3422,19 @@ pub(crate) async fn scan_data_folder_scoped(
|
||||
close_disk_guard.close().await;
|
||||
return Err(ScannerError::NamespaceNotFoundCache(Box::new(partial_cache)));
|
||||
}
|
||||
if matches!(&e, ScannerError::ResourceDeferred) {
|
||||
// Admission pressure is not disk failure or an empty census.
|
||||
// Keep the previous tree when no new useful progress exists.
|
||||
let mut partial_cache = scanner.old_cache;
|
||||
if partial_cache.cache.is_empty() {
|
||||
partial_cache
|
||||
.cache
|
||||
.insert(hash_path(&bucket_name).0, DataUsageEntry::default());
|
||||
}
|
||||
partial_cache.info.snapshot_complete = false;
|
||||
close_disk_guard.close().await;
|
||||
return Err(ScannerError::PartialCache(Box::new(partial_cache)));
|
||||
}
|
||||
close_disk_guard.close().await;
|
||||
// No useful information, return original cache
|
||||
Err(e)
|
||||
|
||||
@@ -31,6 +31,7 @@ pub(super) mod checkpoint_fixture;
|
||||
pub(super) mod enumeration_restart;
|
||||
mod incremental_enumeration;
|
||||
mod metadata_visits;
|
||||
mod node_admission;
|
||||
|
||||
/// Reset the process-global alert cooldown map; test-only.
|
||||
fn reset_alert_cooldowns() {
|
||||
@@ -364,6 +365,7 @@ async fn build_test_scanner() -> (FolderScanner, std::path::PathBuf) {
|
||||
failed_object_paths_seen: HashSet::new(),
|
||||
resolved_failed_object_paths: HashSet::new(),
|
||||
sleeper: SCANNER_SLEEPER.clone(),
|
||||
node_admission: None,
|
||||
disks: Vec::new(),
|
||||
disks_quorum: 0,
|
||||
updates: None,
|
||||
|
||||
@@ -0,0 +1,458 @@
|
||||
// Copyright 2026 RustFS Team
|
||||
// Licensed under the Apache License, Version 2.0.
|
||||
|
||||
use super::*;
|
||||
use rustfs_concurrency::adaptive::{FeedbackController, FeedbackObservation, NodeAdmission, ResourcePressure};
|
||||
use std::num::NonZeroU16;
|
||||
use std::sync::atomic::AtomicU64;
|
||||
|
||||
fn owner(open: bool) -> Arc<NodeAdmission> {
|
||||
let owner = NodeAdmission::new(NonZeroU16::new(1).expect("one node unit"));
|
||||
if open {
|
||||
owner.update(
|
||||
&mut FeedbackController::new(),
|
||||
Some(FeedbackObservation {
|
||||
generation: 1,
|
||||
max_concurrency: 1,
|
||||
pressure: ResourcePressure::Unknown,
|
||||
completed_units: None,
|
||||
}),
|
||||
);
|
||||
}
|
||||
owner
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scanner_node_admission_blocks_real_walk_before_open_or_metadata() {
|
||||
let (mut scanner, root) = build_test_scanner().await;
|
||||
let _guard = TestGuard {
|
||||
temp_dir: Some(root.clone()),
|
||||
};
|
||||
write_test_object_metadata(&root, "bucket", "object").await;
|
||||
let visits = super::metadata_visits::observe(&mut scanner);
|
||||
let gate = owner(false);
|
||||
scanner.node_admission = Some(gate.clone());
|
||||
scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
|
||||
for name in ["bucket", "does-not-exist"] {
|
||||
let result = scanner
|
||||
.scan_folder(
|
||||
CancellationToken::new(),
|
||||
CachedFolder {
|
||||
name: name.to_string(),
|
||||
parent: None,
|
||||
object_heal_prob_div: 1,
|
||||
},
|
||||
&mut DataUsageEntry::default(),
|
||||
)
|
||||
.await;
|
||||
assert!(
|
||||
matches!(result, Err(ScannerError::ResourceDeferred)),
|
||||
"node gate must precede both real reads and missing-path lookup: {result:?}"
|
||||
);
|
||||
}
|
||||
assert_eq!(visits.load(Ordering::Relaxed), 0);
|
||||
assert_eq!(gate.snapshot().active, 0);
|
||||
assert_eq!(gate.completed(), Some(0));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scanner_node_admission_shares_capacity_with_heal_owner() {
|
||||
let (mut scanner, root) = build_test_scanner().await;
|
||||
let _guard = TestGuard {
|
||||
temp_dir: Some(root.clone()),
|
||||
};
|
||||
write_test_object_metadata(&root, "bucket", "object").await;
|
||||
let gate = owner(true);
|
||||
let heal_unit = gate.acquire().await;
|
||||
scanner.node_admission = Some(gate.clone());
|
||||
scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
|
||||
let folder = CachedFolder {
|
||||
name: "bucket".to_string(),
|
||||
parent: None,
|
||||
object_heal_prob_div: 1,
|
||||
};
|
||||
let result = scanner
|
||||
.scan_folder(CancellationToken::new(), folder.clone(), &mut DataUsageEntry::default())
|
||||
.await;
|
||||
assert!(
|
||||
matches!(result, Err(ScannerError::ResourceDeferred)),
|
||||
"scanner must not create a separate node budget: {result:?}"
|
||||
);
|
||||
assert_eq!(gate.snapshot().active, 1);
|
||||
heal_unit.complete();
|
||||
let mut entry = DataUsageEntry::default();
|
||||
scanner
|
||||
.scan_folder(CancellationToken::new(), folder, &mut entry)
|
||||
.await
|
||||
.expect("released heal capacity admits the real scan");
|
||||
assert_eq!(
|
||||
scanner
|
||||
.new_cache
|
||||
.size_recursive("bucket")
|
||||
.expect("complete bucket tree")
|
||||
.objects,
|
||||
1
|
||||
);
|
||||
assert_eq!(gate.snapshot().active, 0);
|
||||
assert!(
|
||||
gate.completed().is_some_and(|count| count > 1),
|
||||
"successful scanner work must reach shared feedback"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn scanner_node_admission_precancellation_never_polls_storage() {
|
||||
let gate = owner(false);
|
||||
let cancel = CancellationToken::new();
|
||||
let polled = std::cell::Cell::new(false);
|
||||
cancel.cancel();
|
||||
let operation = run_scanner_node_operation(Some(&gate), &cancel, async {
|
||||
polled.set(true);
|
||||
Ok::<_, std::io::Error>(())
|
||||
});
|
||||
assert!(matches!(operation.await, Err(ScannerError::Other(message)) if message == "Operation cancelled"));
|
||||
assert!(!polled.get());
|
||||
assert_eq!(gate.completed(), Some(0));
|
||||
assert_eq!(gate.snapshot().active, 0);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn scanner_node_admission_shrink_keeps_running_unit_and_preserves_errors() {
|
||||
let gate = owner(true);
|
||||
let cancel = CancellationToken::new();
|
||||
let mut running = Box::pin(run_scanner_node_operation(
|
||||
Some(&gate),
|
||||
&cancel,
|
||||
std::future::pending::<Result<(), std::io::Error>>(),
|
||||
));
|
||||
assert!(futures::poll!(&mut running).is_pending());
|
||||
assert_eq!(gate.snapshot().active, 1);
|
||||
gate.update(&mut FeedbackController::new(), None);
|
||||
let polled = std::cell::Cell::new(false);
|
||||
let deferred = run_scanner_node_operation(Some(&gate), &cancel, async {
|
||||
polled.set(true);
|
||||
Ok::<_, std::io::Error>(())
|
||||
})
|
||||
.await;
|
||||
assert!(matches!(deferred, Err(ScannerError::ResourceDeferred)));
|
||||
assert!(!polled.get());
|
||||
assert_eq!(gate.snapshot().active, 1, "shrink cannot revoke admitted work");
|
||||
drop(running);
|
||||
assert_eq!(gate.snapshot().active, 0);
|
||||
assert_eq!(gate.completed(), Some(0), "abandoned work is not progress");
|
||||
gate.update(
|
||||
&mut FeedbackController::new(),
|
||||
Some(FeedbackObservation {
|
||||
generation: 2,
|
||||
max_concurrency: 1,
|
||||
pressure: ResourcePressure::Unknown,
|
||||
completed_units: None,
|
||||
}),
|
||||
);
|
||||
let error = run_scanner_node_operation(Some(&gate), &cancel, async {
|
||||
Err::<(), _>(std::io::Error::from(std::io::ErrorKind::PermissionDenied))
|
||||
})
|
||||
.await
|
||||
.expect("resource admission succeeds")
|
||||
.expect_err("storage error survives");
|
||||
assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
|
||||
assert_eq!(gate.completed(), Some(0));
|
||||
run_scanner_node_operation(Some(&gate), &cancel, async { Ok::<_, std::io::Error>(()) })
|
||||
.await
|
||||
.expect("admitted")
|
||||
.expect("successful work");
|
||||
assert_eq!(gate.completed(), Some(1));
|
||||
assert_eq!(gate.snapshot().active, 0);
|
||||
}
|
||||
|
||||
struct ShrinkOnMetadata {
|
||||
owner: Arc<NodeAdmission>,
|
||||
visited: AtomicU64,
|
||||
}
|
||||
impl metrics::CounterFn for ShrinkOnMetadata {
|
||||
fn increment(&self, value: u64) {
|
||||
if self.visited.fetch_add(value, Ordering::SeqCst) == 0 {
|
||||
self.owner.update(&mut FeedbackController::new(), None);
|
||||
assert_eq!(self.owner.snapshot().active, 1, "shrink occurs during the admitted object");
|
||||
}
|
||||
}
|
||||
fn absolute(&self, _: u64) {
|
||||
panic!("visit counter must be cumulative");
|
||||
}
|
||||
}
|
||||
struct ShrinkRecorder(Arc<ShrinkOnMetadata>);
|
||||
impl metrics::Recorder for ShrinkRecorder {
|
||||
fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
|
||||
fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
|
||||
fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
|
||||
fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
|
||||
if key.name() == "rustfs_scanner_object_metadata_visits_total" {
|
||||
metrics::Counter::from_arc(self.0.clone())
|
||||
} else {
|
||||
metrics::Counter::noop()
|
||||
}
|
||||
}
|
||||
fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge {
|
||||
metrics::Gauge::noop()
|
||||
}
|
||||
fn register_histogram(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Histogram {
|
||||
metrics::Histogram::noop()
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scanner_node_admission_mid_walk_shrink_preserves_partial_and_resumes_all_objects() {
|
||||
let (scanner, root) = build_test_scanner().await;
|
||||
let _guard = TestGuard {
|
||||
temp_dir: Some(root.clone()),
|
||||
};
|
||||
for path in ["prefix/a/first", "prefix/b/second"] {
|
||||
write_test_object_metadata(&root, "bucket", path).await;
|
||||
}
|
||||
let gate = owner(true);
|
||||
let recorder = ShrinkRecorder(Arc::new(ShrinkOnMetadata {
|
||||
owner: gate.clone(),
|
||||
visited: AtomicU64::new(0),
|
||||
}));
|
||||
let scan = |cache| {
|
||||
let cancel = CancellationToken::new();
|
||||
scan_data_folder(
|
||||
cancel.clone(),
|
||||
ScannerCycleBudget::new(&cancel, Default::default()),
|
||||
vec![scanner.local_disk.clone()],
|
||||
scanner.local_disk.clone(),
|
||||
cache,
|
||||
None,
|
||||
HealScanMode::Normal,
|
||||
DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
|
||||
)
|
||||
};
|
||||
let cache = DataUsageCache {
|
||||
info: crate::DataUsageCacheInfo {
|
||||
name: "bucket".to_string(),
|
||||
skip_healing: true,
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
};
|
||||
let baseline = scan(cache.clone()).await.expect("independent full walk");
|
||||
let partial = crate::workload_admission::TEST_SCANNER_NODE_ADMISSION
|
||||
.scope(gate.clone(), async {
|
||||
let mut scanning = std::pin::pin!(scan(cache));
|
||||
std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || scanning.as_mut().poll(cx))).await
|
||||
})
|
||||
.await;
|
||||
let Err(ScannerError::PartialCache(partial)) = partial else {
|
||||
panic!("deferral below root cannot certify completion: {partial:?}");
|
||||
};
|
||||
assert!(!partial.info.snapshot_complete);
|
||||
assert_eq!(
|
||||
partial
|
||||
.size_recursive("bucket")
|
||||
.expect("partial keeps accounted work")
|
||||
.objects,
|
||||
1
|
||||
);
|
||||
assert_eq!(recorder.0.visited.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(gate.snapshot().active, 0);
|
||||
gate.update(
|
||||
&mut FeedbackController::new(),
|
||||
Some(FeedbackObservation {
|
||||
generation: 2,
|
||||
max_concurrency: 1,
|
||||
pressure: ResourcePressure::Unknown,
|
||||
completed_units: None,
|
||||
}),
|
||||
);
|
||||
let resumed = crate::workload_admission::TEST_SCANNER_NODE_ADMISSION
|
||||
.scope(gate.clone(), scan(*partial))
|
||||
.await
|
||||
.expect("resume completes with restored node capacity");
|
||||
assert!(resumed.info.snapshot_complete);
|
||||
let expected = baseline.size_recursive("bucket").expect("baseline bucket");
|
||||
let actual = resumed.size_recursive("bucket").expect("resumed bucket");
|
||||
assert_eq!(
|
||||
(actual.objects, actual.size, actual.versions),
|
||||
(expected.objects, expected.size, expected.versions)
|
||||
);
|
||||
assert_eq!(actual.objects, 2);
|
||||
assert_eq!(gate.snapshot().active, 0);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn scanner_node_admission_does_not_park_while_cache_lease_may_be_held() {
|
||||
let gate = owner(false);
|
||||
let cancel = CancellationToken::new();
|
||||
let polled = std::cell::Cell::new(false);
|
||||
let mut operation = Box::pin(run_scanner_node_operation(Some(&gate), &cancel, async {
|
||||
polled.set(true);
|
||||
Ok::<_, std::io::Error>(())
|
||||
}));
|
||||
assert!(matches!(
|
||||
futures::poll!(&mut operation),
|
||||
std::task::Poll::Ready(Err(ScannerError::ResourceDeferred))
|
||||
));
|
||||
assert!(!polled.get());
|
||||
assert_eq!(gate.snapshot().active, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scanner_node_admission_early_deferral_keeps_old_tree_incomplete() {
|
||||
let (scanner, root) = build_test_scanner().await;
|
||||
let _guard = TestGuard { temp_dir: Some(root) };
|
||||
let mut cache = DataUsageCache {
|
||||
info: crate::DataUsageCacheInfo {
|
||||
name: "bucket".to_string(),
|
||||
skip_healing: true,
|
||||
snapshot_complete: true,
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
};
|
||||
cache.replace(
|
||||
"bucket",
|
||||
"",
|
||||
DataUsageEntry {
|
||||
objects: 7,
|
||||
size: 19,
|
||||
..Default::default()
|
||||
},
|
||||
);
|
||||
let gate = owner(false);
|
||||
let cancel = CancellationToken::new();
|
||||
let result = crate::workload_admission::TEST_SCANNER_NODE_ADMISSION
|
||||
.scope(
|
||||
gate.clone(),
|
||||
scan_data_folder(
|
||||
cancel.clone(),
|
||||
ScannerCycleBudget::new(&cancel, Default::default()),
|
||||
vec![scanner.local_disk.clone()],
|
||||
scanner.local_disk,
|
||||
cache,
|
||||
None,
|
||||
HealScanMode::Normal,
|
||||
DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest),
|
||||
),
|
||||
)
|
||||
.await;
|
||||
let Err(ScannerError::PartialCache(partial)) = result else {
|
||||
panic!("resource pressure is a partial census, not a disk error: {result:?}");
|
||||
};
|
||||
assert!(!partial.info.snapshot_complete);
|
||||
let totals = partial.size_recursive("bucket").expect("old tree retained");
|
||||
assert_eq!((totals.objects, totals.size), (7, 19));
|
||||
assert_eq!(gate.completed(), Some(0));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scanner_node_admission_compacted_heal_child_preserves_deferral() {
|
||||
let (mut scanner, root) = build_test_scanner().await;
|
||||
let _guard = TestGuard {
|
||||
temp_dir: Some(root.clone()),
|
||||
};
|
||||
let bucket = "bucket";
|
||||
let object = "prefix/abandoned";
|
||||
for index in 0..DATA_SCANNER_COMPACT_AT_FOLDERS {
|
||||
tokio::fs::create_dir_all(root.join(bucket).join("prefix").join(format!("child-{index:04}")))
|
||||
.await
|
||||
.expect("create real compaction threshold");
|
||||
}
|
||||
let mut disks = vec![scanner.local_disk.clone()];
|
||||
for (index, version) in [(2, 0x1935), (3, 0x1935), (4, 0x1934)] {
|
||||
let disk_root = root.join(format!("disk-{index}"));
|
||||
// Two partial frames exceed the coordinator channel's capacity, so
|
||||
// discovery must consume the first before the finished frame can race it.
|
||||
for child in ["first", "second"] {
|
||||
let child_object = format!("{object}/{child}");
|
||||
let metadata = metadata_for_object_version(bucket, &child_object, Some(Uuid::from_u128(version)));
|
||||
write_test_object_metadata_bytes(&disk_root, bucket, &child_object, &metadata).await;
|
||||
}
|
||||
let endpoint = Endpoint::try_from(disk_root.to_string_lossy().as_ref()).expect("remote fixture endpoint");
|
||||
disks.push(
|
||||
new_disk(
|
||||
&endpoint,
|
||||
&DiskOption {
|
||||
cleanup: false,
|
||||
health_check: false,
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("remote fixture disk"),
|
||||
);
|
||||
}
|
||||
scanner.old_cache.info.name = bucket.to_string();
|
||||
scanner.new_cache.info.name = bucket.to_string();
|
||||
scanner.update_cache.info.name = bucket.to_string();
|
||||
scanner.old_cache.replace(
|
||||
&format!("{bucket}/{object}"),
|
||||
"bucket/prefix",
|
||||
DataUsageEntry {
|
||||
objects: 1,
|
||||
..Default::default()
|
||||
},
|
||||
);
|
||||
scanner.heal_object_select = 1;
|
||||
scanner.disks = disks;
|
||||
scanner.disks_quorum = 2;
|
||||
scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
|
||||
let gate = owner(true);
|
||||
scanner.node_admission = Some(gate.clone());
|
||||
let shrunk = Arc::new(AtomicBool::new(false));
|
||||
let response_gate = gate.clone();
|
||||
let response_shrunk = shrunk.clone();
|
||||
let (_commands, mut heal_rx) = rustfs_heal_contracts::heal_channel::init_heal_channels().expect("isolated heal channel");
|
||||
let responder = tokio::spawn(async move {
|
||||
while let Some(command) = heal_rx.recv().await {
|
||||
let rustfs_heal_contracts::heal_channel::HealReceiptCommand { request, response_tx } = command;
|
||||
if request.object_version_id.is_some() {
|
||||
response_shrunk.store(true, Ordering::SeqCst);
|
||||
response_gate.update(&mut FeedbackController::new(), None);
|
||||
}
|
||||
let _ = response_tx.send(Ok(rustfs_heal_contracts::heal_channel::HealAdmissionReceipt {
|
||||
result: HealAdmissionResult::Accepted,
|
||||
task_id: request.id,
|
||||
}));
|
||||
}
|
||||
});
|
||||
let mut into = DataUsageEntry::default();
|
||||
let folder = CachedFolder {
|
||||
name: "bucket/prefix".to_string(),
|
||||
parent: Some(hash_path(bucket)),
|
||||
object_heal_prob_div: 1,
|
||||
};
|
||||
let result = {
|
||||
let mut walk = std::pin::pin!(scanner.scan_folder(CancellationToken::new(), folder, &mut into));
|
||||
let mut controller = FeedbackController::new();
|
||||
let mut generation = 0;
|
||||
std::future::poll_fn(|cx| {
|
||||
// Keep telemetry fresh until the real heal-discovery receipt shrinks it.
|
||||
if !shrunk.load(Ordering::SeqCst) {
|
||||
generation += 1;
|
||||
gate.update(
|
||||
&mut controller,
|
||||
Some(FeedbackObservation {
|
||||
generation,
|
||||
max_concurrency: 1,
|
||||
pressure: ResourcePressure::Unknown,
|
||||
completed_units: None,
|
||||
}),
|
||||
);
|
||||
}
|
||||
walk.as_mut().poll(cx)
|
||||
})
|
||||
.await
|
||||
};
|
||||
responder.abort();
|
||||
assert!(into.compacted, "real directory count must enter the compacted recursion branch");
|
||||
assert!(shrunk.load(Ordering::SeqCst), "remote orphan discovery must submit an exact-version heal");
|
||||
assert!(
|
||||
matches!(result, Err(ScannerError::ResourceDeferred)),
|
||||
"compacted recursion must retain typed resource pressure for partial-cache handling: {result:?}"
|
||||
);
|
||||
assert_eq!(gate.snapshot().active, 0);
|
||||
}
|
||||
@@ -12,9 +12,56 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use std::sync::{Arc, LazyLock, RwLock};
|
||||
use std::sync::{Arc, LazyLock, OnceLock, RwLock};
|
||||
|
||||
use rustfs_concurrency::adaptive::NodeAdmission;
|
||||
use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshotProvider, WorkloadClass};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
|
||||
use crate::ScannerError;
|
||||
|
||||
static SCANNER_NODE_ADMISSION: OnceLock<Arc<NodeAdmission>> = OnceLock::new();
|
||||
|
||||
#[cfg(test)]
|
||||
tokio::task_local! {
|
||||
pub(crate) static TEST_SCANNER_NODE_ADMISSION: Arc<NodeAdmission>;
|
||||
}
|
||||
|
||||
/// Install the heal manager's node owner before starting scanner workers.
|
||||
/// Refuse replacement: existing scans must not outlive an independently renewed owner.
|
||||
pub fn set_scanner_node_admission(owner: Arc<NodeAdmission>) -> bool {
|
||||
SCANNER_NODE_ADMISSION.set(owner).is_ok()
|
||||
}
|
||||
|
||||
pub(crate) fn scanner_node_admission() -> Option<Arc<NodeAdmission>> {
|
||||
#[cfg(test)]
|
||||
if let Ok(owner) = TEST_SCANNER_NODE_ADMISSION.try_with(Arc::clone) {
|
||||
return Some(owner);
|
||||
}
|
||||
SCANNER_NODE_ADMISSION.get().cloned()
|
||||
}
|
||||
|
||||
/// Scanner callers can already hold a bucket cache lease: try once, never park
|
||||
/// under that lock. Keep the operation's error type and cancellation boundary.
|
||||
/// Units end before recursion, pacing or checkpoint publication.
|
||||
pub(crate) async fn run_scanner_node_operation<T, E>(
|
||||
owner: Option<&Arc<NodeAdmission>>,
|
||||
cancel: &CancellationToken,
|
||||
operation: impl std::future::Future<Output = Result<T, E>>,
|
||||
) -> Result<Result<T, E>, ScannerError> {
|
||||
let Some(owner) = owner else {
|
||||
return Ok(operation.await);
|
||||
};
|
||||
if cancel.is_cancelled() {
|
||||
return Err(ScannerError::Other("Operation cancelled".to_string()));
|
||||
}
|
||||
let permit = owner.try_acquire().ok_or(ScannerError::ResourceDeferred)?;
|
||||
let result = operation.await;
|
||||
if result.is_ok() {
|
||||
permit.complete();
|
||||
}
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
type WorkloadSnapshotProviderRef = Arc<dyn WorkloadAdmissionSnapshotProvider + Send + Sync>;
|
||||
|
||||
|
||||
@@ -486,17 +486,19 @@ The OTLP exporter records these node-local histograms for completed manager admi
|
||||
|
||||
The two new histograms aggregate all observed admission sources and results without per-request labels. An ordinary uninterrupted admission partitions the phase into wait and hold. Admin forceStart releases these guards while cancelling overlaps, then reacquires them: the cancellation gap is included only in the original total phase, and the second acquisition contributes to wait. Cancellation subroutine work is outside these outer admission guard intervals. Recording happens after all three guards are released. They measure manager admission ownership, not namespace/object-lock waiting during repair. The status field `maxLockPhaseMicros` remains a manager-lifetime maximum, not a p95 sample. Keep same-window histogram counts and bucket resolution when deriving percentiles; missing observations do not mean zero latency. These metrics do not establish performance acceptance by themselves.
|
||||
|
||||
## Experimental heal node admission
|
||||
## Experimental shared node admission
|
||||
|
||||
`RUSTFS_HEAL_ADAPTIVE_ADMISSION=true` enables Linux cgroup-v2 feedback for the four object-heal entry points owned by the heal manager. It is off by default while full SSD performance acceptance is pending. Unsupported platforms reject an enabled setting. This does not yet control scanner walks, bucket/format repair or physical devices, and it is not a replacement for existing worker, disk or byte budgets.
|
||||
`RUSTFS_HEAL_ADAPTIVE_ADMISSION=true` enables Linux cgroup-v2 feedback shared by the heal manager’s four object-heal entry points and local Scanner namespace reads and object accounting. It is off by default while full SSD performance acceptance is pending. Unsupported platforms reject an enabled setting. This does not yet control the abandoned-child streaming list coordinator, bucket/format repair or physical devices, and it is not a replacement for existing worker, disk or byte budgets.
|
||||
|
||||
The node samples its effective CPU quota/cpuset, ancestor memory headroom, CPU usage/throttling and CPU/memory/I/O pressure once per second. Sampling uses one blocking worker at a time. Resource reads retain their real start/end times: reads spanning over 20 ms or older than one second after all inputs return cannot renew admission. Counter rates use both read brackets conservatively. The concurrency cap is at most the rounded-up effective CPU count and 256 scheduling units; existing operator limits still apply independently. At most 1% memory headroom blocks new work. This is a pressure safeguard, not a memory reservation or a claim that concurrent repair buffers fit in the remaining bytes.
|
||||
|
||||
Each complete sample leases admission for three seconds. Missing resource caps block new work; missing or low-confidence feedback retains at most one maintenance unit. CPU use above 90%, throttling or pressure above 1% of elapsed time, memory headroom at most 10%, foreground errors/cancellations, a foreground queue or a comparable foreground p99 regression reduce concurrency immediately. Recovery requires CPU use at most 70%, throttling/stalls at most 0.1% of elapsed time, at least 20% memory headroom, fresh foreground evidence and actual successful work. The existing controller probes one additional unit only after eight consecutive healthy seconds and cooldown. Limits affect new admission and do not revoke active work.
|
||||
|
||||
The Scanner captures the same node owner once per bucket walk. Directory open/refill, entry type and size lookup, and object accounting try admission before their first storage poll, releasing before sleeps, child recursion and checkpoint publication. Scanner callers already hold a bucket cache lease, so they never queue for node capacity. Resource deferral immediately unwinds the walk and preserves useful partial cache/cursors without certifying completion; when no new progress exists it retains the old tree as incomplete. Existing remote single-session ownership and local-disk admission shares remain in force. No per-entry global lookup, new task or heap allocation is added by this admission layer. The streaming abandoned-child coordinator requires a separate ownership boundary: retaining a node unit while it waits for a bounded-channel consumer that needs the same unit would deadlock at capacity one.
|
||||
|
||||
Foreground comparison uses the existing complete response-body windows and shared admission snapshots. Each active class needs at least 100 completions and a finite p99 interval; this permits the percentile rank but is not a statistical confidence guarantee. Comparisons require the same active classes and rates within 80–125% of a retained reference. A changed workload resets recovery. The reference is an observed workload window, not an independent background-off baseline. A p99 interval can justify stability only when its upper bound is within 5% of the reference lower bound; a lower bound above 105% of the reference upper bound is pressure. Overlapping/uncertain comparisons cannot justify growth. Missing queues, stale or repeated cuts, uncertain cut boundaries and low counts are not idle evidence.
|
||||
|
||||
Admission waits at most 250 ms before returning the existing typed transient deferral, before polling storage or taking namespace locks. No-lock callers never wait. Queued read repair uses the same node gate; direct foreground ECStore calls remain outside this heal-manager adapter. Cancellation releases the non-cloneable node scheduling unit. Native work retains separate physical-I/O units, and unsuccessful/cancelled/dry-run work cannot increment the node's completed-work counter. This scheduling counter is not an authoritative heal receipt and cannot discharge repair responsibility.
|
||||
Heal admission waits at most 250 ms before returning the existing typed transient deferral, before polling storage or taking namespace locks. No-lock callers never wait. Queued read repair uses the same node gate; direct foreground ECStore calls remain outside this heal-manager adapter. Cancellation releases the non-cloneable node scheduling unit. Native work retains separate physical-I/O units, and unsuccessful/cancelled/dry-run work cannot increment the node's completed-work counter. This scheduling counter is not an authoritative heal receipt and cannot discharge repair responsibility.
|
||||
|
||||
The `rustfs_background_node_admission_limit` and `rustfs_background_node_admission_active` gauges describe the last published sample, not a live reservation proof. A blocked sampler can leave these exports stale even though the admission lease has expired. Full disk-owner feedback, shared scanner scheduling, escaping-buffer byte ownership and the matched-release SSD matrix/hot-shrink/72-hour soak remain required before enabling this by default or claiming optimal performance.
|
||||
|
||||
|
||||
@@ -81,6 +81,14 @@ pub(crate) async fn init_background_service_runtime(store: Arc<ECStore>) -> Resu
|
||||
let admission = None;
|
||||
let heal_storage = Arc::new(ECStoreHealStorage::new(store.clone()).with_admission(admission.clone()));
|
||||
init_heal_manager_with_workload_provider(heal_storage, None, Some(workload_provider.clone())).await?;
|
||||
if let Some(admission) = &admission
|
||||
&& !rustfs_scanner::set_scanner_node_admission(admission.clone())
|
||||
{
|
||||
return Err(std::io::Error::new(
|
||||
std::io::ErrorKind::AlreadyExists,
|
||||
"scanner node admission is already initialized",
|
||||
));
|
||||
}
|
||||
#[cfg(any(target_os = "linux", all(test, unix)))]
|
||||
if let Some(admission) = admission {
|
||||
crate::background_admission::start(admission, background_cancel.clone(), workload_provider);
|
||||
|
||||
Reference in New Issue
Block a user