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)