volume: resolve the disk IO slow-latency threshold per disk (#10976)

* volume: resolve the disk IO slow-latency threshold per disk

volume.toml keys [volume.disk.io.slow.latency] by disk type, but the
threshold was chosen once per server by switching on the raw -disk flag.
-disk is comma-separated, one entry per -dir, so a multi-disk server
matched no case and silently took the hdd threshold.

Carry the table on DiskIOProbeConfig and resolve it in CheckDiskSpace
from the location's own DiskType. A type with no entry keeps falling
back to the hdd threshold.

* volume: run the disk IO probe on multi-directory volume servers

The probe was disabled whenever more than one -dir was configured,
because a single server-wide slow-latency threshold could not describe
disks of different types. The threshold is per disk now, and the rest of
the probe already is: diskRegistry is keyed by directory, each
DiskLocation runs its own CheckDiskSpace, and Store consults
isDiskUnavailable per location.

* volume: reject duplicate -dir entries

Nothing deduplicated -dir, so the same directory listed twice produced two
DiskLocations that each loaded every volume in it, appending to the same .dat
under two independent locks. Compare directory identity with os.SameFile
rather than the path, so a symlink or bind mount aliasing an earlier entry is
rejected as well.

* volume: cover the per-disk slow-latency handoff

SlowLatencyFor has a test, but nothing asserted that CheckDiskSpace feeds it
the location's own disk type. Probe through a seam so the resolved threshold
is observable, and check hdd, ssd, nvme, the empty type, and an unlisted tag.

weed/command/scaffold/volume.toml

@@ -21,6 +21,8 @@ percent = 20
 [volume.disk.io.error]
 percent = 10
 
+# slow threshold per disk type, matched against each -dir's -disk tag.
+# a disk whose type is not listed here uses the hdd threshold.
 [volume.disk.io.slow.latency]
 hdd = "500ms"
 ssd = "100ms"

weed/command/volume.go

@@ -293,11 +293,29 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 
 	// Set multiple folders and each folder's max volume count limit'
 	v.folders = strings.Split(volumeFolders, ",")
+	// Two locations on one directory would load every volume twice, appending to
+	// the same .dat under two independent locks. Compare identity rather than the
+	// path, so a symlink or bind mount aliasing an earlier -dir is caught too.
+	type seenFolder struct {
+		path string
+		info os.FileInfo
+	}
+	var seenFolders []seenFolder
 	for i, folder := range v.folders {
 		v.folders[i] = util.ResolvePath(folder)
 		if err := util.TestFolderWritable(v.folders[i]); err != nil {
 			glog.Fatalf("Check Data Folder(-dir) Writable %s : %s", v.folders[i], err)
 		}
+		folderInfo, err := os.Stat(v.folders[i])
+		if err != nil {
+			glog.Fatalf("Check Data Folder(-dir) %s : %s", v.folders[i], err)
+		}
+		for _, seen := range seenFolders {
+			if os.SameFile(seen.info, folderInfo) {
+				glog.Fatalf("Data Folder(-dir) %s and %s are the same directory", seen.path, v.folders[i])
+			}
+		}
+		seenFolders = append(seenFolders, seenFolder{v.folders[i], folderInfo})
 	}
 
 	// set max
@@ -396,24 +414,18 @@ 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,
+		SlowLatency: *v.diskHDDIOSlowLatency,
+		SlowLatencyByDiskType: map[string]time.Duration{
+			types.HddType:  *v.diskHDDIOSlowLatency,
+			types.SsdType:  *v.diskSSDIOSlowLatency,
+			types.NvmeType: *v.diskNVMEIOSlowLatency,
+		},
 
 		Window:     *v.diskIOWindow,
 		MinSamples: *v.diskIOMinSamples,
@@ -425,10 +437,6 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 
 		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,

weed/stats/disk.go

@@ -27,6 +27,9 @@ type DiskIOProbeConfig struct {
 	// latency above this threshold is considered slow
 	SlowLatency time.Duration
 
+	// per disk type overrides for SlowLatency, keyed by the -disk tag
+	SlowLatencyByDiskType map[string]time.Duration
+
 	// rolling observation window
 	Window time.Duration
 
@@ -64,6 +67,15 @@ func DefaultDiskIOProbeConfig() DiskIOProbeConfig {
 	}
 }
 
+// SlowLatencyFor returns the slow threshold configured for a disk type,
+// falling back to SlowLatency when the type has no entry of its own.
+func (config DiskIOProbeConfig) SlowLatencyFor(diskType string) time.Duration {
+	if slowLatency, found := config.SlowLatencyByDiskType[diskType]; found {
+		return slowLatency
+	}
+	return config.SlowLatency
+}
+
 type ioSample struct {
 	ts      time.Time
 	latency time.Duration

weed/stats/disk_test.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 )
@@ -67,3 +68,29 @@ func TestSetDiskStatusRequiresRepeatedSuccessesToRecover(t *testing.T) {
 		t.Fatalf("expected disk error to recover after %d successful checks: %s", recoveryChecks, disk.Error)
 	}
 }
+
+func TestSlowLatencyForFallsBackToDefault(t *testing.T) {
+	config := DiskIOProbeConfig{
+		SlowLatency: 500 * time.Millisecond,
+		SlowLatencyByDiskType: map[string]time.Duration{
+			"hdd":  500 * time.Millisecond,
+			"nvme": 50 * time.Millisecond,
+		},
+	}
+
+	for diskType, want := range map[string]time.Duration{
+		"hdd":       500 * time.Millisecond,
+		"nvme":      50 * time.Millisecond,
+		"ssd":       500 * time.Millisecond,
+		"nvme-gen5": 500 * time.Millisecond,
+	} {
+		if got := config.SlowLatencyFor(diskType); got != want {
+			t.Errorf("disk type %q: got %v, want %v", diskType, got, want)
+		}
+	}
+
+	empty := DiskIOProbeConfig{SlowLatency: 100 * time.Millisecond}
+	if got := empty.SlowLatencyFor("nvme"); got != 100*time.Millisecond {
+		t.Errorf("unconfigured table: got %v, want %v", got, 100*time.Millisecond)
+	}
+}

weed/storage/disk_location.go

@@ -671,9 +671,13 @@ func (l *DiskLocation) UnUsedSpace(volumeSizeLimit uint64) (unUsedSpace uint64)
 	return
 }
 
+// newDiskStatus is a seam letting a test observe the config CheckDiskSpace probes with.
+var newDiskStatus = stats.NewDiskStatusOnStart
+
 func (l *DiskLocation) CheckDiskSpace(config stats.DiskIOProbeConfig) {
+	config.SlowLatency = config.SlowLatencyFor(l.DiskType.ReadableString())
 	if dir, e := filepath.Abs(l.Directory); e == nil {
-		s := stats.NewDiskStatusOnStart(dir, config)
+		s := newDiskStatus(dir, config)
 		if len(s.Error) != 0 {
 			l.isDiskUnavailable.Store(true)
 			stats.VolumeServerDiskErrorGauge.WithLabelValues(l.Directory, "error").Set(1)

weed/storage/disk_location_test.go

@@ -5,6 +5,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
 	"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
@@ -100,3 +102,40 @@ func TestResolveVolumeIDs(t *testing.T) {
 		t.Errorf("wanted EC volume IDs %v, got %v", want, got)
 	}
 }
+
+// The threshold a disk is probed against is picked from that disk's own type,
+// so a server mixing media holds each of them to its own latency.
+func TestCheckDiskSpaceProbesWithTheDiskTypeThreshold(t *testing.T) {
+	original := newDiskStatus
+	defer func() { newDiskStatus = original }()
+
+	config := stats.DiskIOProbeConfig{
+		SlowLatency: 500 * time.Millisecond,
+		SlowLatencyByDiskType: map[string]time.Duration{
+			types.HddType:  500 * time.Millisecond,
+			types.SsdType:  100 * time.Millisecond,
+			types.NvmeType: 50 * time.Millisecond,
+		},
+	}
+
+	for diskType, want := range map[string]time.Duration{
+		"hdd":       500 * time.Millisecond,
+		"":          500 * time.Millisecond,
+		"ssd":       100 * time.Millisecond,
+		"nvme":      50 * time.Millisecond,
+		"nvme-gen5": 500 * time.Millisecond,
+	} {
+		var probed time.Duration
+		newDiskStatus = func(path string, probeConfig stats.DiskIOProbeConfig) *volume_server_pb.DiskStatus {
+			probed = probeConfig.SlowLatency
+			return &volume_server_pb.DiskStatus{Dir: path}
+		}
+
+		location := &DiskLocation{Directory: t.TempDir(), DiskType: types.ToDiskType(diskType)}
+		location.CheckDiskSpace(config)
+
+		if probed != want {
+			t.Errorf("-disk %q: probed with %v, want %v", diskType, probed, want)
+		}
+	}
+}

weed/storage/types/volume_disk_type.go

@@ -16,6 +16,7 @@ const (
 	HardDriveType DiskType = ""
 	HddType                = "hdd"
 	SsdType                = "ssd"
+	NvmeType               = "nvme"
 )
 
 func ToDiskType(vt string) (diskType DiskType) {

weed/storage/types/volume_disk_type_test.go

@@ -0,0 +1,21 @@
+package types
+
+import "testing"
+
+// The disk IO probe keys its per-disk-type settings by ReadableString, so a
+// disk configured as "hdd" has to come back out as "hdd" and not as the empty
+// string HardDriveType is stored as.
+func TestReadableStringRoundTrip(t *testing.T) {
+	for configured, want := range map[string]string{
+		"":          HddType,
+		"hdd":       HddType,
+		"HDD":       HddType,
+		"ssd":       SsdType,
+		"nvme":      NvmeType,
+		"nvme-gen5": "nvme-gen5",
+	} {
+		if got := ToDiskType(configured).ReadableString(); got != want {
+			t.Errorf("-disk %q: got %q, want %q", configured, got, want)
+		}
+	}
+}