diff --git a/.docker/observability/grafana/dashboards/rustfs.json b/.docker/observability/grafana/dashboards/rustfs.json index 5b7f047a8..5a72306c3 100644 --- a/.docker/observability/grafana/dashboards/rustfs.json +++ b/.docker/observability/grafana/dashboards/rustfs.json @@ -11888,7 +11888,24 @@ }, "unit": "ops" }, - "overrides": [] + "overrides": [ + { + "matcher": { + "id": "byFrameRefID", + "options": "B" + }, + "properties": [ + { + "id": "unit", + "value": "short" + }, + { + "id": "custom.axisPlacement", + "value": "right" + } + ] + } + ] }, "gridPos": { "h": 8, @@ -11997,7 +12014,24 @@ }, "unit": "ops" }, - "overrides": [] + "overrides": [ + { + "matcher": { + "id": "byFrameRefID", + "options": "B" + }, + "properties": [ + { + "id": "unit", + "value": "short" + }, + { + "id": "custom.axisPlacement", + "value": "right" + } + ] + } + ] }, "gridPos": { "h": 8, diff --git a/crates/common/src/metrics.rs b/crates/common/src/metrics.rs index 61879b591..68f0bfde5 100644 --- a/crates/common/src/metrics.rs +++ b/crates/common/src/metrics.rs @@ -17,7 +17,7 @@ use crate::last_minute::{AccElem, LastMinuteLatency}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use std::{ - collections::HashMap, + collections::{BTreeSet, HashMap}, fmt::Display, future::Future, pin::Pin, @@ -719,7 +719,7 @@ pub struct ScannerDiskBucketScanSnapshot { pub active: u64, } -#[derive(Clone, Debug, PartialEq, Eq, Hash)] +#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] struct ScannerBucketDriveResultKey { bucket: String, drive: String, @@ -738,6 +738,12 @@ impl ScannerBucketDriveResultKey { const MAX_SCANNER_BUCKET_DRIVE_RESULT_KEYS: usize = 4096; +#[derive(Debug, Default)] +struct ScannerBucketDriveResults { + counts: HashMap, + eviction_index: BTreeSet<(u64, ScannerBucketDriveResultKey)>, +} + #[derive(Clone, Copy, Debug)] struct ScannerBucketDriveResultValue { count: u64, @@ -775,7 +781,7 @@ pub struct Metrics { scanner_set_scans_queued: AtomicU64, scanner_set_scans_active: AtomicU64, scanner_disk_bucket_scan_states: Mutex>, - scanner_bucket_drive_results: Mutex>, + scanner_bucket_drive_results: Mutex, scanner_bucket_drive_result_clock: AtomicU64, current_scan_cycle_bucket_drive_results_start: Mutex>, last_scan_cycle_bucket_drive_results: Mutex>, @@ -1785,7 +1791,7 @@ impl Metrics { scanner_set_scans_queued: AtomicU64::new(0), scanner_set_scans_active: AtomicU64::new(0), scanner_disk_bucket_scan_states: Mutex::new(HashMap::new()), - scanner_bucket_drive_results: Mutex::new(HashMap::new()), + scanner_bucket_drive_results: Mutex::new(ScannerBucketDriveResults::default()), scanner_bucket_drive_result_clock: AtomicU64::new(0), current_scan_cycle_bucket_drive_results_start: Mutex::new(HashMap::new()), last_scan_cycle_bucket_drive_results: Mutex::new(Vec::new()), @@ -2386,31 +2392,28 @@ impl Metrics { .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()); let last_seen = self.scanner_bucket_drive_result_clock.fetch_add(1, Ordering::Relaxed); - if !results.contains_key(&key) - && results.len() >= MAX_SCANNER_BUCKET_DRIVE_RESULT_KEYS - && let Some(stale_key) = results - .iter() - .min_by(|left, right| { - left.1 - .last_seen - .cmp(&right.1.last_seen) - .then_with(|| left.0.bucket.cmp(&right.0.bucket)) - .then_with(|| left.0.drive.cmp(&right.0.drive)) - .then_with(|| left.0.result.cmp(&right.0.result)) - }) - .map(|(key, _)| key.clone()) - { - results.remove(&stale_key); + if let Some(previous_last_seen) = results.counts.get_mut(&key).map(|value| { + let previous_last_seen = value.last_seen; + value.count = value.count.saturating_add(1); + value.last_seen = last_seen; + previous_last_seen + }) { + results.eviction_index.remove(&(previous_last_seen, key.clone())); + results.eviction_index.insert((last_seen, key)); + return; } - match results.entry(key) { - std::collections::hash_map::Entry::Occupied(mut entry) => { - let value = entry.get_mut(); - value.count = value.count.saturating_add(1); - value.last_seen = last_seen; - } - std::collections::hash_map::Entry::Vacant(entry) => { - entry.insert(ScannerBucketDriveResultValue { count: 1, last_seen }); - } + + if results.counts.len() >= MAX_SCANNER_BUCKET_DRIVE_RESULT_KEYS + && let Some((_, stale_key)) = results.eviction_index.pop_first() + { + results.counts.remove(&stale_key); + } + + if results.counts.len() < MAX_SCANNER_BUCKET_DRIVE_RESULT_KEYS { + results + .counts + .insert(key.clone(), ScannerBucketDriveResultValue { count: 1, last_seen }); + results.eviction_index.insert((last_seen, key)); } } @@ -2694,6 +2697,7 @@ impl Metrics { self.scanner_bucket_drive_results .lock() .unwrap_or_else(|poisoned| poisoned.into_inner()) + .counts .iter() .map(|(key, value)| (key.clone(), value.count)) .collect() @@ -4392,6 +4396,60 @@ mod tests { ); } + #[tokio::test] + async fn scanner_bucket_drive_result_eviction_keeps_recent_keys() { + let metrics = Metrics::new(); + for index in 0..MAX_SCANNER_BUCKET_DRIVE_RESULT_KEYS { + metrics.record_scanner_bucket_drive_result(&format!("bucket-{index}"), "/data1", "success"); + } + metrics.record_scanner_bucket_drive_result("bucket-0", "/data1", "success"); + metrics.record_scanner_bucket_drive_result("overflow", "/data1", "success"); + + let report = metrics.scanner_runtime_details_report(); + + assert_eq!(report.bucket_drive_results.len(), MAX_SCANNER_BUCKET_DRIVE_RESULT_KEYS); + assert!( + report + .bucket_drive_results + .iter() + .any(|snapshot| snapshot.bucket == "bucket-0" && snapshot.count == 2) + ); + assert!( + report + .bucket_drive_results + .iter() + .all(|snapshot| snapshot.bucket != "bucket-1") + ); + } + + #[tokio::test] + async fn scanner_bucket_drive_result_eviction_survives_full_refresh() { + let metrics = Metrics::new(); + for index in 0..MAX_SCANNER_BUCKET_DRIVE_RESULT_KEYS { + metrics.record_scanner_bucket_drive_result(&format!("bucket-{index}"), "/data1", "success"); + } + for index in 0..MAX_SCANNER_BUCKET_DRIVE_RESULT_KEYS { + metrics.record_scanner_bucket_drive_result(&format!("bucket-{index}"), "/data1", "success"); + } + metrics.record_scanner_bucket_drive_result("overflow", "/data1", "success"); + + let report = metrics.scanner_runtime_details_report(); + + assert_eq!(report.bucket_drive_results.len(), MAX_SCANNER_BUCKET_DRIVE_RESULT_KEYS); + assert!( + report + .bucket_drive_results + .iter() + .any(|snapshot| snapshot.bucket == "overflow") + ); + assert!( + report + .bucket_drive_results + .iter() + .all(|snapshot| snapshot.bucket != "bucket-0") + ); + } + #[tokio::test] async fn report_includes_usage_freshness_status() { let metrics = Metrics::new(); diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index fb8d43299..415362aff 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -1551,6 +1551,7 @@ impl LocalDiskWrapper { F: FnOnce() -> Fut, Fut: std::future::Future>, { + self.metrics.record_operation_call(op); // Check if disk is faulty if self.health.is_faulty() { self.metrics.record_availability_error(); @@ -1575,7 +1576,6 @@ impl LocalDiskWrapper { self.health.last_started.store(current_unix_nanos(), Ordering::Relaxed); let _waiting_guard = self.health.waiting_guard(); let _metric_waiting_guard = self.metrics.waiting_guard(); - self.metrics.record_operation_call(op); let started = Instant::now(); if timeout_duration == Duration::ZERO { @@ -1704,6 +1704,7 @@ impl DiskAPI for LocalDiskWrapper { async fn disk_info(&self, opts: &DiskInfoOptions) -> Result { if opts.noop && opts.metrics { + self.metrics.record_operation_call("disk_info"); let info = DiskInfo { metrics: self.metrics_snapshot(), ..Default::default() @@ -1716,6 +1717,7 @@ impl DiskAPI for LocalDiskWrapper { } if self.health.is_faulty() { + self.metrics.record_operation_call("disk_info"); self.metrics.record_availability_error(); return Err(DiskError::FaultyDisk); } @@ -1826,6 +1828,7 @@ impl DiskAPI for LocalDiskWrapper { } async fn delete_versions(&self, volume: &str, versions: Vec, opts: DeleteOptions) -> Vec> { + self.metrics.record_operation_call("delete_versions"); // Check if disk is faulty before proceeding if self.health.is_faulty() { self.metrics.record_availability_error(); @@ -1842,7 +1845,6 @@ impl DiskAPI for LocalDiskWrapper { self.health.last_started.store(current_unix_nanos(), Ordering::Relaxed); self.health.increment_waiting(); let metric_waiting_guard = self.metrics.waiting_guard(); - self.metrics.record_operation_call("delete_versions"); let started = Instant::now(); // Execute the operation @@ -2331,7 +2333,79 @@ mod tests { .expect_err("returned availability error should propagate"); assert_eq!(err, DiskError::DiskNotFound); - assert_eq!(wrapper.metrics_snapshot().total_errors_availability, 1); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.api_calls.get("read_all"), Some(&1)); + assert_eq!(snapshot.total_errors_availability, 1); + } + + #[tokio::test] + async fn local_disk_health_wrapper_counts_faulty_precheck_rejections() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + let wrapper = LocalDiskWrapper::new(disk, false); + wrapper.health.force_runtime_state_for_test(RuntimeDriveHealthState::Offline); + let operation_ran = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let operation_ran_in_call = Arc::clone(&operation_ran); + + let err = wrapper + .track_disk_health_with_op( + "read_all", + || async move { + operation_ran_in_call.store(true, Ordering::Relaxed); + Ok(()) + }, + Duration::ZERO, + ) + .await + .expect_err("faulty generic wrapper call should be rejected before operation runs"); + + assert_eq!(err, DiskError::FaultyDisk); + assert!(!operation_ran.load(Ordering::Relaxed)); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.api_calls.get("read_all"), Some(&1)); + assert_eq!(snapshot.total_errors_availability, 1); + } + + #[tokio::test] + async fn local_disk_health_wrapper_counts_stale_precheck_rejections() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = + Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")).expect("endpoint should parse"); + let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created")); + { + let mut format_info = disk.format_info.write().await; + format_info.id = Some(Uuid::new_v4()); + format_info.file_info = Some( + tokio::fs::metadata(dir.path()) + .await + .expect("temp dir metadata should be readable"), + ); + format_info.last_check = Some(::time::OffsetDateTime::now_utc()); + } + let wrapper = LocalDiskWrapper::new(disk, false); + wrapper.set_disk_id_state(Some(Uuid::new_v4())).await; + let operation_ran = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let operation_ran_in_call = Arc::clone(&operation_ran); + + let err = wrapper + .track_disk_health_with_op( + "write_all", + || async move { + operation_ran_in_call.store(true, Ordering::Relaxed); + Ok(()) + }, + Duration::ZERO, + ) + .await + .expect_err("stale generic wrapper call should be rejected before operation runs"); + + assert_eq!(err, DiskError::DiskNotFound); + assert!(!operation_ran.load(Ordering::Relaxed)); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.api_calls.get("write_all"), Some(&1)); + assert_eq!(snapshot.total_errors_availability, 1); } #[tokio::test] @@ -2392,7 +2466,9 @@ mod tests { assert_eq!(result.len(), 1); assert!(matches!(result.first(), Some(Some(DiskError::FaultyDisk)))); - assert_eq!(wrapper.metrics_snapshot().total_errors_availability, 1); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.api_calls.get("delete_versions"), Some(&1)); + assert_eq!(snapshot.total_errors_availability, 1); } #[tokio::test] @@ -2414,7 +2490,9 @@ mod tests { .expect_err("faulty disk_info should be rejected"); assert_eq!(err, DiskError::FaultyDisk); - assert_eq!(wrapper.metrics_snapshot().total_errors_availability, 1); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.api_calls.get("disk_info"), Some(&1)); + assert_eq!(snapshot.total_errors_availability, 1); } #[tokio::test] @@ -2442,7 +2520,9 @@ mod tests { assert_eq!(result.len(), 1); assert!(matches!(result.first(), Some(Some(DiskError::DiskNotFound)))); - assert_eq!(wrapper.metrics_snapshot().total_errors_availability, 1); + let snapshot = wrapper.metrics_snapshot(); + assert_eq!(snapshot.api_calls.get("delete_versions"), Some(&1)); + assert_eq!(snapshot.total_errors_availability, 1); } impl AsyncWrite for PendingWriter { diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 8d9909a17..c51b1eadf 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -149,6 +149,13 @@ impl Disk { Disk::Remote(_) => None, } } + + pub(crate) async fn cached_disk_id(&self) -> Option { + match self { + Disk::Local(local_disk) => local_disk.get_current_disk_id().await, + Disk::Remote(remote_disk) => remote_disk.get_disk_id().await.ok().flatten(), + } + } } #[async_trait::async_trait] diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 3356cad93..3581a8cba 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -4358,6 +4358,7 @@ async fn get_disks_info(disks: &[Option], eps: &[Endpoint]) -> Vec], eps: &[Endpoint]) -> Vec], eps: &[Endpoint]) -> Vec bool { mod tests { use super::*; use crate::bucket::replication::{replication_statuses_map, version_purge_statuses_map}; + use crate::cluster::rpc::{RemoteDisk, TcpHttpInternodeDataTransport}; use crate::disk::CHECK_PART_UNKNOWN; use crate::disk::CHECK_PART_VOLUME_NOT_FOUND; use crate::disk::DataDirDeleteStatus; @@ -4998,6 +5002,26 @@ mod tests { (dir, endpoint, disk) } + async fn make_remote_disk_for_info_test(disk_idx: usize) -> (Endpoint, DiskStore) { + let endpoint_url = format!("http://remote-server:9000/data{disk_idx}"); + let mut endpoint = Endpoint::try_from(endpoint_url.as_str()).expect("remote endpoint should parse"); + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(disk_idx); + let remote_disk = RemoteDisk::new( + &endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + Arc::new(TcpHttpInternodeDataTransport), + ) + .await + .expect("remote disk should be created"); + + (endpoint, Arc::new(disk::Disk::Remote(Box::new(remote_disk)))) + } + #[tokio::test] async fn test_rename_data_quorum_failure_rolls_back_destination_object() { let dir = tempfile::tempdir().expect("tempdir should be created"); @@ -7707,6 +7731,13 @@ mod tests { .as_ref() .expect("disk 1 should exist") .force_runtime_state_for_test(RuntimeDriveHealthState::Suspect); + let offline_disk_id = Uuid::new_v4(); + disks[2] + .as_ref() + .expect("disk 2 should exist") + .set_disk_id_state(Some(offline_disk_id)) + .await + .expect("offline disk id should be cached"); disks[2] .as_ref() .expect("disk 2 should exist") @@ -7750,12 +7781,52 @@ mod tests { endpoints[2].get_file_path(), "offline disk should keep stable endpoint path" ); + assert_eq!(info[2].uuid, offline_disk_id.to_string()); assert!( info[2].metrics.is_some(), "offline runtime fallback should preserve disk metrics snapshot" ); } + #[tokio::test] + async fn test_get_disks_info_preserves_remote_cached_disk_id_when_offline() { + let (endpoint, disk) = make_remote_disk_for_info_test(0).await; + let remote_disk_id = Uuid::new_v4(); + disk.set_disk_id_state(Some(remote_disk_id)) + .await + .expect("remote disk id should be cached"); + disk.force_runtime_state_for_test(RuntimeDriveHealthState::Offline); + + let info = get_disks_info(&[Some(disk)], &[endpoint]).await; + + assert_eq!(info.len(), 1); + assert_eq!(info[0].state, "offline"); + assert_eq!(info[0].runtime_state.as_deref(), Some("offline")); + assert_eq!(info[0].uuid, remote_disk_id.to_string()); + } + + #[tokio::test] + async fn test_get_disks_info_preserves_cached_disk_id_after_failed_live_probe() { + let format = FormatV3::new(1, 1); + let (temp_dir, endpoint, disk) = make_formatted_local_disk_for_info_test(0, &format).await; + let cached_disk_id = Uuid::new_v4(); + disk.set_disk_id_state(Some(cached_disk_id)) + .await + .expect("disk id should be cached before the failed probe"); + disk.force_runtime_state_for_test(RuntimeDriveHealthState::Suspect); + + let info = get_disks_info(&[Some(disk)], &[endpoint]).await; + + assert_eq!(info.len(), 1); + assert_eq!(info[0].runtime_state.as_deref(), Some("suspect")); + assert_eq!(info[0].uuid, cached_disk_id.to_string()); + assert_eq!( + info[0].drive_path, + temp_dir.path().to_string_lossy(), + "failed live probe should still keep the endpoint path" + ); + } + #[tokio::test] async fn test_get_disks_info_uses_capacity_snapshot_for_offline_disk() { let format = FormatV3::new(1, 1); diff --git a/rustfs/src/app/multipart_usecase.rs b/rustfs/src/app/multipart_usecase.rs index 0a140f272..1b773f1d1 100644 --- a/rustfs/src/app/multipart_usecase.rs +++ b/rustfs/src/app/multipart_usecase.rs @@ -1756,7 +1756,6 @@ mod tests { } #[tokio::test] - #[ignore = "requires isolated global object layer state"] async fn execute_abort_multipart_upload_returns_internal_error_when_store_uninitialized() { let input = AbortMultipartUploadInput::builder() .bucket("bucket".to_string())