diff --git a/weed/storage/store.go b/weed/storage/store.go index 4fe1accac..9d0c19535 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -416,22 +416,29 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat { maxFileKey = curMaxFileKey } - if ioErr, ioCount := v.getIoErrorState(); ioErr != nil && ioCount >= IoErrorTolerance { + ioErr, ioCount, quarantined := v.getIoErrorState() + if quarantined || (ioErr != nil && ioCount >= IoErrorTolerance) { // Sustained EIO: stop announcing this replica so the master - // re-replicates from healthy peers, and mark it read-only so - // further writes fail fast instead of producing more EIOs. - // Never physically delete the data — the disk may be - // transiently bad and this could be the last good copy. - glog.Warningf("volume %d has %d consecutive IO errors, marking read-only and unreporting from master: %v", - v.Id, ioCount, ioErr) + // re-replicates from healthy peers, and mark it read-only + // so further writes fail fast instead of producing more + // EIOs. Never physically delete the data — the disk may + // be transiently bad and this could be the last good + // copy. The quarantine is sticky: a stray successful read + // clears the streak counter but must not silently put a + // known-bad replica back into rotation; recovery is via + // MarkVolumeWritable. + if !quarantined { + glog.Warningf("volume %d quarantined after %d consecutive IO errors: %v", + v.Id, ioCount, ioErr) + v.markIoQuarantined() + } v.noWriteLock.Lock() v.noWriteOrDelete = true v.noWriteLock.Unlock() // Skip per-volume size and read-only bookkeeping: a // quarantined replica should not be summed into the // collection's reported total nor counted in the - // read-only stats. Recovery via MarkVolumeWritable - // resets the error state so it can rejoin. + // read-only stats. continue } @@ -682,10 +689,10 @@ func (s *Store) MarkVolumeWritable(i needle.VolumeId) error { v.noWriteOrDelete = false v.PersistReadOnly(false) v.noWriteLock.Unlock() - // Clear any sustained-EIO state so the next CollectHeartbeat can - // announce the volume again. If the disk is still bad, the next - // failed op will re-arm the streak. - v.clearIoError() + // Clear the EIO streak and the sticky quarantine flag so the next + // CollectHeartbeat can announce the volume again. If the disk is + // still bad, the next failed op will re-arm the streak. + v.resetIoErrorState() return nil } diff --git a/weed/storage/volume.go b/weed/storage/volume.go index c33a806a2..d7bd432e6 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -58,11 +58,22 @@ type Volume struct { // 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. Both fields are guarded together so - // the heartbeat reader sees a consistent (err, count) snapshot. - lastIoError error - lastIoErrorCount int32 - lastIoErrorLock sync.RWMutex + // 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 @@ -74,9 +85,11 @@ func (v *Volume) noteIoError(err error) { v.lastIoErrorCount++ } -// clearIoError resets the EIO streak. 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). +// 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() @@ -84,12 +97,33 @@ func (v *Volume) clearIoError() { v.lastIoErrorCount = 0 } -// getIoErrorState returns the latest EIO and its consecutive-error count -// as a single atomic snapshot. -func (v *Volume) getIoErrorState() (error, int32) { +// 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 + return v.lastIoError, v.lastIoErrorCount, v.ioErrorQuarantined } 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_io_error_test.go b/weed/storage/volume_io_error_test.go index 647ecdf27..855dfdd9c 100644 --- a/weed/storage/volume_io_error_test.go +++ b/weed/storage/volume_io_error_test.go @@ -6,47 +6,67 @@ import ( "sync" "syscall" "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" ) +// eioBackend is a backend.BackendStorageFile that returns EIO from +// every ReadAt — used to verify the streaming read path threads +// errors through checkReadWriteError. +type eioBackend struct{} + +func (eioBackend) ReadAt(p []byte, off int64) (int, error) { return 0, syscall.EIO } +func (eioBackend) WriteAt(p []byte, off int64) (int, error) { + return len(p), nil +} +func (eioBackend) Truncate(int64) error { return nil } +func (eioBackend) Close() error { return nil } +func (eioBackend) GetStat() (int64, time.Time, error) { return 0, time.Time{}, nil } +func (eioBackend) Name() string { return "eio" } +func (eioBackend) Sync() error { return nil } + func TestCheckReadWriteErrorTracksConsecutiveEIO(t *testing.T) { v := &Volume{} // each EIO bumps the counter. for i := int32(1); i <= 5; i++ { v.checkReadWriteError(fmt.Errorf("disk failed: %w", syscall.EIO)) - _, count := v.getIoErrorState() + _, count, _ := v.getIoErrorState() if count != i { t.Fatalf("after %d EIO(s): counter = %d, want %d", i, count, i) } } - // a single success resets both fields. + // a single success resets the streak (but not the quarantine flag, + // which is not set here). v.checkReadWriteError(nil) - if err, count := v.getIoErrorState(); err != nil || count != 0 { - t.Fatalf("success did not reset state: err=%v count=%d", err, count) + if err, count, q := v.getIoErrorState(); err != nil || count != 0 || q { + t.Fatalf("success did not reset state: err=%v count=%d quarantined=%v", err, count, q) } } func TestCheckReadWriteErrorNonEIOResetsStreak(t *testing.T) { v := &Volume{} - // build up a 2-EIO streak. v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) - if _, count := v.getIoErrorState(); count != 2 { + if _, count, _ := v.getIoErrorState(); count != 2 { t.Fatalf("expected count=2 after two EIOs, got %d", count) } // a non-EIO error breaks the streak — only sustained EIOs are // diagnostic of a failing disk. v.checkReadWriteError(fmt.Errorf("other: %w", syscall.ENOSPC)) - if err, count := v.getIoErrorState(); err != nil || count != 0 { + if err, count, _ := v.getIoErrorState(); err != nil || count != 0 { t.Fatalf("non-EIO did not reset streak: err=%v count=%d", err, count) } // a fresh EIO starts the streak from 1, not 3. v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) - if _, count := v.getIoErrorState(); count != 1 { + if _, count, _ := v.getIoErrorState(); count != 1 { t.Fatalf("EIO after non-EIO did not restart streak: count=%d, want 1", count) } } @@ -54,10 +74,8 @@ func TestCheckReadWriteErrorNonEIOResetsStreak(t *testing.T) { func TestCheckReadWriteErrorIgnoresPlainError(t *testing.T) { v := &Volume{} - // non-EIO error with no prior streak should be a no-op (count - // stays 0, no spurious lastIoError). v.checkReadWriteError(errors.New("some other error")) - if err, count := v.getIoErrorState(); err != nil || count != 0 { + if err, count, _ := v.getIoErrorState(); err != nil || count != 0 { t.Fatalf("non-EIO with no prior streak set state: err=%v count=%d", err, count) } } @@ -65,27 +83,79 @@ func TestCheckReadWriteErrorIgnoresPlainError(t *testing.T) { func TestIoErrorToleranceGate(t *testing.T) { v := &Volume{} - // below tolerance: do not act. for i := 0; i < IoErrorTolerance-1; i++ { v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) } - if _, count := v.getIoErrorState(); count >= IoErrorTolerance { + if _, count, _ := v.getIoErrorState(); count >= IoErrorTolerance { t.Fatalf("counter %d already crossed tolerance %d after %d errors", count, IoErrorTolerance, IoErrorTolerance-1) } - // one more crosses the threshold. v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) - if _, count := v.getIoErrorState(); count < IoErrorTolerance { + if _, count, _ := v.getIoErrorState(); count < IoErrorTolerance { t.Fatalf("counter %d below tolerance %d after %d errors", count, IoErrorTolerance, IoErrorTolerance) } } +// Once CollectHeartbeat marks a replica quarantined, a stray successful +// read must NOT silently put a known-bad disk back into rotation. Only +// MarkVolumeWritable (resetIoErrorState) clears the sticky bit. +func TestQuarantineIsSticky(t *testing.T) { + v := &Volume{} + + for i := 0; i < IoErrorTolerance; i++ { + v.checkReadWriteError(fmt.Errorf("eio: %w", syscall.EIO)) + } + v.markIoQuarantined() + + if _, _, q := v.getIoErrorState(); !q { + t.Fatalf("markIoQuarantined did not set the flag") + } + + // A successful read clears the streak counter… + v.checkReadWriteError(nil) + if _, count, q := v.getIoErrorState(); count != 0 { + t.Fatalf("success did not clear streak: count=%d", count) + } else if !q { + t.Fatalf("success cleared the sticky quarantine flag — operator-only recovery violated") + } + + // …and a non-EIO error also clears the streak but not the flag. + v.checkReadWriteError(fmt.Errorf("other: %w", syscall.ENOSPC)) + if _, _, q := v.getIoErrorState(); !q { + t.Fatalf("non-EIO cleared the sticky quarantine flag") + } + + // Only resetIoErrorState (used by MarkVolumeWritable) un-quarantines. + v.resetIoErrorState() + if err, count, q := v.getIoErrorState(); err != nil || count != 0 || q { + t.Fatalf("resetIoErrorState left state: err=%v count=%d quarantined=%v", err, count, q) + } +} + +// Streaming/range reads (ReadNeedleBlob) used to bypass the EIO +// counter, so a failing disk taking range GETs all day would never +// trip IoErrorTolerance. ReadNeedleBlob now threads the backend error +// through checkReadWriteError; one EIO must bump the streak by one. +func TestReadNeedleBlobTracksEIO(t *testing.T) { + v := &Volume{ + DataBackend: eioBackend{}, + SuperBlock: super_block.SuperBlock{Version: needle.GetCurrentVersion()}, + volumeInfo: &volume_server_pb.VolumeInfo{Version: uint32(needle.GetCurrentVersion())}, + } + + if _, err := v.ReadNeedleBlob(0, 1); !errors.Is(err, syscall.EIO) { + t.Fatalf("expected EIO from fake backend, got %v", err) + } + if _, count, _ := v.getIoErrorState(); count != 1 { + t.Fatalf("ReadNeedleBlob did not bump EIO streak: count=%d", count) + } +} + func TestIoErrorStateIsRaceFree(t *testing.T) { - // Drives both writers (checkReadWriteError) and a reader - // (getIoErrorState) concurrently; relies on `go test -race` to - // detect any unprotected access on lastIoError / lastIoErrorCount. + // Relies on `go test -race` to detect any unprotected access on + // lastIoError / lastIoErrorCount / ioErrorQuarantined. v := &Volume{} var wg sync.WaitGroup @@ -116,6 +186,19 @@ func TestIoErrorStateIsRaceFree(t *testing.T) { } }() wg.Add(1) + go func() { + defer wg.Done() + for { + select { + case <-stop: + return + default: + v.markIoQuarantined() + v.resetIoErrorState() + } + } + }() + wg.Add(1) go func() { defer wg.Done() for i := 0; i < 1000; i++ { diff --git a/weed/storage/volume_read.go b/weed/storage/volume_read.go index 702f3ed39..fafdcdabd 100644 --- a/weed/storage/volume_read.go +++ b/weed/storage/volume_read.go @@ -180,6 +180,14 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr if readOption.HasSlowRead { v.dataFileAccessLock.RUnlock() } + // Thread the underlying read error through the EIO tracker. + // Without this, large/range GETs through readNeedleDataInto + // would never trip IoErrorTolerance even on a failing disk. + // io.EOF is treated as a clean end-of-stream below, not an + // error. + if err != nil && err != io.EOF { + v.checkReadWriteError(err) + } toWrite := min(count, int(offset+size-x)) if toWrite > 0 { @@ -201,6 +209,11 @@ func (v *Volume) readNeedleDataInto(n *needle.Needle, readOption *ReadOption, wr break } } + // Whole-needle read completed without a backend error — clear any + // pending EIO streak. If a non-EIO failure happens later (CRC etc.) + // we still return that error to the caller, but the disk itself + // produced clean bytes. + v.checkReadWriteError(nil) if offset == 0 && size == int64(n.DataSize) && (n.Checksum != crc && uint32(n.Checksum) != crc.Value()) { // the crc.Value() function is to be deprecated. this double checking is for backward compatibility // with seaweed version using crc.Value() instead of uint32(crc), which appears in commit 056c480eb @@ -224,7 +237,9 @@ func (v *Volume) ReadNeedleBlob(offset int64, size Size) ([]byte, error) { v.dataFileAccessLock.RLock() defer v.dataFileAccessLock.RUnlock() - return needle.ReadNeedleBlob(v.DataBackend, offset, size, v.Version()) + blob, err := needle.ReadNeedleBlob(v.DataBackend, offset, size, v.Version()) + v.checkReadWriteError(err) + return blob, err } type VolumeFileScanner interface {