diff --git a/weed/worker/tasks/balance/detection.go b/weed/worker/tasks/balance/detection.go index dd75dbc778..d53c219fd1 100644 --- a/weed/worker/tasks/balance/detection.go +++ b/weed/worker/tasks/balance/detection.go @@ -322,36 +322,23 @@ func detectForDiskType(diskType string, diskMetrics []*types.VolumeHealthMetrics break } - // Per-move convergence guard: mirrors weed/shell/command_volume_balance.go - // to prevent oscillation and destination overshoot when MaxVolumeCount - // values are heterogeneous. Without this guard, the greedy max→min - // algorithm can schedule moves that flip which server is most-utilized - // (source becomes min, destination becomes max), producing A→B, B→A - // oscillation within a single detection cycle. - // - // Check: after the move, the destination's utilization must not strictly - // exceed the source's utilization. If it would, no single move can - // improve balance — stop here. This also handles heterogeneous capacity - // correctly by comparing post-move utilization ratios rather than raw - // counts. The integer discretization is handled automatically: when - // counts cannot match the ideal exactly, the check still admits moves - // that reduce the max/min gap without flipping it. - maxCap := serverMaxVolumes[maxServer] - minCap := serverMaxVolumes[minServer] - if allServersHaveMaxInfo && maxCap > 0 && minCap > 0 { - newSrcUtil := float64(effectiveCounts[maxServer]-1) / float64(maxCap) - newDstUtil := float64(effectiveCounts[minServer]+1) / float64(minCap) - if newDstUtil > newSrcUtil { - if len(results) == 0 { - glog.Infof("BALANCE [%s]: No tasks created - no beneficial move available. After move, dest %s util would be %.1f%% vs source %s util %.1f%%", - diskType, minServer, newDstUtil*100, maxServer, newSrcUtil*100) - } else { - glog.Infof("BALANCE [%s]: Created %d task(s), no more beneficial moves available. After move, dest %s util would be %.1f%% vs source %s util %.1f%%", - diskType, len(results), minServer, newDstUtil*100, maxServer, newSrcUtil*100) - } - balanced = true - break + // Per-move convergence guard: after the move, the destination's + // utilization must not strictly exceed the source's. Otherwise the move + // just flips which server is min vs max, producing A→B, B→A oscillation. + // Comparing utilization ratios handles heterogeneous capacities; in + // raw-count mode serverUtilization returns effective counts. + newSrcUtil := serverUtilization(maxServer, effectiveCounts[maxServer]-1) + newDstUtil := serverUtilization(minServer, effectiveCounts[minServer]+1) + if newDstUtil > newSrcUtil { + if len(results) == 0 { + glog.Infof("BALANCE [%s]: No tasks created - no beneficial move available. After move, dest %s util would be %.1f%% vs source %s util %.1f%%", + diskType, minServer, newDstUtil*100, maxServer, newSrcUtil*100) + } else { + glog.Infof("BALANCE [%s]: Created %d task(s), no more beneficial moves available. After move, dest %s util would be %.1f%% vs source %s util %.1f%%", + diskType, len(results), minServer, newDstUtil*100, maxServer, newSrcUtil*100) } + balanced = true + break } // Select a volume from the overloaded server using per-server cursor @@ -398,9 +385,16 @@ 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 !destinationFull[s] && !destinationDiskTooFull(s, 0) { - eligibleTargets[s] = c + if destinationFull[s] || destinationDiskTooFull(s, 0) { + continue } + // The score-based fallback obeys the same convergence guard: a + // destination that ends up more utilized than the source only + // relocates the imbalance instead of reducing it. + if serverUtilization(s, effectiveCounts[s]+1) > newSrcUtil { + continue + } + eligibleTargets[s] = c } task, destServerID := createBalanceTask(diskType, selectedVolume, clusterInfo, minServer, eligibleTargets, balanceConfig.IoBytePerSecond) if task == nil { @@ -491,29 +485,6 @@ func createBalanceTask(diskType string, selectedVolume *types.VolumeHealthMetric glog.Warningf("Failed to plan balance destination for volume %d: %v", selectedVolume.VolumeID, err) return nil, "" } - } - - // Verify the resolved destination. If it falls outside the filtered scope - // or violates replica placement, fall back to the score-based planner and - // pick the best candidate that is actually valid. - if !isValidBalanceDestination(destinationPlan, allowedServers, volumeRP, replicas, selectedVolume.Server) { - switch { - case destinationPlan == nil: - glog.V(1).Infof("BALANCE [%s]: Planned destination for volume %d is nil, falling back", - diskType, selectedVolume.VolumeID) - case !isAllowedBalanceTarget(destinationPlan.TargetNode, allowedServers): - glog.V(1).Infof("BALANCE [%s]: Planned destination %s for volume %d is outside filtered scope, falling back", - diskType, destinationPlan.TargetNode, selectedVolume.VolumeID) - case volumeRP != nil && len(replicas) > 0: - glog.V(1).Infof("BALANCE [%s]: Destination %s violates replica placement for volume %d (rp=%03d), falling back", - diskType, destinationPlan.TargetNode, selectedVolume.VolumeID, selectedVolume.ExpectedReplicas) - } - - destinationPlan, err = planBalanceDestination(clusterInfo.ActiveTopology, selectedVolume, volumeRP, replicas, allowedServers) - if err != nil { - glog.Warningf("BALANCE [%s]: Failed to plan fallback destination for volume %d: %v", diskType, selectedVolume.VolumeID, err) - return nil, "" - } if !isValidBalanceDestination(destinationPlan, allowedServers, volumeRP, replicas, selectedVolume.Server) { if destinationPlan == nil { glog.V(1).Infof("BALANCE [%s]: Fallback destination for volume %d is nil", @@ -524,6 +495,15 @@ func createBalanceTask(diskType string, selectedVolume *types.VolumeHealthMetric } return nil, "" } + } else if !isValidBalanceDestination(destinationPlan, allowedServers, volumeRP, replicas, selectedVolume.Server) { + // The loop-chosen target is out of scope or already holds a replica of + // this volume. Skip it like shell volume.balance does instead of + // redirecting the move to a third server: with equally full servers the + // redirected move never shrinks the imbalance and ping-pongs the same + // volumes back on the next detection cycle. + glog.V(1).Infof("BALANCE [%s]: target %s is not a valid destination for volume %d, trying next volume", + diskType, targetServer, selectedVolume.VolumeID) + return nil, "" } // Find the actual disk containing the volume on the source server diff --git a/weed/worker/tasks/balance/detection_test.go b/weed/worker/tasks/balance/detection_test.go index e74add194b..ae8febcf5a 100644 --- a/weed/worker/tasks/balance/detection_test.go +++ b/weed/worker/tasks/balance/detection_test.go @@ -1485,3 +1485,125 @@ func TestResolveBalanceDestination_UsesEffectiveCapacity(t *testing.T) { t.Errorf("unexpected target node: got %q want %q", plan.TargetNode, "dst-node") } } + +// TestDetection_SkipsVolumeWhenTargetHoldsReplica reproduces issue #11839. +// The emptiest server (node-d) already holds a replica of the lowest-ID +// volumes on the fuller servers, so those volumes cannot legally move to it. +// Detection must skip them and move on to volumes that can, instead of +// redirecting the move to a third, equally full server — which ping-pongs +// the same volumes back and forth on every detection cycle. +func TestDetection_SkipsVolumeWhenTargetHoldsReplica(t *testing.T) { + servers := []serverSpec{ + {id: "node-a", diskType: "hdd", diskID: 1, dc: "dc1", rack: "rack1", maxVolumes: 100}, + {id: "node-b", diskType: "hdd", diskID: 2, dc: "dc1", rack: "rack1", maxVolumes: 100}, + {id: "node-c", diskType: "hdd", diskID: 3, dc: "dc1", rack: "rack1", maxVolumes: 100}, + {id: "node-d", diskType: "hdd", diskID: 4, dc: "dc1", rack: "rack1", maxVolumes: 100}, + } + + var metrics []*types.VolumeHealthMetrics + replicaMap := make(map[uint32][]types.ReplicaLocation) + addReplica := func(id uint32, node string) { + replicaMap[id] = append(replicaMap[id], types.ReplicaLocation{ + DataCenter: "dc1", Rack: "rack1", NodeID: node, + }) + } + // place registers a replica of volume id on server (a metric); the first + // placement of a volume also records its second replica on peer. + place := func(server, peer string, lo, hi uint32) { + for id := lo; id <= hi; id++ { + metrics = append(metrics, &types.VolumeHealthMetrics{ + VolumeID: id, Server: server, ServerAddress: server + ":8080", + DiskType: "hdd", Collection: "c1", Size: 1024, + DataCenter: "dc1", Rack: "rack1", ExpectedReplicas: 1, // rp=001 + }) + if _, seen := replicaMap[id]; !seen { + addReplica(id, server) + addReplica(id, peer) + } + } + } + + // The lowest-ID volumes on each full server already have their second + // replica on node-d, so node-d is never a legal target for them. + place("node-a", "node-d", 1, 10) + place("node-b", "node-d", 11, 20) + place("node-c", "node-d", 21, 30) + + // Higher-ID volumes replicate within {a,b,c} and can move to node-d. + place("node-a", "node-b", 31, 55) + place("node-b", "node-c", 56, 70) + place("node-c", "node-a", 81, 105) + place("node-c", "node-b", 106, 120) + place("node-b", "node-a", 31, 55) // second copies + place("node-c", "node-b", 56, 70) // second copies + place("node-a", "node-c", 81, 105) // second copies + place("node-b", "node-c", 106, 120) // second copies + + // Per-server counts: a=60, b=65, c=65 of cap 100. node-d has no metrics of + // its own (it only holds the replica-map copies), so it is the clear min. + at := buildTopology(servers, metrics) + clusterInfo := &types.ClusterInfo{ActiveTopology: at, VolumeReplicaMap: replicaMap} + + tasks, _, err := Detection(metrics, clusterInfo, defaultConf(), 100) + if err != nil { + t.Fatalf("Detection failed: %v", err) + } + if len(tasks) == 0 { + t.Fatal("expected balance tasks: 60/65/65 vs 0 is far above the imbalance threshold") + } + + addrToNode := map[string]string{ + "node-a:8080": "node-a", "node-b:8080": "node-b", + "node-c:8080": "node-c", "node-d:8080": "node-d", + } + srcOf := make(map[uint32]string) + dstOf := make(map[uint32]string) + for _, task := range tasks { + for _, tgt := range task.TypedParams.Targets { + dst := addrToNode[tgt.Node] + if dst != "node-d" { + t.Errorf("volume %d moved %s -> %s: only node-d is under-utilized", task.VolumeID, task.Server, tgt.Node) + } + dstOf[task.VolumeID] = dst + } + srcOf[task.VolumeID] = task.Server + } + assertNoDuplicateVolumes(t, tasks) + + // Apply the planned moves and re-run detection: a volume already moved + // must never be picked again (the reported oscillation). + for _, m := range metrics { + if m.Server == srcOf[m.VolumeID] { + m.Server = dstOf[m.VolumeID] + m.ServerAddress = m.Server + ":8080" + } + } + for id, dst := range dstOf { + for i, loc := range replicaMap[id] { + if loc.NodeID == srcOf[id] { + replicaMap[id][i].NodeID = dst + } + } + } + for id, locs := range replicaMap { + seen := make(map[string]bool) + for _, loc := range locs { + if seen[loc.NodeID] { + t.Fatalf("volume %d ended with two replicas on %s", id, loc.NodeID) + } + seen[loc.NodeID] = true + } + } + + at2 := buildTopology(servers, metrics) + clusterInfo2 := &types.ClusterInfo{ActiveTopology: at2, VolumeReplicaMap: replicaMap} + tasks2, _, err := Detection(metrics, clusterInfo2, defaultConf(), 100) + if err != nil { + t.Fatalf("second Detection failed: %v", err) + } + for _, task := range tasks2 { + if dstOf[task.VolumeID] != "" { + t.Errorf("volume %d moved again in the next detection cycle (%s)", task.VolumeID, task.Server) + } + } +} diff --git a/weed/worker/tasks/balance/replica_placement_test.go b/weed/worker/tasks/balance/replica_placement_test.go index 396ea3b30f..9afd0af9cd 100644 --- a/weed/worker/tasks/balance/replica_placement_test.go +++ b/weed/worker/tasks/balance/replica_placement_test.go @@ -219,6 +219,9 @@ func TestPlanBalanceDestination_ChoosesBestValidCompositeDestination(t *testing. } } +// The score-based fallback runs when the loop-chosen target cannot be +// resolved (here: a node absent from the topology), and must still pick a +// destination that satisfies the volume's replica placement. func TestCreateBalanceTask_FallbackSelectsValidCompositeDestination(t *testing.T) { servers := []serverSpec{ {id: "node-a", diskType: "hdd", diskID: 1, dc: "dc1", rack: "rack1"}, @@ -237,20 +240,49 @@ func TestCreateBalanceTask_FallbackSelectsValidCompositeDestination(t *testing.T }, } + task, destination := createBalanceTask("hdd", volumes[0], clusterInfo, "node-x", map[string]int{ + "node-b": 0, + "node-c": 0, + }, 0) + if task == nil { + t.Fatal("expected a balance task") + } + if destination != "node-c" { + t.Fatalf("expected fallback destination node-c, got %s", destination) + } + if len(task.TypedParams.Targets) != 1 { + t.Fatalf("expected 1 target, got %d", len(task.TypedParams.Targets)) + } + if got := task.TypedParams.Targets[0].Node; got != "node-c:8080" { + t.Fatalf("expected target node-c:8080, got %s", got) + } +} + +// A resolvable but invalid loop-chosen target (rp=011 cannot add a third +// rack) must skip the volume rather than redirect it to another server. +func TestCreateBalanceTask_SkipsWhenTargetViolatesPlacement(t *testing.T) { + servers := []serverSpec{ + {id: "node-a", diskType: "hdd", diskID: 1, dc: "dc1", rack: "rack1"}, + {id: "node-b", diskType: "hdd", diskID: 2, dc: "dc1", rack: "rack3"}, + {id: "node-c", diskType: "hdd", diskID: 3, dc: "dc1", rack: "rack1"}, + } + volumes := makeVolumesWith("node-a", "hdd", "dc1", "rack1", "c1", 1, 1, withReplicas(11)) + clusterInfo := &types.ClusterInfo{ + ActiveTopology: buildTopology(servers, volumes), + VolumeReplicaMap: map[uint32][]types.ReplicaLocation{ + 1: { + loc("dc1", "rack1", "node-a"), + loc("dc1", "rack1", "node-d"), + loc("dc1", "rack2", "node-e"), + }, + }, + } + task, destination := createBalanceTask("hdd", volumes[0], clusterInfo, "node-b", map[string]int{ "node-b": 0, "node-c": 0, }, 0) - if task == nil { - t.Fatal("expected a balance task") - } - if destination != "node-c" { - t.Fatalf("expected fallback destination node-c, got %s", destination) - } - if len(task.TypedParams.Targets) != 1 { - t.Fatalf("expected 1 target, got %d", len(task.TypedParams.Targets)) - } - if got := task.TypedParams.Targets[0].Node; got != "node-c:8080" { - t.Fatalf("expected target node-c:8080, got %s", got) + if task != nil || destination != "" { + t.Fatalf("expected no task when the chosen target violates replica placement, got task=%v dest=%s", task != nil, destination) } }