diff --git a/seaweed-volume/src/metrics.rs b/seaweed-volume/src/metrics.rs index dadc79682..30f53c4d5 100644 --- a/seaweed-volume/src/metrics.rs +++ b/seaweed-volume/src/metrics.rs @@ -3,8 +3,8 @@ //! Mirrors the Go SeaweedFS volume server metrics. use prometheus::{ - self, Encoder, GaugeVec, HistogramOpts, HistogramVec, IntCounterVec, IntGauge, IntGaugeVec, - Opts, Registry, TextEncoder, + self, Encoder, GaugeVec, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge, + IntGaugeVec, Opts, Registry, TextEncoder, }; use std::sync::Once; @@ -171,6 +171,23 @@ lazy_static::lazy_static! { &["mode"], ).expect("metric can be created"); + /// Counter of storage read/write EIO errors on volumes and EC shards. + /// Mirrors Go's VolumeServerStorageIoErrorCounter. + pub static ref STORAGE_IO_ERROR_COUNTER: IntCounter = IntCounter::new( + "SeaweedFS_volumeServer_storage_io_error_total", + "Counter of storage read/write EIO errors on volumes and EC shards.", + ).expect("metric can be created"); + + /// Number of volumes quarantined due to storage IO errors. + /// Mirrors Go's VolumeServerIoQuarantineGauge. + pub static ref IO_QUARANTINE_GAUGE: IntGaugeVec = IntGaugeVec::new( + Opts::new( + "SeaweedFS_volumeServer_io_quarantine", + "Number of volumes or EC shards quarantined due to storage IO errors.", + ), + &["kind"], + ).expect("metric can be created"); + // ---- Legacy aliases for backward compat with existing code ---- /// Total number of volumes on this server (flat gauge). @@ -283,6 +300,8 @@ pub fn register_metrics() { Box::new(SCRUB_LAST_TIME_SECONDS.clone()), Box::new(SCRUB_VOLUME_FAILURES.clone()), Box::new(SCRUB_SHARD_FAILURES.clone()), + Box::new(STORAGE_IO_ERROR_COUNTER.clone()), + Box::new(IO_QUARANTINE_GAUGE.clone()), // Legacy metrics Box::new(VOLUMES_TOTAL.clone()), Box::new(DISK_SIZE_BYTES.clone()), diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 74a15b680..c0d789b8c 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -3801,9 +3801,14 @@ impl VolumeServer for VolumeGrpcService { while bytes_read < total_size { let chunk_size = std::cmp::min(BUFFER_SIZE_LIMIT, total_size - bytes_read); let mut buf = vec![0u8; chunk_size]; - let n = shard - .read_at(&mut buf, current_offset) - .map_err(|e| Status::internal(e.to_string()))?; + let n = match shard.read_at(&mut buf, current_offset) { + Ok(n) => n, + Err(e) => { + ec_vol.check_read_write_error(Some(&e)); + return Err(Status::internal(e.to_string())); + } + }; + ec_vol.check_read_write_error(None); if n == 0 { break; } diff --git a/seaweed-volume/src/server/handlers.rs b/seaweed-volume/src/server/handlers.rs index 0d20fd1d0..da7513810 100644 --- a/seaweed-volume/src/server/handlers.rs +++ b/seaweed-volume/src/server/handlers.rs @@ -3127,6 +3127,11 @@ pub async fn healthz_handler(State(state): State>) -> Res if !state.is_heartbeating.load(Ordering::Relaxed) { return StatusCode::SERVICE_UNAVAILABLE.into_response(); } + // A server with quarantined local replicas has faulty storage media; + // report degraded so a load balancer can drain it. + if state.store.read().unwrap().has_io_quarantine() { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + } StatusCode::OK.into_response() } diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index f1cf2b775..667fb9807 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -19,11 +19,12 @@ use crate::pb::master_pb::seaweed_client::SeaweedClient; use crate::pb::volume_server_pb; use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig}; use crate::storage::store::Store; -use crate::storage::types::NeedleId; +use crate::storage::types::{NeedleId, VolumeId}; 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. @@ -315,6 +316,10 @@ fn collect_ec_shard_delta_messages( for (disk_id, loc) in store.locations.iter().enumerate() { for (_, ec_vol) in loc.ec_volumes() { + let (_, _, quarantined) = ec_vol.get_io_error_state(); + if quarantined { + continue; + } for shard in ec_vol.shards.iter().flatten() { messages.insert( ( @@ -895,6 +900,7 @@ fn build_heartbeat_with_ec_status( // master can tell whether applying what it was sent leaves it current. // Volumes skipped below -- quarantined, phantom, expired -- are in neither. let mut volume_digest: u64 = 0; + let mut quarantined_volumes: u32 = 0; let (send_full_list, report_generation, report_pass) = store.volume_report.begin(); let mut changed_volumes = Vec::new(); let mut max_file_key = NeedleId(0); @@ -937,6 +943,7 @@ fn build_heartbeat_with_ec_status( loc.disk_free_bytes.load(Ordering::Relaxed); let mut delete_vids = Vec::new(); + let mut quarantine_vids: Vec = Vec::new(); for (_, vol) in loc.iter_volumes() { let cur_max = vol.max_file_key(); if cur_max > max_file_key { @@ -946,9 +953,18 @@ fn build_heartbeat_with_ec_status( let volume_size = vol.dat_file_size().unwrap_or(0); let mut should_delete_volume = false; - if vol.last_io_error().is_some() { - delete_vids.push(vol.id); - should_delete_volume = true; + let (_, io_count, io_quarantined) = vol.get_io_error_state(); + if io_quarantined || io_count >= VOLUME_IO_ERROR_TOLERANCE { + if !io_quarantined { + vol.mark_io_quarantined(); + warn!( + "Volume {} quarantined after {} consecutive IO errors", + vol.id.0, io_count + ); + } + quarantined_volumes += 1; + quarantine_vids.push(vol.id); + continue; } else if !vol.is_expired(volume_size, volume_size_limit) { // Detect phantom volumes: the .dat was unlinked from disk but is still // held open as a deleted FD, so the volume keeps serving and heartbeating @@ -1046,6 +1062,12 @@ fn build_heartbeat_with_ec_status( for vid in delete_vids { let _ = loc.delete_volume(vid, false, false); } + + for vid in quarantine_vids { + if let Some(vol) = loc.find_volume_mut(vid) { + vol.set_no_write_or_delete(true); + } + } } // Update disk size and read-only gauges @@ -1102,6 +1124,23 @@ fn build_heartbeat_with_ec_status( }; let (location_uuids, disk_tags) = collect_location_metadata(store, &disk_max_by_id); + let mut quarantined_ec_shards: u32 = 0; + for loc in &store.locations { + for (_, ec_vol) in loc.ec_volumes() { + let (_, _, quarantined) = ec_vol.get_io_error_state(); + if quarantined { + quarantined_ec_shards += ec_vol.shard_count() as u32; + } + } + } + + crate::metrics::IO_QUARANTINE_GAUGE + .with_label_values(&["volume"]) + .set(quarantined_volumes as i64); + crate::metrics::IO_QUARANTINE_GAUGE + .with_label_values(&["ec_shard"]) + .set(quarantined_ec_shards as i64); + let heartbeat = master_pb::Heartbeat { id: store.id.clone(), ip: config.ip.clone(), @@ -1137,6 +1176,10 @@ fn collect_live_ec_shards( for (disk_id, loc) in store.locations.iter().enumerate() { for (_, ec_vol) in loc.ec_volumes() { + let (_, _, quarantined) = ec_vol.get_io_error_state(); + if quarantined { + continue; + } for message in ec_vol.to_volume_ec_shard_information_messages(disk_id as u32) { if update_metrics { let total_size: u64 = message @@ -1976,8 +2019,13 @@ mod tests { let heartbeat = build_heartbeat(&test_config(), &mut store); + // A sustained IO error quarantines the volume: it stays mounted + // (so healthz can observe the quarantine state) but is not + // advertised to the master. assert!(heartbeat.volumes.is_empty()); - assert!(!store.has_volume(VolumeId(51))); + assert!(store.has_volume(VolumeId(51))); + let (_, volume) = store.find_volume_mut(VolumeId(51)).unwrap(); + assert!(volume.is_no_write_or_delete()); } #[test] diff --git a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs index f46f578d9..19853caec 100644 --- a/seaweed-volume/src/storage/erasure_coding/ec_volume.rs +++ b/seaweed-volume/src/storage/erasure_coding/ec_volume.rs @@ -17,6 +17,8 @@ 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; + pub struct EcVolume { pub volume_id: VolumeId, pub collection: String, @@ -79,6 +81,10 @@ pub struct EcVolume { /// so `bitrot_protection()` can return the `Off`/`Invalid` distinction without /// re-reading, mirroring Go's `EcVolume.bitrotStatus`. pub(crate) bitrot_status: crate::storage::erasure_coding::ec_bitrot::BitrotStatus, + + io_error_count: std::sync::atomic::AtomicI32, + io_error_quarantined: std::sync::atomic::AtomicBool, + last_io_error: std::sync::Mutex>, } /// Locate the `.vif` for a (collection, vid) by preferring the data dir @@ -430,6 +436,9 @@ impl EcVolume { encode_ts_ns, bitrot: None, bitrot_status: crate::storage::erasure_coding::ec_bitrot::BitrotStatus::Off, + 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), }; // Open .ecx file (sorted index) in read/write mode for in-place deletion marking. @@ -953,6 +962,50 @@ impl EcVolume { .unwrap_or_default() } + // ---- I/O error tracking (mirrors Go's EcVolume IoErrorTracker) ---- + + pub fn check_read_write_error(&self, err: Option<&io::Error>) { + use std::sync::atomic::Ordering; + if let Some(e) = err { + if 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() { + if guard.is_some() { + *guard = None; + } + } + } + + 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) + } + + pub fn mark_io_quarantined(&self) { + self.io_error_quarantined + .store(true, std::sync::atomic::Ordering::Relaxed); + } + + 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; + } + } + // ---- Index operations ---- /// Find a needle's offset and size in the sorted .ecx index via binary search. @@ -979,7 +1032,22 @@ impl EcVolume { #[cfg(unix)] { use std::os::unix::fs::FileExt; - ecx_file.read_exact_at(&mut entry_buf, file_offset)?; + if let Err(e) = ecx_file.read_exact_at(&mut entry_buf, file_offset) { + self.check_read_write_error(Some(&e)); + return Err(e); + } + } + #[cfg(not(unix))] + { + use std::io::{Read, Seek, SeekFrom}; + if let Err(e) = ecx_file.seek(SeekFrom::Start(file_offset)) { + self.check_read_write_error(Some(&e)); + return Err(e); + } + if let Err(e) = ecx_file.read_exact(&mut entry_buf) { + self.check_read_write_error(Some(&e)); + return Err(e); + } } let (key, offset, size) = idx_entry_from_bytes(&entry_buf); @@ -989,8 +1057,10 @@ impl EcVolume { // reported with TOMBSTONE_FILE_SIZE even though the .ecx // record itself is untouched. if self.is_needle_deleted(needle_id) { + self.check_read_write_error(None); return Ok(Some((offset, TOMBSTONE_FILE_SIZE))); } + self.check_read_write_error(None); return Ok(Some((offset, size))); } else if key < needle_id { lo = mid + 1; @@ -999,6 +1069,7 @@ impl EcVolume { } } + self.check_read_write_error(None); Ok(None) } diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index dd3a8eddf..8086e7c1b 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -427,6 +427,26 @@ impl Store { false } + /// Reports whether any local volume or EC shard is currently quarantined + /// due to sustained storage-media EIO. Mirrors Go's Store.HasIoQuarantine. + pub fn has_io_quarantine(&self) -> bool { + for loc in &self.locations { + for (_, vol) in loc.iter_volumes() { + let (_, _, quarantined) = vol.get_io_error_state(); + if quarantined { + return true; + } + } + for (_, ec_vol) in loc.ec_volumes() { + let (_, _, quarantined) = ec_vol.get_io_error_state(); + if quarantined { + return true; + } + } + } + false + } + /// Mount a volume from an existing .dat file. pub fn mount_volume( &mut self, @@ -990,12 +1010,19 @@ impl Store { for (disk_id, loc) in self.locations.iter_mut().enumerate() { let mut expired_vids = Vec::new(); + let mut io_quarantined_vids = Vec::new(); for (vid, ec_vol) in loc.ec_volumes() { if ec_vol.is_time_to_destroy() { expired_vids.push(*vid); } else { - ec_shards - .extend(ec_vol.to_volume_ec_shard_information_messages(disk_id as u32)); + 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)); + } } } @@ -1016,6 +1043,20 @@ impl Store { ec_shards.extend(messages); } } + + for vid in io_quarantined_vids { + if let Some(ec_vol) = loc.find_ec_volume(vid) { + let (_, io_count, quarantined) = ec_vol.get_io_error_state(); + if !quarantined { + ec_vol.mark_io_quarantined(); + tracing::warn!( + volume_id = vid.0, + io_count, + "ec volume quarantined after consecutive IO errors" + ); + } + } + } } (ec_shards, deleted) diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 3963d373f..229315986 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -94,6 +94,27 @@ fn is_skippable_needle_read_error(e: &VolumeError) -> bool { } } +/// 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)] + { + return 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) // ============================================================================ @@ -547,6 +568,11 @@ pub struct Volume { /// 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, /// Protobuf VolumeInfo for tiered storage (.vif file). pub volume_info: PbVolumeInfo, @@ -617,6 +643,8 @@ impl Volume { 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), volume_info: PbVolumeInfo::default(), has_remote_file: false, }; @@ -655,6 +683,8 @@ impl Volume { 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), volume_info: PbVolumeInfo::default(), has_remote_file: false, } @@ -1874,7 +1904,14 @@ impl Volume { self.last_modified_ts_seconds = n.last_modified; } - self.maybe_checkpoint_index(fsync); + let checkpoint_ok = self.maybe_checkpoint_index(fsync); + + // Clear the EIO streak only after the full write (data + flush + + // index + checkpoint) succeeds, so a failed fsync or checkpoint + // does not get its EIO erased by the success reset. + if checkpoint_ok { + self.check_read_write_error(None); + } // Return Size(n.DataSize) as the logical size, matching Go's doWriteRequest Ok((offset, Size(n.data_size as i32), false)) @@ -1889,10 +1926,10 @@ impl Volume { /// When `idx_already_synced` is true the .idx has already been fsynced by /// `flush_idx` on the fsync=true write path, so the checkpoint skips its /// own .idx fsync to avoid a redundant one. - fn maybe_checkpoint_index(&mut self, idx_already_synced: bool) { + fn maybe_checkpoint_index(&mut self, idx_already_synced: bool) -> bool { let due = self.nm.as_ref().is_some_and(|nm| nm.checkpoint_due()); if !due { - return; + return true; } if let Err(e) = self.flush_dat() { self.check_read_write_error(Some(&e)); @@ -1901,7 +1938,7 @@ impl Volume { self.id.0, e ); - return; + return false; } let checkpointed = match self.nm.as_mut() { Some(nm) => nm.checkpoint(!idx_already_synced), @@ -1910,7 +1947,9 @@ impl Volume { if let Err(e) = checkpointed { self.check_read_write_error(Some(&e)); tracing::warn!("volume {}: index checkpoint failed: {}", self.id.0, e); + return false; } + true } fn read_needle_header_unlocked(&self, n: &mut Needle, offset: i64) -> Result<(), VolumeError> { @@ -1993,7 +2032,6 @@ impl Volume { self.check_read_write_error(Some(&e)); return Err(VolumeError::Io(e)); } - self.check_read_write_error(None); Ok((offset, n.size, actual_size)) } @@ -2049,7 +2087,13 @@ impl Volume { if let Some(nm) = &mut self.nm { nm.delete(n.id, Offset::from_actual_offset(offset as i64))?; } - self.maybe_checkpoint_index(false); + let checkpoint_ok = self.maybe_checkpoint_index(false); + + // Clear the EIO streak after a successful delete (tombstone append + + // index update + checkpoint), mirroring do_write_request. + if checkpoint_ok { + self.check_read_write_error(None); + } Ok(size) } @@ -2086,6 +2130,10 @@ impl Volume { self.no_write_or_delete } + pub fn set_no_write_or_delete(&mut self, value: bool) { + self.no_write_or_delete = value; + } + pub fn is_no_write_can_delete(&self) -> bool { self.no_write_can_delete } @@ -2844,6 +2892,7 @@ impl Volume { self.no_write_can_delete = was_no_write_can_delete; return Err(e); } + self.reset_io_error_state(); Ok(()) } @@ -4159,23 +4208,25 @@ impl Volume { && has_ecx(&volume_file_name(&self.dir, &self.collection, self.id)) } - /// Check if an I/O error is EIO (errno 5) and record it for health monitoring. - /// On success (None), clears any previously recorded EIO error. - /// Matches Go's `checkReadWriteError` in volume_write.go. + /// 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. fn check_read_write_error(&self, err: Option<&io::Error>) { + use std::sync::atomic::Ordering; if let Some(e) = err { - if e.raw_os_error() == Some(5) { - // EIO — record it + if 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; } - } else { - // Success — clear any previous EIO - if let Ok(mut guard) = self.last_io_error.lock() { - if guard.is_some() { - *guard = None; - } + } + self.io_error_count.store(0, Ordering::Relaxed); + if let Ok(mut guard) = self.last_io_error.lock() { + if guard.is_some() { + *guard = None; } } } @@ -4186,11 +4237,41 @@ impl Volume { self.last_io_error.lock().ok()?.clone() } + 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) + } + + pub fn mark_io_quarantined(&self) { + self.io_error_quarantined + .store(true, std::sync::atomic::Ordering::Relaxed); + } + + 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; + } + } + #[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); + } } #[cfg(test)] diff --git a/weed/server/volume_grpc_erasure_coding.go b/weed/server/volume_grpc_erasure_coding.go index cd520b287..1251045c4 100644 --- a/weed/server/volume_grpc_erasure_coding.go +++ b/weed/server/volume_grpc_erasure_coding.go @@ -933,6 +933,12 @@ func (vs *VolumeServer) VolumeEcShardRead(req *volume_server_pb.VolumeEcShardRea } bytesread, err := ecShard.ReadAt(buffer[0:bufferSize], startOffset) + if err != nil && err != io.EOF { + ecVolume.CheckReadWriteError(err) + } else { + ecVolume.CheckReadWriteError(nil) + } + // println("read", ecShard.FileName(), "startOffset", startOffset, bytesread, "bytes, with target", bufferSize) if bytesread > 0 { diff --git a/weed/server/volume_server_handlers_admin.go b/weed/server/volume_server_handlers_admin.go index dfb90befd..c1eaa5c30 100644 --- a/weed/server/volume_server_handlers_admin.go +++ b/weed/server/volume_server_handlers_admin.go @@ -31,6 +31,13 @@ func (vs *VolumeServer) healthzHandler(w http.ResponseWriter, r *http.Request) { return } + // A server with quarantined local replicas has faulty storage media; + // report degraded so a load balancer can drain it. + if vs.store.HasIoQuarantine() { + w.WriteHeader(http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusOK) } diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index 5846d0adf..0c238c711 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -487,6 +487,22 @@ var ( Help: "Disk error status", }, []string{"name", "type"}) + VolumeServerStorageIoErrorCounter = prometheus.NewCounter( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: subsystemVolumeServer, + Name: "storage_io_error_total", + Help: "Counter of storage read/write EIO errors on volumes and EC shards.", + }) + + VolumeServerIoQuarantineGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: subsystemVolumeServer, + Name: "io_quarantine", + Help: "Number of volumes or EC shards quarantined due to storage IO errors.", + }, []string{"kind"}) + VolumeServerConcurrentDownloadLimit = prometheus.NewGauge( prometheus.GaugeOpts{ Namespace: Namespace, @@ -1016,6 +1032,8 @@ func init() { Gather.MustRegister(VolumeServerDiskSizeGauge) Gather.MustRegister(VolumeServerResourceGauge) Gather.MustRegister(VolumeServerDiskErrorGauge) + Gather.MustRegister(VolumeServerStorageIoErrorCounter) + Gather.MustRegister(VolumeServerIoQuarantineGauge) Gather.MustRegister(VolumeServerConcurrentDownloadLimit) Gather.MustRegister(VolumeServerConcurrentUploadLimit) Gather.MustRegister(VolumeServerInFlightDownloadSize) diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go index e3a8d6e96..77b09d73d 100644 --- a/weed/storage/erasure_coding/ec_volume.go +++ b/weed/storage/erasure_coding/ec_volume.go @@ -6,12 +6,14 @@ import ( "os" "slices" "sync" + "syscall" "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/storage/backend" "github.com/seaweedfs/seaweedfs/weed/storage/idx" "github.com/seaweedfs/seaweedfs/weed/storage/needle" @@ -75,6 +77,58 @@ type EcVolume struct { bitrotLock sync.RWMutex bitrot *volume_server_pb.EcBitrotProtection bitrotStatus BitrotStatus + + lastIoError error + lastIoErrorCount int32 + ioErrorQuarantined bool + lastIoErrorLock sync.RWMutex +} + +func (ev *EcVolume) CheckReadWriteError(err error) { + if err == nil { + ev.clearIoError() + return + } + if errors.Is(err, syscall.EIO) { + ev.noteIoError(err) + return + } + ev.clearIoError() +} + +func (ev *EcVolume) noteIoError(err error) { + ev.lastIoErrorLock.Lock() + defer ev.lastIoErrorLock.Unlock() + ev.lastIoError = err + ev.lastIoErrorCount++ + stats.VolumeServerStorageIoErrorCounter.Inc() +} + +func (ev *EcVolume) clearIoError() { + ev.lastIoErrorLock.Lock() + defer ev.lastIoErrorLock.Unlock() + ev.lastIoError = nil + ev.lastIoErrorCount = 0 +} + +func (ev *EcVolume) ResetIoErrorState() { + ev.lastIoErrorLock.Lock() + defer ev.lastIoErrorLock.Unlock() + ev.lastIoError = nil + ev.lastIoErrorCount = 0 + ev.ioErrorQuarantined = false +} + +func (ev *EcVolume) MarkIoQuarantined() { + ev.lastIoErrorLock.Lock() + defer ev.lastIoErrorLock.Unlock() + ev.ioErrorQuarantined = true +} + +func (ev *EcVolume) GetIoErrorState() (error, int32, bool) { + ev.lastIoErrorLock.RLock() + defer ev.lastIoErrorLock.RUnlock() + return ev.lastIoError, ev.lastIoErrorCount, ev.ioErrorQuarantined } // statEcxSize returns the size of an .ecx file, os.ErrNotExist when it is absent @@ -634,9 +688,10 @@ func (ev *EcVolume) IntervalToShardIdAndOffset(interval Interval) (ShardId, int6 func (ev *EcVolume) FindNeedleFromEcx(needleId types.NeedleId) (offset types.Offset, size types.Size, err error) { offset, size, err = SearchNeedleFromSortedIndex(ev.ecxFile, ev.ecxFileSize, needleId, nil) if err != nil { + ev.CheckReadWriteError(err) return } - // Apply runtime deletion state on top of the sealed .ecx lookup. + ev.CheckReadWriteError(nil) if ev.IsNeedleDeleted(needleId) { size = types.TombstoneFileSize } @@ -651,7 +706,7 @@ func SearchNeedleFromSortedIndex(ecxFile *os.File, ecxFileSize int64, needleId t m := (l + h) / 2 if n, err := ecxFile.ReadAt(buf, m*types.NeedleMapEntrySize); err != nil { if n != types.NeedleMapEntrySize { - return types.Offset{}, types.TombstoneFileSize, fmt.Errorf("ecx file %d read at %d: %v", ecxFileSize, m*types.NeedleMapEntrySize, err) + return types.Offset{}, types.TombstoneFileSize, fmt.Errorf("ecx file %d read at %d: %w", ecxFileSize, m*types.NeedleMapEntrySize, err) } } key, offset, size = idx.IdxFileEntry(buf) diff --git a/weed/storage/io_error.go b/weed/storage/io_error.go new file mode 100644 index 000000000..75f99bbb6 --- /dev/null +++ b/weed/storage/io_error.go @@ -0,0 +1,75 @@ +package storage + +import ( + "errors" + "sync" + "syscall" + + "github.com/seaweedfs/seaweedfs/weed/stats" +) + +const IoErrorTolerance = 3 + +type IoErrorTracker struct { + lastIoError error + lastIoErrorCount int32 + ioErrorQuarantined bool + lastIoErrorLock sync.RWMutex +} + +func (t *IoErrorTracker) checkReadWriteError(err error) { + if err == nil { + t.clearIoError() + return + } + if isStorageIoError(err) { + t.noteIoError(err) + return + } + t.clearIoError() +} + +func isStorageIoError(err error) bool { + if errors.Is(err, syscall.EIO) { + return true + } + if isWindowsStorageIoError(err) { + return true + } + return false +} + +func (t *IoErrorTracker) noteIoError(err error) { + t.lastIoErrorLock.Lock() + defer t.lastIoErrorLock.Unlock() + t.lastIoError = err + t.lastIoErrorCount++ + stats.VolumeServerStorageIoErrorCounter.Inc() +} + +func (t *IoErrorTracker) clearIoError() { + t.lastIoErrorLock.Lock() + defer t.lastIoErrorLock.Unlock() + t.lastIoError = nil + t.lastIoErrorCount = 0 +} + +func (t *IoErrorTracker) resetIoErrorState() { + t.lastIoErrorLock.Lock() + defer t.lastIoErrorLock.Unlock() + t.lastIoError = nil + t.lastIoErrorCount = 0 + t.ioErrorQuarantined = false +} + +func (t *IoErrorTracker) markIoQuarantined() { + t.lastIoErrorLock.Lock() + defer t.lastIoErrorLock.Unlock() + t.ioErrorQuarantined = true +} + +func (t *IoErrorTracker) getIoErrorState() (error, int32, bool) { + t.lastIoErrorLock.RLock() + defer t.lastIoErrorLock.RUnlock() + return t.lastIoError, t.lastIoErrorCount, t.ioErrorQuarantined +} diff --git a/weed/storage/io_error_other.go b/weed/storage/io_error_other.go new file mode 100644 index 000000000..ed5aa013f --- /dev/null +++ b/weed/storage/io_error_other.go @@ -0,0 +1,7 @@ +//go:build !windows + +package storage + +func isWindowsStorageIoError(err error) bool { + return false +} diff --git a/weed/storage/io_error_windows.go b/weed/storage/io_error_windows.go new file mode 100644 index 000000000..4ebc0af56 --- /dev/null +++ b/weed/storage/io_error_windows.go @@ -0,0 +1,23 @@ +//go:build windows + +package storage + +import ( + "errors" + "syscall" +) + +const ( + errERROR_CRC = syscall.Errno(23) + errERROR_IO_DEVICE = syscall.Errno(1117) +) + +func isWindowsStorageIoError(err error) bool { + if errors.Is(err, errERROR_CRC) { + return true + } + if errors.Is(err, errERROR_IO_DEVICE) { + return true + } + return false +} diff --git a/weed/storage/store.go b/weed/storage/store.go index 3b7818e11..af005f3d2 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -451,6 +451,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { // master can tell whether applying what it was sent leaves it current. // Volumes skipped below -- quarantined, phantom, expired -- are in neither. var volumeDigest uint64 + var quarantinedVolumes int sendFullList, reportGeneration, reportPass := s.volumeReport.begin() maxVolumeCounts := make(map[string]uint32) // Per-disk effective max for DiskTag, captured alongside the per-type sum. @@ -514,6 +515,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { v.Id, ioCount, ioErr) v.markIoQuarantined() } + quarantinedVolumes++ v.noWriteLock.Lock() v.noWriteOrDelete = true v.noWriteLock.Unlock() @@ -656,6 +658,19 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { state = s.State.Proto() } + quarantinedEcShards := 0 + for _, location := range s.Locations { + location.ecVolumesLock.RLock() + for _, ev := range location.ecVolumes { + if _, _, quarantined := ev.GetIoErrorState(); quarantined { + quarantinedEcShards += len(ev.Shards) + } + } + location.ecVolumesLock.RUnlock() + } + stats.VolumeServerIoQuarantineGauge.WithLabelValues("volume").Set(float64(quarantinedVolumes)) + stats.VolumeServerIoQuarantineGauge.WithLabelValues("ec_shard").Set(float64(quarantinedEcShards)) + return &master_pb.Heartbeat{ Ip: s.Ip, Port: uint32(s.Port), @@ -720,10 +735,13 @@ func (s *Store) deleteExpiredEcVolumes() (ecShards, deleted []*master_pb.VolumeE // Collect ecVolume to be deleted var toDeleteEvs []*erasure_coding.EcVolume + var ioQuarantinedEvs []*erasure_coding.EcVolume location.ecVolumesLock.RLock() for _, ev := range location.ecVolumes { if ev.IsTimeToDestroy() { toDeleteEvs = append(toDeleteEvs, ev) + } else if ioErr, ioCount, quarantined := ev.GetIoErrorState(); quarantined || (ioErr != nil && ioCount >= IoErrorTolerance) { + ioQuarantinedEvs = append(ioQuarantinedEvs, ev) } else { messages := ev.ToVolumeEcShardInformationMessage(uint32(diskId)) ecShards = append(ecShards, messages...) @@ -745,6 +763,19 @@ func (s *Store) deleteExpiredEcVolumes() (ecShards, deleted []*master_pb.VolumeE // from volumes that were already collected deleted = append(deleted, messages...) } + + // Quarantine EC volumes on faulty media: keep them in memory (so + // healthz can observe the quarantine state and an operator can + // recover) but stop reporting them so the master re-replicates + // from healthy peers. Mirrors the regular volume quarantine. + for _, ev := range ioQuarantinedEvs { + ioErr, ioCount, quarantined := ev.GetIoErrorState() + if !quarantined { + ev.MarkIoQuarantined() + glog.Warningf("ec volume %d quarantined after %d consecutive IO errors: %v", + ev.VolumeId, ioCount, ioErr) + } + } } return } @@ -760,6 +791,31 @@ func (s *Store) IsStopping() bool { return s.isStopping.Load() } +// HasIoQuarantine reports whether any local volume or EC shard is currently +// quarantined due to sustained storage-media EIO. Used by /healthz so a +// load balancer can drain a server whose underlying media is faulty. +func (s *Store) HasIoQuarantine() bool { + for _, location := range s.Locations { + location.volumesLock.RLock() + for _, v := range location.volumes { + if _, _, quarantined := v.getIoErrorState(); quarantined { + location.volumesLock.RUnlock() + return true + } + } + location.volumesLock.RUnlock() + location.ecVolumesLock.RLock() + for _, ev := range location.ecVolumes { + if _, _, quarantined := ev.GetIoErrorState(); quarantined { + location.ecVolumesLock.RUnlock() + return true + } + } + location.ecVolumesLock.RUnlock() + } + return false +} + func (s *Store) LoadNewVolumes() { for _, location := range s.Locations { location.loadExistingVolumes(s.NeedleMapKind, 0) diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 588417896..5ec1c9f87 100644 --- a/weed/storage/store_ec.go +++ b/weed/storage/store_ec.go @@ -210,6 +210,9 @@ func (s *Store) CollectErasureCodingHeartbeat() *master_pb.Heartbeat { for diskId, location := range s.Locations { location.ecVolumesLock.RLock() for _, ecShards := range location.ecVolumes { + if _, _, quarantined := ecShards.GetIoErrorState(); quarantined { + continue + } ecShardMessages = append(ecShardMessages, ecShards.ToVolumeEcShardInformationMessage(uint32(diskId))...) for _, ecShard := range ecShards.Shards { @@ -771,8 +774,10 @@ func (s *Store) readLocalEcShardInterval(ecVolume *erasure_coding.EcVolume, shar readBytes, err := shard.ReadAt(buf, offset) if err != nil { - return fmt.Errorf("failed to read local EC shard %d for volume %d: %v", shardId, ecVolume.VolumeId, err) + ownerVolume.CheckReadWriteError(err) + return fmt.Errorf("failed to read local EC shard %d for volume %d: %w", shardId, ecVolume.VolumeId, err) } + ownerVolume.CheckReadWriteError(nil) if got, want := readBytes, len(buf); got != want { return fmt.Errorf("expected %d bytes for local EC shard %d on volume %d, got %d", want, shardId, ecVolume.VolumeId, got) } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 404d3ba43..1f5b582bd 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -65,77 +65,7 @@ type Volume struct { location *DiskLocation diskId uint32 // ID of this volume's disk in Store.Locations array - // lastIoError is the most recent EIO from a read/write/delete; cleared - // on the next successful or non-EIO op. lastIoErrorCount tracks - // consecutive EIOs so CollectHeartbeat can require a sustained failure - // before unmounting the replica — protects against a transient - // hardware/network blip hitting multiple replicas at once and - // stranding the only good copy. - // - // ioErrorQuarantined is sticky: once CollectHeartbeat sees the streak - // cross IoErrorTolerance it sets this and never clears it on its own. - // A subsequent successful read clears the streak counter but must NOT - // un-quarantine the volume — only MarkVolumeWritable does that, after - // an operator has decided the disk is healthy. Without the sticky - // bit, one good read between heartbeats would silently put a known- - // bad replica back into rotation. - // - // All four fields are guarded together so the heartbeat reader sees - // a consistent snapshot. - lastIoError error - lastIoErrorCount int32 - ioErrorQuarantined bool - lastIoErrorLock sync.RWMutex -} - -// noteIoError records an EIO and increments the consecutive-error -// counter. Caller has already verified errors.Is(err, syscall.EIO). -func (v *Volume) noteIoError(err error) { - v.lastIoErrorLock.Lock() - defer v.lastIoErrorLock.Unlock() - v.lastIoError = err - v.lastIoErrorCount++ -} - -// clearIoError resets the EIO streak counter only. The sticky quarantine -// bit set by CollectHeartbeat is intentionally left alone — recovery is -// an operator decision via MarkVolumeWritable. Called on any successful -// op or on a non-EIO error (which still breaks the EIO streak; only -// sustained EIOs are diagnostic of a failing volume). -func (v *Volume) clearIoError() { - v.lastIoErrorLock.Lock() - defer v.lastIoErrorLock.Unlock() - v.lastIoError = nil - v.lastIoErrorCount = 0 -} - -// resetIoErrorState clears both the EIO streak and the sticky quarantine -// flag. Used by MarkVolumeWritable to rejoin a previously-quarantined -// replica; if the disk is still bad, the next failed op re-arms the -// streak. -func (v *Volume) resetIoErrorState() { - v.lastIoErrorLock.Lock() - defer v.lastIoErrorLock.Unlock() - v.lastIoError = nil - v.lastIoErrorCount = 0 - v.ioErrorQuarantined = false -} - -// markIoQuarantined sets the sticky quarantine flag. Idempotent; safe -// to call from CollectHeartbeat each pass while the volume remains -// quarantined. -func (v *Volume) markIoQuarantined() { - v.lastIoErrorLock.Lock() - defer v.lastIoErrorLock.Unlock() - v.ioErrorQuarantined = true -} - -// getIoErrorState returns the latest EIO, the consecutive-EIO count, -// and the sticky quarantine flag as one consistent snapshot. -func (v *Volume) getIoErrorState() (error, int32, bool) { - v.lastIoErrorLock.RLock() - defer v.lastIoErrorLock.RUnlock() - return v.lastIoError, v.lastIoErrorCount, v.ioErrorQuarantined + IoErrorTracker } func NewVolume(dirname string, dirIdx string, collection string, id needle.VolumeId, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, ver needle.Version, memoryMapMaxSizeMb uint32, ldbTimeout int64) (v *Volume, e error) { diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 77e216ae2..b1d8818cd 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "os" - "syscall" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/storage/backend" @@ -17,27 +16,6 @@ var ErrorNotFound = errors.New("not found") var ErrorDeleted = errors.New("already deleted") var ErrorSizeMismatch = errors.New("size mismatch") -// IoErrorTolerance is the number of consecutive EIOs a volume must -// see before CollectHeartbeat treats the replica as broken. A single -// transient error is forgiven so a brief NFS / fabric / power blip -// affecting several replicas at once does not cascade into removal of -// the last healthy copy. -const IoErrorTolerance = 3 - -func (v *Volume) checkReadWriteError(err error) { - if err == nil { - v.clearIoError() - return - } - if errors.Is(err, syscall.EIO) { - v.noteIoError(err) - return - } - // non-EIO error breaks the EIO streak — only sustained EIOs should - // be treated as a failing volume. - v.clearIoError() -} - // isFileUnchanged checks whether this needle to write is same as last one. // It requires serialized access in the same volume. func (v *Volume) isFileUnchanged(n *needle.Needle) bool {