diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 5da2c65ab..f209faf5b 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -15,6 +15,7 @@ use axum::http::{HeaderMap, Method, Request, StatusCode, header}; use axum::response::{IntoResponse, Response}; use serde::{Deserialize, Serialize}; +use super::absolute_display_path; use super::grpc_client::{GRPC_MAX_MESSAGE_SIZE, build_grpc_endpoint}; use super::volume_server::{VolumeServerState, normalize_outgoing_http_url, to_http_address}; use crate::config::ReadMode; @@ -3621,16 +3622,6 @@ async fn read_remote_chunk_needle( // Helpers // ============================================================================ -fn absolute_display_path(path: &str) -> String { - let p = std::path::Path::new(path); - if p.is_absolute() { - return path.to_string(); - } - std::env::current_dir() - .map(|cwd| cwd.join(p).to_string_lossy().to_string()) - .unwrap_or_else(|_| path.to_string()) -} - fn build_disk_statuses(store: &crate::storage::store::Store) -> Vec { let mut disk_statuses = Vec::new(); for loc in &store.locations { diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 8a5879697..eb19ec78e 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -24,7 +24,6 @@ use crate::storage::volume_report::VolumeReportKey; use crate::storage::volume_report_hash::report_hash; const DUPLICATE_UUID_RETRY_MESSAGE: &str = "duplicate UUIDs detected, retrying connection"; -const VOLUME_IO_ERROR_TOLERANCE: i32 = 3; const MAX_DUPLICATE_UUID_RETRIES: u32 = 3; /// Configuration for the heartbeat client. @@ -952,8 +951,8 @@ fn build_heartbeat_with_ec_status( let volume_size = vol.dat_file_size().unwrap_or(0); let mut should_delete_volume = false; - let (_, io_count, io_quarantined) = vol.get_io_error_state(); - if io_quarantined || io_count >= VOLUME_IO_ERROR_TOLERANCE { + if vol.should_quarantine() { + let (_, io_count, io_quarantined) = vol.get_io_error_state(); if !io_quarantined { vol.mark_io_quarantined(); warn!( diff --git a/seaweed-volume/src/server/mod.rs b/seaweed-volume/src/server/mod.rs index 16877f064..de34ebe71 100644 --- a/seaweed-volume/src/server/mod.rs +++ b/seaweed-volume/src/server/mod.rs @@ -38,6 +38,19 @@ pub fn status_with_context(context: &str, err: VolumeError) -> Status { Status::new(status.code(), format!("{context}: {}", status.message())) } +/// Render a configured disk directory as an absolute path for display, so the +/// status JSON and the UI show the same thing for a relative `-dir`. Falls +/// back to the configured spelling when the current directory cannot be read. +pub(crate) fn absolute_display_path(path: &str) -> String { + let p = std::path::Path::new(path); + if p.is_absolute() { + return path.to_string(); + } + std::env::current_dir() + .map(|cwd| cwd.join(p).to_string_lossy().to_string()) + .unwrap_or_else(|_| path.to_string()) +} + #[cfg(test)] mod tests { use super::*; diff --git a/seaweed-volume/src/server/ui.rs b/seaweed-volume/src/server/ui.rs index 9a0db81e6..451a49249 100644 --- a/seaweed-volume/src/server/ui.rs +++ b/seaweed-volume/src/server/ui.rs @@ -1,5 +1,6 @@ use std::fmt::Write as _; +use crate::server::absolute_display_path; use crate::server::server_stats; use crate::server::volume_server::VolumeServerState; use crate::storage::store::Store; @@ -450,16 +451,6 @@ fn collect_ui_data( (disk_rows, volumes, remote_volumes, ec_volumes) } -fn absolute_display_path(path: &str) -> String { - let p = std::path::Path::new(path); - if p.is_absolute() { - return path.to_string(); - } - std::env::current_dir() - .map(|cwd| cwd.join(p).to_string_lossy().to_string()) - .unwrap_or_else(|_| path.to_string()) -} - fn join_i64(values: &[i64]) -> String { values .iter() diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index 50a2dccdc..2d19be1f8 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -13,13 +13,11 @@ use crate::pb::master_pb; use crate::storage::erasure_coding::ec_locate; use crate::storage::erasure_coding::ec_shard::*; use crate::storage::io::read_exact_at; +use crate::storage::io_error::IoErrorTracker; use crate::storage::needle::needle::{Needle, NeedleError, get_actual_size}; use crate::storage::types::*; use crate::storage::volume_open::open_volume_file; -/// An erasure-coded volume managing its local shards and index. -pub const IO_ERROR_TOLERANCE: i32 = 3; - /// The shard-location cache: where each shard lives, when that was last learned /// from the master, and whether a read has since disproved it. One struct behind /// one lock, because the freshness heuristic judges all three together (the map @@ -46,6 +44,7 @@ pub(crate) struct ShardLocationCache { stale: bool, } +/// An erasure-coded volume managing its local shards and index. pub struct EcVolume { pub volume_id: VolumeId, pub collection: String, @@ -104,9 +103,9 @@ pub struct EcVolume { /// sidecar other than the one those two fields actually hold. pub(crate) bitrot_source_dir: String, - io_error_count: std::sync::atomic::AtomicI32, - io_error_quarantined: std::sync::atomic::AtomicBool, - last_io_error: std::sync::Mutex>, + /// Consecutive storage-media errors and the quarantine they lead to, + /// for EC volume health monitoring. + io_errors: IoErrorTracker, } /// Locate the `.vif` for a (collection, vid) by preferring the data dir @@ -459,9 +458,7 @@ impl EcVolume { bitrot: None, bitrot_status: crate::storage::erasure_coding::ec_bitrot::BitrotStatus::Off, bitrot_source_dir: String::new(), - io_error_count: std::sync::atomic::AtomicI32::new(0), - io_error_quarantined: std::sync::atomic::AtomicBool::new(false), - last_io_error: std::sync::Mutex::new(None), + io_errors: IoErrorTracker::default(), }; // Open .ecx file (sorted index) in read/write mode for in-place deletion marking. @@ -1006,48 +1003,26 @@ impl EcVolume { refresh } - // ---- I/O error tracking (mirrors Go's EcVolume IoErrorTracker) ---- + // ---- I/O error tracking ---- pub fn check_read_write_error(&self, err: Option<&io::Error>) { - use std::sync::atomic::Ordering; - if let Some(e) = err - && crate::storage::volume::is_storage_io_error(e) - { - self.io_error_count.fetch_add(1, Ordering::Relaxed); - if let Ok(mut guard) = self.last_io_error.lock() { - *guard = Some(e.to_string()); - } - crate::metrics::STORAGE_IO_ERROR_COUNTER.inc(); - return; - } - self.io_error_count.store(0, Ordering::Relaxed); - if let Ok(mut guard) = self.last_io_error.lock() - && guard.is_some() - { - *guard = None; - } + self.io_errors.check_read_write_error(err); } pub fn get_io_error_state(&self) -> (Option, i32, bool) { - use std::sync::atomic::Ordering; - let err = self.last_io_error.lock().ok().and_then(|g| g.clone()); - let count = self.io_error_count.load(Ordering::Relaxed); - let quarantined = self.io_error_quarantined.load(Ordering::Relaxed); - (err, count, quarantined) + self.io_errors.get_io_error_state() + } + + pub fn should_quarantine(&self) -> bool { + self.io_errors.should_quarantine() } pub fn mark_io_quarantined(&self) { - self.io_error_quarantined - .store(true, std::sync::atomic::Ordering::Relaxed); + self.io_errors.mark_io_quarantined(); } pub fn reset_io_error_state(&self) { - use std::sync::atomic::Ordering; - self.io_error_count.store(0, Ordering::Relaxed); - self.io_error_quarantined.store(false, Ordering::Relaxed); - if let Ok(mut guard) = self.last_io_error.lock() { - *guard = None; - } + self.io_errors.reset_io_error_state(); } // ---- Index operations ---- diff --git a/seaweed-volume/src/storage/io_error.rs b/seaweed-volume/src/storage/io_error.rs new file mode 100644 index 000000000..d1cf03c88 --- /dev/null +++ b/seaweed-volume/src/storage/io_error.rs @@ -0,0 +1,202 @@ +//! Consecutive storage-media error tracking shared by `Volume` and +//! `EcVolume`. Mirrors Go's `weed/storage/io_error.go`. + +use std::io; +use std::sync::Mutex; +use std::sync::atomic::{AtomicBool, AtomicI32, Ordering}; + +/// Consecutive storage-media errors allowed before the volume is quarantined. +pub(crate) const IO_ERROR_TOLERANCE: i32 = 3; + +/// Returns true for I/O errors that indicate faulty storage media, not +/// transient/network failures. On Unix this is EIO; on Windows it covers +/// ERROR_CRC and ERROR_IO_DEVICE, which the kernel returns for failing disks. +pub(crate) fn is_storage_io_error(e: &io::Error) -> bool { + #[cfg(unix)] + { + e.raw_os_error() == Some(libc::EIO) + } + #[cfg(windows)] + { + const ERROR_CRC: i32 = 23; + const ERROR_IO_DEVICE: i32 = 1117; + return e.raw_os_error() == Some(ERROR_CRC) || e.raw_os_error() == Some(ERROR_IO_DEVICE); + } + #[cfg(not(any(unix, windows)))] + { + false + } +} + +/// Consecutive storage-media error state for one volume. `quarantined` is +/// sticky: once set it survives later successful I/O and is lifted only by +/// `reset_io_error_state`. +#[derive(Default)] +pub(crate) struct IoErrorTracker { + last: Mutex>, + count: AtomicI32, + quarantined: AtomicBool, +} + +impl IoErrorTracker { + /// `Some(e)` records a failure, `None` a success. Only storage-media + /// failures count; every other outcome clears the count and last error. + pub(crate) fn check_read_write_error(&self, err: Option<&io::Error>) { + if let Some(e) = err + && is_storage_io_error(e) + { + self.count.fetch_add(1, Ordering::Relaxed); + if let Ok(mut guard) = self.last.lock() { + *guard = Some(e.to_string()); + } + crate::metrics::STORAGE_IO_ERROR_COUNTER.inc(); + return; + } + self.count.store(0, Ordering::Relaxed); + if let Ok(mut guard) = self.last.lock() + && guard.is_some() + { + *guard = None; + } + } + + /// The last recorded error, the consecutive count, and the quarantine flag. + pub(crate) fn get_io_error_state(&self) -> (Option, i32, bool) { + let err = self.last.lock().ok().and_then(|g| g.clone()); + let count = self.count.load(Ordering::Relaxed); + let quarantined = self.quarantined.load(Ordering::Relaxed); + (err, count, quarantined) + } + + pub(crate) fn should_quarantine(&self) -> bool { + self.quarantined.load(Ordering::Relaxed) + || self.count.load(Ordering::Relaxed) >= IO_ERROR_TOLERANCE + } + + pub(crate) fn mark_io_quarantined(&self) { + self.quarantined.store(true, Ordering::Relaxed); + } + + pub(crate) fn reset_io_error_state(&self) { + self.count.store(0, Ordering::Relaxed); + self.quarantined.store(false, Ordering::Relaxed); + if let Ok(mut guard) = self.last.lock() { + *guard = None; + } + } + + #[cfg(test)] + pub(crate) fn set_last_io_error_for_test(&self, err: Option<&str>) { + if let Ok(mut guard) = self.last.lock() { + *guard = err.map(|value| value.to_string()); + } + if err.is_some() { + self.count.store(IO_ERROR_TOLERANCE, Ordering::Relaxed); + } else { + self.count.store(0, Ordering::Relaxed); + } + } +} + +// The tracker only reacts to errors `is_storage_io_error` recognises, which is +// nothing at all on a platform that is neither Unix nor Windows. +#[cfg(all(test, any(unix, windows)))] +mod tests { + use super::*; + + /// An OS error the platform reports for failing storage media. + #[cfg(unix)] + fn media_error() -> io::Error { + io::Error::from_raw_os_error(libc::EIO) + } + + /// An OS error the platform reports for failing storage media. + #[cfg(windows)] + fn media_error() -> io::Error { + const ERROR_IO_DEVICE: i32 = 1117; + io::Error::from_raw_os_error(ERROR_IO_DEVICE) + } + + #[test] + fn check_read_write_error_counts_consecutive_media_errors() { + let tracker = IoErrorTracker::default(); + tracker.check_read_write_error(Some(&media_error())); + tracker.check_read_write_error(Some(&media_error())); + + let (last, count, quarantined) = tracker.get_io_error_state(); + assert_eq!(last, Some(media_error().to_string())); + assert_eq!(count, 2); + assert!(!quarantined); + } + + #[test] + fn success_clears_the_count_and_the_last_error() { + let tracker = IoErrorTracker::default(); + tracker.check_read_write_error(Some(&media_error())); + tracker.check_read_write_error(None); + + assert_eq!(tracker.get_io_error_state(), (None, 0, false)); + } + + #[test] + fn non_media_error_clears_the_count() { + let tracker = IoErrorTracker::default(); + tracker.check_read_write_error(Some(&media_error())); + tracker.check_read_write_error(Some(&io::Error::new( + io::ErrorKind::NotFound, + "no such file", + ))); + + assert_eq!(tracker.get_io_error_state(), (None, 0, false)); + } + + #[test] + fn should_quarantine_only_once_the_tolerance_is_reached() { + let tracker = IoErrorTracker::default(); + for _ in 1..IO_ERROR_TOLERANCE { + tracker.check_read_write_error(Some(&media_error())); + assert!(!tracker.should_quarantine()); + } + tracker.check_read_write_error(Some(&media_error())); + assert!(tracker.should_quarantine()); + } + + #[test] + fn quarantine_survives_later_successful_io() { + let tracker = IoErrorTracker::default(); + tracker.mark_io_quarantined(); + tracker.check_read_write_error(None); + + assert_eq!(tracker.get_io_error_state(), (None, 0, true)); + assert!(tracker.should_quarantine()); + } + + #[test] + fn reset_io_error_state_lifts_the_quarantine() { + let tracker = IoErrorTracker::default(); + tracker.check_read_write_error(Some(&media_error())); + tracker.mark_io_quarantined(); + tracker.reset_io_error_state(); + + assert_eq!(tracker.get_io_error_state(), (None, 0, false)); + assert!(!tracker.should_quarantine()); + } + + #[test] + fn test_helper_arms_a_sustained_error() { + let tracker = IoErrorTracker::default(); + tracker.set_last_io_error_for_test(Some("input/output error")); + assert!(tracker.should_quarantine()); + assert_eq!( + tracker.get_io_error_state(), + ( + Some("input/output error".to_string()), + IO_ERROR_TOLERANCE, + false + ) + ); + + tracker.set_last_io_error_for_test(None); + assert_eq!(tracker.get_io_error_state(), (None, 0, false)); + } +} diff --git a/seaweed-volume/src/storage/mod.rs b/seaweed-volume/src/storage/mod.rs index 2a62e55f4..38220cc62 100644 --- a/seaweed-volume/src/storage/mod.rs +++ b/seaweed-volume/src/storage/mod.rs @@ -2,6 +2,7 @@ pub mod disk_location; pub mod erasure_coding; pub mod idx; pub(crate) mod io; +pub(crate) mod io_error; pub mod needle; pub mod needle_map; pub mod store; diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index df2fa3403..ee5edd638 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -1144,16 +1144,11 @@ impl Store { for (vid, ec_vol) in loc.ec_volumes() { if ec_vol.is_time_to_destroy() { expired_vids.push(*vid); + } else if ec_vol.should_quarantine() { + io_quarantined_vids.push(*vid); } else { - let (_, io_count, quarantined) = ec_vol.get_io_error_state(); - if quarantined - || io_count >= crate::storage::erasure_coding::ec_volume::IO_ERROR_TOLERANCE - { - io_quarantined_vids.push(*vid); - } else { - ec_shards - .extend(ec_vol.to_volume_ec_shard_information_messages(disk_id as u32)); - } + ec_shards + .extend(ec_vol.to_volume_ec_shard_information_messages(disk_id as u32)); } } diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 7e0f811a5..ae3127f22 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -22,6 +22,7 @@ use tracing::{debug, error, info, warn}; use crate::storage::idx; use crate::storage::io::read_exact_at; +use crate::storage::io_error::IoErrorTracker; use crate::storage::needle::needle::{self, Needle, NeedleError, get_actual_size}; use crate::storage::needle_map::sorted_file::SortedFileNeedleMap; use crate::storage::needle_map::{CompactNeedleMap, NeedleMap, NeedleMapKind, RedbNeedleMap}; @@ -112,26 +113,6 @@ fn exceeds_expected_compacted_size(expected_live_bytes: u64, dst_dat_size: u64) expected_live_bytes > dst_dat_size } -/// Returns true for I/O errors that indicate faulty storage media, not -/// transient/network failures. On Unix this is EIO; on Windows it covers -/// ERROR_CRC and ERROR_IO_DEVICE, which the kernel returns for failing disks. -pub fn is_storage_io_error(e: &io::Error) -> bool { - #[cfg(unix)] - { - e.raw_os_error() == Some(libc::EIO) - } - #[cfg(windows)] - { - const ERROR_CRC: i32 = 23; - const ERROR_IO_DEVICE: i32 = 1117; - return e.raw_os_error() == Some(ERROR_CRC) || e.raw_os_error() == Some(ERROR_IO_DEVICE); - } - #[cfg(not(any(unix, windows)))] - { - false - } -} - // ============================================================================ // VolumeInfo (.vif persistence) // ============================================================================ @@ -710,14 +691,9 @@ pub struct Volume { /// Compaction speed limit in bytes per second (0 = unlimited). pub compaction_byte_per_second: i64, - /// Tracks the last I/O error (EIO) for volume health monitoring. - /// Uses Mutex for interior mutability so reads (&self) can clear/set it. - last_io_error: Mutex>, - /// Consecutive EIO count; reset on success or non-EIO errors. - io_error_count: std::sync::atomic::AtomicI32, - /// Sticky quarantine flag set after sustained EIO; cleared only by - /// explicit recovery (mirrors Go's markIoQuarantined). - io_error_quarantined: std::sync::atomic::AtomicBool, + /// Consecutive storage-media errors and the quarantine they lead to, + /// for volume health monitoring. + io_errors: IoErrorTracker, /// Protobuf VolumeInfo for tiered storage (.vif file). /// @@ -799,9 +775,7 @@ impl Volume { last_compact_revision: 0, is_compacting: false, compaction_byte_per_second: 0, - last_io_error: Mutex::new(None), - io_error_count: std::sync::atomic::AtomicI32::new(0), - io_error_quarantined: std::sync::atomic::AtomicBool::new(false), + io_errors: IoErrorTracker::default(), volume_info: PbVolumeInfo::default(), }; @@ -838,9 +812,7 @@ impl Volume { last_compact_revision: 0, is_compacting: false, compaction_byte_per_second: 0, - last_io_error: Mutex::new(None), - io_error_count: std::sync::atomic::AtomicI32::new(0), - io_error_quarantined: std::sync::atomic::AtomicBool::new(false), + io_errors: IoErrorTracker::default(), volume_info: PbVolumeInfo::default(), } } @@ -4400,69 +4372,30 @@ impl Volume { self.dir != self.dir_idx && has_ecx(&volume_file_name(&self.dir, &self.collection, self.id)) } - /// Check if an I/O error is a storage-media failure and record it for - /// health monitoring. On success (None), clears any previously recorded - /// EIO error. Matches Go's `checkReadWriteError` in volume_write.go. + /// Matches Go's `checkReadWriteError` in `weed/storage/io_error.go`. fn check_read_write_error(&self, err: Option<&io::Error>) { - use std::sync::atomic::Ordering; - if let Some(e) = err - && is_storage_io_error(e) - { - self.io_error_count.fetch_add(1, Ordering::Relaxed); - if let Ok(mut guard) = self.last_io_error.lock() { - *guard = Some(e.to_string()); - } - crate::metrics::STORAGE_IO_ERROR_COUNTER.inc(); - return; - } - self.io_error_count.store(0, Ordering::Relaxed); - if let Ok(mut guard) = self.last_io_error.lock() - && guard.is_some() - { - *guard = None; - } - } - - /// Returns the last recorded I/O error string, if any. - pub fn last_io_error(&self) -> Option { - self.last_io_error.lock().ok()?.clone() + self.io_errors.check_read_write_error(err); } pub fn get_io_error_state(&self) -> (Option, i32, bool) { - use std::sync::atomic::Ordering; - let err = self.last_io_error.lock().ok().and_then(|g| g.clone()); - let count = self.io_error_count.load(Ordering::Relaxed); - let quarantined = self.io_error_quarantined.load(Ordering::Relaxed); - (err, count, quarantined) + self.io_errors.get_io_error_state() + } + + pub fn should_quarantine(&self) -> bool { + self.io_errors.should_quarantine() } pub fn mark_io_quarantined(&self) { - self.io_error_quarantined - .store(true, std::sync::atomic::Ordering::Relaxed); + self.io_errors.mark_io_quarantined(); } pub fn reset_io_error_state(&self) { - use std::sync::atomic::Ordering; - self.io_error_count.store(0, Ordering::Relaxed); - self.io_error_quarantined.store(false, Ordering::Relaxed); - if let Ok(mut guard) = self.last_io_error.lock() { - *guard = None; - } + self.io_errors.reset_io_error_state(); } #[cfg(test)] pub(crate) fn set_last_io_error_for_test(&self, err: Option<&str>) { - use std::sync::atomic::Ordering; - if let Ok(mut guard) = self.last_io_error.lock() { - *guard = err.map(|value| value.to_string()); - } - // Set count at/above the heartbeat tolerance (3) so the test - // helper reflects a sustained error, not a single transient one. - if err.is_some() { - self.io_error_count.store(3, Ordering::Relaxed); - } else { - self.io_error_count.store(0, Ordering::Relaxed); - } + self.io_errors.set_last_io_error_for_test(err); } #[cfg(test)]