From e3e02d3364203b9ea4b152915f6f4b95592aa81c Mon Sep 17 00:00:00 2001 From: Aleksey <48918167+MilanFun@users.noreply.github.com> Date: Tue, 2 Jun 2026 19:02:05 +0300 Subject: [PATCH] [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 Co-authored-by: Chris Lu --- weed/command/mini.go | 3 + weed/command/scaffold.go | 6 +- weed/command/scaffold/example.go | 3 + weed/command/scaffold/volume.toml | 27 + weed/command/server.go | 3 + weed/command/volume.go | 144 ++++++ weed/pb/volume_server.proto | 1 + weed/pb/volume_server_pb/volume_server.pb.go | 13 +- .../server/volume_grpc_erasure_coding_test.go | 3 + weed/server/volume_server.go | 3 +- weed/stats/disk.go | 487 ++++++++++++++++++ weed/stats/disk_notsupported.go | 4 +- weed/stats/disk_openbsd.go | 9 +- weed/stats/disk_solaris.go | 9 +- weed/stats/disk_supported.go | 14 +- weed/stats/disk_test.go | 69 +++ weed/stats/disk_windows.go | 12 +- weed/stats/metrics.go | 9 + weed/storage/disk_location.go | 29 +- weed/storage/store.go | 24 +- weed/storage/store_disk_space_test.go | 6 +- weed/storage/store_ec.go | 2 +- .../store_ec_9plus3_reboot_repro_test.go | 7 +- weed/storage/store_ec_disk_type_test.go | 3 + weed/storage/store_ec_mirror_test.go | 7 + .../storage/store_ec_mount_cross_disk_test.go | 7 + weed/storage/store_ec_orphan_shard_test.go | 10 + weed/storage/store_ec_target_location_test.go | 4 +- weed/storage/store_load_balancing_test.go | 5 +- weed/storage/store_maxvolume_deadzone_test.go | 2 +- weed/storage/volume.go | 2 +- weed/storage/volume_vacuum_test.go | 5 +- 32 files changed, 871 insertions(+), 61 deletions(-) create mode 100644 weed/command/scaffold/volume.toml create mode 100644 weed/stats/disk_test.go diff --git a/weed/command/mini.go b/weed/command/mini.go index 9eb1f6f585..c8cefd91cf 100644 --- a/weed/command/mini.go +++ b/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. diff --git a/weed/command/scaffold.go b/weed/command/scaffold.go index b667dcb4bb..3530bab844 100644 --- a/weed/command/scaffold.go +++ b/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": diff --git a/weed/command/scaffold/example.go b/weed/command/scaffold/example.go index 7dfc5032c7..55259fae83 100644 --- a/weed/command/scaffold/example.go +++ b/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 diff --git a/weed/command/scaffold/volume.toml b/weed/command/scaffold/volume.toml new file mode 100644 index 0000000000..64c350d3a9 --- /dev/null +++ b/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" diff --git a/weed/command/server.go b/weed/command/server.go index b52ad0f874..b878c97ff8 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -156,6 +156,7 @@ func init() { serverOptions.v.hasSlowRead = cmdServer.Flag.Bool("volume.hasSlowRead", true, " 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, " 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) diff --git a/weed/command/volume.go b/weed/command/volume.go index 4895cd103a..bca02d9403 100644 --- a/weed/command/volume.go +++ b/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) diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 4f71016d4d..bb05725890 100644 --- a/weed/pb/volume_server.proto +++ b/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 { diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index c05774b067..3d2ef4266b 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/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" + diff --git a/weed/server/volume_grpc_erasure_coding_test.go b/weed/server/volume_grpc_erasure_coding_test.go index 4c42e8fe1c..d42ab547f5 100644 --- a/weed/server/volume_grpc_erasure_coding_test.go +++ b/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() { diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go index 8fa56ac5de..a3692509d5 100644 --- a/weed/server/volume_server.go +++ b/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) diff --git a/weed/stats/disk.go b/weed/stats/disk.go index 5a8699a54a..260901ef81 100644 --- a/weed/stats/disk.go +++ b/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 +} diff --git a/weed/stats/disk_notsupported.go b/weed/stats/disk_notsupported.go index 8dbe14c37d..7ad19fff80 100644 --- a/weed/stats/disk_notsupported.go +++ b/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 } diff --git a/weed/stats/disk_openbsd.go b/weed/stats/disk_openbsd.go index ae71180e32..da32d97e3f 100644 --- a/weed/stats/disk_openbsd.go +++ b/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 } diff --git a/weed/stats/disk_solaris.go b/weed/stats/disk_solaris.go index b3a3e75acb..93c1e784d6 100644 --- a/weed/stats/disk_solaris.go +++ b/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 } diff --git a/weed/stats/disk_supported.go b/weed/stats/disk_supported.go index 807f62bce6..95d9596e5e 100644 --- a/weed/stats/disk_supported.go +++ b/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 } diff --git a/weed/stats/disk_test.go b/weed/stats/disk_test.go new file mode 100644 index 0000000000..cc34573563 --- /dev/null +++ b/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) + } +} diff --git a/weed/stats/disk_windows.go b/weed/stats/disk_windows.go index 694347c32f..7e3422558e 100644 --- a/weed/stats/disk_windows.go +++ b/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 } diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index 0d4aec244b..a30463504c 100644 --- a/weed/stats/metrics.go +++ b/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) diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go index bfeff4033f..46558b27a6 100644 --- a/weed/storage/disk_location.go +++ b/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) } diff --git a/weed/storage/store.go b/weed/storage/store.go index 0ec82b81df..e65f69d901 100644 --- a/weed/storage/store.go +++ b/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() diff --git a/weed/storage/store_disk_space_test.go b/weed/storage/store_disk_space_test.go index 884b8dda1c..953ad4bb45 100644 --- a/weed/storage/store_disk_space_test.go +++ b/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{} } diff --git a/weed/storage/store_ec.go b/weed/storage/store_ec.go index 935e7b90cb..018f18d72c 100644 --- a/weed/storage/store_ec.go +++ b/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) diff --git a/weed/storage/store_ec_9plus3_reboot_repro_test.go b/weed/storage/store_ec_9plus3_reboot_repro_test.go index f657160c10..b9646d47df 100644 --- a/weed/storage/store_ec_9plus3_reboot_repro_test.go +++ b/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] diff --git a/weed/storage/store_ec_disk_type_test.go b/weed/storage/store_ec_disk_type_test.go index 6695d58766..7119865199 100644 --- a/weed/storage/store_ec_disk_type_test.go +++ b/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{} diff --git a/weed/storage/store_ec_mirror_test.go b/weed/storage/store_ec_mirror_test.go index 6f43afb5a7..96ccbdac71 100644 --- a/weed/storage/store_ec_mirror_test.go +++ b/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() { diff --git a/weed/storage/store_ec_mount_cross_disk_test.go b/weed/storage/store_ec_mount_cross_disk_test.go index 20790574ae..53ed4a64d4 100644 --- a/weed/storage/store_ec_mount_cross_disk_test.go +++ b/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() { diff --git a/weed/storage/store_ec_orphan_shard_test.go b/weed/storage/store_ec_orphan_shard_test.go index 2b8f37660e..851c3c8998 100644 --- a/weed/storage/store_ec_orphan_shard_test.go +++ b/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() { diff --git a/weed/storage/store_ec_target_location_test.go b/weed/storage/store_ec_target_location_test.go index 92283647ef..9b79422975 100644 --- a/weed/storage/store_ec_target_location_test.go +++ b/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() { diff --git a/weed/storage/store_load_balancing_test.go b/weed/storage/store_load_balancing_test.go index 404f905922..90fa7ff7fc 100644 --- a/weed/storage/store_load_balancing_test.go +++ b/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) diff --git a/weed/storage/store_maxvolume_deadzone_test.go b/weed/storage/store_maxvolume_deadzone_test.go index e819395551..8dc9d462bc 100644 --- a/weed/storage/store_maxvolume_deadzone_test.go +++ b/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) diff --git a/weed/storage/volume.go b/weed/storage/volume.go index d7bd432e62..10249482ff 100644 --- a/weed/storage/volume.go +++ b/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) { diff --git a/weed/storage/volume_vacuum_test.go b/weed/storage/volume_vacuum_test.go index 5baec284d9..764528e5a6 100644 --- a/weed/storage/volume_vacuum_test.go +++ b/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)