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

fix(scanner): anchor the bitrot cadence and budget per-disk I/O (#8701)

Browse Source 
* test(scanner): pin Normal mode when no bitrot start is recorded

Regression for rustfs/backlog#2905: three consecutive cycles that find no
recorded bitrot start currently run Deep through the selection window, and a
Normal cycle never records the cadence start.

* fix(scanner): anchor the bitrot cadence and budget per-disk I/O

Decide the scan mode from a recorded Deep window start only. A cycle that
finds no recorded start runs Normal and anchors the bitrot cadence in
process memory instead of opening the selection window; the Deep cycle that
follows the elapsed cadence records the window start exactly as before. A
missing record used to be read as a window that started at cycle 0, which
ran Deep for deep_window_cycles cycles whenever the start could not be read
back (rustfs/rustfs#8015). The recorded start is also remembered across a
skipped or failed save, and /v3/scanner/status reports the cadence state
under a new bitrot object.

Add a per-local-disk background I/O budget (operations and bytes per second,
defaults by detected media, auto|off|<rate>) that DynamicSleeper charges
before directory listings and object metadata reads as a hard ceiling
layered on the foreground-priority disk admission gate, with runtime config
keys, per-disk status, cycle metrics, pacing pressure, usage_freshness,
Prometheus series, Helm values, admin config help, and operator docs.

Refs rustfs/backlog#2905, rustfs/rustfs#8015, rustfs/rustfs#8377.

* test(scanner): expect Normal retries on erasure sets without a bitrot start

The pending-PUT retry fixture asserted that a 4-drive set revisits every
bucket because the cycle ran Deep. Without a recorded Deep window the cycle
now runs Normal on every layout and reuses the cold inventory.
This commit is contained in:
Chris
2026-10-19 19:15:03 +09:00
committed by GitHub
parent 2b57fa8758
commit f3cef98afc
23 changed files with 1825 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/config/src/constants/scanner.rs
+30
View File
@@ -62,6 +62,12 @@ pub const SCANNER_MAX_CONCURRENT_DISK_SCANS: &str = "max_concurrent_disk_scans";
/// Scanner config key setting how many bucket walkers each local disk runs.
pub const SCANNER_WALKERS_PER_DISK: &str = "walkers_per_disk";
/// Per-local-disk scanner I/O ceiling in operations per second.
pub const SCANNER_DISK_IO_BUDGET_IOPS: &str = "disk_io_budget_iops";
/// Per-local-disk scanner I/O ceiling in bytes per second.
pub const SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC: &str = "disk_io_budget_bytes_per_sec";
/// Scanner config key controlling how often object loops yield.
pub const SCANNER_YIELD_EVERY_N_OBJECTS: &str = "yield_every_n_objects";
@@ -90,6 +96,8 @@ pub const SCANNER_KEYS: &[&str] = &[
SCANNER_MAX_CONCURRENT_SET_SCANS,
SCANNER_MAX_CONCURRENT_DISK_SCANS,
SCANNER_WALKERS_PER_DISK,
SCANNER_DISK_IO_BUDGET_IOPS,
SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC,
SCANNER_YIELD_EVERY_N_OBJECTS,
SCANNER_ALERT_EXCESS_VERSIONS,
SCANNER_ALERT_EXCESS_VERSION_SIZE,
@@ -226,6 +234,10 @@ pub const ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS: &str = "RUSTFS_SCANNER_MAX_CONC
/// - Example: `export RUSTFS_SCANNER_WALKERS_PER_DISK=4`
pub const ENV_SCANNER_WALKERS_PER_DISK: &str = "RUSTFS_SCANNER_WALKERS_PER_DISK";
pub const ENV_SCANNER_DISK_IO_BUDGET_IOPS: &str = "RUSTFS_SCANNER_DISK_IO_BUDGET_IOPS";
pub const ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC: &str = "RUSTFS_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC";
/// Environment variable that controls how often scanner object loops yield to the async runtime.
/// A value of `0` disables this extra object-count yield.
/// - Example: `export RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS=32`
@@ -252,6 +264,24 @@ pub const DEFAULT_SCANNER_WALKERS_PER_DISK: usize = 1;
/// contention on one spindle or NVMe queue without more disks to spread it.
pub const MAX_SCANNER_WALKERS_PER_DISK: usize = 16;
/// Default spelling of both per-disk scanner I/O budget keys: resolve the
/// rate from the detected media. `off` removes the cap; a number is explicit.
pub const DEFAULT_SCANNER_DISK_IO_BUDGET: &str = "auto";
/// Rotational media budget, also used when media cannot be detected: about
/// half of a 7200 rpm spindle's random reads stay available to foreground
/// requests, and the byte ceiling sits well above the steady ~2 MB/s a paced
/// single-disk scan was measured at once admission landed (rustfs/rustfs#8377).
pub const DEFAULT_SCANNER_DISK_IO_BUDGET_ROTATIONAL_IOPS: u64 = 100;
pub const DEFAULT_SCANNER_DISK_IO_BUDGET_ROTATIONAL_BYTES_PER_SEC: u64 = 16 * 1024 * 1024;
/// Flash media budget: a few percent of a SATA SSD's random-read capacity,
/// enough to keep metadata walks far from saturating the device.
pub const DEFAULT_SCANNER_DISK_IO_BUDGET_FLASH_IOPS: u64 = 4000;
pub const DEFAULT_SCANNER_DISK_IO_BUDGET_FLASH_BYTES_PER_SEC: u64 = 256 * 1024 * 1024;
/// Default object interval for cooperative scanner yields.
pub const DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS: u64 = 128;
crates/ecstore/src/config/scanner.rs
+16 -5
View File
@@ -15,13 +15,14 @@
use rustfs_config::server_config::{KV, KVS};
use rustfs_config::{
DEFAULT_SCANNER_BITROT_CYCLE_SECS, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES,
DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_IDLE_MODE,
DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_SPEED,
DEFAULT_SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, SCANNER_ALERT_EXCESS_FOLDERS,
DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_DISK_IO_BUDGET,
DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS,
DEFAULT_SCANNER_SPEED, DEFAULT_SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, SCANNER_ALERT_EXCESS_FOLDERS,
SCANNER_ALERT_EXCESS_VERSION_SIZE, SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT,
SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY,
SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED,
SCANNER_START_DELAY, SCANNER_WALKERS_PER_DISK, SCANNER_YIELD_EVERY_N_OBJECTS,
SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, SCANNER_DISK_IO_BUDGET_IOPS, SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS,
SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED, SCANNER_START_DELAY, SCANNER_WALKERS_PER_DISK,
SCANNER_YIELD_EVERY_N_OBJECTS,
};
use std::sync::LazyLock;
@@ -97,6 +98,16 @@ pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
value: DEFAULT_SCANNER_WALKERS_PER_DISK.to_string(),
hidden_if_empty: false,
},
KV {
key: SCANNER_DISK_IO_BUDGET_IOPS.to_owned(),
value: DEFAULT_SCANNER_DISK_IO_BUDGET.to_owned(),
hidden_if_empty: false,
},
KV {
key: SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC.to_owned(),
value: DEFAULT_SCANNER_DISK_IO_BUDGET.to_owned(),
hidden_if_empty: false,
},
KV {
key: SCANNER_YIELD_EVERY_N_OBJECTS.to_owned(),
value: DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS.to_string(),
crates/ecstore/src/services/metrics_realtime.rs
+12
View File
@@ -144,6 +144,8 @@ fn to_madmin_scanner_metrics(metrics: rustfs_scanner_metrics::metrics::ScannerMe
current_cycle_yield_duration_seconds: metrics.current_cycle_yield_duration_seconds,
current_cycle_throttle_sleep_events: metrics.current_cycle_throttle_sleep_events,
current_cycle_throttle_sleep_duration_seconds: metrics.current_cycle_throttle_sleep_duration_seconds,
current_cycle_disk_io_budget_wait_events: metrics.current_cycle_disk_io_budget_wait_events,
current_cycle_disk_io_budget_wait_duration_seconds: metrics.current_cycle_disk_io_budget_wait_duration_seconds,
current_cycle_ilm_actions: metrics.current_cycle_ilm_actions,
current_cycle_lifecycle_expiry_actions: metrics.current_cycle_lifecycle_expiry_actions,
current_cycle_lifecycle_transition_actions: metrics.current_cycle_lifecycle_transition_actions,
@@ -167,6 +169,8 @@ fn to_madmin_scanner_metrics(metrics: rustfs_scanner_metrics::metrics::ScannerMe
last_cycle_yield_duration_seconds: metrics.last_cycle_yield_duration_seconds,
last_cycle_throttle_sleep_events: metrics.last_cycle_throttle_sleep_events,
last_cycle_throttle_sleep_duration_seconds: metrics.last_cycle_throttle_sleep_duration_seconds,
last_cycle_disk_io_budget_wait_events: metrics.last_cycle_disk_io_budget_wait_events,
last_cycle_disk_io_budget_wait_duration_seconds: metrics.last_cycle_disk_io_budget_wait_duration_seconds,
last_cycle_ilm_actions: metrics.last_cycle_ilm_actions,
last_cycle_heal_objects: metrics.last_cycle_heal_objects,
last_cycle_replication_checks: metrics.last_cycle_replication_checks,
@@ -184,6 +188,7 @@ fn to_madmin_scanner_metrics(metrics: rustfs_scanner_metrics::metrics::ScannerMe
last_cycle_budget_limited: metrics.pacing_pressure.last_cycle_budget_limited,
last_cycle_pause_observed: metrics.pacing_pressure.last_cycle_pause_observed,
last_cycle_throttle_sleep_ratio: metrics.pacing_pressure.last_cycle_throttle_sleep_ratio,
last_cycle_disk_io_budget_wait_ratio: metrics.pacing_pressure.last_cycle_disk_io_budget_wait_ratio,
last_cycle_yield_ratio: metrics.pacing_pressure.last_cycle_yield_ratio,
last_cycle_total_pause_ratio: metrics.pacing_pressure.last_cycle_total_pause_ratio,
},
@@ -234,6 +239,9 @@ fn to_madmin_scanner_metrics(metrics: rustfs_scanner_metrics::metrics::ScannerMe
guardrail_veto_total: metrics.usage_freshness.guardrail_veto_total,
last_guardrail_veto_unix_secs: metrics.usage_freshness.last_guardrail_veto_unix_secs,
last_guardrail_veto_reason: metrics.usage_freshness.last_guardrail_veto_reason,
disk_io_budget_wait_events: metrics.usage_freshness.disk_io_budget_wait_events,
disk_io_budget_wait_millis: metrics.usage_freshness.disk_io_budget_wait_millis,
last_disk_io_budget_wait_unix_secs: metrics.usage_freshness.last_disk_io_budget_wait_unix_secs,
},
maintenance_control: MadminScannerMaintenanceControlSnapshot {
primary_control: metrics.maintenance_control.primary_control,
@@ -750,6 +758,7 @@ mod test {
last_cycle_budget_limited: true,
last_cycle_pause_observed: true,
last_cycle_throttle_sleep_ratio: 0.25,
last_cycle_disk_io_budget_wait_ratio: 0.15,
last_cycle_yield_ratio: 0.05,
last_cycle_total_pause_ratio: 0.3,
},
@@ -892,6 +901,9 @@ mod test {
guardrail_veto_total: 4,
last_guardrail_veto_unix_secs: 17,
last_guardrail_veto_reason: "publication_blocked@<local>".to_string(),
disk_io_budget_wait_events: 5,
disk_io_budget_wait_millis: 2_500,
last_disk_io_budget_wait_unix_secs: 18,
},
..Default::default()
});
crates/madmin/src/metrics.rs
+39
View File
@@ -325,6 +325,12 @@ pub struct ScannerUsageFreshnessSnapshot {
pub last_guardrail_veto_unix_secs: u64,
#[serde(rename = "last_guardrail_veto_reason", default)]
pub last_guardrail_veto_reason: String,
#[serde(rename = "disk_io_budget_wait_events", default)]
pub disk_io_budget_wait_events: u64,
#[serde(rename = "disk_io_budget_wait_millis", default)]
pub disk_io_budget_wait_millis: u64,
#[serde(rename = "last_disk_io_budget_wait_unix_secs", default)]
pub last_disk_io_budget_wait_unix_secs: u64,
}
impl ScannerUsageFreshnessSnapshot {
@@ -351,6 +357,15 @@ impl ScannerUsageFreshnessSnapshot {
}
self.deferred_total = self.deferred_total.saturating_add(other.deferred_total);
self.guardrail_veto_total = self.guardrail_veto_total.saturating_add(other.guardrail_veto_total);
self.disk_io_budget_wait_events = self
.disk_io_budget_wait_events
.saturating_add(other.disk_io_budget_wait_events);
self.disk_io_budget_wait_millis = self
.disk_io_budget_wait_millis
.saturating_add(other.disk_io_budget_wait_millis);
self.last_disk_io_budget_wait_unix_secs = self
.last_disk_io_budget_wait_unix_secs
.max(other.last_disk_io_budget_wait_unix_secs);
if other.last_guardrail_veto_unix_secs > self.last_guardrail_veto_unix_secs {
self.last_guardrail_veto_unix_secs = other.last_guardrail_veto_unix_secs;
self.last_guardrail_veto_reason = other.last_guardrail_veto_reason.clone();
@@ -443,6 +458,8 @@ pub struct ScannerPacingPressureSnapshot {
pub last_cycle_pause_observed: bool,
#[serde(rename = "last_cycle_throttle_sleep_ratio", default)]
pub last_cycle_throttle_sleep_ratio: f64,
#[serde(rename = "last_cycle_disk_io_budget_wait_ratio", default)]
pub last_cycle_disk_io_budget_wait_ratio: f64,
#[serde(rename = "last_cycle_yield_ratio", default)]
pub last_cycle_yield_ratio: f64,
#[serde(rename = "last_cycle_total_pause_ratio", default)]
@@ -458,6 +475,7 @@ impl Default for ScannerPacingPressureSnapshot {
last_cycle_budget_limited: false,
last_cycle_pause_observed: false,
last_cycle_throttle_sleep_ratio: 0.0,
last_cycle_disk_io_budget_wait_ratio: 0.0,
last_cycle_yield_ratio: 0.0,
last_cycle_total_pause_ratio: 0.0,
}
@@ -470,6 +488,8 @@ impl ScannerPacingPressureSnapshot {
"queued_scans"
} else if self.last_cycle_budget_limited {
"cycle_budget"
} else if self.last_cycle_disk_io_budget_wait_ratio > 0.0 {
"disk_io_budget"
} else if self.last_cycle_pause_observed {
"throttle_pause"
} else if self.current_active_scans > 0 {
@@ -492,6 +512,9 @@ impl ScannerPacingPressureSnapshot {
self.last_cycle_throttle_sleep_ratio = self
.last_cycle_throttle_sleep_ratio
.max(other.last_cycle_throttle_sleep_ratio);
self.last_cycle_disk_io_budget_wait_ratio = self
.last_cycle_disk_io_budget_wait_ratio
.max(other.last_cycle_disk_io_budget_wait_ratio);
self.last_cycle_yield_ratio = self.last_cycle_yield_ratio.max(other.last_cycle_yield_ratio);
self.last_cycle_total_pause_ratio = self.last_cycle_total_pause_ratio.max(other.last_cycle_total_pause_ratio);
self.refresh_primary_pressure();
@@ -641,6 +664,10 @@ pub struct ScannerMetrics {
pub current_cycle_throttle_sleep_events: u64,
#[serde(rename = "current_cycle_throttle_sleep_duration_seconds", default)]
pub current_cycle_throttle_sleep_duration_seconds: f64,
#[serde(rename = "current_cycle_disk_io_budget_wait_events", default)]
pub current_cycle_disk_io_budget_wait_events: u64,
#[serde(rename = "current_cycle_disk_io_budget_wait_duration_seconds", default)]
pub current_cycle_disk_io_budget_wait_duration_seconds: f64,
#[serde(rename = "current_cycle_ilm_actions", default)]
pub current_cycle_ilm_actions: u64,
#[serde(rename = "current_cycle_lifecycle_expiry_actions", default)]
@@ -687,6 +714,10 @@ pub struct ScannerMetrics {
pub last_cycle_throttle_sleep_events: u64,
#[serde(rename = "last_cycle_throttle_sleep_duration_seconds", default)]
pub last_cycle_throttle_sleep_duration_seconds: f64,
#[serde(rename = "last_cycle_disk_io_budget_wait_events", default)]
pub last_cycle_disk_io_budget_wait_events: u64,
#[serde(rename = "last_cycle_disk_io_budget_wait_duration_seconds", default)]
pub last_cycle_disk_io_budget_wait_duration_seconds: f64,
#[serde(rename = "last_cycle_ilm_actions", default)]
pub last_cycle_ilm_actions: u64,
#[serde(rename = "last_cycle_heal_objects", default)]
@@ -868,6 +899,10 @@ impl ScannerMetrics {
.current_cycle_throttle_sleep_events
.saturating_add(other.current_cycle_throttle_sleep_events);
self.current_cycle_throttle_sleep_duration_seconds += other.current_cycle_throttle_sleep_duration_seconds;
self.current_cycle_disk_io_budget_wait_events = self
.current_cycle_disk_io_budget_wait_events
.saturating_add(other.current_cycle_disk_io_budget_wait_events);
self.current_cycle_disk_io_budget_wait_duration_seconds += other.current_cycle_disk_io_budget_wait_duration_seconds;
self.current_cycle_ilm_actions = self.current_cycle_ilm_actions.saturating_add(other.current_cycle_ilm_actions);
self.current_cycle_lifecycle_expiry_actions = self
.current_cycle_lifecycle_expiry_actions
@@ -900,6 +935,10 @@ impl ScannerMetrics {
.last_cycle_throttle_sleep_events
.saturating_add(other.last_cycle_throttle_sleep_events);
self.last_cycle_throttle_sleep_duration_seconds += other.last_cycle_throttle_sleep_duration_seconds;
self.last_cycle_disk_io_budget_wait_events = self
.last_cycle_disk_io_budget_wait_events
.saturating_add(other.last_cycle_disk_io_budget_wait_events);
self.last_cycle_disk_io_budget_wait_duration_seconds += other.last_cycle_disk_io_budget_wait_duration_seconds;
self.last_cycle_ilm_actions = self.last_cycle_ilm_actions.saturating_add(other.last_cycle_ilm_actions);
self.last_cycle_lifecycle_expiry_actions = self
.last_cycle_lifecycle_expiry_actions
crates/scanner-metrics/src/metrics.rs
+122 -2
View File
@@ -919,6 +919,8 @@ pub struct Metrics {
current_scan_cycle_yield_duration_millis_start: AtomicU64,
current_scan_cycle_throttle_sleep_events_start: AtomicU64,
current_scan_cycle_throttle_sleep_duration_millis_start: AtomicU64,
current_scan_cycle_disk_io_budget_wait_events_start: AtomicU64,
current_scan_cycle_disk_io_budget_wait_duration_millis_start: AtomicU64,
current_scan_cycle_ilm_actions_start: AtomicU64,
current_scan_cycle_lifecycle_expiry_actions_start: AtomicU64,
current_scan_cycle_lifecycle_transition_actions_start: AtomicU64,
@@ -956,6 +958,8 @@ pub struct Metrics {
last_scan_cycle_yield_duration_millis: AtomicU64,
last_scan_cycle_throttle_sleep_events: AtomicU64,
last_scan_cycle_throttle_sleep_duration_millis: AtomicU64,
last_scan_cycle_disk_io_budget_wait_events: AtomicU64,
last_scan_cycle_disk_io_budget_wait_duration_millis: AtomicU64,
last_scan_cycle_ilm_actions: AtomicU64,
last_scan_cycle_lifecycle_expiry_actions: AtomicU64,
last_scan_cycle_lifecycle_transition_actions: AtomicU64,
@@ -971,6 +975,9 @@ pub struct Metrics {
partial_scan_cycles_by_source: Vec<AtomicU64>,
scanner_yield_duration_millis: AtomicU64,
scanner_throttle_sleep_duration_millis: AtomicU64,
scanner_disk_io_budget_wait_events: AtomicU64,
scanner_disk_io_budget_wait_duration_millis: AtomicU64,
scanner_disk_io_budget_last_wait_unix_secs: AtomicU64,
scanner_ilm_actions: AtomicU64,
scanner_lifecycle_expiry_actions: AtomicU64,
scanner_lifecycle_transition_actions: AtomicU64,
@@ -1141,6 +1148,8 @@ pub struct ScanCycleWorkSnapshot {
yield_duration_millis: u64,
throttle_sleep_events: u64,
throttle_sleep_duration_millis: u64,
disk_io_budget_wait_events: u64,
disk_io_budget_wait_duration_millis: u64,
ilm_actions: u64,
lifecycle_expiry_actions: u64,
lifecycle_transition_actions: u64,
@@ -1261,6 +1270,8 @@ pub struct ScannerPacingPressureSnapshot {
pub last_cycle_budget_limited: bool,
pub last_cycle_pause_observed: bool,
pub last_cycle_throttle_sleep_ratio: f64,
#[serde(default)]
pub last_cycle_disk_io_budget_wait_ratio: f64,
pub last_cycle_yield_ratio: f64,
pub last_cycle_total_pause_ratio: f64,
}
@@ -1274,6 +1285,7 @@ impl Default for ScannerPacingPressureSnapshot {
last_cycle_budget_limited: false,
last_cycle_pause_observed: false,
last_cycle_throttle_sleep_ratio: 0.0,
last_cycle_disk_io_budget_wait_ratio: 0.0,
last_cycle_yield_ratio: 0.0,
last_cycle_total_pause_ratio: 0.0,
}
@@ -1370,6 +1382,14 @@ pub struct ScannerUsageFreshnessSnapshot {
pub last_guardrail_veto_unix_secs: u64,
#[serde(default)]
pub last_guardrail_veto_reason: String,
/// Waits imposed by the per-disk background I/O budget; sustained waits
/// explain slow usage freshness on a budgeted disk.
#[serde(default)]
pub disk_io_budget_wait_events: u64,
#[serde(default)]
pub disk_io_budget_wait_millis: u64,
#[serde(default)]
pub last_disk_io_budget_wait_unix_secs: u64,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
@@ -1438,6 +1458,10 @@ pub struct ScannerMetricsReport {
#[serde(default)]
pub current_cycle_throttle_sleep_duration_seconds: f64,
#[serde(default)]
pub current_cycle_disk_io_budget_wait_events: u64,
#[serde(default)]
pub current_cycle_disk_io_budget_wait_duration_seconds: f64,
#[serde(default)]
pub current_cycle_ilm_actions: u64,
#[serde(default)]
pub current_cycle_lifecycle_expiry_actions: u64,
@@ -1477,6 +1501,10 @@ pub struct ScannerMetricsReport {
#[serde(default)]
pub last_cycle_throttle_sleep_duration_seconds: f64,
#[serde(default)]
pub last_cycle_disk_io_budget_wait_events: u64,
#[serde(default)]
pub last_cycle_disk_io_budget_wait_duration_seconds: f64,
#[serde(default)]
pub last_cycle_ilm_actions: u64,
#[serde(default)]
pub last_cycle_lifecycle_expiry_actions: u64,
@@ -1638,6 +1666,7 @@ fn scanner_primary_pressure(
current_queued_scans: u64,
current_active_scans: u64,
last_cycle_budget_limited: bool,
last_cycle_disk_io_budget_wait_events: u64,
last_cycle_throttle_sleep_events: u64,
last_cycle_yield_events: u64,
) -> &'static str {
@@ -1645,6 +1674,8 @@ fn scanner_primary_pressure(
"queued_scans"
} else if last_cycle_budget_limited {
"cycle_budget"
} else if last_cycle_disk_io_budget_wait_events > 0 {
"disk_io_budget"
} else if last_cycle_throttle_sleep_events > 0 || last_cycle_yield_events > 0 {
"throttle_pause"
} else if current_active_scans > 0 {
@@ -1664,12 +1695,20 @@ fn scanner_pacing_pressure(metrics: &ScannerMetricsReport) -> ScannerPacingPress
.max(usize_to_u64_saturated(metrics.active_scan_paths));
let last_cycle_budget_limited =
scanner_last_cycle_budget_limited(metrics.last_cycle_result_code, metrics.last_cycle_partial_reason.as_str());
let last_cycle_pause_observed = metrics.last_cycle_throttle_sleep_events > 0 || metrics.last_cycle_yield_events > 0;
let last_cycle_pause_observed = metrics.last_cycle_throttle_sleep_events > 0
|| metrics.last_cycle_disk_io_budget_wait_events > 0
|| metrics.last_cycle_yield_events > 0;
let last_cycle_throttle_sleep_ratio =
scanner_ratio(metrics.last_cycle_throttle_sleep_duration_seconds, metrics.last_cycle_duration_seconds);
let last_cycle_disk_io_budget_wait_ratio = scanner_ratio(
metrics.last_cycle_disk_io_budget_wait_duration_seconds,
metrics.last_cycle_duration_seconds,
);
let last_cycle_yield_ratio = scanner_ratio(metrics.last_cycle_yield_duration_seconds, metrics.last_cycle_duration_seconds);
let last_cycle_total_pause_ratio = scanner_ratio(
metrics.last_cycle_throttle_sleep_duration_seconds.max(0.0) + metrics.last_cycle_yield_duration_seconds.max(0.0),
metrics.last_cycle_throttle_sleep_duration_seconds.max(0.0)
+ metrics.last_cycle_disk_io_budget_wait_duration_seconds.max(0.0)
+ metrics.last_cycle_yield_duration_seconds.max(0.0),
metrics.last_cycle_duration_seconds,
);
@@ -1678,6 +1717,7 @@ fn scanner_pacing_pressure(metrics: &ScannerMetricsReport) -> ScannerPacingPress
current_queued_scans,
current_active_scans,
last_cycle_budget_limited,
metrics.last_cycle_disk_io_budget_wait_events,
metrics.last_cycle_throttle_sleep_events,
metrics.last_cycle_yield_events,
)
@@ -1687,6 +1727,7 @@ fn scanner_pacing_pressure(metrics: &ScannerMetricsReport) -> ScannerPacingPress
last_cycle_budget_limited,
last_cycle_pause_observed,
last_cycle_throttle_sleep_ratio,
last_cycle_disk_io_budget_wait_ratio,
last_cycle_yield_ratio,
last_cycle_total_pause_ratio,
}
@@ -2005,6 +2046,8 @@ impl Metrics {
current_scan_cycle_yield_duration_millis_start: AtomicU64::new(0),
current_scan_cycle_throttle_sleep_events_start: AtomicU64::new(0),
current_scan_cycle_throttle_sleep_duration_millis_start: AtomicU64::new(0),
current_scan_cycle_disk_io_budget_wait_events_start: AtomicU64::new(0),
current_scan_cycle_disk_io_budget_wait_duration_millis_start: AtomicU64::new(0),
current_scan_cycle_ilm_actions_start: AtomicU64::new(0),
current_scan_cycle_lifecycle_expiry_actions_start: AtomicU64::new(0),
current_scan_cycle_lifecycle_transition_actions_start: AtomicU64::new(0),
@@ -2038,6 +2081,8 @@ impl Metrics {
last_scan_cycle_yield_duration_millis: AtomicU64::new(0),
last_scan_cycle_throttle_sleep_events: AtomicU64::new(0),
last_scan_cycle_throttle_sleep_duration_millis: AtomicU64::new(0),
last_scan_cycle_disk_io_budget_wait_events: AtomicU64::new(0),
last_scan_cycle_disk_io_budget_wait_duration_millis: AtomicU64::new(0),
last_scan_cycle_ilm_actions: AtomicU64::new(0),
last_scan_cycle_lifecycle_expiry_actions: AtomicU64::new(0),
last_scan_cycle_lifecycle_transition_actions: AtomicU64::new(0),
@@ -2053,6 +2098,9 @@ impl Metrics {
partial_scan_cycles_by_source: ScannerWorkSource::all().iter().map(|_| AtomicU64::new(0)).collect(),
scanner_yield_duration_millis: AtomicU64::new(0),
scanner_throttle_sleep_duration_millis: AtomicU64::new(0),
scanner_disk_io_budget_wait_events: AtomicU64::new(0),
scanner_disk_io_budget_wait_duration_millis: AtomicU64::new(0),
scanner_disk_io_budget_last_wait_unix_secs: AtomicU64::new(0),
scanner_ilm_actions: AtomicU64::new(0),
scanner_lifecycle_expiry_actions: AtomicU64::new(0),
scanner_lifecycle_transition_actions: AtomicU64::new(0),
@@ -2276,6 +2324,19 @@ impl Metrics {
});
}
/// Record one wait imposed by a per-disk background I/O budget.
pub fn record_scanner_disk_io_budget_wait(&self, duration: Duration) {
self.scanner_disk_io_budget_wait_events.fetch_add(1, Ordering::Relaxed);
let duration_millis = duration_millis_saturated(duration);
let _ = self
.scanner_disk_io_budget_wait_duration_millis
.try_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
Some(current.saturating_add(duration_millis))
});
self.scanner_disk_io_budget_last_wait_unix_secs
.store(unix_now_secs(), Ordering::Relaxed);
}
pub fn record_scanner_ilm_action(&self, count: u64) {
self.scanner_ilm_actions.fetch_add(count, Ordering::Relaxed);
self.record_scanner_source_executed(ScannerWorkSource::Lifecycle, count);
@@ -2998,6 +3059,10 @@ impl Metrics {
.store(snapshot.throttle_sleep_events, Ordering::Relaxed);
self.current_scan_cycle_throttle_sleep_duration_millis_start
.store(snapshot.throttle_sleep_duration_millis, Ordering::Relaxed);
self.current_scan_cycle_disk_io_budget_wait_events_start
.store(snapshot.disk_io_budget_wait_events, Ordering::Relaxed);
self.current_scan_cycle_disk_io_budget_wait_duration_millis_start
.store(snapshot.disk_io_budget_wait_duration_millis, Ordering::Relaxed);
self.current_scan_cycle_ilm_actions_start
.store(snapshot.ilm_actions, Ordering::Relaxed);
self.current_scan_cycle_lifecycle_expiry_actions_start
@@ -3062,6 +3127,8 @@ impl Metrics {
yield_duration_millis: self.scanner_yield_duration_millis.load(Ordering::Relaxed),
throttle_sleep_events: self.lifetime(Metric::ThrottleSleep),
throttle_sleep_duration_millis: self.scanner_throttle_sleep_duration_millis.load(Ordering::Relaxed),
disk_io_budget_wait_events: self.scanner_disk_io_budget_wait_events.load(Ordering::Relaxed),
disk_io_budget_wait_duration_millis: self.scanner_disk_io_budget_wait_duration_millis.load(Ordering::Relaxed),
ilm_actions: self.scanner_ilm_actions.load(Ordering::Relaxed),
lifecycle_expiry_actions: self.scanner_lifecycle_expiry_actions.load(Ordering::Relaxed),
lifecycle_transition_actions: self.scanner_lifecycle_transition_actions.load(Ordering::Relaxed),
@@ -3083,6 +3150,12 @@ impl Metrics {
throttle_sleep_duration_millis: self
.current_scan_cycle_throttle_sleep_duration_millis_start
.load(Ordering::Relaxed),
disk_io_budget_wait_events: self
.current_scan_cycle_disk_io_budget_wait_events_start
.load(Ordering::Relaxed),
disk_io_budget_wait_duration_millis: self
.current_scan_cycle_disk_io_budget_wait_duration_millis_start
.load(Ordering::Relaxed),
ilm_actions: self.current_scan_cycle_ilm_actions_start.load(Ordering::Relaxed),
lifecycle_expiry_actions: self.current_scan_cycle_lifecycle_expiry_actions_start.load(Ordering::Relaxed),
lifecycle_transition_actions: self
@@ -3120,6 +3193,12 @@ impl Metrics {
throttle_sleep_duration_millis: current
.throttle_sleep_duration_millis
.saturating_sub(start.throttle_sleep_duration_millis),
disk_io_budget_wait_events: current
.disk_io_budget_wait_events
.saturating_sub(start.disk_io_budget_wait_events),
disk_io_budget_wait_duration_millis: current
.disk_io_budget_wait_duration_millis
.saturating_sub(start.disk_io_budget_wait_duration_millis),
ilm_actions: current.ilm_actions.saturating_sub(start.ilm_actions),
lifecycle_expiry_actions: current
.lifecycle_expiry_actions
@@ -3393,6 +3472,10 @@ impl Metrics {
.store(work.throttle_sleep_events, Ordering::Relaxed);
self.last_scan_cycle_throttle_sleep_duration_millis
.store(work.throttle_sleep_duration_millis, Ordering::Relaxed);
self.last_scan_cycle_disk_io_budget_wait_events
.store(work.disk_io_budget_wait_events, Ordering::Relaxed);
self.last_scan_cycle_disk_io_budget_wait_duration_millis
.store(work.disk_io_budget_wait_duration_millis, Ordering::Relaxed);
self.last_scan_cycle_ilm_actions.store(work.ilm_actions, Ordering::Relaxed);
self.last_scan_cycle_lifecycle_expiry_actions
.store(work.lifecycle_expiry_actions, Ordering::Relaxed);
@@ -3470,6 +3553,9 @@ impl Metrics {
m.current_cycle_yield_duration_seconds = current_work.yield_duration_millis as f64 / 1000.0;
m.current_cycle_throttle_sleep_events = current_work.throttle_sleep_events;
m.current_cycle_throttle_sleep_duration_seconds = current_work.throttle_sleep_duration_millis as f64 / 1000.0;
m.current_cycle_disk_io_budget_wait_events = current_work.disk_io_budget_wait_events;
m.current_cycle_disk_io_budget_wait_duration_seconds =
current_work.disk_io_budget_wait_duration_millis as f64 / 1000.0;
m.current_cycle_ilm_actions = current_work.ilm_actions;
m.current_cycle_lifecycle_expiry_actions = current_work.lifecycle_expiry_actions;
m.current_cycle_lifecycle_transition_actions = current_work.lifecycle_transition_actions;
@@ -3540,6 +3626,11 @@ impl Metrics {
m.last_cycle_throttle_sleep_events = self.last_scan_cycle_throttle_sleep_events.load(Ordering::Relaxed);
m.last_cycle_throttle_sleep_duration_seconds =
self.last_scan_cycle_throttle_sleep_duration_millis.load(Ordering::Relaxed) as f64 / 1000.0;
m.last_cycle_disk_io_budget_wait_events = self.last_scan_cycle_disk_io_budget_wait_events.load(Ordering::Relaxed);
m.last_cycle_disk_io_budget_wait_duration_seconds = self
.last_scan_cycle_disk_io_budget_wait_duration_millis
.load(Ordering::Relaxed) as f64
/ 1000.0;
m.last_cycle_ilm_actions = self.last_scan_cycle_ilm_actions.load(Ordering::Relaxed);
m.last_cycle_lifecycle_expiry_actions = self.last_scan_cycle_lifecycle_expiry_actions.load(Ordering::Relaxed);
m.last_cycle_lifecycle_transition_actions = self.last_scan_cycle_lifecycle_transition_actions.load(Ordering::Relaxed);
@@ -3621,6 +3712,9 @@ impl Metrics {
Err(poisoned) => poisoned.into_inner().clone(),
},
guardrail_veto_total: self.scanner_guardrail_veto_total.load(Ordering::Relaxed),
disk_io_budget_wait_events: self.scanner_disk_io_budget_wait_events.load(Ordering::Relaxed),
disk_io_budget_wait_millis: self.scanner_disk_io_budget_wait_duration_millis.load(Ordering::Relaxed),
last_disk_io_budget_wait_unix_secs: self.scanner_disk_io_budget_last_wait_unix_secs.load(Ordering::Relaxed),
last_guardrail_veto_unix_secs: self.scanner_guardrail_last_veto_unix_secs.load(Ordering::Relaxed),
last_guardrail_veto_reason: match self.scanner_guardrail_last_veto_reason.lock() {
Ok(reason) => reason.clone(),
@@ -4098,6 +4192,32 @@ mod tests {
assert_eq!(lifecycle_checked, 4, "ILM-checked versions tracked on the Lifecycle source");
}
#[tokio::test]
async fn report_includes_disk_io_budget_waits_in_pacing_and_freshness() {
let metrics = Metrics::new();
metrics.record_scanner_disk_io_budget_wait(Duration::from_millis(250));
metrics.record_scanner_disk_io_budget_wait(Duration::from_millis(750));
metrics.record_scan_cycle_complete(true, Duration::from_secs(10));
metrics.record_scan_cycle_work(ScanCycleWorkSnapshot {
disk_io_budget_wait_events: 2,
disk_io_budget_wait_duration_millis: 1_000,
..Default::default()
});
let report = metrics.report().await;
assert_eq!(report.usage_freshness.disk_io_budget_wait_events, 2);
assert_eq!(report.usage_freshness.disk_io_budget_wait_millis, 1_000);
assert!(report.usage_freshness.last_disk_io_budget_wait_unix_secs > 0);
assert_eq!(report.last_cycle_disk_io_budget_wait_events, 2);
assert_eq!(report.last_cycle_disk_io_budget_wait_duration_seconds, 1.0);
assert_eq!(report.pacing_pressure.primary_pressure, "disk_io_budget");
assert!(report.pacing_pressure.last_cycle_pause_observed);
assert_eq!(report.pacing_pressure.last_cycle_disk_io_budget_wait_ratio, 0.1);
assert_eq!(report.pacing_pressure.last_cycle_total_pause_ratio, 0.1);
assert_eq!(report.pacing_pressure.last_cycle_throttle_sleep_ratio, 0.0);
}
#[tokio::test]
async fn report_derives_scanner_pacing_pressure() {
let metrics = Metrics::new();
crates/scanner/src/lib.rs
+13 -10
View File
@@ -86,15 +86,15 @@ pub use remote_scanner::{
pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_status, validate_scanner_runtime_config};
pub use rustfs_scanner_metrics::last_minute;
pub use scanner::{
SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD, ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus,
ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase, ScannerPauseBacklogReplicaId,
ScannerPauseBacklogReplicaStatus, ScannerPauseBacklogReplicaStatusState, ScannerPauseBacklogStatus,
ScannerPauseBacklogThresholds, ScannerRecoveryIntentAcceptResult, ScannerRecoveryIntentConflict, ScannerRecoveryIntentRecord,
ScannerRecoveryIntentRequest, ScannerUsageStateResetResult, accept_scanner_usage_recovery_intent,
get_scanner_usage_recovery_intent, init_data_scanner, init_scanner_with_recovery, register_scanner_pause_backlog_retirement,
reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild, run_scanner_usage_recovery_intent,
scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_recovery_actor_sha256,
scanner_topology_digest,
SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD, ScannerBitrotStartSource, ScannerBitrotStatus, ScannerCycleRecoveryMarker,
ScannerCycleRecoveryStatus, ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase,
ScannerPauseBacklogReplicaId, ScannerPauseBacklogReplicaStatus, ScannerPauseBacklogReplicaStatusState,
ScannerPauseBacklogStatus, ScannerPauseBacklogThresholds, ScannerRecoveryIntentAcceptResult, ScannerRecoveryIntentConflict,
ScannerRecoveryIntentRecord, ScannerRecoveryIntentRequest, ScannerUsageStateResetResult,
accept_scanner_usage_recovery_intent, get_scanner_usage_recovery_intent, init_data_scanner, init_scanner_with_recovery,
register_scanner_pause_backlog_retirement, reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild,
run_scanner_usage_recovery_intent, scanner_bitrot_status, scanner_cycle_recovery_status, scanner_cycle_schedule_status,
scanner_pause_backlog_status, scanner_recovery_actor_sha256, scanner_topology_digest,
};
pub use scanner_io::{
ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageMutationObserver,
@@ -108,7 +108,10 @@ pub use scanner_io::{
scanner_maintenance_generation, set_scanner_dirty_usage_clear_observer, set_scanner_dirty_usage_mutation_observer,
};
pub use segment_invalidation::SegmentInvalidationProducerIdentity;
pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER};
pub use sleeper::{
DynamicSleeper, SCANNER_DISK_IO_BUDGET_MAX_WAIT, SCANNER_IDLE_MODE, SCANNER_SLEEPER, ScannerDiskIoBudgetStatus,
ScannerDiskMedia, scanner_disk_io_budget_status,
};
use std::sync::atomic::{AtomicU64, Ordering};
pub use storage_api::ScannerReplicationConfig as ReplicationConfig;
pub use storage_api::scan::{
crates/scanner/src/runtime_config.rs
+128 -12
View File
@@ -14,23 +14,25 @@
use crate::resolve_scanner_server_config;
use crate::scanner_budget::ScannerCycleBudgetConfig;
use crate::sleeper::{SCANNER_SLEEPER, scanner_default_speed};
use crate::sleeper::{SCANNER_SLEEPER, ScannerDiskIoBudgetConfig, ScannerIoBudgetLimit, scanner_default_speed};
use rustfs_config::{
DEFAULT_DELIMITER, DEFAULT_HEAL_BITROT_CYCLE_SECS, DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS,
DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE, DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS,
DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS,
DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS,
DEFAULT_SCANNER_SPEED, DEFAULT_SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS,
ENV_SCANNER_ALERT_EXCESS_FOLDERS, ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, ENV_SCANNER_ALERT_EXCESS_VERSIONS,
ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES,
ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, ENV_SCANNER_IDLE_MODE,
DEFAULT_SCANNER_DISK_IO_BUDGET, DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS,
DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_SPEED, DEFAULT_SCANNER_WALKERS_PER_DISK,
DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, ENV_SCANNER_ALERT_EXCESS_FOLDERS, ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE,
ENV_SCANNER_ALERT_EXCESS_VERSIONS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE,
ENV_SCANNER_CYCLE_MAX_DIRECTORIES, ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY,
ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, ENV_SCANNER_DISK_IO_BUDGET_IOPS, ENV_SCANNER_IDLE_MODE,
ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED,
ENV_SCANNER_START_DELAY_SECS, ENV_SCANNER_WALKERS_PER_DISK, ENV_SCANNER_YIELD_EVERY_N_OBJECTS, HEAL_BITROT_CYCLE,
HEAL_SUB_SYS, MAX_SCANNER_WALKERS_PER_DISK, SCANNER_ALERT_EXCESS_FOLDERS, SCANNER_ALERT_EXCESS_VERSION_SIZE,
SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE,
SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE,
SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED, SCANNER_START_DELAY,
SCANNER_SUB_SYS, SCANNER_WALKERS_PER_DISK, SCANNER_YIELD_EVERY_N_OBJECTS, ScannerSpeed,
SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY,
SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, SCANNER_DISK_IO_BUDGET_IOPS, SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS,
SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED, SCANNER_START_DELAY, SCANNER_SUB_SYS,
SCANNER_WALKERS_PER_DISK, SCANNER_YIELD_EVERY_N_OBJECTS, ScannerSpeed,
server_config::{Config as ServerConfig, KVS},
};
use serde::Serialize;
@@ -54,6 +56,7 @@ const NO_DEFAULT_CYCLE_OVERRIDE: u64 = 0;
const MAX_SCANNER_DELAY_FACTOR: f64 = 10_000.0;
const SCANNER_DELAY_RANGE_REASON: &str = "expected scanner delay between 0 and 10000";
const SCANNER_WALKERS_PER_DISK_RANGE_REASON: &str = "expected walkers per disk between 1 and 16";
const SCANNER_DISK_IO_BUDGET_REASON: &str = "expected auto, off, or a non-negative integer rate";
static SCANNER_DEFAULT_CYCLE_SECS: AtomicU64 = AtomicU64::new(NO_DEFAULT_CYCLE_OVERRIDE);
static SCANNER_RUNTIME_CONFIG_GENERATION: AtomicU64 = AtomicU64::new(0);
@@ -99,6 +102,10 @@ pub(crate) struct ScannerRuntimeConfig {
pub(crate) max_concurrent_disk_scans_source: ScannerRuntimeConfigSource,
pub(crate) walkers_per_disk: usize,
pub(crate) walkers_per_disk_source: ScannerRuntimeConfigSource,
pub(crate) disk_io_budget_iops: ScannerIoBudgetLimit,
pub(crate) disk_io_budget_iops_source: ScannerRuntimeConfigSource,
pub(crate) disk_io_budget_bytes_per_sec: ScannerIoBudgetLimit,
pub(crate) disk_io_budget_bytes_per_sec_source: ScannerRuntimeConfigSource,
pub(crate) yield_every_n_objects: u64,
pub(crate) yield_every_n_objects_source: ScannerRuntimeConfigSource,
pub(crate) alert_excess_versions: u64,
@@ -143,6 +150,10 @@ impl Default for ScannerRuntimeConfig {
max_concurrent_disk_scans_source: ScannerRuntimeConfigSource::Default,
walkers_per_disk: DEFAULT_SCANNER_WALKERS_PER_DISK,
walkers_per_disk_source: ScannerRuntimeConfigSource::Default,
disk_io_budget_iops: ScannerIoBudgetLimit::Auto,
disk_io_budget_iops_source: ScannerRuntimeConfigSource::Default,
disk_io_budget_bytes_per_sec: ScannerIoBudgetLimit::Auto,
disk_io_budget_bytes_per_sec_source: ScannerRuntimeConfigSource::Default,
yield_every_n_objects: DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS,
yield_every_n_objects_source: ScannerRuntimeConfigSource::Default,
alert_excess_versions: DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS,
@@ -187,6 +198,9 @@ pub struct ScannerRuntimeConfigStatus {
pub max_concurrent_set_scans: ScannerRuntimeConfigValue<usize>,
pub max_concurrent_disk_scans: ScannerRuntimeConfigValue<usize>,
pub walkers_per_disk: ScannerRuntimeConfigValue<usize>,
/// `auto`, `off`, or an explicit rate, as configured.
pub disk_io_budget_iops: ScannerRuntimeConfigValue<String>,
pub disk_io_budget_bytes_per_sec: ScannerRuntimeConfigValue<String>,
pub yield_every_n_objects: ScannerRuntimeConfigValue<u64>,
pub alert_excess_versions: ScannerRuntimeConfigValue<u64>,
pub alert_excess_version_size: ScannerRuntimeConfigValue<u64>,
@@ -280,6 +294,24 @@ fn parse_config_speed(value: String) -> Result<ScannerSpeed, ScannerRuntimeConfi
ScannerSpeed::parse_str(&value).ok_or_else(|| invalid_value(SCANNER_SPEED, value, "expected scanner speed preset"))
}
fn parse_config_io_budget(key: &'static str, value: String) -> Result<ScannerIoBudgetLimit, ScannerRuntimeConfigError> {
ScannerIoBudgetLimit::parse(&value).ok_or_else(|| invalid_value(key, value, SCANNER_DISK_IO_BUDGET_REASON))
}
fn lookup_io_budget(
kvs: Option<&KVS>,
key: &'static str,
env_key: &'static str,
) -> Result<(ScannerIoBudgetLimit, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
if let Some(value) = rustfs_utils::get_env_opt_str(env_key) {
return parse_config_io_budget(env_key, value).map(|limit| (limit, ScannerRuntimeConfigSource::Env));
}
if let Some(value) = config_value(kvs, key, DEFAULT_SCANNER_DISK_IO_BUDGET) {
return parse_config_io_budget(key, value).map(|limit| (limit, ScannerRuntimeConfigSource::Config));
}
Ok((ScannerIoBudgetLimit::Auto, ScannerRuntimeConfigSource::Default))
}
fn parse_config_delay(key: &'static str, value: String) -> Result<f64, ScannerRuntimeConfigError> {
let parsed = value
.parse::<f64>()
@@ -401,6 +433,12 @@ fn validate_persisted_scanner_runtime_config(config: &ServerConfig) -> Result<()
if let Some(value) = config_value(scanner_kvs, SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_WALKERS_PER_DISK) {
walkers_per_disk_from_count(SCANNER_WALKERS_PER_DISK, parse_config_usize(SCANNER_WALKERS_PER_DISK, value)?)?;
}
if let Some(value) = config_value(scanner_kvs, SCANNER_DISK_IO_BUDGET_IOPS, DEFAULT_SCANNER_DISK_IO_BUDGET) {
parse_config_io_budget(SCANNER_DISK_IO_BUDGET_IOPS, value)?;
}
if let Some(value) = config_value(scanner_kvs, SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, DEFAULT_SCANNER_DISK_IO_BUDGET) {
parse_config_io_budget(SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, value)?;
}
validate_optional_config_u64(scanner_kvs, SCANNER_YIELD_EVERY_N_OBJECTS, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS)?;
validate_optional_config_u64(scanner_kvs, SCANNER_ALERT_EXCESS_VERSIONS, DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS)?;
validate_optional_config_u64(scanner_kvs, SCANNER_ALERT_EXCESS_VERSION_SIZE, DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE)?;
@@ -682,6 +720,13 @@ pub(crate) fn lookup_scanner_runtime_config(
DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS,
)?;
let (walkers_per_disk, walkers_per_disk_source) = lookup_walkers_per_disk(scanner_kvs)?;
let (disk_io_budget_iops, disk_io_budget_iops_source) =
lookup_io_budget(scanner_kvs, SCANNER_DISK_IO_BUDGET_IOPS, ENV_SCANNER_DISK_IO_BUDGET_IOPS)?;
let (disk_io_budget_bytes_per_sec, disk_io_budget_bytes_per_sec_source) = lookup_io_budget(
scanner_kvs,
SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC,
ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC,
)?;
let (yield_every_n_objects, yield_every_n_objects_source) = lookup_u64(
scanner_kvs,
SCANNER_YIELD_EVERY_N_OBJECTS,
@@ -734,6 +779,10 @@ pub(crate) fn lookup_scanner_runtime_config(
max_concurrent_disk_scans_source,
walkers_per_disk,
walkers_per_disk_source,
disk_io_budget_iops,
disk_io_budget_iops_source,
disk_io_budget_bytes_per_sec,
disk_io_budget_bytes_per_sec_source,
yield_every_n_objects,
yield_every_n_objects_source,
alert_excess_versions,
@@ -747,6 +796,11 @@ pub(crate) fn lookup_scanner_runtime_config(
fn apply_resolved_runtime_config(config: ScannerRuntimeConfig) {
SCANNER_SLEEPER.update_from_runtime_config(config.delay, config.max_wait, config.idle_mode, config.yield_every_n_objects);
SCANNER_SLEEPER.update_disk_io_budget(ScannerDiskIoBudgetConfig {
iops: config.disk_io_budget_iops,
bytes_per_sec: config.disk_io_budget_bytes_per_sec,
speed: config.speed,
});
if let Ok(mut guard) = SCANNER_RUNTIME_CONFIG.write() {
*guard = config;
}
@@ -874,6 +928,14 @@ pub fn scanner_runtime_config_status() -> ScannerRuntimeConfigStatus {
value: config.walkers_per_disk,
source: config.walkers_per_disk_source,
},
disk_io_budget_iops: ScannerRuntimeConfigValue {
value: config.disk_io_budget_iops.to_string(),
source: config.disk_io_budget_iops_source,
},
disk_io_budget_bytes_per_sec: ScannerRuntimeConfigValue {
value: config.disk_io_budget_bytes_per_sec.to_string(),
source: config.disk_io_budget_bytes_per_sec_source,
},
yield_every_n_objects: ScannerRuntimeConfigValue {
value: config.yield_every_n_objects,
source: config.yield_every_n_objects_source,
@@ -941,13 +1003,15 @@ pub(crate) fn scanner_alert_excess_folders() -> u64 {
mod tests {
use super::{ScannerRuntimeConfigSource, lookup_scanner_runtime_config, validate_scanner_runtime_config};
use crate::init_ecstore_config_for_scanner_tests;
use crate::sleeper::ScannerIoBudgetLimit;
use rustfs_config::server_config::{Config as ServerConfig, KVS};
use rustfs_config::{
DEFAULT_DELIMITER, DEFAULT_HEAL_BITROT_CYCLE_SECS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS,
ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY,
ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED, ENV_SCANNER_WALKERS_PER_DISK, HEAL_BITROT_CYCLE, HEAL_SUB_SYS,
SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES,
SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_IDLE_MODE, SCANNER_SPEED, SCANNER_SUB_SYS,
ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, ENV_SCANNER_DISK_IO_BUDGET_IOPS, ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED,
ENV_SCANNER_WALKERS_PER_DISK, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT,
SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY,
SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, SCANNER_DISK_IO_BUDGET_IOPS, SCANNER_IDLE_MODE, SCANNER_SPEED, SCANNER_SUB_SYS,
SCANNER_WALKERS_PER_DISK, ScannerSpeed,
};
use serial_test::serial;
@@ -1217,6 +1281,58 @@ mod tests {
});
}
#[test]
#[serial]
fn scanner_runtime_config_disk_io_budget_defaults_to_auto_and_parses_overrides() {
let config = server_config_with_scanner(&[
(SCANNER_DISK_IO_BUDGET_IOPS, "250"),
(SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, "off"),
]);
with_var_unset(ENV_SCANNER_DISK_IO_BUDGET_IOPS, || {
with_var_unset(ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, || {
let resolved =
lookup_scanner_runtime_config(Some(&server_config_with_scanner(&[]))).expect("default budget config");
assert_eq!(resolved.disk_io_budget_iops, ScannerIoBudgetLimit::Auto);
assert_eq!(resolved.disk_io_budget_iops_source, ScannerRuntimeConfigSource::Default);
assert_eq!(resolved.disk_io_budget_bytes_per_sec, ScannerIoBudgetLimit::Auto);
assert_eq!(resolved.disk_io_budget_bytes_per_sec_source, ScannerRuntimeConfigSource::Default);
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("persisted budget config");
assert_eq!(resolved.disk_io_budget_iops, ScannerIoBudgetLimit::Explicit(250));
assert_eq!(resolved.disk_io_budget_iops_source, ScannerRuntimeConfigSource::Config);
assert_eq!(resolved.disk_io_budget_bytes_per_sec, ScannerIoBudgetLimit::Unlimited);
assert_eq!(resolved.disk_io_budget_bytes_per_sec_source, ScannerRuntimeConfigSource::Config);
let invalid = server_config_with_scanner(&[(SCANNER_DISK_IO_BUDGET_IOPS, "fast")]);
let err = validate_scanner_runtime_config(&invalid).expect_err("a non-numeric budget must be rejected");
assert!(err.to_string().contains(SCANNER_DISK_IO_BUDGET_IOPS), "{err}");
assert!(lookup_scanner_runtime_config(Some(&invalid)).is_err());
});
});
with_var(ENV_SCANNER_DISK_IO_BUDGET_IOPS, Some("0"), || {
with_var(ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, Some("auto"), || {
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("env budget config");
assert_eq!(resolved.disk_io_budget_iops, ScannerIoBudgetLimit::Explicit(0));
assert_eq!(resolved.disk_io_budget_iops_source, ScannerRuntimeConfigSource::Env);
assert_eq!(resolved.disk_io_budget_bytes_per_sec, ScannerIoBudgetLimit::Auto);
assert_eq!(resolved.disk_io_budget_bytes_per_sec_source, ScannerRuntimeConfigSource::Env);
super::apply_resolved_runtime_config(resolved);
let status = super::scanner_runtime_config_status();
assert_eq!(status.disk_io_budget_iops.value, "0");
assert_eq!(status.disk_io_budget_iops.source, ScannerRuntimeConfigSource::Env);
assert_eq!(status.disk_io_budget_bytes_per_sec.value, "auto");
assert_eq!(status.disk_io_budget_bytes_per_sec.source, ScannerRuntimeConfigSource::Env);
});
});
with_var(ENV_SCANNER_DISK_IO_BUDGET_IOPS, Some("-5"), || {
assert!(lookup_scanner_runtime_config(Some(&config)).is_err());
});
super::refresh_scanner_runtime_config_for_tests();
}
#[test]
#[serial]
fn scanner_runtime_config_walkers_per_disk_accepts_persisted_range_and_rejects_zero_or_excess() {
crates/scanner/src/scanner.rs
+46 -30
View File
@@ -18,7 +18,10 @@ use std::sync::Mutex as StdMutex;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, LazyLock, RwLock};
use self::heal_info::{BackgroundHealInfoReadStatus, read_background_heal_info_with_epoch, save_background_heal_info_for_epoch};
use self::heal_info::{
BackgroundHealInfoReadStatus, ScannerBitrotStart, read_background_heal_info_with_epoch, record_scanner_bitrot_status,
resolve_background_heal_info_for_cycle, save_background_heal_info_for_epoch, scanner_bitrot_start_for_cycle,
};
use crate::data_usage_define::{
BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH, DATA_USAGE_OBSERVED_OBJ_NAME_PATH,
DATA_USAGE_RECOVERY_PATH, DataUsageCache, DataUsageCacheRevision, LEGACY_DATA_USAGE_OBJ_NAME_PATH, read_config_revision,
@@ -1462,12 +1465,7 @@ impl ScannerBitrotPolicy {
}
}
fn get_cycle_scan_mode(
current_cycle: u64,
bitrot_start_cycle: u64,
bitrot_start_time: Option<DateTime<Utc>>,
policy: ScannerBitrotPolicy,
) -> HealScanMode {
fn get_cycle_scan_mode(current_cycle: u64, start: ScannerBitrotStart, policy: ScannerBitrotPolicy) -> HealScanMode {
let Some(bitrot_cycle) = policy.cycle else {
return HealScanMode::Normal;
};
@@ -1480,18 +1478,23 @@ fn get_cycle_scan_mode(
return HealScanMode::Deep;
}
if current_cycle.saturating_sub(bitrot_start_cycle) < deep_window_cycles {
return HealScanMode::Deep;
}
let Some(bitrot_start_time) = bitrot_start_time else {
return HealScanMode::Deep;
// Only a recorded Deep window keeps the selection window open. A missing
// record used to be read as a window that started at cycle 0, which ran
// Deep for `deep_window_cycles` cycles whenever the start could not be
// read back (rustfs/rustfs#8015); an in-memory anchor only counts the
// cadence.
let started = match start {
ScannerBitrotStart::None => return HealScanMode::Normal,
ScannerBitrotStart::Window { cycle, started } => {
if current_cycle.saturating_sub(cycle) < deep_window_cycles {
return HealScanMode::Deep;
}
started
}
ScannerBitrotStart::Anchor(anchored) => anchored,
};
let elapsed = Utc::now()
.signed_duration_since(bitrot_start_time)
.to_std()
.unwrap_or(Duration::ZERO);
let elapsed = Utc::now().signed_duration_since(started).to_std().unwrap_or(Duration::ZERO);
if elapsed >= bitrot_cycle {
HealScanMode::Deep
} else {
@@ -1795,20 +1798,18 @@ where
}
BackgroundHealInfoReadStatus::Loaded | BackgroundHealInfoReadStatus::Missing => {}
}
let mut background_heal_info = background_heal_read.info;
let background_heal_epoch = background_heal_read.expected_epoch;
let bitrot_policy = ScannerBitrotPolicy::new(
configured_bitrot_cycle,
!storeapi.setup_is_erasure_sd().await,
heal_object_select_prob() as u64,
);
let deep_scan_supported = !storeapi.setup_is_erasure_sd().await;
let bitrot_policy = ScannerBitrotPolicy::new(configured_bitrot_cycle, deep_scan_supported, heal_object_select_prob() as u64);
// The persisted record wins when it carries a start; the in-memory record
// covers a save that was skipped or lost since this process recorded it.
let (mut background_heal_info, mut bitrot_start_source) = resolve_background_heal_info_for_cycle(&background_heal_read);
let bitrot_start = scanner_bitrot_start_for_cycle(&background_heal_info, Utc::now());
if matches!(bitrot_start, ScannerBitrotStart::Anchor(_)) {
bitrot_start_source = ScannerBitrotStartSource::Memory;
}
let scan_mode = get_cycle_scan_mode(
cycle_info.current,
background_heal_info.bitrot_start_cycle,
background_heal_info.bitrot_start_time,
bitrot_policy,
);
let scan_mode = get_cycle_scan_mode(cycle_info.current, bitrot_start, bitrot_policy);
info!(
target: "rustfs::scanner",
event = EVENT_SCANNER_CYCLE_STATE,
@@ -1816,6 +1817,7 @@ where
subsystem = LOG_SUBSYSTEM_RUNTIME,
cycle = cycle_info.current,
scan_mode = ?scan_mode,
bitrot_start_source = bitrot_start_source.as_str(),
state = "started",
"Scanner cycle started"
);
@@ -1828,8 +1830,19 @@ where
bitrot_policy,
) {
background_heal_info = new_heal_info.clone();
if new_heal_info.bitrot_start_time != background_heal_read.info.bitrot_start_time {
// A start recorded by this cycle is only in memory until a later
// read returns it from storage.
bitrot_start_source = ScannerBitrotStartSource::Memory;
}
save_background_heal_info_for_epoch(storeapi.clone(), new_heal_info, background_heal_epoch).await;
}
record_scanner_bitrot_status(
scanner_bitrot_start_for_cycle(&background_heal_info, Utc::now()),
bitrot_start_source,
deep_scan_supported,
scan_mode,
);
let cycle_start = std::time::Instant::now();
// Baseline reads are part of the same publication proof as the eventual
@@ -4099,7 +4112,10 @@ pub use cycle_state::{
pub(crate) use cycle_state::{
current_scanner_leader_epoch, decode_persisted_scanner_cycle_fence, load_scanner_cycle_state_for_startup,
};
pub use heal_info::{BackgroundHealInfo, read_background_heal_info, save_background_heal_info};
pub use heal_info::{
BackgroundHealInfo, ScannerBitrotStartSource, ScannerBitrotStatus, read_background_heal_info, save_background_heal_info,
scanner_bitrot_status,
};
pub use usage_store::store_data_usage_in_backend;
#[cfg(test)]
crates/scanner/src/scanner/heal_info.rs
+193
View File
@@ -13,6 +13,8 @@
// limitations under the License.
/// The background-heal info object persisted between scanner cycles.
use super::*;
use std::sync::LazyLock as StdLazyLock;
use std::sync::RwLock as StdRwLock;
/// Background healing information
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
@@ -136,6 +138,10 @@ pub(super) async fn save_background_heal_info_for_epoch<S>(
) where
S: ScannerStorage,
{
// Remember the record before any durability check: a skipped or failed
// save must not erase the cadence start this process decided on.
remember_background_heal_info(&info);
// Serialize to JSON
let data = match serde_json::to_vec(&info) {
Ok(data) => data,
@@ -186,3 +192,190 @@ pub(super) async fn save_background_heal_info_for_epoch<S>(
);
}
}
/// Where the bitrot cadence start used by a cycle came from.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ScannerBitrotStartSource {
/// No cycle has recorded a start yet.
None,
/// This process recorded the start; storage has not returned it yet.
Memory,
/// The start was read back from the persisted background-heal object.
Persisted,
}
impl ScannerBitrotStartSource {
pub fn as_str(self) -> &'static str {
match self {
Self::None => "none",
Self::Memory => "memory",
Self::Persisted => "persisted",
}
}
}
/// The bitrot cadence reference a cycle decides its scan mode from.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(super) enum ScannerBitrotStart {
/// Nothing recorded and no anchor; a cycle never decides from this.
None,
/// A Deep selection window opened at `cycle` / `started`, from the
/// persisted or the remembered record.
Window { cycle: u64, started: DateTime<Utc> },
/// No Deep window is recorded anywhere; the cadence counts from the first
/// cycle this process ran without one. Never persisted, so it opens no
/// selection window and old readers see no new state.
Anchor(DateTime<Utc>),
}
impl ScannerBitrotStart {
pub(super) fn kind(self) -> &'static str {
match self {
Self::None => "none",
Self::Window { .. } => "deep_window",
Self::Anchor(_) => "anchor",
}
}
fn cycle(self) -> u64 {
match self {
Self::Window { cycle, .. } => cycle,
Self::None | Self::Anchor(_) => 0,
}
}
fn started(self) -> Option<DateTime<Utc>> {
match self {
Self::None => None,
Self::Window { started, .. } => Some(started),
Self::Anchor(anchored) => Some(anchored),
}
}
}
/// How the scanner decided the scan mode of the running or latest cycle.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
pub struct ScannerBitrotStatus {
/// `false` on an erasure single-disk deployment, where scanner object
/// heal sampling is unsupported and Deep is never selected.
pub deep_scan_supported: bool,
pub bitrot_start_source: ScannerBitrotStartSource,
/// `deep_window` when a recorded Deep window start is in effect,
/// `anchor` when only this process's in-memory cadence anchor exists.
pub bitrot_start_kind: &'static str,
pub bitrot_start_cycle: u64,
pub bitrot_start_unix_secs: u64,
pub current_scan_mode: &'static str,
}
impl Default for ScannerBitrotStatus {
fn default() -> Self {
Self {
deep_scan_supported: false,
bitrot_start_source: ScannerBitrotStartSource::None,
bitrot_start_kind: ScannerBitrotStart::None.kind(),
bitrot_start_cycle: 0,
bitrot_start_unix_secs: 0,
current_scan_mode: HealScanMode::Unknown.as_str(),
}
}
}
/// Process-lifetime copy of the last background-heal record this node
/// decided on, so a skipped or lost save does not erase the cadence start.
static REMEMBERED_BACKGROUND_HEAL_INFO: StdRwLock<Option<BackgroundHealInfo>> = StdRwLock::new(None);
static SCANNER_BITROT_STATUS: StdLazyLock<StdRwLock<ScannerBitrotStatus>> =
StdLazyLock::new(|| StdRwLock::new(ScannerBitrotStatus::default()));
/// When this process first ran a cycle without any recorded Deep window.
static SCANNER_BITROT_ANCHOR: StdRwLock<Option<DateTime<Utc>>> = StdRwLock::new(None);
fn bitrot_cadence_anchor(now: DateTime<Utc>) -> DateTime<Utc> {
let mut anchor = SCANNER_BITROT_ANCHOR.write().unwrap_or_else(|poisoned| poisoned.into_inner());
*anchor.get_or_insert(now)
}
#[cfg(test)]
pub(super) fn reset_bitrot_cadence_anchor_for_test() {
*SCANNER_BITROT_ANCHOR.write().unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
}
/// The cadence reference for a cycle: the recorded Deep window start when the
/// record carries one, otherwise this process's in-memory anchor.
pub(super) fn scanner_bitrot_start_for_cycle(info: &BackgroundHealInfo, now: DateTime<Utc>) -> ScannerBitrotStart {
match info.bitrot_start_time {
Some(started) => ScannerBitrotStart::Window {
cycle: info.bitrot_start_cycle,
started,
},
None => ScannerBitrotStart::Anchor(bitrot_cadence_anchor(now)),
}
}
pub(super) fn remember_background_heal_info(info: &BackgroundHealInfo) {
*REMEMBERED_BACKGROUND_HEAL_INFO
.write()
.unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(info.clone());
}
fn remembered_background_heal_info() -> Option<BackgroundHealInfo> {
REMEMBERED_BACKGROUND_HEAL_INFO
.read()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.clone()
}
/// Pick the record whose cadence start is newest. Ties go to storage so a
/// start that was persisted successfully reports as `persisted`; a record
/// without a start cannot win over one that has it.
pub(super) fn resolve_background_heal_info(
persisted: Option<&BackgroundHealInfo>,
remembered: Option<&BackgroundHealInfo>,
) -> (BackgroundHealInfo, ScannerBitrotStartSource) {
let persisted_start = persisted.filter(|info| info.bitrot_start_time.is_some());
let remembered_start = remembered.filter(|info| info.bitrot_start_time.is_some());
match (persisted_start, remembered_start) {
(Some(stored), Some(memory)) if memory.bitrot_start_time > stored.bitrot_start_time => {
(memory.clone(), ScannerBitrotStartSource::Memory)
}
(Some(stored), _) => (stored.clone(), ScannerBitrotStartSource::Persisted),
(None, Some(memory)) => (memory.clone(), ScannerBitrotStartSource::Memory),
(None, None) => (persisted.cloned().unwrap_or_default(), ScannerBitrotStartSource::None),
}
}
/// Resolve the record a cycle decides from: the persisted read when it was
/// loaded, otherwise what this process remembered.
pub(super) fn resolve_background_heal_info_for_cycle(
read: &BackgroundHealInfoRead,
) -> (BackgroundHealInfo, ScannerBitrotStartSource) {
let persisted = (read.status == BackgroundHealInfoReadStatus::Loaded).then_some(&read.info);
resolve_background_heal_info(persisted, remembered_background_heal_info().as_ref())
}
pub(super) fn record_scanner_bitrot_status(
start: ScannerBitrotStart,
source: ScannerBitrotStartSource,
deep_scan_supported: bool,
scan_mode: HealScanMode,
) {
let status = ScannerBitrotStatus {
deep_scan_supported,
bitrot_start_source: source,
bitrot_start_kind: start.kind(),
bitrot_start_cycle: start.cycle(),
bitrot_start_unix_secs: start
.started()
.and_then(|started| u64::try_from(started.timestamp()).ok())
.unwrap_or(0),
current_scan_mode: scan_mode.as_str(),
};
*SCANNER_BITROT_STATUS.write().unwrap_or_else(|poisoned| poisoned.into_inner()) = status;
}
/// Deep cadence state of the running or latest cycle on this node.
pub fn scanner_bitrot_status() -> ScannerBitrotStatus {
*SCANNER_BITROT_STATUS.read().unwrap_or_else(|poisoned| poisoned.into_inner())
}
crates/scanner/src/scanner/tests.rs
+246 -10
View File
@@ -12,7 +12,10 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::heal_info::{classify_background_heal_read_error, decode_background_heal_info};
use super::heal_info::{
ScannerBitrotStart, classify_background_heal_read_error, decode_background_heal_info, record_scanner_bitrot_status,
reset_bitrot_cadence_anchor_for_test, resolve_background_heal_info, scanner_bitrot_start_for_cycle,
};
use super::*;
use crate::EcstoreResult;
use crate::storage_api::owner::{WriteCommitGuard, WriteCompletion, ecstore_hold_namespace_commit};
@@ -1681,10 +1684,13 @@ async fn assert_bootstrap_observation_retry(drives_per_set: usize) {
)
.await
.expect("the same cycle must converge after the pending PUT drains");
// Without a recorded Deep window both layouts run Normal (an erasure set
// used to open a Deep window here and revisit every bucket), so the
// retry reuses the cold inventory and only walks the dirty bucket.
assert_eq!(
retry_budget.progress().0,
if drives_per_set == 1 { 1 } else { 2 },
"Normal retries reuse cold inventory; Deep retries still visit every bucket"
1,
"Normal retries reuse cold inventory (drives_per_set={drives_per_set})"
);
assert!(matches!(
outcome,
@@ -10926,7 +10932,11 @@ async fn scanner_activity_probe_wait_stops_after_leader_lock_loss() {
#[serial]
fn test_get_cycle_scan_mode_runs_deep_until_selection_window_completes() {
with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("3600"), || {
let mode = get_cycle_scan_mode(10, 0, Some(Utc::now()), ScannerBitrotPolicy::new(bitrot_scan_cycle(), true, 1024));
let start = ScannerBitrotStart::Window {
cycle: 0,
started: Utc::now(),
};
let mode = get_cycle_scan_mode(10, start, ScannerBitrotPolicy::new(bitrot_scan_cycle(), true, 1024));
assert_eq!(mode, HealScanMode::Deep);
});
}
@@ -10938,12 +10948,17 @@ fn test_get_cycle_scan_mode_respects_elapsed_bitrot_cycle() {
let recent = Utc::now() - chrono::Duration::minutes(30);
let old = Utc::now() - chrono::Duration::hours(2);
let recent = ScannerBitrotStart::Window {
cycle: 0,
started: recent,
};
let old = ScannerBitrotStart::Window { cycle: 0, started: old };
assert_eq!(
get_cycle_scan_mode(2048, 0, Some(recent), ScannerBitrotPolicy::new(bitrot_scan_cycle(), true, 1024)),
get_cycle_scan_mode(2048, recent, ScannerBitrotPolicy::new(bitrot_scan_cycle(), true, 1024)),
HealScanMode::Normal
);
assert_eq!(
get_cycle_scan_mode(2048, 0, Some(old), ScannerBitrotPolicy::new(bitrot_scan_cycle(), true, 1024)),
get_cycle_scan_mode(2048, old, ScannerBitrotPolicy::new(bitrot_scan_cycle(), true, 1024)),
HealScanMode::Deep
);
});
@@ -10954,22 +10969,243 @@ fn test_get_cycle_scan_mode_respects_elapsed_bitrot_cycle() {
fn test_get_cycle_scan_mode_can_disable_periodic_deep_scan() {
with_var(ENV_SCANNER_BITROT_CYCLE_SECS, Some("off"), || {
assert_eq!(
get_cycle_scan_mode(1, 0, None, ScannerBitrotPolicy::new(bitrot_scan_cycle(), true, 1)),
get_cycle_scan_mode(1, ScannerBitrotStart::None, ScannerBitrotPolicy::new(bitrot_scan_cycle(), true, 1)),
HealScanMode::Normal
);
assert_eq!(
get_cycle_scan_mode(
1,
ScannerBitrotStart::Anchor(Utc::now() - chrono::Duration::days(400)),
ScannerBitrotPolicy::new(bitrot_scan_cycle(), true, 1)
),
HealScanMode::Normal
);
});
}
#[test]
fn scan_mode_without_recorded_bitrot_start_stays_normal_until_the_cycle_elapses() {
// Regression for rustfs/backlog#2905 (rustfs/rustfs#8015): a cycle that
// finds no recorded bitrot start (missing heal object, or a save that
// keeps failing) must not open a Deep window. Three consecutive cycles
// with the default `BackgroundHealInfo` used to run Deep through the
// selection window because `bitrot_start_cycle` defaults to 0.
let policy = ScannerBitrotPolicy::new(Some(Duration::from_secs(3600)), true, 1024);
let modes: Vec<_> = (1..=3)
.map(|cycle| get_cycle_scan_mode(cycle, ScannerBitrotStart::None, policy))
.collect();
assert_eq!(modes, vec![HealScanMode::Normal; 3]);
// An in-memory anchor counts the cadence without opening a selection
// window: Normal until the cycle elapses, Deep afterwards.
let anchored = ScannerBitrotStart::Anchor(Utc::now());
let modes: Vec<_> = (1..=3).map(|cycle| get_cycle_scan_mode(cycle, anchored, policy)).collect();
assert_eq!(modes, vec![HealScanMode::Normal; 3]);
assert_eq!(
get_cycle_scan_mode(4, ScannerBitrotStart::Anchor(Utc::now() - chrono::Duration::hours(2)), policy),
HealScanMode::Deep
);
// A recorded Deep window keeps its selection window, and an elapsed
// cadence still triggers the next window.
let started = Utc::now();
let window = ScannerBitrotStart::Window { cycle: 1, started };
assert_eq!(get_cycle_scan_mode(1, window, policy), HealScanMode::Deep);
assert_eq!(
get_cycle_scan_mode(
2048,
ScannerBitrotStart::Window {
cycle: 1,
started: started - chrono::Duration::hours(2),
},
policy
),
HealScanMode::Deep
);
assert_eq!(get_cycle_scan_mode(2048, window, policy), HealScanMode::Normal);
}
#[test]
#[serial]
fn bitrot_cadence_anchor_is_remembered_in_memory_without_a_persisted_write() {
reset_bitrot_cadence_anchor_for_test();
let now = Utc::now();
let policy = ScannerBitrotPolicy::new(Some(Duration::from_secs(3600)), true, 1024);
// A Normal cycle without a recorded window writes nothing to storage ...
assert!(
background_heal_info_for_scan_start(BackgroundHealInfo::default(), 3, HealScanMode::Normal, now, policy).is_none(),
"a Normal cycle must not persist a start that would open a Deep window"
);
// ... but the process anchors the cadence at its first such cycle and
// keeps that anchor for later cycles.
let first = scanner_bitrot_start_for_cycle(&BackgroundHealInfo::default(), now);
assert_eq!(first, ScannerBitrotStart::Anchor(now));
let later = scanner_bitrot_start_for_cycle(&BackgroundHealInfo::default(), now + chrono::Duration::hours(2));
assert_eq!(later, ScannerBitrotStart::Anchor(now));
assert_eq!(get_cycle_scan_mode(4, later, policy), HealScanMode::Normal);
// A recorded window start takes precedence over the anchor.
let recorded = BackgroundHealInfo {
bitrot_start_time: Some(now),
bitrot_start_cycle: 7,
current_scan_mode: HealScanMode::Normal,
};
assert_eq!(
scanner_bitrot_start_for_cycle(&recorded, now),
ScannerBitrotStart::Window { cycle: 7, started: now }
);
// Once the cadence elapses on the anchor, the Deep cycle records the
// window start for persistence (and for the in-memory record).
let opened = background_heal_info_for_scan_start(BackgroundHealInfo::default(), 9, HealScanMode::Deep, now, policy)
.expect("opening a Deep window records its start");
assert_eq!(opened.bitrot_start_cycle, 9);
assert_eq!(opened.bitrot_start_time, Some(now));
assert_eq!(opened.current_scan_mode, HealScanMode::Deep);
reset_bitrot_cadence_anchor_for_test();
}
#[test]
#[serial]
fn bitrot_start_memory_record_outlives_a_skipped_persisted_save() {
reset_bitrot_cadence_anchor_for_test();
let now = Utc::now();
let policy = ScannerBitrotPolicy::new(Some(Duration::from_secs(3600)), true, 1024);
let recorded = BackgroundHealInfo {
bitrot_start_time: Some(now),
bitrot_start_cycle: 4,
current_scan_mode: HealScanMode::Normal,
};
// Storage never returned a start (missing object, or a save skipped
// without movement admission): the in-memory record keeps the cadence.
let (info, source) = resolve_background_heal_info(None, Some(&recorded));
assert_eq!(source, ScannerBitrotStartSource::Memory);
assert_eq!(info.bitrot_start_cycle, 4);
let start = scanner_bitrot_start_for_cycle(&info, now);
assert_eq!(start, ScannerBitrotStart::Window { cycle: 4, started: now });
assert_eq!(
get_cycle_scan_mode(5, start, policy),
HealScanMode::Deep,
"a remembered start keeps the Deep selection window it opened"
);
assert_eq!(
get_cycle_scan_mode(2048, start, policy),
HealScanMode::Normal,
"and stays Normal after the window until the cadence elapses"
);
// A persisted start wins ties and older in-memory records.
let stored = BackgroundHealInfo {
bitrot_start_time: Some(now),
bitrot_start_cycle: 4,
current_scan_mode: HealScanMode::Deep,
};
let (info, source) = resolve_background_heal_info(Some(&stored), Some(&recorded));
assert_eq!(source, ScannerBitrotStartSource::Persisted);
assert_eq!(info.current_scan_mode, HealScanMode::Deep);
// A newer in-memory start (a Deep window opened after the last
// successful save) wins over the stale persisted record.
let newer = BackgroundHealInfo {
bitrot_start_time: Some(now + chrono::Duration::seconds(1)),
bitrot_start_cycle: 9,
current_scan_mode: HealScanMode::Deep,
};
let (info, source) = resolve_background_heal_info(Some(&stored), Some(&newer));
assert_eq!(source, ScannerBitrotStartSource::Memory);
assert_eq!(info.bitrot_start_cycle, 9);
// Nothing recorded anywhere: the persisted record is used as-is, the
// source says so, and the cycle runs Normal instead of opening a window.
let legacy = BackgroundHealInfo {
bitrot_start_time: None,
bitrot_start_cycle: 0,
current_scan_mode: HealScanMode::Deep,
};
let (info, source) = resolve_background_heal_info(Some(&legacy), None);
assert_eq!(source, ScannerBitrotStartSource::None);
assert_eq!(info.current_scan_mode, HealScanMode::Deep);
let start = scanner_bitrot_start_for_cycle(&info, now);
assert_eq!(start, ScannerBitrotStart::Anchor(now));
assert_eq!(get_cycle_scan_mode(1, start, policy), HealScanMode::Normal);
reset_bitrot_cadence_anchor_for_test();
}
#[test]
#[serial]
fn scan_mode_status_reports_bitrot_start_source_and_topology_support() {
let now = Utc::now();
let info = BackgroundHealInfo {
bitrot_start_time: Some(now),
bitrot_start_cycle: 2,
current_scan_mode: HealScanMode::Normal,
};
record_scanner_bitrot_status(
scanner_bitrot_start_for_cycle(&info, now),
ScannerBitrotStartSource::Memory,
false,
HealScanMode::Normal,
);
let status = scanner_bitrot_status();
assert!(!status.deep_scan_supported);
assert_eq!(status.bitrot_start_source, ScannerBitrotStartSource::Memory);
assert_eq!(status.bitrot_start_source.as_str(), "memory");
assert_eq!(status.bitrot_start_kind, "deep_window");
assert_eq!(status.bitrot_start_cycle, 2);
assert_eq!(
status.bitrot_start_unix_secs,
u64::try_from(now.timestamp()).expect("current time is after the epoch")
);
assert_eq!(status.current_scan_mode, "normal");
let encoded = serde_json::to_value(status).expect("bitrot status serializes");
assert_eq!(encoded["bitrot_start_source"], "memory");
assert_eq!(encoded["deep_scan_supported"], false);
record_scanner_bitrot_status(ScannerBitrotStart::None, ScannerBitrotStartSource::None, true, HealScanMode::Deep);
let status = scanner_bitrot_status();
assert!(status.deep_scan_supported);
assert_eq!(status.bitrot_start_source, ScannerBitrotStartSource::None);
assert_eq!(status.bitrot_start_kind, "none");
assert_eq!(status.bitrot_start_unix_secs, 0);
assert_eq!(status.current_scan_mode, "deep");
record_scanner_bitrot_status(
ScannerBitrotStart::Anchor(now),
ScannerBitrotStartSource::Memory,
true,
HealScanMode::Normal,
);
let status = scanner_bitrot_status();
assert_eq!(status.bitrot_start_kind, "anchor");
assert_eq!(status.bitrot_start_cycle, 0);
assert_eq!(
status.bitrot_start_unix_secs,
u64::try_from(now.timestamp()).expect("current time is after the epoch")
);
}
#[test]
fn test_erasure_sd_does_not_enter_deep_scan_mode() {
let started = Utc::now();
let cycle = Some(Duration::from_secs(3600));
let unsupported_policy = ScannerBitrotPolicy::new(cycle, false, 1024);
assert_eq!(get_cycle_scan_mode(10, 10, Some(started), unsupported_policy), HealScanMode::Normal);
assert_eq!(get_cycle_scan_mode(11, 10, None, unsupported_policy), HealScanMode::Normal);
let window = ScannerBitrotStart::Window { cycle: 10, started };
assert_eq!(get_cycle_scan_mode(10, window, unsupported_policy), HealScanMode::Normal);
assert_eq!(
get_cycle_scan_mode(10, 10, None, ScannerBitrotPolicy::new(Some(Duration::ZERO), false, 1024)),
get_cycle_scan_mode(11, ScannerBitrotStart::None, unsupported_policy),
HealScanMode::Normal
);
assert_eq!(
get_cycle_scan_mode(11, ScannerBitrotStart::Anchor(started), unsupported_policy),
HealScanMode::Normal
);
assert_eq!(
get_cycle_scan_mode(10, ScannerBitrotStart::None, ScannerBitrotPolicy::new(Some(Duration::ZERO), false, 1024)),
HealScanMode::Normal
);
crates/scanner/src/scanner_folder.rs
+13
View File
@@ -1978,6 +1978,19 @@ impl FolderScanner {
continue;
}
// Charge the disk's background I/O budget before the metadata
// read. The `xl.meta` size is only measured when bytes are
// budgeted, so an uncapped byte axis costs no extra stat.
let budgeted_bytes = if self.sleeper.budgets_disk_io_bytes() {
tokio::fs::metadata(&item.path)
.await
.map(|metadata| metadata.len())
.unwrap_or(0)
} else {
0
};
self.sleeper.admit_disk_io(1, budgeted_bytes).await;
let sz = match self
.local_disk
.get_size_with_tier_names(item.clone(), &self.tier_registry.names)
crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs
+3
View File
@@ -31,6 +31,9 @@ const SOURCE: DataUsageCacheSource = DataUsageCacheSource::new(0, 0);
const PLAN: DataUsageScanPlanDigest = DataUsageScanPlanDigest([17; 32]);
fn scan_options(scan_mode: HealScanMode) -> ScannerDiskScanOptions {
// These fixtures walk thousands of synthetic objects through the real
// local-disk path; the per-disk I/O budget would only slow them down.
crate::SCANNER_SLEEPER.disable_disk_io_budget_for_tests();
ScannerDiskScanOptions {
scan_mode,
prefix_scan_scope: None,
crates/scanner/src/scanner_io/io_disk.rs
+2 -1
View File
@@ -230,7 +230,8 @@ impl ScannerIODisk for Disk {
cache,
updates,
scan_mode,
SCANNER_SLEEPER.clone(),
// Every walker of this local disk shares its background I/O budget.
SCANNER_SLEEPER.for_disk(&self.path()),
prefix_scan_scope,
checkpoint_tx,
),
crates/scanner/src/sleeper.rs
+147 -3
View File
@@ -12,17 +12,25 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU8, Ordering};
use std::sync::{Arc, LazyLock, RwLock};
use std::time::Instant;
use rustfs_config::{
DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, ENV_SCANNER_IDLE_MODE, ENV_SCANNER_SPEED,
ENV_SCANNER_YIELD_EVERY_N_OBJECTS, ScannerSpeed,
DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC,
ENV_SCANNER_DISK_IO_BUDGET_IOPS, ENV_SCANNER_IDLE_MODE, ENV_SCANNER_SPEED, ENV_SCANNER_YIELD_EVERY_N_OBJECTS, ScannerSpeed,
};
use rustfs_scanner_metrics::metrics::global_metrics;
use tokio::time::Duration;
mod io_budget;
pub use io_budget::{
SCANNER_DISK_IO_BUDGET_MAX_WAIT, ScannerDiskIoBudget, ScannerDiskIoBudgetConfig, ScannerDiskIoBudgetLimits,
ScannerDiskIoBudgetStatus, ScannerDiskMedia, ScannerIoBudgetLimit, resolve_disk_io_budget, scanner_disk_io_budget_status,
};
const MIN_SLEEP: Duration = Duration::from_millis(1);
const SCANNER_SPEED_FASTEST: u8 = 0;
const SCANNER_SPEED_FAST: u8 = 1;
@@ -74,6 +82,19 @@ fn scanner_env_config() -> (ScannerSpeed, bool) {
(speed, idle_mode)
}
fn disk_io_budget_env_config(speed: ScannerSpeed) -> ScannerDiskIoBudgetConfig {
let limit = |key: &str| {
rustfs_utils::get_env_opt_str(key)
.and_then(|value| ScannerIoBudgetLimit::parse(&value))
.unwrap_or(ScannerIoBudgetLimit::Auto)
};
ScannerDiskIoBudgetConfig {
iops: limit(ENV_SCANNER_DISK_IO_BUDGET_IOPS),
bytes_per_sec: limit(ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC),
speed,
}
}
pub(crate) fn scanner_yield_every_n_objects() -> u64 {
rustfs_utils::get_env_u64(ENV_SCANNER_YIELD_EVERY_N_OBJECTS, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS)
}
@@ -102,6 +123,7 @@ pub static SCANNER_SLEEPER: LazyLock<DynamicSleeper> = LazyLock::new(|| {
let sleeper = DynamicSleeper::new(speed);
sleeper.record_throttle_config();
sleeper.update_disk_io_budget(disk_io_budget_env_config(speed));
sleeper
});
@@ -111,9 +133,12 @@ pub static SCANNER_SLEEPER: LazyLock<DynamicSleeper> = LazyLock::new(|| {
/// `elapsed_work_time * factor`, clamped to `[MIN_SLEEP, max_sleep]`.
/// For folder-level gaps, [`sleep_folder`] uses `MIN_SLEEP * factor`,
/// clamped only by `max_sleep`. All sleeps are gated on [`SCANNER_IDLE_MODE`];
/// the per-disk background I/O budget bound with [`DynamicSleeper::for_disk`]
/// is a hard ceiling that applies regardless of idle mode or preset.
#[derive(Clone)]
pub struct DynamicSleeper {
inner: Arc<SleeperParams>,
disk_io_budget: Option<Arc<ScannerDiskIoBudget>>,
}
struct SleeperParams {
@@ -128,9 +153,52 @@ impl DynamicSleeper {
factor: RwLock::new(speed.sleep_factor()),
max_sleep: RwLock::new(speed.max_sleep()),
}),
disk_io_budget: None,
}
}
/// Bind this sleeper to the background I/O budget of one local disk.
/// Walkers that share a disk path share its buckets.
pub fn for_disk(&self, disk_path: &Path) -> Self {
Self {
inner: self.inner.clone(),
disk_io_budget: Some(io_budget::disk_io_budget_for(disk_path)),
}
}
/// Whether byte charges can change pacing on the bound disk.
pub fn budgets_disk_io_bytes(&self) -> bool {
self.disk_io_budget.as_ref().is_some_and(|budget| budget.budgets_bytes())
}
/// Charge the bound disk's background I/O budget and wait it out.
///
/// This is a hard ceiling layered on the local-disk foreground-priority
/// admission gate: it runs regardless of idle mode or speed preset, and
/// it only ever delays, so it can never admit more than the gate allows.
pub async fn admit_disk_io(&self, ops: u64, bytes: u64) {
if let Some(budget) = self.disk_io_budget.as_ref() {
budget.admit(ops, bytes).await;
}
}
pub(crate) fn update_disk_io_budget(&self, config: ScannerDiskIoBudgetConfig) {
io_budget::update_disk_io_budget_config(config);
}
/// Remove the per-disk budget for fixtures that walk large synthetic
/// namespaces through the local-disk path and test checkpoint semantics,
/// not pacing. nextest runs each test in its own process, so this does
/// not leak into other tests.
#[cfg(test)]
pub(crate) fn disable_disk_io_budget_for_tests(&self) {
self.update_disk_io_budget(ScannerDiskIoBudgetConfig {
iops: ScannerIoBudgetLimit::Unlimited,
bytes_per_sec: ScannerIoBudgetLimit::Unlimited,
speed: ScannerSpeed::Default,
});
}
fn should_sleep() -> bool {
SCANNER_IDLE_MODE.load(Ordering::Relaxed)
}
@@ -141,7 +209,14 @@ impl DynamicSleeper {
(factor, max_sleep)
}
/// Pace one directory listing: the preset folder gap, then one
/// operation charged to the bound disk's background I/O budget.
pub async fn sleep_folder(&self) {
self.sleep_folder_preset().await;
self.admit_disk_io(1, 0).await;
}
async fn sleep_folder_preset(&self) {
if !Self::should_sleep() {
return;
}
@@ -187,10 +262,12 @@ impl DynamicSleeper {
self.update_params(speed.sleep_factor(), speed.max_sleep());
}
/// Reload speed and idle-mode settings from the current environment.
/// Reload speed, idle-mode, and disk I/O budget settings from the
/// current environment.
pub fn refresh_from_env(&self) {
let (speed, idle_mode) = scanner_env_config();
self.update_from_runtime_config(speed.sleep_factor(), speed.max_sleep(), idle_mode, scanner_yield_every_n_objects());
self.update_disk_io_budget(disk_io_budget_env_config(speed));
}
pub(crate) fn update_from_runtime_config(
@@ -377,6 +454,73 @@ mod tests {
SCANNER_IDLE_MODE.store(prev_mode, Ordering::Relaxed);
}
#[tokio::test(start_paused = true)]
#[serial]
async fn sleeper_without_a_disk_binding_never_charges_a_budget() {
let s = DynamicSleeper::new(ScannerSpeed::Fastest);
s.update_disk_io_budget(ScannerDiskIoBudgetConfig {
iops: ScannerIoBudgetLimit::Explicit(0),
bytes_per_sec: ScannerIoBudgetLimit::Explicit(0),
speed: ScannerSpeed::Fastest,
});
let start = tokio::time::Instant::now();
for _ in 0..10 {
s.sleep_folder().await;
s.admit_disk_io(1, 1 << 20).await;
}
assert_eq!(start.elapsed(), Duration::ZERO);
assert!(!s.budgets_disk_io_bytes());
s.update_disk_io_budget(ScannerDiskIoBudgetConfig::DEFAULT);
}
#[tokio::test(start_paused = true)]
#[serial]
async fn sleeper_bound_to_a_disk_charges_folder_and_object_io_as_a_hard_ceiling() {
let prev_mode = SCANNER_IDLE_MODE.load(Ordering::Relaxed);
// Sleeps are disabled and the preset is `fastest`: only the explicit
// budget can pace this walk.
SCANNER_IDLE_MODE.store(false, Ordering::Relaxed);
let s = DynamicSleeper::new(ScannerSpeed::Fastest);
s.update_disk_io_budget(ScannerDiskIoBudgetConfig {
iops: ScannerIoBudgetLimit::Explicit(2),
bytes_per_sec: ScannerIoBudgetLimit::Explicit(1_000),
speed: ScannerSpeed::Fastest,
});
let disk = std::path::PathBuf::from(format!("/rustfs-scanner-sleeper-test-{}", std::process::id()));
let bound = s.for_disk(&disk);
assert!(bound.budgets_disk_io_bytes());
// Two listings fit the burst; the third owes one operation at 2/s.
let start = tokio::time::Instant::now();
bound.sleep_folder().await;
bound.sleep_folder().await;
assert_eq!(start.elapsed(), Duration::ZERO);
bound.sleep_folder().await;
assert_eq!(start.elapsed(), Duration::from_millis(500));
// Bytes pace independently: 1500 bytes at 1000 B/s owes 500 ms on
// top of the operation debt already repaid.
tokio::time::advance(Duration::from_secs(5)).await;
let start = tokio::time::Instant::now();
bound.admit_disk_io(1, 1_500).await;
assert_eq!(start.elapsed(), Duration::from_millis(500));
let status = scanner_disk_io_budget_status()
.into_iter()
.find(|entry| entry.disk == disk.to_string_lossy())
.expect("bound disk reports a budget status");
assert_eq!(status.iops, Some(2));
assert_eq!(status.bytes_per_sec, Some(1_000));
assert_eq!(status.wait_events, 2);
assert_eq!(status.charged_ops, 4);
assert_eq!(status.charged_bytes, 1_500);
s.update_disk_io_budget(ScannerDiskIoBudgetConfig::DEFAULT);
SCANNER_IDLE_MODE.store(prev_mode, Ordering::Relaxed);
}
#[tokio::test(start_paused = true)]
#[serial]
async fn test_idle_mode_off_skips_sleep() {
crates/scanner/src/sleeper/io_budget.rs
+714
View File
@@ -0,0 +1,714 @@
// 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.
//! Per-disk background I/O budget for the data scanner.
//!
//! Each local disk owns two token buckets, one for operations per second and
//! one for bytes per second. Every scanner directory listing and object
//! metadata read charges them before it runs; an exhausted bucket delays the
//! walk until the rate allows the charge again. The budget is a hard ceiling
//! layered on top of the local-disk foreground-priority admission gate:
//! admission decides *when* background I/O may start while foreground
//! requests are active, the budget bounds *how much* background I/O one disk
//! sees per second even while it is idle. A charge can never bypass the gate.
//!
//! A zero or tiny budget still admits one charge per
//! [`SCANNER_DISK_IO_BUDGET_MAX_WAIT`], so a walk keeps making progress
//! instead of starving.
use std::collections::BTreeMap;
use std::path::Path;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, LazyLock, Mutex, RwLock};
use rustfs_config::{
DEFAULT_SCANNER_DISK_IO_BUDGET_FLASH_BYTES_PER_SEC, DEFAULT_SCANNER_DISK_IO_BUDGET_FLASH_IOPS,
DEFAULT_SCANNER_DISK_IO_BUDGET_ROTATIONAL_BYTES_PER_SEC, DEFAULT_SCANNER_DISK_IO_BUDGET_ROTATIONAL_IOPS, ScannerSpeed,
};
use rustfs_scanner_metrics::metrics::global_metrics;
use serde::Serialize;
use tokio::time::{Duration, Instant};
use tracing::info;
const LOG_COMPONENT_SCANNER: &str = "scanner";
const LOG_SUBSYSTEM_RUNTIME: &str = "runtime";
const EVENT_SCANNER_DISK_IO_BUDGET: &str = "scanner_disk_io_budget";
const OTEL_SCANNER_DISK_IO_BUDGET_WAIT_SECONDS: &str = "rustfs_scanner_disk_io_budget_wait_seconds";
const OTEL_SCANNER_DISK_IO_BUDGET_WAITS_TOTAL: &str = "rustfs_scanner_disk_io_budget_waits_total";
/// Longest wait one budget charge may impose.
///
/// This is the minimum service guarantee: whatever the configured rate, a
/// walk is admitted at least once per window.
pub const SCANNER_DISK_IO_BUDGET_MAX_WAIT: Duration = Duration::from_secs(1);
/// Burst allowance expressed in seconds of the configured rate. One second
/// lets a short directory finish without pacing while a long walk converges
/// on the configured rate.
const BURST_WINDOW_SECS: f64 = 1.0;
const BUDGET_LIMIT_AUTO: &str = "auto";
const BUDGET_LIMIT_OFF: &str = "off";
/// Media class of a local disk as far as the scanner can tell.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ScannerDiskMedia {
Rotational,
Flash,
/// Media could not be detected; budgets fall back to the rotational
/// defaults because a spindle is the case that needs protection.
Unknown,
}
impl ScannerDiskMedia {
pub fn as_str(self) -> &'static str {
match self {
Self::Rotational => "rotational",
Self::Flash => "flash",
Self::Unknown => "unknown",
}
}
}
/// One configured budget axis: media default, no cap, or an explicit rate.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ScannerIoBudgetLimit {
Auto,
Unlimited,
Explicit(u64),
}
impl ScannerIoBudgetLimit {
/// Parse the `auto` | `off` | `<non-negative integer>` spelling shared by
/// the persisted keys and the environment variables.
pub fn parse(value: &str) -> Option<Self> {
match value.trim().to_ascii_lowercase().as_str() {
BUDGET_LIMIT_AUTO => Some(Self::Auto),
BUDGET_LIMIT_OFF | "disabled" | "unlimited" | "none" => Some(Self::Unlimited),
number => number.parse::<u64>().ok().map(Self::Explicit),
}
}
}
impl std::fmt::Display for ScannerIoBudgetLimit {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::Auto => f.write_str(BUDGET_LIMIT_AUTO),
Self::Unlimited => f.write_str(BUDGET_LIMIT_OFF),
Self::Explicit(value) => write!(f, "{value}"),
}
}
}
/// Configured budget inputs, resolved per disk once its media is known.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ScannerDiskIoBudgetConfig {
pub iops: ScannerIoBudgetLimit,
pub bytes_per_sec: ScannerIoBudgetLimit,
/// The `fastest` preset means "do not pace"; `auto` budgets resolve to
/// unlimited under it while explicit values still apply.
pub speed: ScannerSpeed,
}
impl ScannerDiskIoBudgetConfig {
pub const DEFAULT: Self = Self {
iops: ScannerIoBudgetLimit::Auto,
bytes_per_sec: ScannerIoBudgetLimit::Auto,
speed: ScannerSpeed::Default,
};
}
impl Default for ScannerDiskIoBudgetConfig {
fn default() -> Self {
Self::DEFAULT
}
}
/// Effective per-disk rates; `None` means that axis is not capped.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct ScannerDiskIoBudgetLimits {
pub iops: Option<u64>,
pub bytes_per_sec: Option<u64>,
}
fn resolve_budget_axis(limit: ScannerIoBudgetLimit, speed: ScannerSpeed, media_default: u64) -> Option<u64> {
match limit {
ScannerIoBudgetLimit::Explicit(value) => Some(value),
ScannerIoBudgetLimit::Unlimited => None,
ScannerIoBudgetLimit::Auto if speed == ScannerSpeed::Fastest => None,
ScannerIoBudgetLimit::Auto => Some(media_default),
}
}
/// Resolve the configured budget for one disk's media. Unknown media uses
/// the rotational defaults.
pub fn resolve_disk_io_budget(config: ScannerDiskIoBudgetConfig, media: ScannerDiskMedia) -> ScannerDiskIoBudgetLimits {
let (default_iops, default_bytes) = match media {
ScannerDiskMedia::Flash => (
DEFAULT_SCANNER_DISK_IO_BUDGET_FLASH_IOPS,
DEFAULT_SCANNER_DISK_IO_BUDGET_FLASH_BYTES_PER_SEC,
),
ScannerDiskMedia::Rotational | ScannerDiskMedia::Unknown => (
DEFAULT_SCANNER_DISK_IO_BUDGET_ROTATIONAL_IOPS,
DEFAULT_SCANNER_DISK_IO_BUDGET_ROTATIONAL_BYTES_PER_SEC,
),
};
ScannerDiskIoBudgetLimits {
iops: resolve_budget_axis(config.iops, config.speed, default_iops),
bytes_per_sec: resolve_budget_axis(config.bytes_per_sec, config.speed, default_bytes),
}
}
/// Rates, charges, and wait totals are counts; precision loss above 2^53
/// is irrelevant for a per-second budget.
fn count_as_f64(value: u64) -> f64 {
value as f64
}
#[derive(Debug)]
struct TokenBucket {
rate_per_sec: f64,
capacity: f64,
tokens: f64,
refilled_at: Instant,
}
impl TokenBucket {
fn new(rate_per_sec: u64, now: Instant) -> Self {
let rate_per_sec = count_as_f64(rate_per_sec);
let capacity = Self::capacity_for(rate_per_sec);
Self {
rate_per_sec,
capacity,
tokens: capacity,
refilled_at: now,
}
}
fn capacity_for(rate_per_sec: f64) -> f64 {
(rate_per_sec * BURST_WINDOW_SECS).max(1.0)
}
fn reconfigure(&mut self, rate_per_sec: u64, now: Instant) {
self.refill(now);
self.rate_per_sec = count_as_f64(rate_per_sec);
self.capacity = Self::capacity_for(self.rate_per_sec);
self.tokens = self.tokens.min(self.capacity);
}
fn refill(&mut self, now: Instant) {
let elapsed = now.saturating_duration_since(self.refilled_at).as_secs_f64();
self.tokens = (self.tokens + elapsed * self.rate_per_sec).min(self.capacity);
self.refilled_at = now;
}
/// Charge `amount` and return how long the caller must wait before the
/// bucket is non-negative again, capped at the max wait. Debt is bounded
/// to one burst window so an oversized charge costs at most the max wait
/// plus one window, never a permanent stall.
fn charge(&mut self, amount: f64, now: Instant) -> Duration {
self.refill(now);
self.tokens -= amount;
if self.tokens >= 0.0 {
return Duration::ZERO;
}
let max_wait_secs = SCANNER_DISK_IO_BUDGET_MAX_WAIT.as_secs_f64();
let wait_secs = if self.rate_per_sec > 0.0 {
(-self.tokens / self.rate_per_sec).min(max_wait_secs)
} else {
max_wait_secs
};
self.tokens = self.tokens.max(-self.capacity);
// `wait_secs` is finite and within `[0, max_wait_secs]`.
Duration::from_secs_f64(wait_secs)
}
}
#[derive(Debug)]
struct BudgetState {
limits: ScannerDiskIoBudgetLimits,
iops: Option<TokenBucket>,
bytes: Option<TokenBucket>,
}
impl BudgetState {
fn new(limits: ScannerDiskIoBudgetLimits, now: Instant) -> Self {
Self {
limits,
iops: limits.iops.map(|rate| TokenBucket::new(rate, now)),
bytes: limits.bytes_per_sec.map(|rate| TokenBucket::new(rate, now)),
}
}
fn reconfigure(&mut self, limits: ScannerDiskIoBudgetLimits, now: Instant) {
Self::reconfigure_axis(&mut self.iops, limits.iops, now);
Self::reconfigure_axis(&mut self.bytes, limits.bytes_per_sec, now);
self.limits = limits;
}
fn reconfigure_axis(bucket: &mut Option<TokenBucket>, rate: Option<u64>, now: Instant) {
match (bucket.as_mut(), rate) {
(Some(bucket), Some(rate)) => bucket.reconfigure(rate, now),
(None, Some(rate)) => *bucket = Some(TokenBucket::new(rate, now)),
(_, None) => *bucket = None,
}
}
}
/// The budget of one local disk, shared by every walker on that disk.
#[derive(Debug)]
pub struct ScannerDiskIoBudget {
disk: String,
media: ScannerDiskMedia,
state: Mutex<BudgetState>,
wait_events: AtomicU64,
wait_millis: AtomicU64,
charged_ops: AtomicU64,
charged_bytes: AtomicU64,
}
impl ScannerDiskIoBudget {
fn new(disk: String, media: ScannerDiskMedia, limits: ScannerDiskIoBudgetLimits, now: Instant) -> Self {
Self {
disk,
media,
state: Mutex::new(BudgetState::new(limits, now)),
wait_events: AtomicU64::new(0),
wait_millis: AtomicU64::new(0),
charged_ops: AtomicU64::new(0),
charged_bytes: AtomicU64::new(0),
}
}
pub fn media(&self) -> ScannerDiskMedia {
self.media
}
pub fn limits(&self) -> ScannerDiskIoBudgetLimits {
self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner()).limits
}
/// Whether byte charges can change pacing; callers skip measuring sizes
/// when they cannot.
pub fn budgets_bytes(&self) -> bool {
self.limits().bytes_per_sec.is_some()
}
fn reconfigure(&self, limits: ScannerDiskIoBudgetLimits) {
let now = Instant::now();
self.state
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.reconfigure(limits, now);
}
/// Charge `ops` operations and `bytes` bytes, returning the wait the
/// budget imposes before the caller may proceed.
pub fn charge(&self, ops: u64, bytes: u64) -> Duration {
self.charged_ops.fetch_add(ops, Ordering::Relaxed);
self.charged_bytes.fetch_add(bytes, Ordering::Relaxed);
let now = Instant::now();
let mut state = self.state.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
let mut wait = Duration::ZERO;
if ops > 0
&& let Some(bucket) = state.iops.as_mut()
{
wait = wait.max(bucket.charge(count_as_f64(ops), now));
}
if bytes > 0
&& let Some(bucket) = state.bytes.as_mut()
{
wait = wait.max(bucket.charge(count_as_f64(bytes), now));
}
wait.min(SCANNER_DISK_IO_BUDGET_MAX_WAIT)
}
/// Charge and wait out the budget. Runs regardless of idle mode or
/// speed preset: the budget is a ceiling, not a preset sleep.
pub async fn admit(&self, ops: u64, bytes: u64) {
let wait = self.charge(ops, bytes);
if wait.is_zero() {
return;
}
tokio::time::sleep(wait).await;
self.record_wait(wait);
}
fn record_wait(&self, wait: Duration) {
self.wait_events.fetch_add(1, Ordering::Relaxed);
self.wait_millis
.fetch_add(u64::try_from(wait.as_millis()).unwrap_or(u64::MAX), Ordering::Relaxed);
metrics::histogram!(OTEL_SCANNER_DISK_IO_BUDGET_WAIT_SECONDS, "disk" => self.disk.clone()).record(wait.as_secs_f64());
metrics::counter!(OTEL_SCANNER_DISK_IO_BUDGET_WAITS_TOTAL, "disk" => self.disk.clone()).increment(1);
global_metrics().record_scanner_disk_io_budget_wait(wait);
}
pub fn status(&self) -> ScannerDiskIoBudgetStatus {
let limits = self.limits();
ScannerDiskIoBudgetStatus {
disk: self.disk.clone(),
media: self.media,
iops: limits.iops,
bytes_per_sec: limits.bytes_per_sec,
charged_ops: self.charged_ops.load(Ordering::Relaxed),
charged_bytes: self.charged_bytes.load(Ordering::Relaxed),
wait_events: self.wait_events.load(Ordering::Relaxed),
wait_seconds: count_as_f64(self.wait_millis.load(Ordering::Relaxed)) / 1000.0,
}
}
}
/// Status of one disk budget as reported by the scanner status endpoint.
#[derive(Clone, Debug, PartialEq, Serialize)]
pub struct ScannerDiskIoBudgetStatus {
pub disk: String,
pub media: ScannerDiskMedia,
/// `null` when that axis is uncapped.
pub iops: Option<u64>,
pub bytes_per_sec: Option<u64>,
pub charged_ops: u64,
pub charged_bytes: u64,
pub wait_events: u64,
pub wait_seconds: f64,
}
static DISK_IO_BUDGET_CONFIG: RwLock<ScannerDiskIoBudgetConfig> = RwLock::new(ScannerDiskIoBudgetConfig::DEFAULT);
static DISK_IO_BUDGETS: LazyLock<Mutex<BTreeMap<String, Arc<ScannerDiskIoBudget>>>> =
LazyLock::new(|| Mutex::new(BTreeMap::new()));
pub(crate) fn current_disk_io_budget_config() -> ScannerDiskIoBudgetConfig {
*DISK_IO_BUDGET_CONFIG.read().unwrap_or_else(|poisoned| poisoned.into_inner())
}
/// Look up or create the budget for one local disk path. Walkers that share
/// a disk share its buckets; the media is detected once per path.
pub(crate) fn disk_io_budget_for(disk_path: &Path) -> Arc<ScannerDiskIoBudget> {
let key = disk_path.to_string_lossy().into_owned();
let mut registry = DISK_IO_BUDGETS.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
if let Some(budget) = registry.get(&key) {
return budget.clone();
}
let media = detect_disk_media(disk_path);
let config = current_disk_io_budget_config();
let limits = resolve_disk_io_budget(config, media);
let budget = Arc::new(ScannerDiskIoBudget::new(key.clone(), media, limits, Instant::now()));
info!(
target: "rustfs::scanner",
event = EVENT_SCANNER_DISK_IO_BUDGET,
component = LOG_COMPONENT_SCANNER,
subsystem = LOG_SUBSYSTEM_RUNTIME,
state = "resolved",
disk = %key,
media = media.as_str(),
iops = ?limits.iops,
bytes_per_sec = ?limits.bytes_per_sec,
"Scanner disk I/O budget resolved"
);
registry.insert(key, budget.clone());
budget
}
/// Apply a new configuration to every known disk budget.
pub(crate) fn update_disk_io_budget_config(config: ScannerDiskIoBudgetConfig) {
*DISK_IO_BUDGET_CONFIG.write().unwrap_or_else(|poisoned| poisoned.into_inner()) = config;
let registry = DISK_IO_BUDGETS.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
for budget in registry.values() {
budget.reconfigure(resolve_disk_io_budget(config, budget.media));
}
}
/// Per-disk budget status, ordered by disk path.
pub fn scanner_disk_io_budget_status() -> Vec<ScannerDiskIoBudgetStatus> {
let registry = DISK_IO_BUDGETS.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
registry.values().map(|budget| budget.status()).collect()
}
#[cfg(target_os = "linux")]
fn detect_disk_media(disk_path: &Path) -> ScannerDiskMedia {
let Ok(devices) = rustfs_utils::os::get_physical_device_ids(disk_path.to_string_lossy().as_ref()) else {
return ScannerDiskMedia::Unknown;
};
let mut detected = ScannerDiskMedia::Unknown;
for device in devices {
let candidates = [
format!("/sys/block/{device}/queue/rotational"),
format!("/sys/dev/block/{device}/queue/rotational"),
];
let Some(content) = candidates
.iter()
.find_map(|candidate| std::fs::read_to_string(candidate).ok())
else {
continue;
};
match content.trim() {
// Any spindle in the backing stack makes the path rotational.
"1" => return ScannerDiskMedia::Rotational,
"0" => detected = ScannerDiskMedia::Flash,
_ => {}
}
}
detected
}
#[cfg(not(target_os = "linux"))]
fn detect_disk_media(_disk_path: &Path) -> ScannerDiskMedia {
ScannerDiskMedia::Unknown
}
#[cfg(test)]
mod tests {
use super::*;
use serial_test::serial;
fn config(iops: ScannerIoBudgetLimit, bytes_per_sec: ScannerIoBudgetLimit, speed: ScannerSpeed) -> ScannerDiskIoBudgetConfig {
ScannerDiskIoBudgetConfig {
iops,
bytes_per_sec,
speed,
}
}
#[test]
fn disk_io_budget_limit_parses_auto_off_and_numbers() {
assert_eq!(ScannerIoBudgetLimit::parse("auto"), Some(ScannerIoBudgetLimit::Auto));
assert_eq!(ScannerIoBudgetLimit::parse(" AUTO "), Some(ScannerIoBudgetLimit::Auto));
assert_eq!(ScannerIoBudgetLimit::parse("off"), Some(ScannerIoBudgetLimit::Unlimited));
assert_eq!(ScannerIoBudgetLimit::parse("unlimited"), Some(ScannerIoBudgetLimit::Unlimited));
assert_eq!(ScannerIoBudgetLimit::parse("0"), Some(ScannerIoBudgetLimit::Explicit(0)));
assert_eq!(ScannerIoBudgetLimit::parse("4096"), Some(ScannerIoBudgetLimit::Explicit(4096)));
assert_eq!(ScannerIoBudgetLimit::parse("-1"), None);
assert_eq!(ScannerIoBudgetLimit::parse("fast"), None);
assert_eq!(ScannerIoBudgetLimit::Auto.to_string(), "auto");
assert_eq!(ScannerIoBudgetLimit::Unlimited.to_string(), "off");
assert_eq!(ScannerIoBudgetLimit::Explicit(7).to_string(), "7");
}
#[test]
fn disk_io_budget_resolves_media_defaults_and_overrides() {
let auto = config(ScannerIoBudgetLimit::Auto, ScannerIoBudgetLimit::Auto, ScannerSpeed::Default);
assert_eq!(
resolve_disk_io_budget(auto, ScannerDiskMedia::Rotational),
ScannerDiskIoBudgetLimits {
iops: Some(DEFAULT_SCANNER_DISK_IO_BUDGET_ROTATIONAL_IOPS),
bytes_per_sec: Some(DEFAULT_SCANNER_DISK_IO_BUDGET_ROTATIONAL_BYTES_PER_SEC),
}
);
// Unknown media is treated like a spindle.
assert_eq!(
resolve_disk_io_budget(auto, ScannerDiskMedia::Unknown),
resolve_disk_io_budget(auto, ScannerDiskMedia::Rotational)
);
assert_eq!(
resolve_disk_io_budget(auto, ScannerDiskMedia::Flash),
ScannerDiskIoBudgetLimits {
iops: Some(DEFAULT_SCANNER_DISK_IO_BUDGET_FLASH_IOPS),
bytes_per_sec: Some(DEFAULT_SCANNER_DISK_IO_BUDGET_FLASH_BYTES_PER_SEC),
}
);
// Rotational media is paced tighter than flash on both axes.
let rotational = resolve_disk_io_budget(auto, ScannerDiskMedia::Rotational);
let flash = resolve_disk_io_budget(auto, ScannerDiskMedia::Flash);
assert!(rotational.iops < flash.iops, "{rotational:?} vs {flash:?}");
assert!(rotational.bytes_per_sec < flash.bytes_per_sec, "{rotational:?} vs {flash:?}");
// The fastest preset never paces unless a value is explicit.
let fastest = config(ScannerIoBudgetLimit::Auto, ScannerIoBudgetLimit::Explicit(5), ScannerSpeed::Fastest);
assert_eq!(
resolve_disk_io_budget(fastest, ScannerDiskMedia::Rotational),
ScannerDiskIoBudgetLimits {
iops: None,
bytes_per_sec: Some(5),
}
);
let explicit = config(ScannerIoBudgetLimit::Explicit(0), ScannerIoBudgetLimit::Unlimited, ScannerSpeed::Slow);
assert_eq!(
resolve_disk_io_budget(explicit, ScannerDiskMedia::Flash),
ScannerDiskIoBudgetLimits {
iops: Some(0),
bytes_per_sec: None,
}
);
}
#[tokio::test(start_paused = true)]
async fn disk_io_budget_waits_once_the_burst_is_spent() {
let budget = ScannerDiskIoBudget::new(
"disk-a".to_string(),
ScannerDiskMedia::Rotational,
ScannerDiskIoBudgetLimits {
iops: Some(10),
bytes_per_sec: None,
},
Instant::now(),
);
// The burst window admits one second of rate without waiting.
for _ in 0..10 {
assert_eq!(budget.charge(1, 0), Duration::ZERO);
}
// The eleventh operation owes one token at 10/s.
assert_eq!(budget.charge(1, 0), Duration::from_millis(100));
// After the rate has refilled the deficit, the next charge is free again.
tokio::time::advance(Duration::from_millis(200)).await;
assert_eq!(budget.charge(1, 0), Duration::ZERO);
}
#[tokio::test(start_paused = true)]
async fn disk_io_budget_bytes_axis_caps_independently_of_iops() {
let budget = ScannerDiskIoBudget::new(
"disk-b".to_string(),
ScannerDiskMedia::Flash,
ScannerDiskIoBudgetLimits {
iops: Some(1_000),
bytes_per_sec: Some(1_000),
},
Instant::now(),
);
assert_eq!(budget.charge(1, 1_000), Duration::ZERO);
assert_eq!(budget.charge(1, 500), Duration::from_millis(500));
assert!(budget.budgets_bytes());
}
#[tokio::test(start_paused = true)]
async fn disk_io_budget_zero_budget_still_admits_one_charge_per_window() {
let budget = ScannerDiskIoBudget::new(
"disk-c".to_string(),
ScannerDiskMedia::Unknown,
ScannerDiskIoBudgetLimits {
iops: Some(0),
bytes_per_sec: Some(0),
},
Instant::now(),
);
// A zero budget starts with one token of burst, then degrades to the
// floor instead of waiting forever.
assert_eq!(budget.charge(1, 1), Duration::ZERO);
for _ in 0..3 {
assert_eq!(budget.charge(1, 4096), SCANNER_DISK_IO_BUDGET_MAX_WAIT);
}
let started = Instant::now();
budget.admit(1, 4096).await;
assert_eq!(started.elapsed(), SCANNER_DISK_IO_BUDGET_MAX_WAIT);
let status = budget.status();
assert_eq!(status.wait_events, 1);
assert_eq!(status.wait_seconds, 1.0);
assert_eq!(status.iops, Some(0));
assert_eq!(status.bytes_per_sec, Some(0));
}
#[tokio::test(start_paused = true)]
async fn disk_io_budget_oversized_charge_owes_at_most_one_extra_window() {
let budget = ScannerDiskIoBudget::new(
"disk-d".to_string(),
ScannerDiskMedia::Rotational,
ScannerDiskIoBudgetLimits {
iops: None,
bytes_per_sec: Some(1_000),
},
Instant::now(),
);
// A charge far above the rate is capped at the max wait ...
assert_eq!(budget.charge(0, 1_000_000), SCANNER_DISK_IO_BUDGET_MAX_WAIT);
// ... and leaves at most one burst window of debt behind.
assert_eq!(budget.charge(0, 0), Duration::ZERO);
assert_eq!(budget.charge(0, 1), SCANNER_DISK_IO_BUDGET_MAX_WAIT);
tokio::time::advance(Duration::from_secs(2)).await;
assert_eq!(budget.charge(0, 1), Duration::ZERO);
}
#[tokio::test(start_paused = true)]
async fn disk_io_budget_unlimited_axis_never_waits() {
let budget = ScannerDiskIoBudget::new(
"disk-e".to_string(),
ScannerDiskMedia::Flash,
ScannerDiskIoBudgetLimits::default(),
Instant::now(),
);
for _ in 0..1_000 {
assert_eq!(budget.charge(1, 1 << 20), Duration::ZERO);
}
assert!(!budget.budgets_bytes());
let started = Instant::now();
budget.admit(1, 1 << 20).await;
assert_eq!(started.elapsed(), Duration::ZERO);
assert_eq!(budget.status().wait_events, 0);
}
#[tokio::test(start_paused = true)]
async fn disk_io_budget_reconfigure_applies_the_new_rate_to_live_buckets() {
let budget = ScannerDiskIoBudget::new(
"disk-f".to_string(),
ScannerDiskMedia::Rotational,
ScannerDiskIoBudgetLimits {
iops: Some(1),
bytes_per_sec: None,
},
Instant::now(),
);
assert_eq!(budget.charge(1, 0), Duration::ZERO);
assert_eq!(budget.charge(1, 0), SCANNER_DISK_IO_BUDGET_MAX_WAIT);
budget.reconfigure(ScannerDiskIoBudgetLimits {
iops: Some(1_000),
bytes_per_sec: Some(10),
});
// Debt carried over is repaid at the new rate.
assert_eq!(budget.charge(1, 0), Duration::from_millis(2));
assert!(budget.budgets_bytes());
budget.reconfigure(ScannerDiskIoBudgetLimits::default());
assert_eq!(budget.charge(1, 1 << 30), Duration::ZERO);
assert_eq!(budget.limits(), ScannerDiskIoBudgetLimits::default());
}
#[tokio::test(start_paused = true)]
#[serial]
async fn disk_io_budget_registry_shares_one_budget_per_disk_and_follows_config_updates() {
// A path that does not exist resolves to unknown media on every platform.
let path = std::path::PathBuf::from(format!("/rustfs-scanner-budget-test-{}", std::process::id()));
let first = disk_io_budget_for(&path);
let second = disk_io_budget_for(&path);
assert!(Arc::ptr_eq(&first, &second));
let status = scanner_disk_io_budget_status();
assert!(status.iter().any(|entry| entry.disk == path.to_string_lossy()));
update_disk_io_budget_config(config(
ScannerIoBudgetLimit::Explicit(3),
ScannerIoBudgetLimit::Unlimited,
ScannerSpeed::Default,
));
assert_eq!(
first.limits(),
ScannerDiskIoBudgetLimits {
iops: Some(3),
bytes_per_sec: None,
}
);
update_disk_io_budget_config(ScannerDiskIoBudgetConfig::DEFAULT);
assert_eq!(first.limits(), resolve_disk_io_budget(ScannerDiskIoBudgetConfig::DEFAULT, first.media()));
}
}
docs/architecture/scanner-per-bucket-convergence.md
+9
View File
@@ -290,6 +290,15 @@ the speed factor. On a single-disk deployment this is one walker, one bucket,
one object at a time, at roughly one third duty cycle under the default
speed. Walker count is therefore a property of topology, not of the work.
Two later controls bound that walk per disk rather than per object.
rustfs/rustfs#8389 added the local-disk foreground-priority admission gate,
which orders scanner I/O behind foreground requests. rustfs/backlog#2905 added
a per-disk token-bucket budget (operations and bytes per second, defaults by
detected media) that `DynamicSleeper` charges before each directory listing
and metadata read; it is a hard ceiling layered on admission and never lets
the scanner exceed what admission allows. The operator view is in
`docs/operations/scanner-runtime-controls.md`.
A collector's pacer sets collector work from the mutator's rate under a
resource ceiling. The scanner adopts the same shape:
docs/operations/scanner-benchmark-runbook.md
+1
View File
@@ -248,6 +248,7 @@ Collect at least two runs on the same RustFS commit and the same workload. Keep
| Duration budget (when one cycle is too long) | Bound wall-clock time per cycle. | `scanner.cycle_max_duration="1800"`. |
| Object budget | Bound objects processed per cycle. | `scanner.cycle_max_objects="1000000"`. |
| Directory budget | Bound directories entered per cycle. | `scanner.cycle_max_directories="100000"`. |
| Disk I/O budget (single slow disk) | Cap scanner metadata I/O per local disk regardless of idle time; compare foreground p99 and scan completion against the media default. | `scanner.disk_io_budget_iops="50"` and `scanner.disk_io_budget_bytes_per_sec="8388608"`; `off` removes the cap for an A/B leg. |
## Deployment Matrix
docs/operations/scanner-runtime-controls.md
+45 -2
View File
@@ -2,7 +2,7 @@
**Use this when:** tuning scanner pacing or heal runtime knobs, reading `/v3/scanner/status`, deciding whether slow lifecycle/replication/heal progress is a scanner problem or a downstream queue problem, or migrating scanner settings from MinIO.
**Source of truth:** `crates/config/src/constants/scanner.rs` and `crates/config/src/constants/heal.rs` (keys, env names, `DEFAULT_*` constants), `crates/scanner/src/runtime_config.rs` (resolution order, parsing, hot update), `crates/scanner/src/scanner_folder.rs` (env-only scanner knobs, alerts), `crates/scanner/src/sleeper.rs` (throttling), `crates/heal/src/heal/manager.rs` (`HealConfig::default`), `crates/utils/src/envs.rs` (`EXTERNAL_COMPATIBLE_SUFFIXES`, MinIO env aliases).
**Source of truth:** `crates/config/src/constants/scanner.rs` and `crates/config/src/constants/heal.rs` (keys, env names, `DEFAULT_*` constants), `crates/scanner/src/runtime_config.rs` (resolution order, parsing, hot update), `crates/scanner/src/scanner_folder.rs` (env-only scanner knobs, alerts), `crates/scanner/src/sleeper.rs` and `crates/scanner/src/sleeper/io_budget.rs` (throttling and per-disk I/O budget), `crates/heal/src/heal/manager.rs` (`HealConfig::default`), `crates/utils/src/envs.rs` (`EXTERNAL_COMPATIBLE_SUFFIXES`, MinIO env aliases).
For reproducible scanner-pressure validation and before/after evidence, see [Scanner Benchmark Runbook](scanner-benchmark-runbook.md). Alert thresholds and S3 event names are detailed in [Scanner Excess Alerts](scanner-excess-alerts.md).
@@ -45,6 +45,8 @@ Bitrot cycle resolution differs because the canonical persistent key belongs to
| `scanner.max_concurrent_set_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS`) | Caps concurrent set-level scanner tasks. `0` keeps topology-derived concurrency. |
| `scanner.max_concurrent_disk_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS`) | Caps concurrent disk bucket walks per set. `0` keeps walker-count-derived concurrency. |
| `scanner.walkers_per_disk` | `RUSTFS_SCANNER_WALKERS_PER_DISK` | count | `1` (`DEFAULT_SCANNER_WALKERS_PER_DISK`) | Bucket walkers spawned per local disk of a set. The walkers of one disk share its scanner admission, so a peer still sees one scan session per physical disk; remote disks keep one walker. The resulting walker count is what `scanner.max_concurrent_disk_scans` caps, and the disk count stays the floor. Valid range is `1` through `16` (`MAX_SCANNER_WALKERS_PER_DISK`); `1` keeps today's I/O profile and is the right value for rotational media. The `disk_concurrency_budget` log line reports `walkers_per_disk`, `walkers`, the effective `concurrency_limit`, and its `reason` (`available_walkers`, `configured_cap`, or `count_budget`). |
| `scanner.disk_io_budget_iops` | `RUSTFS_SCANNER_DISK_IO_BUDGET_IOPS` | operations/s | `auto` (`DEFAULT_SCANNER_DISK_IO_BUDGET`) | Per-local-disk ceiling on scanner directory listings and object metadata reads. `auto` picks the media default (`100` on rotational or unknown media, `4000` on flash); `off` removes the cap; a number is an explicit rate. See [Per-disk background I/O budget](#per-disk-background-io-budget). |
| `scanner.disk_io_budget_bytes_per_sec` | `RUSTFS_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC` | bytes/s | `auto` (`DEFAULT_SCANNER_DISK_IO_BUDGET`) | Per-local-disk ceiling on the metadata bytes the scanner reads. `auto` picks the media default (`16777216` on rotational or unknown media, `268435456` on flash); `off` removes the cap; a number is an explicit rate. |
| `scanner.yield_every_n_objects` | `RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS` | objects | `128` (`DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS`) | Controls how often object loops yield to the async runtime. `0` disables this extra yield. |
| `scanner.alert_excess_versions` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS` | versions | `100` (`DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS`) | Version count threshold for scanner alerts. |
| `scanner.alert_excess_version_size` | `RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE` | bytes | `1099511627776` (`DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE`) | Retained version byte threshold for scanner alerts. |
@@ -129,6 +131,20 @@ The `rustfs_disk_io_admission_active` foreground gauge is sampled while backgrou
This is logical operation admission, not an OS disk scheduler: foreground and background operations may overlap after a background permit is granted, and the gate does not promise a bandwidth or IOPS limit. It is scoped to each `LocalDiskWrapper`; separate paths to one physical device and remote disk RPCs do not share this gate. It also does not observe or schedule unrelated host processes that share the physical device.
## Per-disk background I/O budget
The admission gate orders scanner I/O behind foreground requests but does not bound how much background I/O an idle disk sees. `scanner.disk_io_budget_iops` and `scanner.disk_io_budget_bytes_per_sec` add that bound. Each local disk owns one token bucket per axis (`crates/scanner/src/sleeper/io_budget.rs`); every directory listing charges one operation, and every object metadata read charges one operation plus the `xl.meta` size. A charge that exceeds the remaining budget delays that walker until the rate allows it. The budget is layered on top of admission: a budgeted operation still queues behind foreground requests, so the budget can only slow the scanner down and never lets it exceed what admission allows. It applies whether or not `idle_mode` is enabled.
- Media is detected once per disk path from the backing block device's `rotational` flag on Linux, through partitions, `dm-*`, and software RAID; any spindle in the stack counts as rotational. Undetectable media, including non-Linux hosts and overlay filesystems without a bind-mounted volume, uses the rotational defaults because a spindle is the case that needs protection.
- `auto` resolves to `100` operations/s and `16 MiB/s` on rotational or unknown media and to `4000` operations/s and `256 MiB/s` on flash. The rotational values leave roughly half of a 7200 rpm spindle's random reads to foreground traffic; the flash values are a few percent of a SATA SSD. The `fastest` speed preset resolves `auto` to no cap; explicit numbers always apply.
- A budget of `0` or any tiny value never starves a walk: one charge is admitted per second (`SCANNER_DISK_IO_BUDGET_MAX_WAIT`), and an oversized charge owes at most one extra second of debt.
- The walkers of one disk (`scanner.walkers_per_disk`) share that disk's budget. Remote disks are paced by their owning node.
- Changes take effect at the next charge without a restart.
The status endpoint lists each disk under `disk_io_budget` with its detected `media`, the effective `iops` and `bytes_per_sec` (`null` when uncapped), charged totals, and accumulated `wait_events` / `wait_seconds`. Budget waits are also reported per cycle as `metrics.current_cycle_disk_io_budget_wait_events` / `_duration_seconds` and `metrics.last_cycle_disk_io_budget_wait_events` / `_duration_seconds`, as `metrics.pacing_pressure.last_cycle_disk_io_budget_wait_ratio` with the `disk_io_budget` primary pressure, and in `metrics.usage_freshness.disk_io_budget_wait_events`, `disk_io_budget_wait_millis`, and `last_disk_io_budget_wait_unix_secs`. Prometheus exports `rustfs_scanner_disk_io_budget_waits_total` and `rustfs_scanner_disk_io_budget_wait_seconds` by disk. The `scanner_disk_io_budget` log line reports the resolved media and rates once per disk.
Like admission, this is in-process pacing: it does not schedule other host processes. Deep bitrot verification reads are issued by the heal subsystem and are paced by heal controls, not by this budget.
## Status Endpoint
```text
@@ -142,6 +158,8 @@ The request must be authenticated with an admin identity that has `ServerInfoAdm
| `runtime_config` | Effective runtime controls and their value sources. |
| `cycle_schedule` | Current effective cycle interval and clean-idle backoff state. |
| `cycle_recovery` | This node's cycle-state recovery observation, retry budget, and reason. |
| `bitrot` | How the current cycle's scan mode was decided: whether Deep is supported on this topology, where the bitrot cadence start came from, and the recorded start. See [Deep scan cadence state](#deep-scan-cadence-state). |
| `disk_io_budget` | One entry per local disk with detected media, effective per-disk I/O budget, charged totals, and accumulated budget waits. |
| `metrics` | Scanner work, pressure, checkpoint, lifecycle, replication, heal, bitrot, and alert counters. |
| `data_movement_pause` | Global-pause policy, current movement reason, operation epoch, start time, duration, and estimated movement work items. |
| `pause_backlog` | Replicated durable pause ledger, post-pause catch-up phase, rate window, retry state, thresholds, and active alert reasons. |
@@ -161,6 +179,14 @@ cycle_schedule.clean_idle_backoff_enabled
cycle_schedule.clean_idle_backoff_multiplier
metrics.pacing_pressure.primary_pressure
metrics.pacing_pressure.last_cycle_budget_limited
metrics.pacing_pressure.last_cycle_disk_io_budget_wait_ratio
metrics.usage_freshness.disk_io_budget_wait_events
bitrot.deep_scan_supported
bitrot.bitrot_start_kind
bitrot.bitrot_start_source
disk_io_budget[].media
disk_io_budget[].iops
disk_io_budget[].wait_seconds
metrics.lifecycle_transition.current_queued
metrics.lifecycle_transition.scanner_missed
metrics.maintenance_control.primary_control
@@ -189,6 +215,20 @@ catch_up_estimate.discovered_expiry_items
catch_up_estimate.discovered_transition_items
```
### Deep scan cadence state
`bitrot` reports how the scanner decided the scan mode of the running or most recent cycle:
| Field | Meaning |
|---|---|
| `deep_scan_supported` | `false` on an erasure single-disk deployment, where scanner-originated object heal sampling is unsupported and Deep mode is never selected (rustfs/rustfs#8126). |
| `bitrot_start_kind` | `deep_window` when a recorded Deep window start is in effect, `anchor` when no Deep window is recorded anywhere and the cadence counts from this process's first cycle without one, `none` before the first cycle. |
| `bitrot_start_source` | Where the reference came from: `persisted` (read back from the background-heal object), `memory` (the in-memory anchor, or a window start this process recorded that storage has not returned, for example because the save was skipped without movement admission), or `none`. |
| `bitrot_start_cycle`, `bitrot_start_unix_secs` | The recorded window start, or the anchor time with cycle `0`; both `0` when none. |
| `current_scan_mode` | `normal` or `deep`. |
A cycle that finds no recorded Deep window runs Normal and anchors the cadence in process memory; nothing is persisted until the configured bitrot cycle has elapsed since that anchor (or on every cycle when the cycle is `0`), at which point the Deep cycle records the window start exactly as before. An anchor never opens the Deep selection window: a missing record used to be read as a window that started at cycle `0`, which made every cycle below the selection window run Deep whenever the start could not be read back (rustfs/rustfs#8015). The window start also stays in process memory across a skipped or failed save, so the next cycle continues the window instead of reopening one; when a persisted and an in-memory start both exist, the newer one wins and ties go to the persisted record. `RUSTFS_SCANNER_BITROT_CYCLE_SECS=off` still never selects Deep, deployments whose background-heal state persists normally keep their existing cadence, and a restart resets the anchor.
### Periodic checkpoint persistence
`metrics.scan_checkpoint` is the in-memory resume hint; it does not confirm a durable save. `metrics.scan_checkpoints_persisted` lists the latest confirmed periodic save for each `(pool_index, set_index)` handled by this process, with its bucket, `saved_unix_secs`, and optional checkpoint cursor. It retains one observation per set, is reset on process restart, and does not include the final bucket save. Query the peer executing the walk for remote-worker observations. `metrics.scan_checkpoint_save_failures` counts failed periodic saves, including failures to refresh CAS revisions after a write; failed attempts do not advance the confirmed save observation.
@@ -311,11 +351,12 @@ Unlabeled Prometheus gauges:
|---|---|---|
| `queued_scans` | Set or disk scan queues are backing up. | Lower scanner concurrency or increase pacing delay if user traffic is affected. |
| `cycle_budget` | The last cycle stopped because a runtime/object/directory budget was reached. | Check `last_cycle_partial_reason` and `last_cycle_partial_source`; increase the specific budget if scans need to finish sooner. |
| `disk_io_budget` | The last cycle waited on the per-disk background I/O budget. | Expected on rotational media; raise `scanner.disk_io_budget_iops` or `scanner.disk_io_budget_bytes_per_sec` only when foreground latency has headroom. |
| `throttle_pause` | Scanner sleeps or cooperative yields were observed. | Expected when `idle_mode` is enabled; inspect pause ratios before tuning. |
| `active_scans` | Scanner work is active but not currently queued or budget-limited. | Usually healthy; correlate with CPU/disk metrics. |
| `none` | No current scanner pressure was observed. | No scanner pacing action needed. |
The ratio fields `last_cycle_throttle_sleep_ratio`, `last_cycle_yield_ratio`, and `last_cycle_total_pause_ratio` are fractions of the last cycle duration. If CPU is high but pause ratios are already high, increasing `scanner.delay` or `scanner.max_wait` may have limited value; check active paths, source work, and disk activity before changing the cycle interval.
The ratio fields `last_cycle_throttle_sleep_ratio`, `last_cycle_disk_io_budget_wait_ratio`, `last_cycle_yield_ratio`, and `last_cycle_total_pause_ratio` are fractions of the last cycle duration. If CPU is high but pause ratios are already high, increasing `scanner.delay` or `scanner.max_wait` may have limited value; check active paths, source work, and disk activity before changing the cycle interval.
## Reading Source Work
@@ -655,6 +696,8 @@ config:
idle_mode: "true"
yield_every_n_objects: "128"
bitrot_cycle_secs: "2592000"
disk_io_budget_iops: "100"
disk_io_budget_bytes_per_sec: "16777216"
```
Use `extraEnv` for environment variables that are not represented by chart values, including every heal knob above.
helm/README.md
+2
View File
@@ -177,6 +177,8 @@ behavior.
| config.rustfs.scanner.max_concurrent_set_scans | string | `""` | Cap concurrent scanner set tasks with `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` (`0` keeps topology-derived concurrency). |
| config.rustfs.scanner.max_concurrent_disk_scans | string | `""` | Cap concurrent scanner disk bucket walks per set with `RUSTFS_SCANNER_MAX_CONCURRENT_DISK_SCANS` (`0` keeps disk-count-derived concurrency). |
| config.rustfs.scanner.walkers_per_disk | string | `""` | Bucket walkers per local disk with `RUSTFS_SCANNER_WALKERS_PER_DISK` (`1` to `16`; the default `1` keeps one bucket walk per disk, raise it on flash so one disk walks several buckets concurrently). |
| config.rustfs.scanner.disk_io_budget_iops | string | `""` | Per-local-disk ceiling on scanner metadata operations per second with `RUSTFS_SCANNER_DISK_IO_BUDGET_IOPS` (`auto` picks the media default, `off` removes the cap, a number is an explicit rate). |
| config.rustfs.scanner.disk_io_budget_bytes_per_sec | string | `""` | Per-local-disk ceiling on scanner metadata bytes per second with `RUSTFS_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC` (`auto` picks the media default, `off` removes the cap, a number is an explicit rate). |
| config.rustfs.scanner.yield_every_n_objects | string | `""` | Yield to the async runtime every N scanned objects with `RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS` (`0` disables extra yield). |
| config.rustfs.scanner.alert_excess_versions | string | `""` | Set version count threshold for scanner alerts with `RUSTFS_SCANNER_ALERT_EXCESS_VERSIONS`. |
| config.rustfs.scanner.alert_excess_version_size | string | `""` | Set retained version byte threshold for scanner alerts with `RUSTFS_SCANNER_ALERT_EXCESS_VERSION_SIZE`. |
helm/rustfs/templates/configmap.yaml
+6
View File
@@ -126,6 +126,12 @@ data:
{{- end }}
{{- if ne (toString .walkers_per_disk) "" }}
RUSTFS_SCANNER_WALKERS_PER_DISK: {{ .walkers_per_disk | quote }}
{{- end }}
{{- if ne (toString .disk_io_budget_iops) "" }}
RUSTFS_SCANNER_DISK_IO_BUDGET_IOPS: {{ .disk_io_budget_iops | quote }}
{{- end }}
{{- if ne (toString .disk_io_budget_bytes_per_sec) "" }}
RUSTFS_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC: {{ .disk_io_budget_bytes_per_sec | quote }}
{{- end }}
{{- if ne (toString .yield_every_n_objects) "" }}
RUSTFS_SCANNER_YIELD_EVERY_N_OBJECTS: {{ .yield_every_n_objects | quote }}
helm/rustfs/values.yaml
+4
View File
@@ -189,6 +189,10 @@ config:
max_concurrent_disk_scans: ""
# Bucket walkers per local disk (1-16; default 1 keeps one bucket walk per disk, raise on flash)
walkers_per_disk: ""
# Per-local-disk scanner I/O ceiling in operations per second (auto, off, or a number)
disk_io_budget_iops: ""
# Per-local-disk scanner I/O ceiling in bytes per second (auto, off, or a number)
disk_io_budget_bytes_per_sec: ""
# Yield to the async runtime every N scanned objects (0 disables extra yield)
yield_every_n_objects: ""
# Version count threshold for scanner alerts
rustfs/src/admin/handlers/config_admin.rs
+21 -5
View File
@@ -66,17 +66,19 @@ use rustfs_config::{
BASE_DSN_STRING, COMMENT_KEY, DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, ENV_SCANNER_ALERT_EXCESS_FOLDERS,
ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, ENV_SCANNER_ALERT_EXCESS_VERSIONS, ENV_SCANNER_BITROT_CYCLE_SECS,
ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES,
ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY, ENV_SCANNER_IDLE_MODE,
ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY,
ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, ENV_SCANNER_DISK_IO_BUDGET_IOPS, ENV_SCANNER_IDLE_MODE,
ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_WAIT_SECS, ENV_SCANNER_SPEED,
ENV_SCANNER_START_DELAY_SECS, ENV_SCANNER_WALKERS_PER_DISK, ENV_SCANNER_YIELD_EVERY_N_OBJECTS, HEAL_BITROT_CYCLE,
HEAL_SUB_SYS, MAX_ADMIN_REQUEST_BODY_SIZE, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR,
MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, SCANNER_ALERT_EXCESS_FOLDERS,
SCANNER_ALERT_EXCESS_VERSION_SIZE, SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT,
SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY,
SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED,
SCANNER_START_DELAY, SCANNER_SUB_SYS, SCANNER_WALKERS_PER_DISK, SCANNER_YIELD_EVERY_N_OBJECTS, WEBHOOK_AUTH_TOKEN,
WEBHOOK_BATCH_SIZE, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, WEBHOOK_MAX_RETRY,
WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL,
SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, SCANNER_DISK_IO_BUDGET_IOPS, SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS,
SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT, SCANNER_SPEED, SCANNER_START_DELAY, SCANNER_SUB_SYS,
SCANNER_WALKERS_PER_DISK, SCANNER_YIELD_EVERY_N_OBJECTS, WEBHOOK_AUTH_TOKEN, WEBHOOK_BATCH_SIZE, WEBHOOK_CLIENT_CERT,
WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, WEBHOOK_MAX_RETRY, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT,
WEBHOOK_RETRY_INTERVAL,
};
use rustfs_credentials::Credentials;
use rustfs_policy::policy::action::{Action, AdminAction};
@@ -292,6 +294,18 @@ const SCANNER_HELP_KEYS: &[HelpKeyMetadata] = &[
description: "bucket walkers per local disk, 1 through 16; remote disks keep one walker",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_DISK_IO_BUDGET_IOPS,
type_name: "string",
description: "per-local-disk ceiling on scanner metadata operations per second: auto picks the media default, off removes the cap, a number is an explicit rate",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC,
type_name: "string",
description: "per-local-disk ceiling on scanner metadata bytes per second: auto picks the media default, off removes the cap, a number is an explicit rate",
optional: true,
},
HelpKeyMetadata {
key: SCANNER_YIELD_EVERY_N_OBJECTS,
type_name: "number",
@@ -1626,6 +1640,8 @@ fn env_help_key(sub_system: &str, key: &str) -> String {
(SCANNER_SUB_SYS, SCANNER_MAX_CONCURRENT_SET_SCANS) => ENV_SCANNER_MAX_CONCURRENT_SET_SCANS.to_string(),
(SCANNER_SUB_SYS, SCANNER_MAX_CONCURRENT_DISK_SCANS) => ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS.to_string(),
(SCANNER_SUB_SYS, SCANNER_WALKERS_PER_DISK) => ENV_SCANNER_WALKERS_PER_DISK.to_string(),
(SCANNER_SUB_SYS, SCANNER_DISK_IO_BUDGET_IOPS) => ENV_SCANNER_DISK_IO_BUDGET_IOPS.to_string(),
(SCANNER_SUB_SYS, SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC) => ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC.to_string(),
(SCANNER_SUB_SYS, SCANNER_YIELD_EVERY_N_OBJECTS) => ENV_SCANNER_YIELD_EVERY_N_OBJECTS.to_string(),
(SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_VERSIONS) => ENV_SCANNER_ALERT_EXCESS_VERSIONS.to_string(),
(SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_VERSION_SIZE) => ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE.to_string(),
rustfs/src/admin/handlers/scanner.rs
+13
View File
@@ -47,6 +47,8 @@ struct ScannerStatusResponse {
cycle_schedule: rustfs_scanner::ScannerCycleScheduleStatus,
runtime_config: rustfs_scanner::runtime_config::ScannerRuntimeConfigStatus,
cycle_recovery: rustfs_scanner::ScannerCycleRecoveryStatus,
bitrot: rustfs_scanner::ScannerBitrotStatus,
disk_io_budget: Vec<rustfs_scanner::ScannerDiskIoBudgetStatus>,
data_movement_pause: ScannerDataMovementPauseStatus,
pause_backlog: rustfs_scanner::ScannerPauseBacklogStatus,
catch_up_estimate: ScannerCatchUpEstimate,
@@ -220,6 +222,8 @@ fn scanner_status_response(
cycle_schedule,
runtime_config,
cycle_recovery: rustfs_scanner::scanner::scanner_cycle_recovery_status(),
bitrot: rustfs_scanner::scanner_bitrot_status(),
disk_io_budget: rustfs_scanner::scanner_disk_io_budget_status(),
data_movement_pause,
pause_backlog,
catch_up_estimate,
@@ -839,6 +843,15 @@ mod tests {
assert_eq!(encoded["data_movement_pause"]["paused"], false);
assert_eq!(encoded["catch_up_estimate"]["estimated"], false);
assert_eq!(encoded["catch_up_estimate"]["undiscovered_ilm_items_known"], true);
assert!(
["none", "memory", "persisted"].contains(&encoded["bitrot"]["bitrot_start_source"].as_str().unwrap_or("")),
"bitrot start source must name where the Deep cadence start came from: {}",
encoded["bitrot"]
);
assert!(encoded["bitrot"]["deep_scan_supported"].is_boolean());
assert!(encoded["bitrot"]["bitrot_start_cycle"].is_u64());
assert!(encoded["bitrot"]["bitrot_start_unix_secs"].is_u64());
assert!(encoded["disk_io_budget"].is_array());
}
#[test]