From 9550b830d05c33b0e89dc8ae1dee46f301290ff0 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 30 Jun 2026 20:08:19 -0700 Subject: [PATCH] worker: project the moved volume when gating on disk fullness (#10171) The disk-fullness gate only rejected destinations already at/above the mark, so a server just under it could take a large volume and overshoot. Project the selected volume's bytes onto the candidate: if the move would cross the mark, drop that destination for the rest of the cycle and re-pick instead of overshooting. Also note the per-location capacity-summing assumption on the Rust heartbeat side, to match the Go store.go comment. --- seaweed-volume/src/server/heartbeat.rs | 3 +++ weed/worker/tasks/balance/detection.go | 27 +++++++++++++++------ weed/worker/tasks/balance/detection_test.go | 27 +++++++++++++++++++++ 3 files changed, 49 insertions(+), 8 deletions(-) diff --git a/seaweed-volume/src/server/heartbeat.rs b/seaweed-volume/src/server/heartbeat.rs index 8c90919f4..1601d7efb 100644 --- a/seaweed-volume/src/server/heartbeat.rs +++ b/seaweed-volume/src/server/heartbeat.rs @@ -831,6 +831,9 @@ fn build_heartbeat_with_ec_status( } *max_volume_counts.entry(disk_type_str.clone()).or_insert(0) += effective_max_count as u32; disk_max_by_id[disk_id] = effective_max_count; + // Sum capacity per disk type; assumes one location per filesystem. Locations + // sharing a mount over-report absolute bytes but not the used ratio the gate + // uses. Mirrors weed/storage/store.go. *disk_total_bytes.entry(disk_type_str.clone()).or_insert(0) += loc.disk_total_bytes.load(Ordering::Relaxed); *disk_free_bytes.entry(disk_type_str).or_insert(0) += diff --git a/weed/worker/tasks/balance/detection.go b/weed/worker/tasks/balance/detection.go index 6483b5441..4ff865b4c 100644 --- a/weed/worker/tasks/balance/detection.go +++ b/weed/worker/tasks/balance/detection.go @@ -205,19 +205,20 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics // planned instead of only its heartbeat-time free space (the shell equivalent // is adjustAfterMove decrementing DiskFreeBytes). plannedBytes := make(map[string]uint64) + // destinationFull excludes servers a projected move would push over, for the cycle. + destinationFull := make(map[string]bool) - // destinationDiskTooFull reports whether a server's physical disk is already - // at/above the high-water mark, making it ineligible as a move destination. - // Judged per server against its own disk, discounting data planned onto it so - // far this cycle; servers not reporting disk bytes are never gated (slot-only). - destinationDiskTooFull := func(server string) bool { + // destinationDiskTooFull reports whether landing incomingBytes on server would put + // its disk at/over the mark, net of bytes already planned this cycle; unreported + // disks are never gated. + destinationDiskTooFull := func(server string, incomingBytes uint64) bool { free := serverDiskFreeBytes[server] if planned := plannedBytes[server]; planned < free { free -= planned } else { free = 0 } - return balancer.DiskTooFullAfter(serverDiskTotalBytes[server], free, 0, balancer.DefaultMaxDiskUsagePercent) + return balancer.DiskTooFullAfter(serverDiskTotalBytes[server], free, incomingBytes, balancer.DefaultMaxDiskUsagePercent) } for len(results) < maxResults { @@ -245,7 +246,7 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics // Min is the emptiest server that can actually receive a volume, so a // physically full server (whose over-set maxVolumeCount makes its slot // utilization look low) is never chosen as the destination. - if !destinationDiskTooFull(server) && util < minUtilization { + if !destinationFull[server] && !destinationDiskTooFull(server, 0) && util < minUtilization { minUtilization = util minServer = server } @@ -373,6 +374,16 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics continue } + // Skip a destination this specific volume would push over the mark, and + // re-pick, rather than overshoot. + if destinationDiskTooFull(minServer, uint64(selectedVolume.Size)) { + glog.V(1).Infof("BALANCE [%s]: skip destination %s: volume %d (%d bytes) would cross %d%% disk usage", + diskType, minServer, selectedVolume.VolumeID, selectedVolume.Size, balancer.DefaultMaxDiskUsagePercent) + destinationFull[minServer] = true + serverCursors[maxServer]-- // retry this volume against another destination + continue + } + // Create task targeting minServer — the greedy algorithm's natural choice. // Using minServer instead of letting planBalanceDestination independently // pick a destination ensures that the detection loop's effective counts @@ -385,7 +396,7 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics // sources are unaffected, so a full server can still be drained. eligibleTargets := make(map[string]int, len(serverVolumeCounts)) for s, c := range serverVolumeCounts { - if !destinationDiskTooFull(s) { + if !destinationFull[s] && !destinationDiskTooFull(s, 0) { eligibleTargets[s] = c } } diff --git a/weed/worker/tasks/balance/detection_test.go b/weed/worker/tasks/balance/detection_test.go index cd22ea24f..e74add194 100644 --- a/weed/worker/tasks/balance/detection_test.go +++ b/weed/worker/tasks/balance/detection_test.go @@ -465,6 +465,33 @@ func TestDetection_NoDestinationWhenAllDisksFull(t *testing.T) { } } +// near-full is at 85% (under the mark) but moving one 100-byte volume onto its +// 1000-byte disk would reach 95%, so it must not be targeted. +func TestDetection_ProjectsVolumeOntoNearFullDestination(t *testing.T) { + servers := []serverSpec{ + {id: "src", diskType: "hdd", dc: "dc1", rack: "rack1", maxVolumes: 10}, + {id: "near-full", diskType: "hdd", dc: "dc1", rack: "rack1", maxVolumes: 1000, diskTotalBytes: 1000, diskFreeBytes: 150}, + } + metrics := make([]*types.VolumeHealthMetrics, 8) + for i := range metrics { + metrics[i] = &types.VolumeHealthMetrics{ + VolumeID: uint32(i + 1), Server: "src", ServerAddress: "src:8080", + DiskType: "hdd", Collection: "c1", Size: 100, DataCenter: "dc1", Rack: "rack1", + } + } + + at := buildTopology(servers, metrics) + clusterInfo := &types.ClusterInfo{ActiveTopology: at} + + tasks, _, err := Detection(metrics, clusterInfo, defaultConf(), 100) + if err != nil { + t.Fatalf("Detection failed: %v", err) + } + if len(tasks) != 0 { + t.Fatalf("expected no tasks: the projected move would cross 90%% disk usage, got %d", len(tasks)) + } +} + func TestDetection_SkipsRemoteTieredVolumes(t *testing.T) { metrics := []*types.VolumeHealthMetrics{}