[CheckDisk]: implement disk health detection (#9560)

* [CheckDisk][GRPC]: implement MVP for disk health detection, added timeout for new grpc connections

* fix(volume): build disk health check on every platform

setDiskStatus only existed behind the statfs build tag, so disk.go failed
to compile on windows, openbsd, solaris, netbsd and plan9. Move the timeout
wrapper and failure tracking into the shared disk.go and have each platform's
fillInDiskStatus return an error, so every platform gets the same protection
from a stuck filesystem.

Also restore the uint64(fs.Bavail) cast: Bavail is int64 on freebsd, so the
unguarded multiply broke the freebsd build.

* fix(volume): keep one outstanding statfs probe per disk

A stuck statfs used to leave isChecking cleared by the timeout path, so the
next check spawned another goroutine while the previous one was still blocked
in the syscall, leaking one goroutine per minute on a hung disk. Clear the
flag only when statfs returns and treat an overlapping check as a failure, so
a hung filesystem keeps a single outstanding probe and still gets reported.

* fix(volume): assume disk available until the first health check

isDiskAvailable defaulted to false, and CollectHeartbeat skips locations that
are not available. A freshly started volume server would therefore omit every
volume from its first heartbeats until the async CheckDiskSpace ran, so the
master could briefly treat all of them as missing.

* fix(volume): label the disk error metric by data directory

The new gauge tagged the series with IdxDirectory while every neighbouring
resource gauge uses Directory, so the error series would not line up with them
in dashboards. Also log the underlying error instead of a generic message.

* test(volume): cover disk health success and repeated-failure paths

* fix(volume): make a healthy disk the zero-value default

Track the disk as isDiskUnavailable instead of isDiskAvailable so the safe
state is the zero value, matching isDiskSpaceLow. CollectHeartbeat only skips a
location once a check has actively marked it unavailable, so any DiskLocation
built without running CheckDiskSpace (tests, future call sites) still reports
its volumes instead of silently dropping them.

* feat(disk): detect degraded disks using IO latency probes

* feat(stats): introduce configurable disk I/O health probe with EWMA-based latency detection

* feat(disk): replace EWMA with sliding window algorithm for disk health detection and added user-friendly options

* feat(disk): improve disk health probing and recovery

* feat(volume): configure disk health checks via volume.toml

* fix(volume): Remove disk IO probe CLI options

---------

Co-authored-by: ptukha <ptukha@tochka.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/command/mini.go

@@ -486,6 +486,7 @@ func initMiniVolumeFlags() {
 	miniOptions.v.readBufferSizeMB = cmdMini.Flag.Int("volume.readBufferSizeMB", 4, "read buffer size in MB")
 	miniOptions.v.allowUntrustedRemoteEndpoints = cmdMini.Flag.Bool("volume.allowUntrustedRemoteEndpoints", false, "if true, FetchAndWriteNeedle accepts arbitrary remote S3 endpoints including loopback / link-local hosts. Default rejects internal / metadata endpoints.")
 	miniOptions.v.preStopSeconds = cmdMini.Flag.Int("volume.preStopSeconds", 1, "number of seconds between stop send heartbeats and stop volume server (default: 1 for mini)")
+	miniOptions.v.setDiskIOProbeDefaults()
 }
 
 // initMiniS3Flags initializes S3 server flag options
@@ -1137,6 +1138,8 @@ func runMini(cmd *Command, args []string) bool {
 
 	util.LoadSecurityConfiguration()
 	util.LoadConfiguration("master", false)
+	util.LoadConfiguration("volume", false)
+	miniOptions.v.applyDiskIOProbeConfig()
 
 	// applyConfigFileOptions above may have overwritten -dir from the
 	// mini.options file, so re-resolve it here alongside the other paths.

weed/command/scaffold.go

@@ -14,7 +14,7 @@ func init() {
 }
 
 var cmdScaffold = &Command{
-	UsageLine: "scaffold -config=[filer|notification|replication|security|master|shell|credential]",
+	UsageLine: "scaffold -config=[filer|notification|replication|security|master|volume|shell|credential]",
 	Short:     "generate basic configuration files",
 	Long: `Generate configuration files with all possible configurations for you to customize.
 
@@ -31,7 +31,7 @@ var cmdScaffold = &Command{
 
 var (
 	outputPath = cmdScaffold.Flag.String("output", "", "if not empty, save the configuration file to this directory")
-	config     = cmdScaffold.Flag.String("config", "filer", "[filer|notification|replication|security|master|shell|credential] the configuration file to generate")
+	config     = cmdScaffold.Flag.String("config", "filer", "[filer|notification|replication|security|master|volume|shell|credential] the configuration file to generate")
 )
 
 func runScaffold(cmd *Command, args []string) bool {
@@ -48,6 +48,8 @@ func runScaffold(cmd *Command, args []string) bool {
 		content = scaffold.Security
 	case "master":
 		content = scaffold.Master
+	case "volume":
+		content = scaffold.Volume
 	case "shell":
 		content = scaffold.Shell
 	case "credential":

weed/command/scaffold/example.go

@@ -27,6 +27,9 @@ var Security string
 var masterTemplate string
 var Master string
 
+//go:embed volume.toml
+var Volume string
+
 //go:embed shell.toml
 var Shell string
 

weed/command/scaffold/volume.toml

@@ -0,0 +1,27 @@
+# Put this file to one of the location, with descending priority
+#    ./volume.toml
+#    $HOME/.seaweedfs/volume.toml
+#    /etc/seaweedfs/volume.toml
+# this file is read by volume servers
+
+[volume.disk.io]
+# Disk IO probing is disabled by default. When enabled, a volume server can
+# stop advertising a disk after sustained IO errors or latency degradation.
+probe = false
+timeout = "2s"
+interval = "60s"
+window = "1m"
+min.samples = 10
+max.stat.failures = 5
+recovery.coef = 0.5
+
+[volume.disk.io.slow]
+percent = 20
+
+[volume.disk.io.error]
+percent = 10
+
+[volume.disk.io.slow.latency]
+hdd = "500ms"
+ssd = "100ms"
+nvme = "50ms"

weed/command/server.go

@@ -156,6 +156,7 @@ func init() {
 	serverOptions.v.hasSlowRead = cmdServer.Flag.Bool("volume.hasSlowRead", true, "<experimental> if true, this prevents slow reads from blocking other requests, but large file read P99 latency will increase.")
 	serverOptions.v.readBufferSizeMB = cmdServer.Flag.Int("volume.readBufferSizeMB", 4, "<experimental> larger values can optimize query performance but will increase some memory usage,Use with hasSlowRead normally")
 	serverOptions.v.allowUntrustedRemoteEndpoints = cmdServer.Flag.Bool("volume.allowUntrustedRemoteEndpoints", false, "if true, FetchAndWriteNeedle accepts arbitrary remote S3 endpoints including loopback / link-local hosts. Default rejects internal / metadata endpoints.")
+	serverOptions.v.setDiskIOProbeDefaults()
 
 	s3Options.port = cmdServer.Flag.Int("s3.port", 8333, "s3 server http listen port")
 	s3Options.portHttps = cmdServer.Flag.Int("s3.port.https", 0, "s3 server https listen port")
@@ -224,6 +225,8 @@ func runServer(cmd *Command, args []string) bool {
 
 	util.LoadSecurityConfiguration()
 	util.LoadConfiguration("master", false)
+	util.LoadConfiguration("volume", false)
+	serverOptions.v.applyDiskIOProbeConfig()
 
 	*serverOptions.cpuprofile = util.ResolvePath(*serverOptions.cpuprofile)
 	*serverOptions.memprofile = util.ResolvePath(*serverOptions.memprofile)

weed/command/volume.go

@@ -78,12 +78,119 @@ type VolumeServerOptions struct {
 	allowUntrustedRemoteEndpoints *bool
 	debug                         *bool
 	debugPort                     *int
+	diskIOProbe                   *bool
+	diskIOTimeout                 *time.Duration
+	diskIOInterval                *time.Duration
+	diskHDDIOSlowLatency          *time.Duration
+	diskSSDIOSlowLatency          *time.Duration
+	diskNVMEIOSlowLatency         *time.Duration
+	diskIOWindow                  *time.Duration
+	diskIOMinSamples              *int
+	diskIOSlowPercent             *float64
+	diskIOErrorPercent            *float64
+	diskIOMaxStatFailures         *int
+	diskRecoveryCoef              *float64
 	// shutdownCtx, when non-nil, tells startVolumeServer to shut down once the
 	// ctx is cancelled. Used by integration tests and by weed mini; nil for
 	// standalone weed volume.
 	shutdownCtx context.Context
 }
 
+func (v *VolumeServerOptions) setDiskIOProbeDefaults() {
+	if v.diskIOProbe == nil {
+		defaultValue := false
+		v.diskIOProbe = &defaultValue
+	}
+	if v.diskIOTimeout == nil {
+		defaultValue := 2 * time.Second
+		v.diskIOTimeout = &defaultValue
+	}
+	if v.diskIOInterval == nil {
+		defaultValue := 60 * time.Second
+		v.diskIOInterval = &defaultValue
+	}
+	if v.diskHDDIOSlowLatency == nil {
+		defaultValue := 500 * time.Millisecond
+		v.diskHDDIOSlowLatency = &defaultValue
+	}
+	if v.diskSSDIOSlowLatency == nil {
+		defaultValue := 100 * time.Millisecond
+		v.diskSSDIOSlowLatency = &defaultValue
+	}
+	if v.diskNVMEIOSlowLatency == nil {
+		defaultValue := 50 * time.Millisecond
+		v.diskNVMEIOSlowLatency = &defaultValue
+	}
+	if v.diskIOWindow == nil {
+		defaultValue := time.Minute
+		v.diskIOWindow = &defaultValue
+	}
+	if v.diskIOMinSamples == nil {
+		defaultValue := 10
+		v.diskIOMinSamples = &defaultValue
+	}
+	if v.diskIOSlowPercent == nil {
+		defaultValue := 20.0
+		v.diskIOSlowPercent = &defaultValue
+	}
+	if v.diskIOErrorPercent == nil {
+		defaultValue := 10.0
+		v.diskIOErrorPercent = &defaultValue
+	}
+	if v.diskIOMaxStatFailures == nil {
+		defaultValue := 5
+		v.diskIOMaxStatFailures = &defaultValue
+	}
+	if v.diskRecoveryCoef == nil {
+		defaultValue := 0.5
+		v.diskRecoveryCoef = &defaultValue
+	}
+}
+
+func (v *VolumeServerOptions) applyDiskIOProbeConfig() {
+	v.setDiskIOProbeDefaults()
+
+	config := util.GetViper()
+
+	if config.IsSet("volume.disk.io.probe") {
+		*v.diskIOProbe = config.GetBool("volume.disk.io.probe")
+	}
+	if config.IsSet("volume.disk.io.timeout") {
+		*v.diskIOTimeout = config.GetDuration("volume.disk.io.timeout")
+	}
+	if config.IsSet("volume.disk.io.slow.latency.hdd") {
+		*v.diskHDDIOSlowLatency = config.GetDuration("volume.disk.io.slow.latency.hdd")
+	}
+	if config.IsSet("volume.disk.io.slow.latency.ssd") {
+		*v.diskSSDIOSlowLatency = config.GetDuration("volume.disk.io.slow.latency.ssd")
+	}
+	if config.IsSet("volume.disk.io.slow.latency.nvme") {
+		*v.diskNVMEIOSlowLatency = config.GetDuration("volume.disk.io.slow.latency.nvme")
+	}
+
+	if config.IsSet("volume.disk.io.interval") {
+		*v.diskIOInterval = config.GetDuration("volume.disk.io.interval")
+	}
+	if config.IsSet("volume.disk.io.window") {
+		*v.diskIOWindow = config.GetDuration("volume.disk.io.window")
+	}
+	if config.IsSet("volume.disk.io.min.samples") {
+		*v.diskIOMinSamples = config.GetInt("volume.disk.io.min.samples")
+	}
+	if config.IsSet("volume.disk.io.slow.percent") {
+		*v.diskIOSlowPercent = config.GetFloat64("volume.disk.io.slow.percent")
+	}
+	if config.IsSet("volume.disk.io.error.percent") {
+		*v.diskIOErrorPercent = config.GetFloat64("volume.disk.io.error.percent")
+	}
+	if config.IsSet("volume.disk.io.max.stat.failures") {
+		*v.diskIOMaxStatFailures = config.GetInt("volume.disk.io.max.stat.failures")
+	}
+	if config.IsSet("volume.disk.io.recovery.coef") {
+		*v.diskRecoveryCoef = config.GetFloat64("volume.disk.io.recovery.coef")
+	}
+}
+
 func init() {
 	cmdVolume.Run = runVolume // break init cycle
 	v.port = cmdVolume.Flag.Int("port", 8080, "http listen port")
@@ -124,6 +231,7 @@ func init() {
 	v.allowUntrustedRemoteEndpoints = cmdVolume.Flag.Bool("volume.allowUntrustedRemoteEndpoints", false, "if true, FetchAndWriteNeedle accepts arbitrary remote S3 endpoints including loopback / link-local hosts. Default rejects internal / metadata endpoints.")
 	v.debug = cmdVolume.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port")
 	v.debugPort = cmdVolume.Flag.Int("debug.port", 6060, "http port for debugging")
+	v.setDiskIOProbeDefaults()
 }
 
 var cmdVolume = &Command{
@@ -151,6 +259,8 @@ func runVolume(cmd *Command, args []string) bool {
 	}
 
 	util.LoadSecurityConfiguration()
+	util.LoadConfiguration("volume", false)
+	v.applyDiskIOProbeConfig()
 
 	// If --pprof is set we assume the caller wants to be able to collect
 	// cpu and memory profiles via go tool pprof
@@ -201,6 +311,7 @@ func runVolume(cmd *Command, args []string) bool {
 }
 
 func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, volumeWhiteListOption string, minFreeSpaces []util.MinFreeSpace) {
+	v.setDiskIOProbeDefaults()
 
 	// Set multiple folders and each folder's max volume count limit'
 	v.folders = strings.Split(volumeFolders, ",")
@@ -306,7 +417,39 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 
 	// Determine volume server ID: if not specified, use ip:port
 	volumeServerId := util.GetVolumeServerId(*v.id, *v.ip, *v.port)
+	var slowLatency time.Duration
 
+	switch *v.diskType {
+	case "hdd":
+		slowLatency = *v.diskHDDIOSlowLatency
+	case "ssd":
+		slowLatency = *v.diskSSDIOSlowLatency
+	case "nvme":
+		slowLatency = *v.diskNVMEIOSlowLatency
+	default:
+		slowLatency = *v.diskHDDIOSlowLatency
+	}
+	diskProbeConfig := stats_collect.DiskIOProbeConfig{
+		Enabled:  *v.diskIOProbe,
+		Timeout:  *v.diskIOTimeout,
+		Interval: *v.diskIOInterval,
+
+		SlowLatency: slowLatency,
+
+		Window:     *v.diskIOWindow,
+		MinSamples: *v.diskIOMinSamples,
+
+		SlowPercent:  *v.diskIOSlowPercent,
+		ErrorPercent: *v.diskIOErrorPercent,
+
+		MaxStatFailures: *v.diskIOMaxStatFailures,
+
+		RecoveryCoef: *v.diskRecoveryCoef,
+	}
+	if diskProbeConfig.Enabled && len(v.folders) > 1 {
+		glog.Warningf("disk IO probe is disabled for multiple volume directories: %v", v.folders)
+		diskProbeConfig.Enabled = false
+	}
 	volumeServer := weed_server.NewVolumeServer(volumeMux, publicVolumeMux,
 		*v.ip, *v.port, *v.portGrpc, *v.publicUrl, volumeServerId,
 		v.folders, v.folderMaxLimits, minFreeSpaces, diskTypes, folderTags,
@@ -326,6 +469,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 		*v.readBufferSizeMB,
 		*v.ldbTimeout,
 		*v.allowUntrustedRemoteEndpoints,
+		diskProbeConfig,
 	)
 	// starting grpc server
 	grpcS := v.startGrpcService(volumeServer)

weed/pb/volume_server.proto

@@ -542,6 +542,7 @@ message DiskStatus {
     float percent_free = 5;
     float percent_used = 6;
     string disk_type = 7;
+    string error = 8;
 }
 
 message MemStatus {

weed/pb/volume_server_pb/volume_server.pb.go

@@ -4565,6 +4565,7 @@ type DiskStatus struct {
 	PercentFree   float32                `protobuf:"fixed32,5,opt,name=percent_free,json=percentFree,proto3" json:"percent_free,omitempty"`
 	PercentUsed   float32                `protobuf:"fixed32,6,opt,name=percent_used,json=percentUsed,proto3" json:"percent_used,omitempty"`
 	DiskType      string                 `protobuf:"bytes,7,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
+	Error         string                 `protobuf:"bytes,8,opt,name=error,proto3" json:"error,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -4648,6 +4649,13 @@ func (x *DiskStatus) GetDiskType() string {
 	return ""
 }
 
+func (x *DiskStatus) GetError() string {
+	if x != nil {
+		return x.Error
+	}
+	return ""
+}
+
 type MemStatus struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Goroutines    int32                  `protobuf:"varint,1,opt,name=goroutines,proto3" json:"goroutines,omitempty"`
@@ -7298,7 +7306,7 @@ const file_volume_server_proto_rawDesc = "" +
 	"\vvolume_info\x18\n" +
 	" \x01(\v2\x1c.volume_server_pb.VolumeInfoR\n" +
 	"volumeInfo\x12\x18\n" +
-	"\aversion\x18\v \x01(\rR\aversion\"\xbb\x01\n" +
+	"\aversion\x18\v \x01(\rR\aversion\"\xd1\x01\n" +
 	"\n" +
 	"DiskStatus\x12\x10\n" +
 	"\x03dir\x18\x01 \x01(\tR\x03dir\x12\x10\n" +
@@ -7307,7 +7315,8 @@ const file_volume_server_proto_rawDesc = "" +
 	"\x04free\x18\x04 \x01(\x04R\x04free\x12!\n" +
 	"\fpercent_free\x18\x05 \x01(\x02R\vpercentFree\x12!\n" +
 	"\fpercent_used\x18\x06 \x01(\x02R\vpercentUsed\x12\x1b\n" +
-	"\tdisk_type\x18\a \x01(\tR\bdiskType\"\xa3\x01\n" +
+	"\tdisk_type\x18\a \x01(\tR\bdiskType\x12\x14\n" +
+	"\x05error\x18\b \x01(\tR\x05error\"\xa3\x01\n" +
 	"\tMemStatus\x12\x1e\n" +
 	"\n" +
 	"goroutines\x18\x01 \x01(\x05R\n" +

weed/server/volume_grpc_erasure_coding_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
@@ -91,6 +92,7 @@ func TestVolumeEcShardsInfo_AggregatesAcrossDisks(t *testing.T) {
 	// cross-disk fallback in NewEcVolume.
 	shardsOnDisk0 := []erasure_coding.ShardId{0, 5}
 	shardsOnDisk1 := []erasure_coding.ShardId{7, 12}
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 
 	store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{dir0, dir1},
@@ -101,6 +103,7 @@ func TestVolumeEcShardsInfo_AggregatesAcrossDisks(t *testing.T) {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {

weed/server/volume_server.go

@@ -83,6 +83,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 	readBufferSizeMB int,
 	ldbTimeout int64,
 	allowUntrustedRemoteEndpoints bool,
+	diskProbeConfig stats.DiskIOProbeConfig,
 ) *VolumeServer {
 
 	v := util.GetViper()
@@ -134,7 +135,7 @@ func NewVolumeServer(adminMux, publicMux *http.ServeMux, ip string,
 
 	vs.checkWithMaster()
 
-	vs.store = storage.NewStore(vs.grpcDialOption, ip, port, grpcPort, publicUrl, id, folders, maxCounts, minFreeSpaces, idxFolder, vs.needleMapKind, diskTypes, diskTags, ldbTimeout)
+	vs.store = storage.NewStore(vs.grpcDialOption, ip, port, grpcPort, publicUrl, id, folders, maxCounts, minFreeSpaces, idxFolder, vs.needleMapKind, diskTypes, diskTags, ldbTimeout, diskProbeConfig)
 	vs.guard = security.NewGuard(whiteList, signingKey, expiresAfterSec, readSigningKey, readExpiresAfterSec)
 
 	handleStaticResources(adminMux)

weed/stats/disk.go

@@ -1,10 +1,105 @@
 package stats
 
 import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
 
+type DiskIOProbeConfig struct {
+	// enable/disable disk io probing
+	Enabled bool
+
+	// timeout for a single io probe
+	Timeout time.Duration
+
+	// probe interval
+	Interval time.Duration
+
+	// latency above this threshold is considered slow
+	SlowLatency time.Duration
+
+	// rolling observation window
+	Window time.Duration
+
+	// minimum number of samples required before evaluating health
+	MinSamples int
+
+	// percentage of slow operations required to mark disk degraded
+	SlowPercent float64
+
+	// percentage of failed operations required to mark disk degraded
+	ErrorPercent float64
+
+	// recovery coef defines the hysteresis threshold for recovering from degraded state.
+	RecoveryCoef float64
+
+	// statfs failures before alerting
+	MaxStatFailures int
+}
+
+func DefaultDiskIOProbeConfig() DiskIOProbeConfig {
+	return DiskIOProbeConfig{
+		Enabled:     false,
+		Timeout:     2 * time.Second,
+		Interval:    30 * time.Second,
+		SlowLatency: 100 * time.Millisecond,
+
+		Window:     time.Minute,
+		MinSamples: 10,
+
+		SlowPercent:  20,
+		ErrorPercent: 10,
+		RecoveryCoef: 0.5,
+
+		MaxStatFailures: 3,
+	}
+}
+
+type ioSample struct {
+	ts      time.Time
+	latency time.Duration
+	failed  bool
+	slow    bool
+}
+
+type diskState struct {
+	mu sync.Mutex
+
+	// statfs probe
+	isChecking       bool
+	statFailureCount int
+	statSuccessCount int
+	lastStatErr      error
+	lastGoodStatus   volume_server_pb.DiskStatus
+	hasLastGood      bool
+
+	samples []ioSample
+
+	// IO probe
+	isIOChecking   bool
+	ioCheckID      uint64
+	lastIOCheck    time.Time
+	ioFailureCount int
+	lastIOErr      error
+	ioDegraded     bool
+}
+
+var diskRegistry sync.Map
+
+const (
+	// statfs timeout
+	diskTimeout = 500 * time.Millisecond
+)
+
 func NewDiskStatus(path string) (disk *volume_server_pb.DiskStatus) {
 	disk = &volume_server_pb.DiskStatus{Dir: path}
 	fillInDiskStatus(disk)
@@ -13,3 +108,395 @@ func NewDiskStatus(path string) (disk *volume_server_pb.DiskStatus) {
 	}
 	return
 }
+
+func NewDiskStatusOnStart(path string, config DiskIOProbeConfig) (disk *volume_server_pb.DiskStatus) {
+	disk = &volume_server_pb.DiskStatus{Dir: path}
+	diskProbe(disk, config)
+	if disk.PercentUsed > 95 {
+		glog.V(0).Infof("disk status: %v", disk)
+	}
+	return
+}
+
+func diskProbe(disk *volume_server_pb.DiskStatus, config DiskIOProbeConfig) {
+	actual, _ := diskRegistry.LoadOrStore(disk.Dir, &diskState{})
+	state := actual.(*diskState)
+
+	state.mu.Lock()
+	if state.isChecking {
+		state.statFailureCount++
+		state.statSuccessCount = 0
+		state.lastStatErr = errors.New("statfs still in progress")
+		state.updateErrorLocked(disk, config)
+		state.applyLastGoodStatusLocked(disk)
+		state.mu.Unlock()
+		return
+	}
+	state.isChecking = true
+	state.mu.Unlock()
+
+	ctx, cancel := context.WithTimeout(context.Background(), diskTimeout)
+	defer cancel()
+
+	probe := &volume_server_pb.DiskStatus{Dir: disk.Dir}
+	ch := make(chan error, 1)
+	go func() {
+		err := fillInDiskStatus(probe)
+		// Clear the flag only once statfs has actually returned, so a stuck disk
+		// keeps a single outstanding probe instead of spawning a new one each tick.
+		state.mu.Lock()
+		state.isChecking = false
+		state.mu.Unlock()
+		ch <- err
+	}()
+
+	var probeErr error
+	select {
+	case probeErr = <-ch:
+	case <-ctx.Done():
+		// Leave isChecking set; the probe goroutine clears it when statfs returns.
+		state.mu.Lock()
+		state.statFailureCount++
+		state.statSuccessCount = 0
+		state.lastStatErr = errors.New("statfs timeout")
+		state.updateErrorLocked(disk, config)
+		state.applyLastGoodStatusLocked(disk)
+		state.mu.Unlock()
+		return
+	}
+
+	state.mu.Lock()
+
+	if probeErr != nil {
+		state.statFailureCount++
+		state.statSuccessCount = 0
+		state.lastStatErr = probeErr
+		state.updateErrorLocked(disk, config)
+		state.applyLastGoodStatusLocked(disk)
+		state.mu.Unlock()
+		return
+	}
+
+	state.observeStatSuccessLocked(config)
+	disk.All = probe.All
+	disk.Free = probe.Free
+	disk.Used = probe.Used
+	disk.PercentFree = probe.PercentFree
+	disk.PercentUsed = probe.PercentUsed
+	state.rememberGoodStatusLocked(disk)
+
+	if !config.Enabled {
+		state.ioFailureCount = 0
+		state.lastIOErr = nil
+		state.ioDegraded = false
+		state.updateErrorLocked(disk, config)
+		state.mu.Unlock()
+		return
+	}
+
+	shouldCheckIO := false
+	ioCheckID := uint64(0)
+
+	if time.Since(state.lastIOCheck) >= config.Interval {
+		if state.isIOChecking {
+			degraded := state.observeLatencyLocked(
+				config.Timeout,
+				true,
+				config,
+			)
+
+			state.ioDegraded = degraded
+			state.lastIOCheck = time.Now()
+
+			if degraded {
+				state.ioFailureCount++
+				state.lastIOErr = errors.New("disk io still in progress")
+			} else {
+				state.ioFailureCount = 0
+				state.lastIOErr = nil
+			}
+		} else {
+			state.isIOChecking = true
+			state.ioCheckID++
+			ioCheckID = state.ioCheckID
+			state.lastIOCheck = time.Now()
+			shouldCheckIO = true
+		}
+	}
+
+	state.updateErrorLocked(disk, config)
+	state.mu.Unlock()
+
+	if shouldCheckIO {
+		checkDiskIOLatency(state, disk, ioCheckID, config)
+	}
+}
+
+type ioProbeResult struct {
+	err     error
+	latency time.Duration
+}
+
+func checkDiskIOLatency(state *diskState, disk *volume_server_pb.DiskStatus, checkID uint64, config DiskIOProbeConfig) {
+	ctx, cancel := context.WithTimeout(context.Background(), config.Timeout)
+	defer cancel()
+
+	ch := make(chan ioProbeResult, 1)
+
+	go func() {
+		start := time.Now()
+		err := runIOProbe(disk.Dir)
+		ch <- ioProbeResult{
+			err:     err,
+			latency: time.Since(start),
+		}
+
+		state.mu.Lock()
+		if state.ioCheckID == checkID {
+			state.isIOChecking = false
+			state.lastIOCheck = time.Now()
+		}
+		state.mu.Unlock()
+	}()
+
+	select {
+	case result := <-ch:
+		state.mu.Lock()
+		defer state.mu.Unlock()
+
+		if state.ioCheckID != checkID {
+			return
+		}
+
+		state.isIOChecking = false
+		state.lastIOCheck = time.Now()
+
+		if result.err != nil {
+			degraded := state.observeLatencyLocked(
+				config.Timeout,
+				true,
+				config,
+			)
+
+			state.ioDegraded = degraded
+
+			if degraded {
+				state.ioFailureCount++
+				state.lastIOErr = result.err
+			} else {
+				state.ioFailureCount = 0
+				state.lastIOErr = nil
+			}
+
+			state.updateErrorLocked(disk, config)
+			return
+		}
+
+		degraded := state.observeLatencyLocked(
+			result.latency,
+			false,
+			config,
+		)
+
+		state.ioDegraded = degraded
+
+		if degraded {
+			state.ioFailureCount++
+			state.lastIOErr = fmt.Errorf(
+				"disk io degradation detected latency=%v",
+				result.latency,
+			)
+		} else {
+			state.ioFailureCount = 0
+			state.lastIOErr = nil
+		}
+
+		state.updateErrorLocked(disk, config)
+
+		glog.V(1).Infof(
+			"disk io latency dir=%s latency=%v ioFailureCount=%d degraded=%v samples=%d error=%v",
+			disk.Dir,
+			result.latency,
+			state.ioFailureCount,
+			degraded,
+			len(state.samples),
+			state.lastIOErr,
+		)
+
+	case <-ctx.Done():
+		state.mu.Lock()
+
+		if state.ioCheckID == checkID {
+			degraded := state.observeLatencyLocked(
+				config.Timeout,
+				true,
+				config,
+			)
+
+			state.ioDegraded = degraded
+
+			if degraded {
+				state.ioFailureCount++
+				state.lastIOErr = errors.New("disk io timeout")
+			} else {
+				state.ioFailureCount = 0
+				state.lastIOErr = nil
+			}
+
+			state.lastIOCheck = time.Now()
+			state.updateErrorLocked(disk, config)
+		}
+
+		state.mu.Unlock()
+	}
+}
+
+func runIOProbe(dir string) error {
+	path := filepath.Join(dir, ".disk-health-check")
+
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0644)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	if _, err := f.WriteAt(make([]byte, 4096), 0); err != nil {
+		return err
+	}
+
+	return f.Sync()
+}
+
+func (s *diskState) observeLatencyLocked(latency time.Duration, failed bool, config DiskIOProbeConfig) bool {
+	now := time.Now()
+
+	sample := ioSample{
+		ts:      now,
+		latency: latency,
+		failed:  failed,
+		slow:    latency >= config.SlowLatency,
+	}
+
+	s.samples = append(s.samples, sample)
+
+	cutoff := now.Add(-config.Window)
+
+	idx := 0
+	for _, sample := range s.samples {
+		if sample.ts.After(cutoff) {
+			s.samples[idx] = sample
+			idx++
+		}
+	}
+
+	s.samples = s.samples[:idx]
+
+	total := len(s.samples)
+
+	if total < config.MinSamples {
+		return s.ioDegraded
+	}
+
+	var slowCount int
+	var failedCount int
+
+	for _, sample := range s.samples {
+		if sample.slow {
+			slowCount++
+		}
+
+		if sample.failed {
+			failedCount++
+		}
+	}
+
+	slowPercent := float64(slowCount) / float64(total) * 100
+	errorPercent := float64(failedCount) / float64(total) * 100
+
+	if !s.ioDegraded {
+		return slowPercent >= config.SlowPercent ||
+			errorPercent >= config.ErrorPercent
+	}
+
+	recoverSlowPercent := config.SlowPercent * config.RecoveryCoef
+	recoverErrorPercent := config.ErrorPercent * config.RecoveryCoef
+
+	return slowPercent > recoverSlowPercent ||
+		errorPercent > recoverErrorPercent
+}
+
+func (s *diskState) observeStatSuccessLocked(config DiskIOProbeConfig) {
+	if s.statFailureCount < config.MaxStatFailures {
+		s.statFailureCount = 0
+		s.statSuccessCount = 0
+		s.lastStatErr = nil
+		return
+	}
+
+	s.statSuccessCount++
+	if s.statSuccessCount >= statRecoverySuccesses(config) {
+		s.statFailureCount = 0
+		s.statSuccessCount = 0
+		s.lastStatErr = nil
+	}
+}
+
+func statRecoverySuccesses(config DiskIOProbeConfig) int {
+	if config.MaxStatFailures <= 1 {
+		return 1
+	}
+	if config.RecoveryCoef <= 0 {
+		return config.MaxStatFailures * 2
+	}
+
+	successes := int(float64(config.MaxStatFailures) * (1/config.RecoveryCoef - 1))
+	if successes < 1 {
+		return 1
+	}
+	return successes
+}
+
+func (s *diskState) updateErrorLocked(disk *volume_server_pb.DiskStatus, config DiskIOProbeConfig) {
+	var problems []string
+
+	if s.statFailureCount >= config.MaxStatFailures &&
+		s.lastStatErr != nil {
+		problems = append(
+			problems,
+			fmt.Sprintf("statfs: %v", s.lastStatErr),
+		)
+	}
+
+	if config.Enabled && s.ioDegraded && s.lastIOErr != nil {
+		problems = append(
+			problems,
+			fmt.Sprintf("io: %v", s.lastIOErr),
+		)
+	}
+
+	if len(problems) == 0 {
+		disk.Error = ""
+		return
+	}
+
+	disk.Error =
+		"disk health check failed: " +
+			strings.Join(problems, "; ")
+}
+
+func (s *diskState) rememberGoodStatusLocked(disk *volume_server_pb.DiskStatus) {
+	s.lastGoodStatus = *disk
+	s.hasLastGood = true
+}
+
+func (s *diskState) applyLastGoodStatusLocked(disk *volume_server_pb.DiskStatus) {
+	if !s.hasLastGood || disk.Error != "" {
+		return
+	}
+
+	dir := disk.Dir
+	errorMessage := disk.Error
+	*disk = s.lastGoodStatus
+	disk.Dir = dir
+	disk.Error = errorMessage
+}

weed/stats/disk_notsupported.go

@@ -4,6 +4,6 @@ package stats
 
 import "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 
-func fillInDiskStatus(status *volume_server_pb.DiskStatus) {
-	return
+func fillInDiskStatus(status *volume_server_pb.DiskStatus) error {
+	return nil
 }

weed/stats/disk_openbsd.go

@@ -8,11 +8,10 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
 
-func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
+func fillInDiskStatus(disk *volume_server_pb.DiskStatus) error {
 	fs := syscall.Statfs_t{}
-	err := syscall.Statfs(disk.Dir, &fs)
-	if err != nil {
-		return
+	if err := syscall.Statfs(disk.Dir, &fs); err != nil {
+		return err
 	}
 
 	disk.All = fs.F_blocks * uint64(fs.F_bsize)
@@ -21,5 +20,5 @@ func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
 	disk.PercentFree = float32((float64(disk.Free) / float64(disk.All)) * 100)
 	disk.PercentUsed = float32((float64(disk.Used) / float64(disk.All)) * 100)
 
-	return
+	return nil
 }

weed/stats/disk_solaris.go

@@ -8,11 +8,10 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
 
-func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
+func fillInDiskStatus(disk *volume_server_pb.DiskStatus) error {
 	var stat unix.Statvfs_t
-	err := unix.Statvfs(disk.Dir, &stat)
-	if err != nil {
-		return
+	if err := unix.Statvfs(disk.Dir, &stat); err != nil {
+		return err
 	}
 
 	disk.All = stat.Blocks * uint64(stat.Bsize)
@@ -21,5 +20,5 @@ func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
 	disk.PercentFree = float32((float64(disk.Free) / float64(disk.All)) * 100)
 	disk.PercentUsed = float32((float64(disk.Used) / float64(disk.All)) * 100)
 
-	return
+	return nil
 }

weed/stats/disk_supported.go

@@ -8,20 +8,18 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
 
-func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
+func fillInDiskStatus(disk *volume_server_pb.DiskStatus) error {
 	fs := syscall.Statfs_t{}
-	err := syscall.Statfs(disk.Dir, &fs)
-	if err != nil {
-		return
+	if err := syscall.Statfs(disk.Dir, &fs); err != nil {
+		return err
 	}
 	disk.All = fs.Blocks * uint64(fs.Bsize)
 	// https://man7.org/linux/man-pages/man3/statvfs.3.html
 	// fs.Bfree: Number of free blocks
 	// fs.Bavail: Number of free blocks for unprivileged users
-	// disk.Free = fs.Bfree * uint64(fs.Bsize)
 	disk.Free = uint64(fs.Bavail) * uint64(fs.Bsize)
 	disk.Used = disk.All - disk.Free
-	disk.PercentFree = float32((float64(disk.Free) / float64(disk.All)) * 100)
-	disk.PercentUsed = float32((float64(disk.Used) / float64(disk.All)) * 100)
-	return
+	disk.PercentFree = float32(float64(disk.Free) / float64(disk.All) * 100)
+	disk.PercentUsed = float32(float64(disk.Used) / float64(disk.All) * 100)
+	return nil
 }

weed/stats/disk_test.go

@@ -0,0 +1,69 @@
+//go:build !netbsd && !plan9
+
+package stats
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+)
+
+func TestSetDiskStatusSuccess(t *testing.T) {
+	disk := &volume_server_pb.DiskStatus{Dir: t.TempDir()}
+	diskProbe(disk, DefaultDiskIOProbeConfig())
+
+	if disk.Error != "" {
+		t.Fatalf("unexpected disk error: %s", disk.Error)
+	}
+	if disk.All == 0 {
+		t.Fatalf("expected non-zero capacity for a real directory")
+	}
+}
+
+func TestSetDiskStatusReportsRepeatedFailures(t *testing.T) {
+	missing := filepath.Join(t.TempDir(), "does-not-exist")
+	config := DefaultDiskIOProbeConfig()
+	var disk *volume_server_pb.DiskStatus
+	for i := 0; i < config.MaxStatFailures; i++ {
+		disk = &volume_server_pb.DiskStatus{Dir: missing}
+		diskProbe(disk, config)
+	}
+
+	if disk.Error == "" {
+		t.Fatalf("expected disk error after %d failed checks", config.MaxStatFailures)
+	}
+}
+
+func TestSetDiskStatusRequiresRepeatedSuccessesToRecover(t *testing.T) {
+	missing := filepath.Join(t.TempDir(), "temporarily-missing")
+	config := DefaultDiskIOProbeConfig()
+	var disk *volume_server_pb.DiskStatus
+	for i := 0; i < config.MaxStatFailures; i++ {
+		disk = &volume_server_pb.DiskStatus{Dir: missing}
+		diskProbe(disk, config)
+	}
+	if disk.Error == "" {
+		t.Fatalf("expected disk error after %d failed checks", config.MaxStatFailures)
+	}
+
+	if err := os.Mkdir(missing, 0755); err != nil {
+		t.Fatalf("mkdir %s: %v", missing, err)
+	}
+
+	recoveryChecks := statRecoverySuccesses(config)
+	for i := 1; i < recoveryChecks; i++ {
+		disk = &volume_server_pb.DiskStatus{Dir: missing}
+		diskProbe(disk, config)
+		if disk.Error == "" {
+			t.Fatalf("expected disk error to persist after %d successful checks", i)
+		}
+	}
+
+	disk = &volume_server_pb.DiskStatus{Dir: missing}
+	diskProbe(disk, config)
+	if disk.Error != "" {
+		t.Fatalf("expected disk error to recover after %d successful checks: %s", recoveryChecks, disk.Error)
+	}
+}

weed/stats/disk_windows.go

@@ -1,6 +1,7 @@
 package stats
 
 import (
+	"fmt"
 	"syscall"
 	"unsafe"
 
@@ -13,12 +14,12 @@ var (
 	getDiskFreeSpaceEx = kernel32.NewProc("GetDiskFreeSpaceExW")
 )
 
-func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
+func fillInDiskStatus(disk *volume_server_pb.DiskStatus) error {
 
 	ptr, err := syscall.UTF16PtrFromString(disk.Dir)
 
 	if err != nil {
-		return
+		return err
 	}
 	var _temp uint64
 	/* #nosec */
@@ -35,14 +36,13 @@ func fillInDiskStatus(disk *volume_server_pb.DiskStatus) {
 
 	if r == 0 {
 		if e != 0 {
-			return
+			return e
 		}
-
-		return
+		return fmt.Errorf("GetDiskFreeSpaceExW failed for %s", disk.Dir)
 	}
 	disk.Used = disk.All - disk.Free
 	disk.PercentFree = float32((float64(disk.Free) / float64(disk.All)) * 100)
 	disk.PercentUsed = float32((float64(disk.Used) / float64(disk.All)) * 100)
 
-	return
+	return nil
 }

weed/stats/metrics.go

@@ -330,6 +330,14 @@ var (
 			Help:      "Resource usage",
 		}, []string{"name", "type"})
 
+	VolumeServerDiskErrorGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: Namespace,
+			Subsystem: "volumeServer",
+			Name:      "disk_error_status",
+			Help:      "Disk error status",
+		}, []string{"name", "type"})
+
 	VolumeServerConcurrentDownloadLimit = prometheus.NewGauge(
 		prometheus.GaugeOpts{
 			Namespace: Namespace,
@@ -759,6 +767,7 @@ func init() {
 	Gather.MustRegister(VolumeServerReadOnlyVolumeGauge)
 	Gather.MustRegister(VolumeServerDiskSizeGauge)
 	Gather.MustRegister(VolumeServerResourceGauge)
+	Gather.MustRegister(VolumeServerDiskErrorGauge)
 	Gather.MustRegister(VolumeServerConcurrentDownloadLimit)
 	Gather.MustRegister(VolumeServerConcurrentUploadLimit)
 	Gather.MustRegister(VolumeServerInFlightDownloadSize)

weed/storage/disk_location.go

@@ -45,8 +45,9 @@ type DiskLocation struct {
 
 	ecShardNotifyHandler func(collection string, vid needle.VolumeId, shardId erasure_coding.ShardId, ecVolume *erasure_coding.EcVolume)
 
-	isDiskSpaceLow bool
-	closeCh        chan struct{}
+	isDiskSpaceLow    atomic.Bool
+	isDiskUnavailable atomic.Bool
+	closeCh           chan struct{}
 }
 
 func GenerateDirUuid(dir string) (dirUuidString string, err error) {
@@ -77,7 +78,7 @@ func writeNewUuid(fileName string) (string, error) {
 	return dirUuidString, nil
 }
 
-func NewDiskLocation(dir string, maxVolumeCount int32, minFreeSpace util.MinFreeSpace, idxDir string, diskType types.DiskType, tags []string) *DiskLocation {
+func NewDiskLocation(dir string, maxVolumeCount int32, minFreeSpace util.MinFreeSpace, idxDir string, diskType types.DiskType, tags []string, config stats.DiskIOProbeConfig) *DiskLocation {
 	glog.V(4).Infof("Added new Disk %s: maxVolumes=%d", dir, maxVolumeCount)
 	dir = util.ResolvePath(dir)
 	if idxDir == "" {
@@ -109,13 +110,13 @@ func NewDiskLocation(dir string, maxVolumeCount int32, minFreeSpace util.MinFree
 	location.ecVolumes = make(map[needle.VolumeId]*erasure_coding.EcVolume)
 	location.closeCh = make(chan struct{})
 	go func() {
-		location.CheckDiskSpace()
+		location.CheckDiskSpace(config)
 		for {
 			select {
 			case <-location.closeCh:
 				return
 			case <-time.After(time.Minute):
-				location.CheckDiskSpace()
+				location.CheckDiskSpace(config)
 			}
 		}
 	}()
@@ -539,9 +540,17 @@ func (l *DiskLocation) UnUsedSpace(volumeSizeLimit uint64) (unUsedSpace uint64)
 	return
 }
 
-func (l *DiskLocation) CheckDiskSpace() {
+func (l *DiskLocation) CheckDiskSpace(config stats.DiskIOProbeConfig) {
 	if dir, e := filepath.Abs(l.Directory); e == nil {
-		s := stats.NewDiskStatus(dir)
+		s := stats.NewDiskStatusOnStart(dir, config)
+		if len(s.Error) != 0 {
+			l.isDiskUnavailable.Store(true)
+			stats.VolumeServerDiskErrorGauge.WithLabelValues(l.Directory, "error").Set(1)
+			glog.V(1).Infof("disk %s is not healthy: %s", dir, s.Error)
+		} else {
+			l.isDiskUnavailable.Store(false)
+			stats.VolumeServerDiskErrorGauge.WithLabelValues(l.Directory, "error").Set(0)
+		}
 		available := l.MinFreeSpace.AvailableSpace(s.Free, s.All)
 		stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "all").Set(float64(s.All))
 		stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "used").Set(float64(s.Used))
@@ -549,12 +558,12 @@ func (l *DiskLocation) CheckDiskSpace() {
 		stats.VolumeServerResourceGauge.WithLabelValues(l.Directory, "avail").Set(float64(available))
 		l.AvailableSpace.Store(available)
 		isLow, desc := l.MinFreeSpace.IsLow(s.Free, s.PercentFree)
-		if isLow != l.isDiskSpaceLow {
-			l.isDiskSpaceLow = !l.isDiskSpaceLow
+		if isLow != l.isDiskSpaceLow.Load() {
+			l.isDiskSpaceLow.Store(isLow)
 		}
 
 		logLevel := glog.Level(4)
-		if l.isDiskSpaceLow {
+		if l.isDiskSpaceLow.Load() {
 			logLevel = glog.Level(0)
 		}
 

weed/storage/store.go

@@ -94,6 +94,7 @@ func NewStore(
 	diskTypes []DiskType,
 	diskTags [][]string,
 	ldbTimeout int64,
+	diskProbeConfig stats.DiskIOProbeConfig,
 ) (s *Store) {
 	s = &Store{
 		grpcDialOption: grpcDialOption,
@@ -118,7 +119,7 @@ func NewStore(
 		if i < len(diskTags) {
 			tags = diskTags[i]
 		}
-		location := NewDiskLocation(dirnames[i], int32(maxVolumeCounts[i]), minFreeSpaces[i], idxFolder, diskTypes[i], tags)
+		location := NewDiskLocation(dirnames[i], int32(maxVolumeCounts[i]), minFreeSpaces[i], idxFolder, diskTypes[i], tags, diskProbeConfig)
 		s.Locations = append(s.Locations, location)
 		stats.VolumeServerMaxVolumeCounter.Add(float64(maxVolumeCounts[i]))
 
@@ -261,7 +262,7 @@ func (s *Store) FindFreeLocation(filterFn func(location *DiskLocation) bool) (re
 		if filterFn != nil && !filterFn(location) {
 			continue
 		}
-		if location.isDiskSpaceLow {
+		if location.isDiskSpaceLow.Load() {
 			continue
 		}
 		currentFreeCount := location.MaxVolumeCount - int32(location.VolumesLen())
@@ -322,7 +323,12 @@ func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind
 // hasFreeDiskLocation checks if a disk location has free space
 func (s *Store) hasFreeDiskLocation(location *DiskLocation) bool {
 	// Check if disk space is low first
-	if location.isDiskSpaceLow {
+	if location.isDiskSpaceLow.Load() {
+		return false
+	}
+
+	// Check if disk is available
+	if location.isDiskUnavailable.Load() {
 		return false
 	}
 
@@ -381,7 +387,6 @@ func collectStatForOneVolume(vid needle.VolumeId, v *Volume) (s *VolumeInfo) {
 	s.DeleteCount = v.nm.DeletedCount()
 	s.DeletedByteCount = v.nm.DeletedSize()
 	s.Size = v.nm.ContentSize()
-
 	return
 }
 
@@ -406,9 +411,12 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 	collectionVolumeDeletedBytes := make(map[string]int64)
 	collectionVolumeReadOnlyCount := make(map[string]map[string]uint8)
 	for _, location := range s.Locations {
+		if location.isDiskUnavailable.Load() {
+			continue
+		}
 		var deleteVids []needle.VolumeId
 		effectiveMaxCount := location.MaxVolumeCount
-		if location.isDiskSpaceLow {
+		if location.isDiskSpaceLow.Load() {
 			usedSlots := int32(location.LocalVolumesLen())
 			ecShardCount := location.EcShardCount()
 			usedSlots += int32((ecShardCount + erasure_coding.DataShardsCount - 1) / erasure_coding.DataShardsCount)
@@ -496,7 +504,7 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 				if v.noWriteCanDelete {
 					collectionVolumeReadOnlyCount[v.Collection][stats.NoWriteCanDelete] += 1
 				}
-				if v.location.isDiskSpaceLow {
+				if v.location.isDiskSpaceLow.Load() {
 					collectionVolumeReadOnlyCount[v.Collection][stats.IsDiskSpaceLow] += 1
 				}
 			}
@@ -571,6 +579,10 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
 
 func (s *Store) deleteExpiredEcVolumes() (ecShards, deleted []*master_pb.VolumeEcShardInformationMessage) {
 	for diskId, location := range s.Locations {
+		if location.isDiskUnavailable.Load() {
+			continue
+		}
+
 		// Collect ecVolume to be deleted
 		var toDeleteEvs []*erasure_coding.EcVolume
 		location.ecVolumesLock.RLock()

weed/storage/store_disk_space_test.go

@@ -72,9 +72,9 @@ func TestHasFreeDiskLocation(t *testing.T) {
 			// setup
 			diskLocation := &DiskLocation{
 				volumes:        make(map[needle.VolumeId]*Volume),
-				isDiskSpaceLow: tc.isDiskSpaceLow,
 				MaxVolumeCount: tc.maxVolumeCount,
 			}
+			diskLocation.isDiskSpaceLow.Store(tc.isDiskSpaceLow)
 			for i := 0; i < tc.currentVolumes; i++ {
 				diskLocation.volumes[needle.VolumeId(i+1)] = &Volume{}
 			}
@@ -89,7 +89,7 @@ func TestHasFreeDiskLocation(t *testing.T) {
 			// assert
 			if result != tc.expected {
 				t.Errorf("Expected hasFreeDiskLocation() = %v; want %v for volumes:%d/%d, lowSpace:%v",
-					result, tc.expected, len(diskLocation.volumes), diskLocation.MaxVolumeCount, diskLocation.isDiskSpaceLow)
+					result, tc.expected, len(diskLocation.volumes), diskLocation.MaxVolumeCount, diskLocation.isDiskSpaceLow.Load())
 			}
 		})
 	}
@@ -101,8 +101,8 @@ func newTestLocation(maxCount int32, isDiskLow bool, volCount int) *DiskLocation
 		ecVolumes:      make(map[needle.VolumeId]*erasure_coding.EcVolume),
 		MaxVolumeCount: maxCount,
 		DiskType:       types.ToDiskType("hdd"),
-		isDiskSpaceLow: isDiskLow,
 	}
+	location.isDiskSpaceLow.Store(isDiskLow)
 	for i := 1; i <= volCount; i++ {
 		location.volumes[needle.VolumeId(i)] = &Volume{}
 	}

weed/storage/store_ec.go

@@ -70,7 +70,7 @@ func (s *Store) FindEcShardTargetLocation(collection string, vid needle.VolumeId
 		bestFree int32
 	)
 	for _, loc := range s.Locations {
-		if loc.isDiskSpaceLow {
+		if loc.isDiskSpaceLow.Load() {
 			continue
 		}
 		freeCount := ecFreeShardCount(loc, dataShardCount)

weed/storage/store_ec_9plus3_reboot_repro_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -114,7 +115,7 @@ func TestEc9plus3MultiDiskRebootLoadsAllShards(t *testing.T) {
 	}
 
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
-		dirs, maxCounts, minFree, "", NeedleMapInMemory, diskTypes, nil, 3)
+		dirs, maxCounts, minFree, "", NeedleMapInMemory, diskTypes, nil, 3, stats.DefaultDiskIOProbeConfig())
 	drainStoreChans(t, store)
 
 	missing := []int{}
@@ -216,7 +217,7 @@ func TestEc9plus3ValidateKeepsVolumeWithDat(t *testing.T) {
 
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "",
-		NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3)
+		NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3, stats.DefaultDiskIOProbeConfig())
 	drainStoreChans(t, store)
 
 	// HEAD: validateEcVolume uses the 9+3 ratio, the size check passes, and the
@@ -306,7 +307,7 @@ func TestEc9plus3PruneKeepsFullDataShardSet(t *testing.T) {
 
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{dir0, dir1}, []int32{100, 100}, []util.MinFreeSpace{{}, {}}, "",
-		NeedleMapInMemory, []types.DiskType{types.HardDriveType, types.HardDriveType}, nil, 3)
+		NeedleMapInMemory, []types.DiskType{types.HardDriveType, types.HardDriveType}, nil, 3, stats.DefaultDiskIOProbeConfig())
 	drainStoreChans(t, store)
 
 	loc0 := store.Locations[0]

weed/storage/store_ec_disk_type_test.go

@@ -8,6 +8,7 @@ import (
 
 	"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/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -41,6 +42,7 @@ func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *[
 	const dataShards, parityShards = 10, 4
 	const datSize int64 = 10 * 1024 * 1024
 	expectedShardSize := calculateExpectedShardSize(datSize, dataShards)
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 
 	store = NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{hddDir, ssdDir},
@@ -51,6 +53,7 @@ func setupECStoreWithMixedDisks(t *testing.T) (store *Store, drainedShardMsgs *[
 		[]types.DiskType{types.HardDriveType, types.SsdType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 
 	captured := []master_pb.VolumeEcShardInformationMessage{}

weed/storage/store_ec_mirror_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -70,6 +71,8 @@ func TestMirrorEcMetadataOnStartup_PhysicallyCopiesSidecars(t *testing.T) {
 		t.Fatalf("save .vif: %v", err)
 	}
 
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
+
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{dir0, dir1},
 		[]int32{100, 100},
@@ -79,6 +82,7 @@ func TestMirrorEcMetadataOnStartup_PhysicallyCopiesSidecars(t *testing.T) {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {
@@ -230,6 +234,8 @@ func TestMirrorEcMetadataOnStartup_NoOpWhenAlreadyMirrored(t *testing.T) {
 
 	base0 := erasure_coding.EcShardFileName(collection, dir0, int(vid))
 
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
+
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{dir0, dir1},
 		[]int32{100, 100},
@@ -239,6 +245,7 @@ func TestMirrorEcMetadataOnStartup_NoOpWhenAlreadyMirrored(t *testing.T) {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {

weed/storage/store_ec_mount_cross_disk_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -45,6 +46,7 @@ func TestMountEcShards_LocatesEcxOnSiblingDisk(t *testing.T) {
 	const datSize int64 = 10 * 1024 * 1024
 	expectedShardSize := calculateExpectedShardSize(datSize, dataShards)
 	const shardOnDisk0 erasure_coding.ShardId = 6
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{dir0, dir1},
@@ -55,6 +57,7 @@ func TestMountEcShards_LocatesEcxOnSiblingDisk(t *testing.T) {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {
@@ -166,6 +169,7 @@ func TestMountEcShards_SameDiskEcxStillWorks(t *testing.T) {
 	const datSize int64 = 10 * 1024 * 1024
 	expectedShardSize := calculateExpectedShardSize(datSize, dataShards)
 	const shardOnDisk0 erasure_coding.ShardId = 3
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{dir0, dir1},
@@ -176,6 +180,7 @@ func TestMountEcShards_SameDiskEcxStillWorks(t *testing.T) {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {
@@ -248,6 +253,7 @@ func startEcMountStore(t *testing.T, dirs []string) *Store {
 			t.Fatalf("mkdir %s: %v", d, err)
 		}
 	}
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		dirs,
 		[]int32{100, 100},
@@ -257,6 +263,7 @@ func startEcMountStore(t *testing.T, dirs []string) *Store {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {

weed/storage/store_ec_orphan_shard_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -99,6 +100,8 @@ func TestLoadEcShardsWhenIndexFilesOnDifferentDisk(t *testing.T) {
 		t.Fatalf("save .vif: %v", err)
 	}
 
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
+
 	// Build the Store with both disks. NewStore triggers per-disk loading
 	// during construction, which is the codepath under test.
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
@@ -110,6 +113,7 @@ func TestLoadEcShardsWhenIndexFilesOnDifferentDisk(t *testing.T) {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 
 	// Drain heartbeat-style channels so loading does not block.
@@ -201,6 +205,7 @@ func TestLoadEcShardsOrphanWithoutSiblingEcx(t *testing.T) {
 		f.Close()
 	}
 
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{dir0, dir1},
 		[]int32{100, 100},
@@ -210,6 +215,7 @@ func TestLoadEcShardsOrphanWithoutSiblingEcx(t *testing.T) {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {
@@ -303,6 +309,7 @@ func TestReconcileNoOpWhenEachDiskIsSelfContained(t *testing.T) {
 	// should leave them untouched.
 	writeFullEcLayout(dir0, []int{0, 4})
 	writeFullEcLayout(dir1, []int{8, 12})
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
 		[]string{dir0, dir1},
@@ -313,6 +320,7 @@ func TestReconcileNoOpWhenEachDiskIsSelfContained(t *testing.T) {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {
@@ -413,6 +421,7 @@ func TestLoadEcShardsWhenOwnerEcxIsInDataDir(t *testing.T) {
 		t.Fatalf("save .vif in data dir: %v", err)
 	}
 
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 	// idxDir is configured but intentionally empty for this volume — we
 	// want IdxDirectory != Directory while the .ecx lives in Directory.
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
@@ -424,6 +433,7 @@ func TestLoadEcShardsWhenOwnerEcxIsInDataDir(t *testing.T) {
 		[]types.DiskType{types.HardDriveType, types.HardDriveType},
 		nil,
 		3,
+		diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {

weed/storage/store_ec_target_location_test.go

@@ -5,6 +5,7 @@ import (
 	"path/filepath"
 	"testing"
 
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -168,8 +169,9 @@ func newEcTargetTestStore(t *testing.T, numDirs int) *Store {
 		minFreeSpaces = append(minFreeSpaces, util.MinFreeSpace{})
 		diskTypes = append(diskTypes, types.HardDriveType)
 	}
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id",
-		dirs, maxCounts, minFreeSpaces, "", NeedleMapInMemory, diskTypes, nil, 3,
+		dirs, maxCounts, minFreeSpaces, "", NeedleMapInMemory, diskTypes, nil, 3, diskIOProbeConfig,
 	)
 	done := make(chan struct{})
 	go func() {

weed/storage/store_load_balancing_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
@@ -30,9 +31,9 @@ func newTestStore(t *testing.T, numDirs int) *Store {
 		minFreeSpaces = append(minFreeSpaces, util.MinFreeSpace{})
 		diskTypes = append(diskTypes, types.HardDriveType)
 	}
-
+	diskIOProbeConfig := stats.DefaultDiskIOProbeConfig()
 	store := NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "",
-		dirs, maxCounts, minFreeSpaces, "", NeedleMapInMemory, diskTypes, nil, 3)
+		dirs, maxCounts, minFreeSpaces, "", NeedleMapInMemory, diskTypes, nil, 3, diskIOProbeConfig)
 
 	// Consume channel messages to prevent blocking
 	done := make(chan bool)

weed/storage/store_maxvolume_deadzone_test.go

@@ -33,7 +33,7 @@ func TestMaybeAdjustVolumeMaxNoDeadZone(t *testing.T) {
 				t.Skip("free disk too small for this ratio")
 			}
 
-			loc := NewDiskLocation(dir, 0 /* auto */, util.MinFreeSpace{}, "", types.HddType, nil)
+			loc := NewDiskLocation(dir, 0 /* auto */, util.MinFreeSpace{}, "", types.HddType, nil, stats.DefaultDiskIOProbeConfig())
 			defer loc.Close()
 			s := &Store{Locations: []*DiskLocation{loc}}
 			s.SetVolumeSizeLimit(volumeSizeLimit)

weed/storage/volume.go

@@ -434,7 +434,7 @@ func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
 func (v *Volume) IsReadOnly() bool {
 	v.noWriteLock.RLock()
 	defer v.noWriteLock.RUnlock()
-	return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow
+	return v.noWriteOrDelete || v.noWriteCanDelete || v.location.isDiskSpaceLow.Load()
 }
 
 func (v *Volume) PersistReadOnly(readOnly bool) {

weed/storage/volume_vacuum_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/idx"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
@@ -201,7 +202,7 @@ func TestCommitCompactDeletionTailKeepsWritable(t *testing.T) {
 
 func TestCompactVolumeFilesOffline(t *testing.T) {
 	dir := t.TempDir()
-	location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil)
+	location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig())
 	defer location.Close()
 
 	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)
@@ -236,7 +237,7 @@ func TestCompactVolumeFilesOffline(t *testing.T) {
 
 func TestCleanupCompactRemovesTempFiles(t *testing.T) {
 	dir := t.TempDir()
-	location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil)
+	location := NewDiskLocation(dir, 10, util.MinFreeSpace{}, dir, "", nil, stats.DefaultDiskIOProbeConfig())
 	defer location.Close()
 
 	v, err := NewVolume(dir, dir, "", 1, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0)