mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-31 18:42:17 +00:00
fix(scanner): report active first-cycle status (#5397)
* fix(scanner): report active first-cycle status * fix(scanner): publish cycle activity consistently * test(common): satisfy Rust 1.97 waker lint * fix(scanner): satisfy Rust 1.97 clippy --------- Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Henry Guo
+145
-33
@@ -1114,6 +1114,8 @@ pub struct ScannerLastMinute {
|
||||
pub struct ScannerMetricsReport {
|
||||
pub collected_at: DateTime<Utc>,
|
||||
pub current_cycle: u64,
|
||||
#[serde(default)]
|
||||
pub current_cycle_active: bool,
|
||||
pub current_started: DateTime<Utc>,
|
||||
pub cycles_completed_at: Vec<DateTime<Utc>>,
|
||||
pub ongoing_buckets: usize,
|
||||
@@ -2051,7 +2053,7 @@ impl Metrics {
|
||||
pub fn record_scanner_transition_failed(&self, count: u64) {
|
||||
self.scanner_transition_failed.fetch_add(count, Ordering::Relaxed);
|
||||
self.record_scanner_source_failed(ScannerWorkSource::Lifecycle, count);
|
||||
if !self.current_scan_cycle_work_active.load(Ordering::Relaxed) {
|
||||
if !self.current_scan_cycle_work_active.load(Ordering::Acquire) {
|
||||
self.record_last_cycle_scanner_source_work(ScannerWorkSource::Lifecycle, ScannerSourceWorkUpdate::failed(count));
|
||||
}
|
||||
}
|
||||
@@ -2336,6 +2338,21 @@ impl Metrics {
|
||||
*self.cycle_info.write().await = cycle;
|
||||
}
|
||||
|
||||
/// Publish a scanner cycle and its work-accounting baseline as one state transition.
|
||||
pub async fn start_scan_cycle_work_with_cycle(&self, cycle: CurrentCycle) -> ScanCycleWorkSnapshot {
|
||||
let mut current_cycle = self.cycle_info.write().await;
|
||||
let snapshot = self.start_scan_cycle_work();
|
||||
*current_cycle = Some(cycle);
|
||||
snapshot
|
||||
}
|
||||
|
||||
/// Publish the completed work snapshot and idle cycle state as one state transition.
|
||||
pub async fn finish_scan_cycle_work_with_cycle(&self, start: ScanCycleWorkSnapshot, cycle: CurrentCycle) {
|
||||
let mut current_cycle = self.cycle_info.write().await;
|
||||
self.finish_scan_cycle_work(start);
|
||||
*current_cycle = Some(cycle);
|
||||
}
|
||||
|
||||
/// Read the current cycle record.
|
||||
pub async fn get_cycle(&self) -> Option<CurrentCycle> {
|
||||
self.cycle_info.read().await.clone()
|
||||
@@ -2464,7 +2481,7 @@ impl Metrics {
|
||||
&self.current_scan_cycle_replication_repair_work_start,
|
||||
&replication_repair_snapshot,
|
||||
);
|
||||
self.current_scan_cycle_work_active.store(true, Ordering::Relaxed);
|
||||
self.current_scan_cycle_work_active.store(true, Ordering::Release);
|
||||
snapshot
|
||||
}
|
||||
|
||||
@@ -2476,11 +2493,11 @@ impl Metrics {
|
||||
self.record_scan_cycle_work(work);
|
||||
self.record_scan_cycle_source_work(&source_work);
|
||||
self.record_scan_cycle_replication_repair_work(&replication_repair_work);
|
||||
self.current_scan_cycle_work_active.store(false, Ordering::Relaxed);
|
||||
self.current_scan_cycle_work_active.store(false, Ordering::Release);
|
||||
}
|
||||
|
||||
pub fn current_scan_cycle_has_unresolved_heal_work(&self) -> bool {
|
||||
if !self.current_scan_cycle_work_active.load(Ordering::Relaxed) {
|
||||
if !self.current_scan_cycle_work_active.load(Ordering::Acquire) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -2746,13 +2763,41 @@ impl Metrics {
|
||||
pub async fn report(&self) -> ScannerMetricsReport {
|
||||
let mut m = ScannerMetricsReport::default();
|
||||
|
||||
let has_cycle = if let Some(cycle) = self.get_cycle().await {
|
||||
m.current_cycle = cycle.current;
|
||||
m.cycles_completed_at = cycle.cycle_completed;
|
||||
m.current_started = cycle.started;
|
||||
true
|
||||
} else {
|
||||
false
|
||||
let has_cycle = {
|
||||
let cycle = self.cycle_info.read().await;
|
||||
let has_cycle = if let Some(cycle) = cycle.as_ref() {
|
||||
m.current_cycle = cycle.current;
|
||||
m.cycles_completed_at = cycle.cycle_completed.clone();
|
||||
m.current_started = cycle.started;
|
||||
true
|
||||
} else {
|
||||
false
|
||||
};
|
||||
m.current_cycle_active = self.current_scan_cycle_work_active.load(Ordering::Acquire);
|
||||
if m.current_cycle_active {
|
||||
let current_work = self.scan_cycle_work_since(self.current_scan_cycle_work_start());
|
||||
let current_source_work = self.scanner_source_work_since(&self.current_scan_cycle_source_work_start_values());
|
||||
let current_replication_repair_work =
|
||||
self.scanner_replication_repair_work_since(&self.current_scan_cycle_replication_repair_work_start_values());
|
||||
m.current_cycle_objects_scanned = current_work.objects_scanned;
|
||||
m.current_cycle_directories_scanned = current_work.directories_scanned;
|
||||
m.current_cycle_bucket_drive_scans = current_work.bucket_drive_scans;
|
||||
m.current_cycle_bucket_drive_failures = current_work.bucket_drive_failures;
|
||||
m.current_cycle_yield_events = current_work.yield_events;
|
||||
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_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;
|
||||
m.current_cycle_heal_objects = current_work.heal_objects;
|
||||
m.current_cycle_replication_checks = current_work.replication_checks;
|
||||
m.current_cycle_usage_saves = current_work.usage_saves;
|
||||
m.current_cycle_source_work = self.scanner_source_work_snapshots(¤t_source_work);
|
||||
m.current_cycle_replication_repair =
|
||||
self.scanner_replication_repair_work_snapshots(¤t_replication_repair_work);
|
||||
}
|
||||
has_cycle
|
||||
};
|
||||
|
||||
if !has_cycle && let Some(init_time) = crate::get_global_init_time().await {
|
||||
@@ -2793,28 +2838,6 @@ impl Metrics {
|
||||
m.current_disk_scan_concurrency_limit = disk_scan_concurrency_limit;
|
||||
m.current_disk_bucket_scans_queued = disk_bucket_scans_queued;
|
||||
m.current_disk_bucket_scans_active = disk_bucket_scans_active;
|
||||
if self.current_scan_cycle_work_active.load(Ordering::Relaxed) {
|
||||
let current_work = self.scan_cycle_work_since(self.current_scan_cycle_work_start());
|
||||
let current_source_work = self.scanner_source_work_since(&self.current_scan_cycle_source_work_start_values());
|
||||
let current_replication_repair_work =
|
||||
self.scanner_replication_repair_work_since(&self.current_scan_cycle_replication_repair_work_start_values());
|
||||
m.current_cycle_objects_scanned = current_work.objects_scanned;
|
||||
m.current_cycle_directories_scanned = current_work.directories_scanned;
|
||||
m.current_cycle_bucket_drive_scans = current_work.bucket_drive_scans;
|
||||
m.current_cycle_bucket_drive_failures = current_work.bucket_drive_failures;
|
||||
m.current_cycle_yield_events = current_work.yield_events;
|
||||
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_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;
|
||||
m.current_cycle_heal_objects = current_work.heal_objects;
|
||||
m.current_cycle_replication_checks = current_work.replication_checks;
|
||||
m.current_cycle_usage_saves = current_work.usage_saves;
|
||||
m.current_cycle_source_work = self.scanner_source_work_snapshots(¤t_source_work);
|
||||
m.current_cycle_replication_repair = self.scanner_replication_repair_work_snapshots(¤t_replication_repair_work);
|
||||
}
|
||||
let last_cycle_result = self.last_scan_cycle_result.load(Ordering::Relaxed);
|
||||
m.last_cycle_result = scan_cycle_result_label(last_cycle_result).to_string();
|
||||
m.last_cycle_result_code = last_cycle_result as u64;
|
||||
@@ -4142,6 +4165,8 @@ mod tests {
|
||||
|
||||
let report = metrics.report().await;
|
||||
|
||||
assert!(report.current_cycle_active);
|
||||
assert_eq!(report.current_cycle, 0);
|
||||
assert_eq!(report.current_cycle_objects_scanned, 7);
|
||||
assert_eq!(report.current_cycle_directories_scanned, 3);
|
||||
assert_eq!(report.current_cycle_bucket_drive_scans, 2);
|
||||
@@ -4158,6 +4183,8 @@ mod tests {
|
||||
metrics.finish_scan_cycle_work(start);
|
||||
let report = metrics.report().await;
|
||||
|
||||
assert!(!report.current_cycle_active);
|
||||
assert_eq!(report.current_cycle, 0);
|
||||
assert_eq!(report.current_cycle_objects_scanned, 0);
|
||||
assert_eq!(report.current_cycle_directories_scanned, 0);
|
||||
assert_eq!(report.current_cycle_bucket_drive_scans, 0);
|
||||
@@ -4184,6 +4211,91 @@ mod tests {
|
||||
assert_eq!(report.last_cycle_usage_saves, 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scan_cycle_activity_and_cycle_state_publish_together() {
|
||||
let metrics = Metrics::new();
|
||||
let cycle_started = Utc::now() - chrono::Duration::seconds(5);
|
||||
let active_cycle = CurrentCycle {
|
||||
current: 12,
|
||||
next: 13,
|
||||
started: cycle_started,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let cycle_state = metrics.cycle_info.read().await;
|
||||
let mut start_transition = Box::pin(metrics.start_scan_cycle_work_with_cycle(active_cycle));
|
||||
let waker = std::task::Waker::noop();
|
||||
let mut context = std::task::Context::from_waker(waker);
|
||||
assert!(start_transition.as_mut().poll(&mut context).is_pending());
|
||||
assert!(!metrics.current_scan_cycle_work_active.load(Ordering::Acquire));
|
||||
drop(cycle_state);
|
||||
|
||||
let start = start_transition.await;
|
||||
let active = metrics.report().await;
|
||||
assert!(active.current_cycle_active);
|
||||
assert_eq!(active.current_cycle, 12);
|
||||
assert_eq!(active.current_started, cycle_started);
|
||||
|
||||
let idle_cycle = CurrentCycle {
|
||||
current: 0,
|
||||
next: 13,
|
||||
started: cycle_started,
|
||||
..Default::default()
|
||||
};
|
||||
let cycle_state = metrics.cycle_info.read().await;
|
||||
let mut finish_transition = Box::pin(metrics.finish_scan_cycle_work_with_cycle(start, idle_cycle));
|
||||
assert!(finish_transition.as_mut().poll(&mut context).is_pending());
|
||||
assert!(metrics.current_scan_cycle_work_active.load(Ordering::Acquire));
|
||||
drop(cycle_state);
|
||||
|
||||
finish_transition.await;
|
||||
let idle = metrics.report().await;
|
||||
assert!(!idle.current_cycle_active);
|
||||
assert_eq!(idle.current_cycle, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn report_keeps_cycle_identity_and_work_in_one_snapshot() {
|
||||
let metrics = Metrics::new();
|
||||
let cycle_ten = CurrentCycle {
|
||||
current: 10,
|
||||
next: 11,
|
||||
started: Utc::now() - chrono::Duration::seconds(10),
|
||||
..Default::default()
|
||||
};
|
||||
let cycle_ten_start = metrics.start_scan_cycle_work_with_cycle(cycle_ten.clone()).await;
|
||||
metrics.operations[Metric::ScanObject as usize].store(1, Ordering::Relaxed);
|
||||
|
||||
let paths = metrics.current_paths.write().await;
|
||||
let mut report = Box::pin(metrics.report());
|
||||
let waker = std::task::Waker::noop();
|
||||
let mut context = std::task::Context::from_waker(waker);
|
||||
assert!(report.as_mut().poll(&mut context).is_pending());
|
||||
|
||||
metrics
|
||||
.finish_scan_cycle_work_with_cycle(cycle_ten_start, CurrentCycle { current: 0, ..cycle_ten })
|
||||
.await;
|
||||
let cycle_eleven_start = metrics
|
||||
.start_scan_cycle_work_with_cycle(CurrentCycle {
|
||||
current: 11,
|
||||
next: 12,
|
||||
started: Utc::now(),
|
||||
..Default::default()
|
||||
})
|
||||
.await;
|
||||
metrics.operations[Metric::ScanObject as usize].store(101, Ordering::Relaxed);
|
||||
|
||||
drop(paths);
|
||||
let snapshot = report.await;
|
||||
|
||||
assert_eq!(snapshot.current_cycle, 10);
|
||||
assert_eq!(snapshot.current_cycle_objects_scanned, 1);
|
||||
|
||||
metrics
|
||||
.finish_scan_cycle_work_with_cycle(cycle_eleven_start, CurrentCycle::default())
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scanner_cycle_ilm_actions_ignore_global_ilm_work() {
|
||||
let metrics = Metrics::new();
|
||||
|
||||
@@ -207,10 +207,6 @@ pub(crate) async fn ensure_boot_time() {
|
||||
GLOBAL_BOOT_TIME.get_or_init(|| async { SystemTime::now() }).await;
|
||||
}
|
||||
|
||||
pub(crate) async fn scanner_init_time() -> Option<chrono::DateTime<chrono::Utc>> {
|
||||
rustfs_common::get_global_init_time().await
|
||||
}
|
||||
|
||||
pub(crate) async fn root_disk_threshold_for_erasure_disk() -> Option<u64> {
|
||||
if is_erasure_sd().await {
|
||||
None
|
||||
|
||||
@@ -71,6 +71,7 @@ fn to_madmin_scanner_metrics(metrics: rustfs_common::metrics::ScannerMetricsRepo
|
||||
MadminScannerMetrics {
|
||||
collected_at: metrics.collected_at,
|
||||
current_cycle: metrics.current_cycle,
|
||||
current_cycle_active: Some(metrics.current_cycle_active),
|
||||
current_started: metrics.current_started,
|
||||
cycles_completed_at: metrics.cycles_completed_at,
|
||||
ongoing_buckets: metrics.ongoing_buckets,
|
||||
@@ -398,10 +399,7 @@ pub async fn collect_local_metrics(types: MetricType, opts: &CollectMetricsOpts)
|
||||
|
||||
if types.contains(&MetricType::SCANNER) {
|
||||
debug!("start get scanner metrics");
|
||||
let mut metrics = global_metrics().report().await;
|
||||
if let Some(init_time) = runtime_sources::scanner_init_time().await {
|
||||
metrics.current_started = init_time;
|
||||
}
|
||||
let metrics = global_metrics().report().await;
|
||||
real_time_metrics.aggregated.scanner = Some(to_madmin_scanner_metrics(metrics));
|
||||
}
|
||||
|
||||
@@ -540,7 +538,9 @@ async fn collect_local_disks_metrics(disks: &HashSet<String>) -> HashMap<String,
|
||||
#[cfg(test)]
|
||||
mod test {
|
||||
use super::*;
|
||||
use rustfs_common::metrics::CurrentCycle;
|
||||
use rustfs_io_metrics::internode_metrics::global_internode_metrics;
|
||||
use serial_test::serial;
|
||||
use std::time::Duration;
|
||||
|
||||
#[test]
|
||||
@@ -588,7 +588,10 @@ mod test {
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_mapping_preserves_partial_source_status() {
|
||||
let current_started = Utc::now() - chrono::Duration::seconds(5);
|
||||
let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {
|
||||
current_cycle_active: true,
|
||||
current_started,
|
||||
last_cycle_partial_source: "usage".to_string(),
|
||||
last_cycle_partial_source_code: 1,
|
||||
partial_cycles_by_source: vec![rustfs_common::metrics::ScannerSourceCycleSnapshot {
|
||||
@@ -598,6 +601,8 @@ mod test {
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
assert_eq!(scanner.current_cycle_active, Some(true));
|
||||
assert_eq!(scanner.current_started, current_started);
|
||||
assert_eq!(scanner.last_cycle_partial_source, "usage");
|
||||
assert_eq!(scanner.last_cycle_partial_source_code, 1);
|
||||
let usage = scanner
|
||||
@@ -608,6 +613,39 @@ mod test {
|
||||
assert_eq!(usage.cycles, 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn collect_local_metrics_preserves_scanner_cycle_started_time() {
|
||||
let previous_init_time = *rustfs_common::globals::GLOBAL_INIT_TIME.read().await;
|
||||
let previous_cycle = global_metrics().get_cycle().await;
|
||||
let init_time = Utc::now() - chrono::Duration::hours(1);
|
||||
let cycle_started = Utc::now() - chrono::Duration::seconds(5);
|
||||
*rustfs_common::globals::GLOBAL_INIT_TIME.write().await = Some(init_time);
|
||||
let cycle = CurrentCycle {
|
||||
current: 0,
|
||||
next: 1,
|
||||
started: cycle_started,
|
||||
..Default::default()
|
||||
};
|
||||
let cycle_start = global_metrics().start_scan_cycle_work_with_cycle(cycle).await;
|
||||
|
||||
let realtime = collect_local_metrics(MetricType::SCANNER, &CollectMetricsOpts::default()).await;
|
||||
|
||||
global_metrics()
|
||||
.finish_scan_cycle_work_with_cycle(cycle_start, previous_cycle.clone().unwrap_or_default())
|
||||
.await;
|
||||
global_metrics().set_cycle(previous_cycle).await;
|
||||
*rustfs_common::globals::GLOBAL_INIT_TIME.write().await = previous_init_time;
|
||||
|
||||
let encoded = rmp_serde::to_vec_named(&realtime).expect("realtime metrics should encode");
|
||||
let decoded: RealtimeMetrics = rmp_serde::from_slice(&encoded).expect("realtime metrics should decode");
|
||||
let mut aggregated = RealtimeMetrics::default();
|
||||
aggregated.merge(decoded);
|
||||
let scanner = aggregated.aggregated.scanner.expect("scanner metrics");
|
||||
assert_eq!(scanner.current_cycle_active, Some(true));
|
||||
assert_eq!(scanner.current_started, cycle_started);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_mapping_preserves_pacing_pressure() {
|
||||
let scanner = to_madmin_scanner_metrics(rustfs_common::metrics::ScannerMetricsReport {
|
||||
|
||||
@@ -545,6 +545,8 @@ pub struct ScannerMetrics {
|
||||
pub collected_at: DateTime<Utc>,
|
||||
#[serde(rename = "current_cycle")]
|
||||
pub current_cycle: u64,
|
||||
#[serde(rename = "current_cycle_active", default, skip_serializing_if = "Option::is_none")]
|
||||
pub current_cycle_active: Option<bool>,
|
||||
#[serde(rename = "current_started")]
|
||||
pub current_started: DateTime<Utc>,
|
||||
#[serde(rename = "cycle_complete_times")]
|
||||
@@ -718,7 +720,31 @@ pub struct ScannerMetrics {
|
||||
}
|
||||
|
||||
impl ScannerMetrics {
|
||||
pub fn is_current_cycle_active(&self) -> bool {
|
||||
self.current_cycle_active.unwrap_or(self.current_cycle > 0)
|
||||
}
|
||||
|
||||
pub fn merge(&mut self, other: &Self) {
|
||||
// Legacy nodes omit the activity field and use a non-zero cycle as
|
||||
// their active signal. New nodes publish explicit first-cycle and idle
|
||||
// states, including cycle zero.
|
||||
let self_cycle_active = self.is_current_cycle_active();
|
||||
let other_cycle_active = other.is_current_cycle_active();
|
||||
let self_cycle_authority = (
|
||||
self_cycle_active,
|
||||
self.current_cycle,
|
||||
self.cycles_completed_at.len(),
|
||||
self.cycles_completed_at.as_slice(),
|
||||
self.current_started,
|
||||
);
|
||||
let other_cycle_authority = (
|
||||
other_cycle_active,
|
||||
other.current_cycle,
|
||||
other.cycles_completed_at.len(),
|
||||
other.cycles_completed_at.as_slice(),
|
||||
other.current_started,
|
||||
);
|
||||
let other_cycle_is_authoritative = self_cycle_authority < other_cycle_authority;
|
||||
let other_is_newer = self.collected_at < other.collected_at;
|
||||
if other_is_newer {
|
||||
self.collected_at = other.collected_at;
|
||||
@@ -854,15 +880,12 @@ impl ScannerMetrics {
|
||||
self.ongoing_buckets = other.ongoing_buckets;
|
||||
}
|
||||
|
||||
if self.current_cycle < other.current_cycle {
|
||||
if other_cycle_is_authoritative {
|
||||
self.current_cycle = other.current_cycle;
|
||||
self.cycles_completed_at = other.cycles_completed_at.clone();
|
||||
self.current_started = other.current_started;
|
||||
}
|
||||
|
||||
if other.cycles_completed_at.len() > self.cycles_completed_at.len() {
|
||||
self.cycles_completed_at = other.cycles_completed_at.clone();
|
||||
}
|
||||
self.current_cycle_active = Some(self_cycle_active || other_cycle_active);
|
||||
|
||||
if !other.life_time_ops.is_empty() && self.life_time_ops.is_empty() {
|
||||
self.life_time_ops = other.life_time_ops.clone();
|
||||
@@ -931,7 +954,13 @@ impl Metrics {
|
||||
if let Some(scanner) = other.scanner.as_ref() {
|
||||
match self.scanner {
|
||||
Some(ref mut s_scanner) => s_scanner.merge(scanner),
|
||||
None => self.scanner = Some(scanner.clone()),
|
||||
None => {
|
||||
let mut scanner = scanner.clone();
|
||||
if scanner.current_cycle_active.is_none() {
|
||||
scanner.current_cycle_active = Some(scanner.is_current_cycle_active());
|
||||
}
|
||||
self.scanner = Some(scanner);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1389,6 +1418,280 @@ pub struct Operations {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_serializes_cycle_active_presence() {
|
||||
let missing_value = serde_json::to_value(ScannerMetrics::default()).expect("scanner metrics should serialize");
|
||||
assert!(missing_value.get("current_cycle_active").is_none());
|
||||
let missing: ScannerMetrics =
|
||||
serde_json::from_value(missing_value).expect("older scanner metrics without cycle-active should decode");
|
||||
assert_eq!(missing.current_cycle_active, None);
|
||||
|
||||
let explicit_false_value = serde_json::to_value(ScannerMetrics {
|
||||
current_cycle_active: Some(false),
|
||||
..Default::default()
|
||||
})
|
||||
.expect("scanner metrics with explicit cycle-active should serialize");
|
||||
assert_eq!(explicit_false_value["current_cycle_active"], serde_json::Value::Bool(false));
|
||||
let explicit_false: ScannerMetrics =
|
||||
serde_json::from_value(explicit_false_value).expect("explicit cycle-active should decode");
|
||||
assert_eq!(explicit_false.current_cycle_active, Some(false));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_merge_prefers_an_active_first_cycle() {
|
||||
let collected_at = Utc::now();
|
||||
let idle_started = collected_at - chrono::Duration::hours(1);
|
||||
let active_started = collected_at - chrono::Duration::seconds(5);
|
||||
let mut scanner = ScannerMetrics {
|
||||
collected_at,
|
||||
current_cycle: 0,
|
||||
current_cycle_active: Some(false),
|
||||
current_started: idle_started,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
scanner.merge(&ScannerMetrics {
|
||||
collected_at: collected_at + chrono::Duration::seconds(1),
|
||||
current_cycle: 0,
|
||||
current_cycle_active: Some(true),
|
||||
current_started: active_started,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
assert_eq!(scanner.current_cycle_active, Some(true));
|
||||
assert_eq!(scanner.current_cycle, 0);
|
||||
assert_eq!(scanner.current_started, active_started);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metrics_merge_preserves_explicit_active_first_cycle() {
|
||||
let mut aggregated = Metrics::default();
|
||||
aggregated.merge(&Metrics {
|
||||
scanner: Some(ScannerMetrics {
|
||||
current_cycle_active: Some(true),
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
let scanner = aggregated.scanner.expect("aggregated scanner metrics");
|
||||
assert_eq!(scanner.current_cycle_active, Some(true));
|
||||
assert_eq!(scanner.current_cycle, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_merge_preserves_legacy_nonzero_active_signal() {
|
||||
let collected_at = Utc::now();
|
||||
let mut scanner = ScannerMetrics {
|
||||
collected_at,
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
scanner.merge(&ScannerMetrics {
|
||||
collected_at: collected_at + chrono::Duration::seconds(1),
|
||||
current_cycle: 7,
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
assert_eq!(scanner.current_cycle_active, Some(true));
|
||||
assert_eq!(scanner.current_cycle, 7);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn metrics_merge_normalizes_first_legacy_scanner_snapshot() {
|
||||
let legacy = Metrics {
|
||||
scanner: Some(ScannerMetrics {
|
||||
current_cycle: 7,
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
};
|
||||
let mut aggregated = Metrics::default();
|
||||
|
||||
aggregated.merge(&legacy);
|
||||
|
||||
let scanner = aggregated.scanner.expect("aggregated scanner metrics");
|
||||
assert_eq!(scanner.current_cycle_active, Some(true));
|
||||
assert_eq!(scanner.current_cycle, 7);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_merge_preserves_explicit_inactive_nonzero_cycle() {
|
||||
let collected_at = Utc::now();
|
||||
let mut scanner = ScannerMetrics::default();
|
||||
|
||||
scanner.merge(&ScannerMetrics {
|
||||
collected_at,
|
||||
current_cycle: 7,
|
||||
current_cycle_active: Some(false),
|
||||
..Default::default()
|
||||
});
|
||||
|
||||
assert_eq!(scanner.current_cycle_active, Some(false));
|
||||
assert_eq!(scanner.current_cycle, 7);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_merge_cycle_active_is_order_independent() {
|
||||
let collected_at = Utc::now();
|
||||
let legacy_active = ScannerMetrics {
|
||||
collected_at,
|
||||
current_cycle: 7,
|
||||
current_started: collected_at - chrono::Duration::seconds(10),
|
||||
..Default::default()
|
||||
};
|
||||
let explicit_idle = ScannerMetrics {
|
||||
collected_at,
|
||||
current_cycle: 0,
|
||||
current_cycle_active: Some(false),
|
||||
current_started: collected_at - chrono::Duration::hours(1),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let mut legacy_first = legacy_active.clone();
|
||||
legacy_first.merge(&explicit_idle);
|
||||
let mut legacy_second = explicit_idle.clone();
|
||||
legacy_second.merge(&legacy_active);
|
||||
assert_eq!(legacy_first.current_cycle_active, Some(true));
|
||||
assert_eq!(legacy_second.current_cycle_active, Some(true));
|
||||
assert_eq!(legacy_first.current_cycle, 7);
|
||||
assert_eq!(legacy_second.current_cycle, 7);
|
||||
|
||||
let explicit_inactive_nonzero = ScannerMetrics {
|
||||
collected_at,
|
||||
current_cycle: 7,
|
||||
current_cycle_active: Some(false),
|
||||
..Default::default()
|
||||
};
|
||||
let mut inactive_first = explicit_inactive_nonzero.clone();
|
||||
inactive_first.merge(&explicit_idle);
|
||||
let mut inactive_second = explicit_idle;
|
||||
inactive_second.merge(&explicit_inactive_nonzero);
|
||||
assert_eq!(inactive_first.current_cycle_active, Some(false));
|
||||
assert_eq!(inactive_second.current_cycle_active, Some(false));
|
||||
assert_eq!(inactive_first.current_cycle, 7);
|
||||
assert_eq!(inactive_second.current_cycle, 7);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_merge_cycle_authority_is_order_independent() {
|
||||
let collected_at = Utc::now();
|
||||
let completion = collected_at - chrono::Duration::minutes(1);
|
||||
let earlier_active = ScannerMetrics {
|
||||
collected_at,
|
||||
current_cycle: 7,
|
||||
current_cycle_active: Some(true),
|
||||
current_started: collected_at - chrono::Duration::seconds(10),
|
||||
cycles_completed_at: vec![completion],
|
||||
..Default::default()
|
||||
};
|
||||
let later_active = ScannerMetrics {
|
||||
collected_at: collected_at + chrono::Duration::seconds(1),
|
||||
current_cycle: 7,
|
||||
current_cycle_active: Some(true),
|
||||
current_started: collected_at - chrono::Duration::seconds(5),
|
||||
cycles_completed_at: vec![completion],
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let mut earlier_first = earlier_active.clone();
|
||||
earlier_first.merge(&later_active);
|
||||
let mut later_first = later_active.clone();
|
||||
later_first.merge(&earlier_active);
|
||||
|
||||
assert_eq!(earlier_first.current_started, later_active.current_started);
|
||||
assert_eq!(later_first.current_started, later_active.current_started);
|
||||
assert_eq!(earlier_first.cycles_completed_at, later_active.cycles_completed_at);
|
||||
assert_eq!(later_first.cycles_completed_at, later_active.cycles_completed_at);
|
||||
|
||||
let stale_idle = ScannerMetrics {
|
||||
collected_at,
|
||||
current_cycle_active: Some(false),
|
||||
current_started: collected_at - chrono::Duration::hours(1),
|
||||
..Default::default()
|
||||
};
|
||||
let completed_idle = ScannerMetrics {
|
||||
collected_at: collected_at + chrono::Duration::seconds(1),
|
||||
current_cycle_active: Some(false),
|
||||
current_started: collected_at - chrono::Duration::seconds(5),
|
||||
cycles_completed_at: vec![completion],
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let mut stale_first = stale_idle.clone();
|
||||
stale_first.merge(&completed_idle);
|
||||
let mut completed_first = completed_idle.clone();
|
||||
completed_first.merge(&stale_idle);
|
||||
|
||||
assert_eq!(stale_first.current_started, completed_idle.current_started);
|
||||
assert_eq!(completed_first.current_started, completed_idle.current_started);
|
||||
assert_eq!(stale_first.cycles_completed_at, completed_idle.cycles_completed_at);
|
||||
assert_eq!(completed_first.cycles_completed_at, completed_idle.cycles_completed_at);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_merge_cycle_authority_is_associative() {
|
||||
let collected_at = Utc::now();
|
||||
let older_completion = collected_at - chrono::Duration::minutes(3);
|
||||
let last_completion = collected_at - chrono::Duration::minutes(1);
|
||||
let cycle_seven = ScannerMetrics {
|
||||
collected_at,
|
||||
current_cycle: 7,
|
||||
current_cycle_active: Some(true),
|
||||
current_started: collected_at - chrono::Duration::seconds(10),
|
||||
cycles_completed_at: vec![older_completion, last_completion],
|
||||
..Default::default()
|
||||
};
|
||||
let cycle_eight = ScannerMetrics {
|
||||
collected_at: collected_at + chrono::Duration::seconds(1),
|
||||
current_cycle: 8,
|
||||
current_cycle_active: Some(true),
|
||||
current_started: collected_at - chrono::Duration::seconds(5),
|
||||
cycles_completed_at: vec![last_completion],
|
||||
..Default::default()
|
||||
};
|
||||
let newer_idle = ScannerMetrics {
|
||||
collected_at: collected_at + chrono::Duration::hours(1),
|
||||
current_cycle_active: Some(false),
|
||||
current_started: collected_at - chrono::Duration::hours(1),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let mut left_associative = cycle_seven.clone();
|
||||
left_associative.merge(&cycle_eight);
|
||||
left_associative.merge(&newer_idle);
|
||||
|
||||
let mut right_group = cycle_eight.clone();
|
||||
right_group.merge(&newer_idle);
|
||||
let mut right_associative = cycle_seven.clone();
|
||||
right_associative.merge(&right_group);
|
||||
|
||||
assert_eq!(left_associative.current_cycle, 8);
|
||||
assert_eq!(right_associative.current_cycle, 8);
|
||||
assert_eq!(left_associative.current_started, cycle_eight.current_started);
|
||||
assert_eq!(right_associative.current_started, cycle_eight.current_started);
|
||||
assert_eq!(left_associative.cycles_completed_at, cycle_eight.cycles_completed_at);
|
||||
assert_eq!(right_associative.cycles_completed_at, cycle_eight.cycles_completed_at);
|
||||
|
||||
for order in [
|
||||
[&cycle_seven, &cycle_eight, &newer_idle],
|
||||
[&cycle_seven, &newer_idle, &cycle_eight],
|
||||
[&cycle_eight, &cycle_seven, &newer_idle],
|
||||
[&cycle_eight, &newer_idle, &cycle_seven],
|
||||
[&newer_idle, &cycle_seven, &cycle_eight],
|
||||
[&newer_idle, &cycle_eight, &cycle_seven],
|
||||
] {
|
||||
let mut merged = ScannerMetrics::default();
|
||||
for scanner in order {
|
||||
merged.merge(scanner);
|
||||
}
|
||||
assert_eq!(merged.current_cycle_active, Some(true));
|
||||
assert_eq!(merged.current_cycle, 8);
|
||||
assert_eq!(merged.current_started, cycle_eight.current_started);
|
||||
assert_eq!(merged.cycles_completed_at, cycle_eight.cycles_completed_at);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scanner_metrics_merge_aggregates_partial_cycles_by_source() {
|
||||
let collected_at = Utc::now();
|
||||
|
||||
@@ -262,11 +262,11 @@ async fn obs_site_replication_stats() -> ReplicationStats {
|
||||
}
|
||||
|
||||
fn current_scanner_cycle_age_seconds(
|
||||
current_cycle: u64,
|
||||
current_cycle_active: bool,
|
||||
current_started: chrono::DateTime<Utc>,
|
||||
now: chrono::DateTime<Utc>,
|
||||
) -> u64 {
|
||||
if current_cycle == 0 {
|
||||
if !current_cycle_active {
|
||||
0
|
||||
} else {
|
||||
now.signed_duration_since(current_started).num_seconds().max(0) as u64
|
||||
@@ -1090,7 +1090,7 @@ pub async fn collect_scanner_metric_stats() -> Option<ScannerStats> {
|
||||
let reference_time = metrics.cycles_completed_at.last().copied().unwrap_or(metrics.current_started);
|
||||
let last_activity_seconds = now.signed_duration_since(reference_time).num_seconds().max(0) as u64;
|
||||
let active_paths = metrics.active_scan_paths as u64;
|
||||
let current_cycle_age_seconds = current_scanner_cycle_age_seconds(metrics.current_cycle, metrics.current_started, now);
|
||||
let current_cycle_age_seconds = current_scanner_cycle_age_seconds(metrics.current_cycle_active, metrics.current_started, now);
|
||||
let current_scan_mode = scanner_scan_mode_code(&metrics.current_scan_mode);
|
||||
let current_cycle_age = current_cycle_age_seconds as f64;
|
||||
let last_cycle_duration = metrics.last_cycle_duration_seconds;
|
||||
@@ -1464,21 +1464,21 @@ mod tests {
|
||||
fn current_scanner_cycle_age_seconds_returns_zero_when_idle() {
|
||||
let now = Utc::now();
|
||||
|
||||
assert_eq!(current_scanner_cycle_age_seconds(0, now - chrono::Duration::seconds(30), now), 0);
|
||||
assert_eq!(current_scanner_cycle_age_seconds(false, now - chrono::Duration::seconds(30), now), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn current_scanner_cycle_age_seconds_clamps_future_start() {
|
||||
let now = Utc::now();
|
||||
|
||||
assert_eq!(current_scanner_cycle_age_seconds(4, now + chrono::Duration::seconds(30), now), 0);
|
||||
assert_eq!(current_scanner_cycle_age_seconds(true, now + chrono::Duration::seconds(30), now), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn current_scanner_cycle_age_seconds_reports_active_elapsed_time() {
|
||||
fn current_scanner_cycle_age_seconds_reports_active_first_cycle_elapsed_time() {
|
||||
let now = Utc::now();
|
||||
|
||||
assert_eq!(current_scanner_cycle_age_seconds(4, now - chrono::Duration::seconds(45), now), 45);
|
||||
assert_eq!(current_scanner_cycle_age_seconds(true, now - chrono::Duration::seconds(45), now), 45);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -47,7 +47,7 @@ rmp-serde = { workspace = true }
|
||||
hmac = { workspace = true }
|
||||
sha2 = { workspace = true }
|
||||
rustfs-filemeta = { workspace = true }
|
||||
tokio-util = { workspace = true, features = ["io", "compat"] }
|
||||
tokio-util = { workspace = true, features = ["io", "compat", "rt"] }
|
||||
rustfs-ecstore = { workspace = true }
|
||||
rustfs-storage-api = { workspace = true }
|
||||
http = { workspace = true }
|
||||
|
||||
+342
-47
@@ -14,6 +14,8 @@
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::future::Future;
|
||||
#[cfg(test)]
|
||||
use std::sync::Mutex as StdMutex;
|
||||
use std::sync::{Arc, LazyLock, RwLock};
|
||||
|
||||
use crate::ScannerObjectIO;
|
||||
@@ -38,8 +40,8 @@ use bytes::Bytes;
|
||||
use chrono::{DateTime, Utc};
|
||||
use rustfs_common::heal_channel::HealScanMode;
|
||||
use rustfs_common::metrics::{
|
||||
CurrentCycle, Metric, Metrics, ScanCyclePartialReason, ScannerUsageSaveResult, ScannerWorkSource, emit_scan_cycle_complete,
|
||||
emit_scan_cycle_partial_with_source, emit_scan_cycle_superseded, global_metrics,
|
||||
CurrentCycle, Metric, Metrics, ScanCyclePartialReason, ScanCycleWorkSnapshot, ScannerUsageSaveResult, ScannerWorkSource,
|
||||
emit_scan_cycle_complete, emit_scan_cycle_partial_with_source, emit_scan_cycle_superseded, global_metrics,
|
||||
};
|
||||
use rustfs_config::ScannerSpeed;
|
||||
#[cfg(test)]
|
||||
@@ -50,9 +52,12 @@ use rustfs_config::{
|
||||
use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sha2::{Digest as _, Sha256};
|
||||
#[cfg(test)]
|
||||
use tokio::sync::Notify;
|
||||
use tokio::sync::mpsc;
|
||||
use tokio::time::{Duration, Instant};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
use tracing::{debug, error, info, instrument, warn};
|
||||
|
||||
use crate::storage_api::scan::{
|
||||
@@ -93,6 +98,44 @@ const SCANNER_CYCLE_STATE_MAGIC: &[u8; 8] = b"RSCYC001";
|
||||
const SCANNER_CYCLE_STATE_HEADER_LEN: usize = 24;
|
||||
#[cfg(test)]
|
||||
const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS";
|
||||
#[cfg(test)]
|
||||
type ScannerCycleStatePersistTestHook = (u64, Arc<Notify>);
|
||||
#[cfg(test)]
|
||||
static SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK: LazyLock<StdMutex<Option<ScannerCycleStatePersistTestHook>>> =
|
||||
LazyLock::new(|| StdMutex::new(None));
|
||||
|
||||
#[cfg(test)]
|
||||
struct ScannerCycleStatePersistTestHookGuard;
|
||||
|
||||
#[cfg(test)]
|
||||
impl Drop for ScannerCycleStatePersistTestHookGuard {
|
||||
fn drop(&mut self) {
|
||||
*SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner()) = None;
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn set_scanner_cycle_state_persist_test_hook(leader_epoch: u64, reached: Arc<Notify>) -> ScannerCycleStatePersistTestHookGuard {
|
||||
*SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner()) = Some((leader_epoch, reached));
|
||||
ScannerCycleStatePersistTestHookGuard
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
fn notify_scanner_cycle_state_persist_test_hook(leader_epoch: u64) {
|
||||
let reached = SCANNER_CYCLE_STATE_PERSIST_TEST_HOOK
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
.as_ref()
|
||||
.filter(|(expected_epoch, _)| *expected_epoch == leader_epoch)
|
||||
.map(|(_, reached)| reached.clone());
|
||||
if let Some(reached) = reached {
|
||||
reached.notify_one();
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
enum ScannerCycleStateError {
|
||||
@@ -1691,10 +1734,10 @@ fn data_usage_persist_timeout() -> Duration {
|
||||
DataUsageCache::persistence_timeout()
|
||||
}
|
||||
|
||||
async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle) {
|
||||
async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle, cycle_metrics_guard: &mut ScannerCycleMetricsGuard) {
|
||||
cycle_info.current = 0;
|
||||
global_metrics().clear_current_scan_mode();
|
||||
global_metrics().set_cycle(Some(cycle_info.clone())).await;
|
||||
cycle_metrics_guard.finish(cycle_info.clone()).await;
|
||||
}
|
||||
|
||||
fn encode_scanner_cycle_state(cycle_info: &CurrentCycle, leader_epoch: u64) -> Result<Vec<u8>, ScannerCycleStateError> {
|
||||
@@ -2218,6 +2261,8 @@ async fn persist_scanner_cycle_state(
|
||||
return false;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
notify_scanner_cycle_state_persist_test_hook(leader_epoch);
|
||||
match save_config_with_preconditions(storeapi.clone(), &DATA_USAGE_BLOOM_NAME_PATH, buf.clone(), revision.preconditions())
|
||||
.await
|
||||
{
|
||||
@@ -2340,7 +2385,6 @@ async fn persist_scanner_cycle_state(
|
||||
|
||||
if persisted_cycle.next >= cycle_info.next {
|
||||
*cycle_info = persisted_cycle;
|
||||
global_metrics().set_cycle(Some(cycle_info.clone())).await;
|
||||
debug!(
|
||||
target: "rustfs::scanner",
|
||||
event = EVENT_SCANNER_PERSIST_STATE,
|
||||
@@ -2406,6 +2450,7 @@ async fn finalize_partial_scan_cycle(
|
||||
cycle_info: &mut CurrentCycle,
|
||||
revision: &mut DataUsageCacheRevision,
|
||||
leader_epoch: u64,
|
||||
cycle_metrics_guard: &mut ScannerCycleMetricsGuard,
|
||||
) -> bool {
|
||||
// A budget-limited cycle is deliberate pacing, not a failure. The cycle counter
|
||||
// must still advance (and persist) because per-bucket next_cycle is stamped from
|
||||
@@ -2422,11 +2467,14 @@ async fn finalize_partial_scan_cycle(
|
||||
error = %err,
|
||||
"Scanner partial cycle could not advance"
|
||||
);
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
mark_scan_cycle_idle(cycle_info, cycle_metrics_guard).await;
|
||||
return false;
|
||||
}
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await
|
||||
cycle_info.current = 0;
|
||||
global_metrics().clear_current_scan_mode();
|
||||
let persisted = persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await;
|
||||
cycle_metrics_guard.finish(cycle_info.clone()).await;
|
||||
persisted
|
||||
}
|
||||
|
||||
async fn persist_required_scanner_cycle_floor(
|
||||
@@ -2436,6 +2484,7 @@ async fn persist_required_scanner_cycle_floor(
|
||||
revision: &mut DataUsageCacheRevision,
|
||||
leader_epoch: u64,
|
||||
required_cycle: u64,
|
||||
cycle_metrics_guard: &mut ScannerCycleMetricsGuard,
|
||||
) -> bool {
|
||||
if required_cycle <= cycle_info.current || required_cycle == u64::MAX {
|
||||
error!(
|
||||
@@ -2448,13 +2497,16 @@ async fn persist_required_scanner_cycle_floor(
|
||||
state = "invalid_cache_cycle_floor",
|
||||
"Scanner cache cycle floor is invalid"
|
||||
);
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
mark_scan_cycle_idle(cycle_info, cycle_metrics_guard).await;
|
||||
return false;
|
||||
}
|
||||
|
||||
cycle_info.next = cycle_info.next.max(required_cycle);
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await
|
||||
cycle_info.current = 0;
|
||||
global_metrics().clear_current_scan_mode();
|
||||
let persisted = persist_scanner_cycle_state(ctx, storeapi, cycle_info, revision, leader_epoch).await;
|
||||
cycle_metrics_guard.finish(cycle_info.clone()).await;
|
||||
persisted
|
||||
}
|
||||
|
||||
async fn await_scanner_cycle_with_lock_fence<Cycle, LockLost>(
|
||||
@@ -2513,7 +2565,7 @@ async fn run_data_scanner_cycle(
|
||||
let now = Instant::now();
|
||||
cycle_info.started = Utc::now();
|
||||
|
||||
global_metrics().set_cycle(Some(cycle_info.clone())).await;
|
||||
let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
|
||||
|
||||
let mut background_heal_info = read_background_heal_info(storeapi.clone()).await;
|
||||
|
||||
@@ -2564,14 +2616,14 @@ async fn run_data_scanner_cycle(
|
||||
"Scanner cycle could not capture the data usage persistence baseline"
|
||||
);
|
||||
emit_scan_cycle_complete(false, cycle_start.elapsed());
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
|
||||
return ScannerCycleOutcome::Failed;
|
||||
}
|
||||
};
|
||||
let (sender, receiver) = mpsc::channel::<DataUsageInfo>(1);
|
||||
let storeapi_clone = storeapi.clone();
|
||||
let ctx_clone = ctx.clone();
|
||||
let mut usage_persist_task = tokio::spawn(async move {
|
||||
let mut usage_persist_task = AbortOnDropHandle::new(tokio::spawn(async move {
|
||||
store_data_usage_in_backend_with_outcome_for_epoch_and_baseline(
|
||||
ctx_clone,
|
||||
storeapi_clone,
|
||||
@@ -2580,10 +2632,9 @@ async fn run_data_scanner_cycle(
|
||||
Some(usage_persist_baseline),
|
||||
)
|
||||
.await
|
||||
});
|
||||
}));
|
||||
|
||||
let done_cycle = Metrics::time(Metric::ScanCycle);
|
||||
let cycle_work_start = global_metrics().start_scan_cycle_work();
|
||||
let cycle_budget = ScannerCycleBudget::new(ctx, cycle_budget_config);
|
||||
let scan_result = storeapi
|
||||
.clone()
|
||||
@@ -2640,7 +2691,6 @@ async fn run_data_scanner_cycle(
|
||||
}
|
||||
};
|
||||
let unresolved_heal_work = global_metrics().current_scan_cycle_has_unresolved_heal_work();
|
||||
global_metrics().finish_scan_cycle_work(cycle_work_start);
|
||||
|
||||
let scan_cycle_result = match scan_result {
|
||||
Ok(result) => result,
|
||||
@@ -2663,7 +2713,7 @@ async fn run_data_scanner_cycle(
|
||||
{
|
||||
save_background_heal_info(storeapi.clone(), new_heal_info).await;
|
||||
}
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
|
||||
return ScannerCycleOutcome::Failed;
|
||||
}
|
||||
};
|
||||
@@ -2678,7 +2728,7 @@ async fn run_data_scanner_cycle(
|
||||
"Scanner cycle stopped before committing cycle state"
|
||||
);
|
||||
emit_scan_cycle_complete(false, cycle_start.elapsed());
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
|
||||
return ScannerCycleOutcome::Failed;
|
||||
}
|
||||
if let Some(required_cycle) = scan_cycle_result.required_cycle_floor() {
|
||||
@@ -2700,6 +2750,7 @@ async fn run_data_scanner_cycle(
|
||||
cycle_revision,
|
||||
leader_epoch,
|
||||
required_cycle,
|
||||
&mut cycle_metrics_guard,
|
||||
)
|
||||
.await
|
||||
{
|
||||
@@ -2719,7 +2770,7 @@ async fn run_data_scanner_cycle(
|
||||
"Scanner cycle completed without a durable data usage snapshot"
|
||||
);
|
||||
emit_scan_cycle_complete(false, cycle_start.elapsed());
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
|
||||
return ScannerCycleOutcome::Failed;
|
||||
}
|
||||
if budget_elapsed {
|
||||
@@ -2743,7 +2794,16 @@ async fn run_data_scanner_cycle(
|
||||
scan_cycle_partial_reason(budget_reason),
|
||||
scan_cycle_partial_source(budget_reason),
|
||||
);
|
||||
return if finalize_partial_scan_cycle(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await {
|
||||
return if finalize_partial_scan_cycle(
|
||||
ctx,
|
||||
storeapi.clone(),
|
||||
cycle_info,
|
||||
cycle_revision,
|
||||
leader_epoch,
|
||||
&mut cycle_metrics_guard,
|
||||
)
|
||||
.await
|
||||
{
|
||||
ScannerCycleOutcome::Partial
|
||||
} else {
|
||||
ScannerCycleOutcome::Failed
|
||||
@@ -2792,7 +2852,7 @@ async fn run_data_scanner_cycle(
|
||||
"Scanner cycle completed without a durable data usage snapshot"
|
||||
);
|
||||
emit_scan_cycle_complete(false, cycle_start.elapsed());
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
|
||||
return ScannerCycleOutcome::Failed;
|
||||
}
|
||||
ScannerCycleOutcome::Partial => {
|
||||
@@ -2818,7 +2878,16 @@ async fn run_data_scanner_cycle(
|
||||
);
|
||||
}
|
||||
emit_scan_cycle_partial_with_source(cycle_start.elapsed(), ScanCyclePartialReason::Unknown, None);
|
||||
return if finalize_partial_scan_cycle(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await {
|
||||
return if finalize_partial_scan_cycle(
|
||||
ctx,
|
||||
storeapi.clone(),
|
||||
cycle_info,
|
||||
cycle_revision,
|
||||
leader_epoch,
|
||||
&mut cycle_metrics_guard,
|
||||
)
|
||||
.await
|
||||
{
|
||||
ScannerCycleOutcome::Partial
|
||||
} else {
|
||||
ScannerCycleOutcome::Failed
|
||||
@@ -2834,7 +2903,16 @@ async fn run_data_scanner_cycle(
|
||||
state = "superseded",
|
||||
"Scanner cycle usage snapshot was superseded by concurrent namespace activity"
|
||||
);
|
||||
if finalize_partial_scan_cycle(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await {
|
||||
if finalize_partial_scan_cycle(
|
||||
ctx,
|
||||
storeapi.clone(),
|
||||
cycle_info,
|
||||
cycle_revision,
|
||||
leader_epoch,
|
||||
&mut cycle_metrics_guard,
|
||||
)
|
||||
.await
|
||||
{
|
||||
emit_scan_cycle_superseded(cycle_start.elapsed());
|
||||
return ScannerCycleOutcome::Superseded;
|
||||
}
|
||||
@@ -2853,7 +2931,7 @@ async fn run_data_scanner_cycle(
|
||||
error = %err,
|
||||
"Scanner completed cycle could not advance"
|
||||
);
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
mark_scan_cycle_idle(cycle_info, &mut cycle_metrics_guard).await;
|
||||
emit_scan_cycle_complete(false, cycle_start.elapsed());
|
||||
return ScannerCycleOutcome::Failed;
|
||||
}
|
||||
@@ -2862,9 +2940,8 @@ async fn run_data_scanner_cycle(
|
||||
global_metrics().clear_current_scan_mode();
|
||||
|
||||
retain_recent_cycle_completions(&mut cycle_info.cycle_completed);
|
||||
global_metrics().set_cycle(Some(cycle_info.clone())).await;
|
||||
if !persist_scanner_cycle_state(ctx, storeapi.clone(), cycle_info, cycle_revision, leader_epoch).await {
|
||||
mark_scan_cycle_idle(cycle_info).await;
|
||||
cycle_metrics_guard.finish(cycle_info.clone()).await;
|
||||
emit_scan_cycle_complete(false, cycle_start.elapsed());
|
||||
return ScannerCycleOutcome::Failed;
|
||||
}
|
||||
@@ -2888,9 +2965,37 @@ async fn run_data_scanner_cycle(
|
||||
"Scanner cycle completed"
|
||||
);
|
||||
|
||||
cycle_metrics_guard.finish(cycle_info.clone()).await;
|
||||
scanner_cycle_outcome_with_pending_maintenance(ScannerCycleOutcome::Completed, pending_maintenance_work)
|
||||
}
|
||||
|
||||
struct ScannerCycleMetricsGuard {
|
||||
start: Option<ScanCycleWorkSnapshot>,
|
||||
}
|
||||
|
||||
impl ScannerCycleMetricsGuard {
|
||||
async fn new(cycle: CurrentCycle) -> Self {
|
||||
Self {
|
||||
start: Some(global_metrics().start_scan_cycle_work_with_cycle(cycle).await),
|
||||
}
|
||||
}
|
||||
|
||||
async fn finish(&mut self, cycle: CurrentCycle) {
|
||||
if let Some(start) = self.start {
|
||||
global_metrics().finish_scan_cycle_work_with_cycle(start, cycle).await;
|
||||
self.start = None;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for ScannerCycleMetricsGuard {
|
||||
fn drop(&mut self) {
|
||||
if let Some(start) = self.start.take() {
|
||||
global_metrics().finish_scan_cycle_work(start);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn record_scanner_leader_lock_lost(message: &'static str) {
|
||||
reset_scanner_cycle_schedule();
|
||||
record_scanner_leader_lock_state("lost");
|
||||
@@ -3443,7 +3548,7 @@ enum DataUsagePersistTaskResult {
|
||||
|
||||
async fn wait_for_data_usage_persist_task(
|
||||
ctx: &CancellationToken,
|
||||
task: &mut tokio::task::JoinHandle<DataUsagePersistOutcome>,
|
||||
task: &mut AbortOnDropHandle<DataUsagePersistOutcome>,
|
||||
timeout: Duration,
|
||||
) -> DataUsagePersistTaskResult {
|
||||
tokio::select! {
|
||||
@@ -3840,8 +3945,9 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::EcstoreResult;
|
||||
use crate::{
|
||||
ScannerGetObjectReader as GetObjectReader, ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions,
|
||||
ScannerPutObjReader as PutObjReader,
|
||||
Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerGetObjectReader as GetObjectReader,
|
||||
ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader,
|
||||
init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx,
|
||||
};
|
||||
use serial_test::serial;
|
||||
use std::collections::HashMap;
|
||||
@@ -3853,6 +3959,47 @@ mod tests {
|
||||
|
||||
const TEST_DEFAULT_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60;
|
||||
|
||||
async fn setup_scanner_cycle_store() -> (tempfile::TempDir, Arc<ECStore>) {
|
||||
init_ecstore_config_for_scanner_tests();
|
||||
let temp_dir = tempfile::tempdir().expect("scanner cycle test directory should be created");
|
||||
let mut endpoints = Vec::new();
|
||||
for disk_index in 0..4 {
|
||||
let disk_path = temp_dir.path().join(format!("disk{disk_index}"));
|
||||
tokio::fs::create_dir_all(&disk_path)
|
||||
.await
|
||||
.expect("scanner cycle test disk should be created");
|
||||
let mut endpoint =
|
||||
Endpoint::try_from(disk_path.to_str().expect("disk path should be utf8")).expect("endpoint should parse");
|
||||
endpoint.set_pool_index(0);
|
||||
endpoint.set_set_index(0);
|
||||
endpoint.set_disk_index(disk_index);
|
||||
endpoints.push(endpoint);
|
||||
}
|
||||
let endpoint_pools = EndpointServerPools::from(vec![PoolEndpoints {
|
||||
legacy: false,
|
||||
set_count: 1,
|
||||
drives_per_set: 4,
|
||||
endpoints: Endpoints::from(endpoints),
|
||||
cmd_line: "scanner-cycle-metrics".to_string(),
|
||||
platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
|
||||
}]);
|
||||
let instance_ctx = Arc::new(InstanceContext::new());
|
||||
init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone())
|
||||
.await
|
||||
.expect("scanner cycle test disks should initialize");
|
||||
let store = ECStore::new_with_instance_ctx(
|
||||
"127.0.0.1:0".parse().expect("test address should parse"),
|
||||
endpoint_pools,
|
||||
CancellationToken::new(),
|
||||
instance_ctx,
|
||||
)
|
||||
.await
|
||||
.expect("scanner cycle test ECStore should initialize");
|
||||
init_bucket_metadata_sys_for_scanner_tests(store.clone()).await;
|
||||
|
||||
(temp_dir, store)
|
||||
}
|
||||
|
||||
fn assert_run_data_scanner_signature<F, Fut>(_run: F)
|
||||
where
|
||||
F: Fn(CancellationToken, Arc<ECStore>) -> Fut,
|
||||
@@ -4312,9 +4459,9 @@ mod tests {
|
||||
};
|
||||
|
||||
global_metrics().set_current_scan_mode(HealScanMode::Deep);
|
||||
global_metrics().set_cycle(Some(cycle_info.clone())).await;
|
||||
let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
|
||||
|
||||
mark_scan_cycle_idle(&mut cycle_info).await;
|
||||
mark_scan_cycle_idle(&mut cycle_info, &mut cycle_metrics_guard).await;
|
||||
|
||||
let published = global_metrics()
|
||||
.get_cycle()
|
||||
@@ -4330,6 +4477,123 @@ mod tests {
|
||||
global_metrics().set_cycle(None).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scanner_cycle_metrics_guard_covers_published_first_cycle_lifetime() {
|
||||
let cycle_started = Utc::now() - chrono::Duration::seconds(5);
|
||||
let mut cycle_info = CurrentCycle {
|
||||
current: 0,
|
||||
next: 1,
|
||||
started: cycle_started,
|
||||
..Default::default()
|
||||
};
|
||||
let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
|
||||
let setup_report = global_metrics().report().await;
|
||||
assert!(setup_report.current_cycle_active);
|
||||
assert_eq!(setup_report.current_cycle, 0);
|
||||
assert_eq!(setup_report.current_started, cycle_started);
|
||||
|
||||
mark_scan_cycle_idle(&mut cycle_info, &mut guard).await;
|
||||
let idle_report = global_metrics().report().await;
|
||||
assert!(!idle_report.current_cycle_active);
|
||||
|
||||
global_metrics().set_cycle(None).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scanner_cycle_metrics_guard_keeps_active_cycle_published_during_finalization() {
|
||||
let mut cycle_info = CurrentCycle {
|
||||
current: 12,
|
||||
next: 13,
|
||||
started: Utc::now(),
|
||||
..Default::default()
|
||||
};
|
||||
let mut guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
|
||||
|
||||
cycle_info.current = 0;
|
||||
tokio::task::yield_now().await;
|
||||
let finalizing_report = global_metrics().report().await;
|
||||
assert!(finalizing_report.current_cycle_active);
|
||||
assert_eq!(finalizing_report.current_cycle, 12);
|
||||
|
||||
guard.finish(cycle_info).await;
|
||||
let idle_report = global_metrics().report().await;
|
||||
assert!(!idle_report.current_cycle_active);
|
||||
assert_eq!(idle_report.current_cycle, 0);
|
||||
|
||||
global_metrics().set_cycle(None).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn scanner_cycle_metrics_guard_drop_clears_activity() {
|
||||
let guard = ScannerCycleMetricsGuard::new(CurrentCycle {
|
||||
current: 12,
|
||||
next: 13,
|
||||
started: Utc::now(),
|
||||
..Default::default()
|
||||
})
|
||||
.await;
|
||||
assert!(global_metrics().report().await.current_cycle_active);
|
||||
|
||||
drop(guard);
|
||||
|
||||
assert!(!global_metrics().report().await.current_cycle_active);
|
||||
global_metrics().set_cycle(None).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn run_data_scanner_cycle_publishes_activity_for_owner_lifetime() {
|
||||
let (_temp_dir, store) = setup_scanner_cycle_store().await;
|
||||
let ctx = CancellationToken::new();
|
||||
let mut cycle_info = CurrentCycle::default();
|
||||
let mut revision = DataUsageCacheRevision::Missing;
|
||||
let leader_epoch = u64::MAX - 1;
|
||||
let state_persist_reached = Arc::new(Notify::new());
|
||||
let _state_persist_hook = set_scanner_cycle_state_persist_test_hook(leader_epoch, state_persist_reached.clone());
|
||||
let state_lock = store
|
||||
.new_ns_lock(RUSTFS_META_BUCKET, DATA_USAGE_BLOOM_NAME_PATH.as_str())
|
||||
.await
|
||||
.expect("scanner cycle state lock should be created");
|
||||
let state_guard = state_lock
|
||||
.get_write_lock(Duration::from_secs(1))
|
||||
.await
|
||||
.expect("scanner cycle state lock should be acquired");
|
||||
let mut cycle = Box::pin(run_data_scanner_cycle(&ctx, &store, &mut cycle_info, &mut revision, leader_epoch));
|
||||
let waker = std::task::Waker::noop();
|
||||
let mut context = std::task::Context::from_waker(waker);
|
||||
|
||||
assert!(cycle.as_mut().poll(&mut context).is_pending());
|
||||
let active = global_metrics().report().await;
|
||||
assert!(active.current_cycle_active);
|
||||
assert_eq!(active.current_cycle, 0);
|
||||
|
||||
tokio::time::timeout(Duration::from_secs(30), async {
|
||||
tokio::select! {
|
||||
outcome = &mut cycle => panic!("scanner cycle finished before state persistence was released: {outcome:?}"),
|
||||
_ = state_persist_reached.notified() => {}
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("scanner cycle should reach state persistence");
|
||||
let finalizing = global_metrics().report().await;
|
||||
assert!(finalizing.current_cycle_active);
|
||||
|
||||
drop(state_guard);
|
||||
let outcome = tokio::time::timeout(Duration::from_secs(30), cycle)
|
||||
.await
|
||||
.expect("scanner cycle should finish");
|
||||
assert!(matches!(
|
||||
outcome,
|
||||
ScannerCycleOutcome::Completed | ScannerCycleOutcome::CompletedWithPendingMaintenance
|
||||
));
|
||||
assert!(!global_metrics().report().await.current_cycle_active);
|
||||
|
||||
global_metrics().set_cycle(None).await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_finalize_partial_scan_cycle_advances_and_persists_counter() {
|
||||
@@ -4342,8 +4606,11 @@ mod tests {
|
||||
cycle_completed: vec![],
|
||||
started: Utc::now(),
|
||||
};
|
||||
let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
|
||||
|
||||
assert!(finalize_partial_scan_cycle(&ctx, store.clone(), &mut cycle_info, &mut revision, 1).await);
|
||||
assert!(
|
||||
finalize_partial_scan_cycle(&ctx, store.clone(), &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await
|
||||
);
|
||||
|
||||
assert_eq!(cycle_info.next, 13);
|
||||
assert_eq!(cycle_info.current, 0);
|
||||
@@ -4377,8 +4644,20 @@ mod tests {
|
||||
cycle_completed: vec![],
|
||||
started: Utc::now(),
|
||||
};
|
||||
let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
|
||||
|
||||
assert!(persist_required_scanner_cycle_floor(&ctx, store.clone(), &mut cycle_info, &mut revision, 7, 19).await);
|
||||
assert!(
|
||||
persist_required_scanner_cycle_floor(
|
||||
&ctx,
|
||||
store.clone(),
|
||||
&mut cycle_info,
|
||||
&mut revision,
|
||||
7,
|
||||
19,
|
||||
&mut cycle_metrics_guard,
|
||||
)
|
||||
.await
|
||||
);
|
||||
assert_eq!(cycle_info.current, 0);
|
||||
assert_eq!(cycle_info.next, 19);
|
||||
|
||||
@@ -4405,22 +4684,37 @@ mod tests {
|
||||
cycle_completed: vec![],
|
||||
started: Utc::now(),
|
||||
};
|
||||
let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
|
||||
|
||||
assert!(!persist_required_scanner_cycle_floor(&ctx, store.clone(), &mut cycle_info, &mut revision, 7, 12).await);
|
||||
assert_eq!(cycle_info.next, 12);
|
||||
assert_eq!(revision, DataUsageCacheRevision::Missing);
|
||||
assert!(
|
||||
!persist_required_scanner_cycle_floor(
|
||||
&ctx,
|
||||
store.clone(),
|
||||
&mut CurrentCycle {
|
||||
current: 12,
|
||||
next: 12,
|
||||
..Default::default()
|
||||
},
|
||||
&mut cycle_info,
|
||||
&mut revision,
|
||||
7,
|
||||
12,
|
||||
&mut cycle_metrics_guard,
|
||||
)
|
||||
.await
|
||||
);
|
||||
assert_eq!(cycle_info.next, 12);
|
||||
assert_eq!(revision, DataUsageCacheRevision::Missing);
|
||||
let mut max_cycle_info = CurrentCycle {
|
||||
current: 12,
|
||||
next: 12,
|
||||
..Default::default()
|
||||
};
|
||||
let mut max_cycle_metrics_guard = ScannerCycleMetricsGuard::new(max_cycle_info.clone()).await;
|
||||
assert!(
|
||||
!persist_required_scanner_cycle_floor(
|
||||
&ctx,
|
||||
store.clone(),
|
||||
&mut max_cycle_info,
|
||||
&mut revision,
|
||||
7,
|
||||
u64::MAX,
|
||||
&mut max_cycle_metrics_guard,
|
||||
)
|
||||
.await
|
||||
);
|
||||
@@ -4685,8 +4979,9 @@ mod tests {
|
||||
cycle_completed: vec![],
|
||||
started: Utc::now(),
|
||||
};
|
||||
let mut cycle_metrics_guard = ScannerCycleMetricsGuard::new(cycle_info.clone()).await;
|
||||
|
||||
assert!(!finalize_partial_scan_cycle(&ctx, store, &mut cycle_info, &mut revision, 1).await);
|
||||
assert!(!finalize_partial_scan_cycle(&ctx, store, &mut cycle_info, &mut revision, 1, &mut cycle_metrics_guard,).await);
|
||||
assert_eq!(cycle_info.next, 13);
|
||||
assert_eq!(cycle_info.current, 0);
|
||||
assert_eq!(revision, DataUsageCacheRevision::Missing);
|
||||
@@ -5943,10 +6238,10 @@ mod tests {
|
||||
#[tokio::test]
|
||||
async fn data_usage_persist_wait_aborts_when_scanner_is_cancelled() {
|
||||
let ctx = CancellationToken::new();
|
||||
let mut task = tokio::spawn(async {
|
||||
let mut task = AbortOnDropHandle::new(tokio::spawn(async {
|
||||
std::future::pending::<()>().await;
|
||||
DataUsagePersistOutcome::Saved
|
||||
});
|
||||
}));
|
||||
ctx.cancel();
|
||||
|
||||
let result = wait_for_data_usage_persist_task(&ctx, &mut task, Duration::from_secs(60)).await;
|
||||
@@ -5958,10 +6253,10 @@ mod tests {
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn data_usage_persist_wait_aborts_after_timeout() {
|
||||
let ctx = CancellationToken::new();
|
||||
let mut task = tokio::spawn(async {
|
||||
let mut task = AbortOnDropHandle::new(tokio::spawn(async {
|
||||
std::future::pending::<()>().await;
|
||||
DataUsagePersistOutcome::Saved
|
||||
});
|
||||
}));
|
||||
|
||||
let result = wait_for_data_usage_persist_task(&ctx, &mut task, Duration::from_secs(30)).await;
|
||||
|
||||
|
||||
Reference in New Issue
Block a user