mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-20 15:04:37 +00:00
fix(volume): handle faulty storage media (Go + Rust) (#11233)
* fix(volume): track EC shard read errors and unmount on faulty media Extract the volume EIO tracker into a reusable IoErrorTracker and add the same tracking to EcVolume. Sustained EIO on .ecx lookups or .ecd shard reads now unmounts the EC volume in the heartbeat (without deleting files) so the master re-replicates from healthy peers, mirroring the existing volume replica quarantine. Closes #11227 (EC shard unmount). * rust(volume): mirror EC shard read error tracking and unmount Add EIO tracking to the Rust EcVolume mirroring Go: a streak counter with IO_ERROR_TOLERANCE, a sticky quarantine flag, and unmount (not file deletion) in the heartbeat so the master re-replicates from healthy peers. * feat(metrics): expose storage IO error counter and quarantine gauge Add a storage_io_error_total counter incremented on every EIO recorded by the volume or EC shard tracker, and an io_quarantine gauge labelled by kind (volume/ec_shard) reflecting the count of replicas suppressed in the heartbeat. Mirrored in Go and Rust. * feat(healthz): report 503 when local replicas are IO-quarantined Add Store.HasIoQuarantine (Go) / Store::has_io_quarantine (Rust) and have /healthz return 503 when any local volume or EC shard is quarantined due to sustained storage-media EIO, so a load balancer can drain a server whose underlying media is faulty. Mirrored in Go and Rust. * fix(volume): keep quarantined EC volumes in memory and reset EIO on success Address review feedback: instead of unloading quarantined EC volumes (which discards the quarantine state healthz needs), keep them in memory and just skip them from heartbeat reporting, mirroring the regular volume quarantine. Also clear the EIO streak on successful .ecx reads in Rust so a transient error does not accumulate, and add an ec_shard label to the io_quarantine gauge in both Go and Rust. * fix(volume): exclude quarantined EC shards from heartbeat and add Rust volume tolerance Address review feedback: - Filter quarantined EC volumes from CollectErasureCodingHeartbeat (Go) and collect_ec_shard_delta_messages / collect_live_ec_shards (Rust) so the master stops advertising faulty shards and re-replicates from healthy peers. - Add consecutive EIO count and sticky quarantine to the Rust regular Volume, mirroring Go IoErrorTracker: a single EIO no longer deletes the replica; the heartbeat quarantines after the tolerance threshold and keeps the volume in memory. - Use the quarantine flag (not last_io_error) in has_io_quarantine so /healthz reflects sustained, not transient, failures. * fix(volume): make Rust quarantined volumes read-only and wire recovery Address Devin review: - Set no_write_or_delete on Rust volumes when quarantined in the heartbeat, so cached or direct clients cannot mutate a faulty replica after the master removes it (mirrors Go). - Wire reset_io_error_state into Volume::set_writable so an operator making a volume writable again clears the sticky quarantine and the volume re-enters heartbeat rotation. * fix(volume): clear EC quarantine on shard re-mount for operator recovery Address Greptile review: re-mounting EC shards (Go loadEcShardWithIdxDir / Rust mount_ec_shards_with_idx_dir) now calls ResetIoErrorState on the existing EcVolume, giving operators a documented recovery path that clears the sticky quarantine and returns the EC volume to heartbeat rotation. Mirrored in Go and Rust. * fix(volume): do not clear EC quarantine on routine shard mounts Address review feedback: clearing the EC IO quarantine on every mount (including duplicate, retry, sibling-shard, and reconciliation mounts) is too aggressive and can re-advertise known-bad shards before the storage media has been validated. Remove the automatic reset from the mount path; quarantine clears naturally on restart or full unmount when a fresh EcVolume is created with clean state. * test(volume): update Rust IO error test for quarantine semantics The heartbeat now quarantines a volume with sustained EIO (keeps it mounted, makes it read-only, omits it from heartbeat) instead of deleting it. Update test_collect_heartbeat_deletes_io_error_volume to assert the volume stays in the store with no_write_or_delete set, and update set_last_io_error_for_test to set the consecutive error count at the tolerance threshold so the test reflects a sustained error. * fix(volume): reset EIO streak after full write and match Windows media errors Move the success-side EIO reset from append_needle (after write_all only) to the end of do_write_request, after flush_dat/flush_idx complete, so a successful write_all followed by a failed fsync no longer resets the counter before the EIO is recorded. Repeated fsync EIOs now accumulate toward the quarantine threshold as intended. Recognize Windows storage-media failure codes ERROR_CRC (23) and ERROR_IO_DEVICE (1117) in addition to Unix EIO (errno 5), so quarantined heartbeat behavior is preserved on Windows. Mirrors the change in both Go and Rust volume servers. * fix(volume): preserve checkpoint EIO and clear streak on successful delete maybe_checkpoint_index now returns whether the checkpoint succeeded; the success-side EIO reset in do_write_request and do_delete_request only fires when it did, so a checkpoint media failure is no longer erased by the unconditional reset that followed it. do_delete_request also gains the success reset that was lost when append_needle stopped clearing the streak, so a successful delete still clears an earlier failure streak. is_storage_io_error now uses libc::EIO on Unix instead of a hard-coded 5, and the ECX binary-search read path gains a Windows fallback (seek + read_exact) so the buffer is no longer zeroed on non-Unix targets.
This commit is contained in:
Chris Lu
@@ -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()),
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -3127,6 +3127,11 @@ pub async fn healthz_handler(State(state): State<Arc<VolumeServerState>>) -> 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()
|
||||
}
|
||||
|
||||
|
||||
@@ -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<VolumeId> = 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]
|
||||
|
||||
@@ -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<Option<String>>,
|
||||
}
|
||||
|
||||
/// 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<String>, 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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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<Option<String>>,
|
||||
/// 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<String>, 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)]
|
||||
|
||||
@@ -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 {
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
//go:build !windows
|
||||
|
||||
package storage
|
||||
|
||||
func isWindowsStorageIoError(err error) bool {
|
||||
return false
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
+1
-71
@@ -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) {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user