diff --git a/weed/storage/erasure_coding/ecbalancer/balancer.go b/weed/storage/erasure_coding/ecbalancer/balancer.go index 2e8ad0ffa5..b310837d14 100644 --- a/weed/storage/erasure_coding/ecbalancer/balancer.go +++ b/weed/storage/erasure_coding/ecbalancer/balancer.go @@ -937,115 +937,190 @@ func detectGlobalImbalance(nodes map[string]*Node, racks map[string]*rack, diskT if iterations <= 0 { iterations = totalShards } + spread := newRackSpread(r) + blocked := make(blockedPairs) for i := 0; i < iterations; i++ { - var minNode, maxNode *Node - minUtil := math.Inf(1) - maxUtil := -1.0 - var minCount, maxCount int - for _, nodeID := range sortedNodeKeys(r.nodes) { - count := nodeShardCounts[nodeID] - node := r.nodes[nodeID] - capacity := nodeCapacity[nodeID] - if capacity <= 0 { - continue - } - util := float64(count) / float64(capacity) - if util < minUtil && node.receiveSlots() > 0 { - minUtil, minCount, minNode = util, count, node - } - if util > maxUtil { - maxUtil, maxCount, maxNode = util, count, node - } - } - if maxNode == nil || minNode == nil || maxNode.id == minNode.id { - break - } - - maxCap := nodeCapacity[maxNode.id] - minCap := nodeCapacity[minNode.id] - if maxCap <= 0 || minCap <= 0 { - break - } - if float64(minCount+1)/float64(minCap) > float64(maxCount-1)/float64(maxCap) { - break - } - - // Prefer a volume absent from the destination's machine (pass 0) before - // adding to one already there (pass 1), to keep volumes spread. - moved := false - for pass := 0; pass < 2 && !moved; pass++ { - for _, vk := range sortedVolumeKeys(maxNode.shards) { - if moved { - break - } - info := maxNode.shards[vk] - minInfo := minNode.shards[vk] - volumeOnMinMachine := machineHoldsVolume(r, minNode.host, vk) - if pass == 0 && volumeOnMinMachine { - continue // pass 0: only volumes absent from the destination machine - } - if pass == 1 { - if !volumeOnMinMachine { - continue - } - // Protect the volume's machine spread only where it's achievable - // (enough machines for this rack's shards to each stay within - // parity); there a cross-machine load move is allowed only if it - // doesn't raise the destination machine's count past the source's. - // Where it isn't achievable, capacity rules and any leveling move - // is fine. Feasibility uses the rack's shards, not the whole volume. - parity := parityShardsByVolume[vk] - spreadFeasible := parity > 0 && rackMachineCount >= ceilDivide(rackVolumeShardCount(r, vk), parity) - if spreadFeasible && minNode.host != maxNode.host && - machineVolumeCount(r, minNode.host, vk) >= machineVolumeCount(r, maxNode.host, vk) { - continue - } - } - // Walk the volume's actual shard bitmap so custom ratios with more - // than the standard total (ids 14..MaxShardCount-1) are candidates too. - for sid := range info.shardBits.All() { - shardID := int(sid) - if minInfo != nil && minInfo.shardBits.Has(sid) { - continue - } - dataShards := dataShardsByVolume[vk] - if dataShards <= 0 { - dataShards = erasure_coding.DataShardsCount - } - destDisk := pickBestDiskOnNode(minNode, vk, diskType, shardID, dataShards) - moves = append(moves, &move{ - volumeID: vk.vid, - shardID: shardID, - collection: vk.collection, - source: maxNode, - sourceDisk: shardDiskID(maxNode, vk, shardID), - target: minNode, - targetDisk: destDisk, - phase: "global", - }) - info.shardBits = info.shardBits.Clear(sid) - for diskID := range info.diskShardBits { - info.diskShardBits[diskID] = info.diskShardBits[diskID].Clear(sid) - } - reserveShard(minNode, vk, shardID, destDisk) - nodeShardCounts[maxNode.id]-- - nodeShardCounts[minNode.id]++ - maxNode.freeSlots++ - minNode.freeSlots-- - moved = true - break - } - } - } - if !moved { + m := planGlobalMove(r, spread, blocked, rackMachineCount, nodeShardCounts, nodeCapacity, diskType, dataShardsByVolume, parityShardsByVolume) + if m == nil { break } + moves = append(moves, m) + vk := volKey{collection: m.collection, vid: m.volumeID} + releaseShard(m.source, vk, m.shardID) + reserveShard(m.target, vk, m.shardID, m.targetDisk) + spread.move(vk, m.source.host, m.target.host) + blocked.forget(m.source, m.target) + nodeShardCounts[m.source.id]-- + nodeShardCounts[m.target.id]++ + m.source.freeSlots++ + m.target.freeSlots-- } } return moves } +// rackSpread indexes each volume's shard counts in a rack, per machine and in +// total, so the global phase can check spread without rescanning the rack. +type rackSpread struct { + machine map[volKey]map[string]int + rack map[volKey]int +} + +func newRackSpread(r *rack) rackSpread { + s := rackSpread{machine: make(map[volKey]map[string]int), rack: make(map[volKey]int)} + for _, n := range r.nodes { + for vk, info := range n.shards { + c := info.shardBits.Count() + if c == 0 { + continue + } + if s.machine[vk] == nil { + s.machine[vk] = make(map[string]int) + } + s.machine[vk][n.host] += c + s.rack[vk] += c + } + } + return s +} + +// move records one shard of vk leaving fromHost for toHost within the rack. +func (s rackSpread) move(vk volKey, fromHost, toHost string) { + s.machine[vk][fromHost]-- + s.machine[vk][toHost]++ +} + +// blockedPairs remembers source/target pairs with no movable shard, so a long +// drain does not rescan them every iteration. +type blockedPairs map[[2]*Node]bool + +// forget drops the pairs a move from source to target can unblock. Losing a +// shard only narrows source's options and gaining one only narrows target's, so +// within a machine only target as a source (it holds a new shard) and source as +// a target (it no longer holds one) can open up. A move between machines also +// changes the volume's count on both machines, which every pair touching either +// machine checks. +func (b blockedPairs) forget(source, target *Node) { + crossMachine := source.host != target.host + for pair := range b { + from, to := pair[0], pair[1] + if from == target || to == source || + (crossMachine && (from.host == source.host || from.host == target.host || to.host == source.host || to.host == target.host)) { + delete(b, pair) + } + } +} + +// planGlobalMove picks one shard to move from a fuller node to an emptier one in +// the rack, or returns nil when no move would narrow the gap. Sources are tried +// fullest first and destinations emptiest first once they hold the shard, so a +// node whose shards cannot go to the emptiest node still drains to the next one +// that can take them, such as another disk on its own machine. +func planGlobalMove(r *rack, spread rackSpread, blocked blockedPairs, rackMachineCount int, counts, capacity map[string]int, diskType string, dataShardsByVolume, parityShardsByVolume map[volKey]int) *move { + util := func(n *Node, delta int) float64 { + return float64(counts[n.id]+delta) / float64(capacity[n.id]) + } + var sources, targets []*Node + for _, n := range sortedNodeSlice(r.nodes) { + if capacity[n.id] <= 0 { + continue + } + sources = append(sources, n) + if n.receiveSlots() > 0 { + targets = append(targets, n) + } + } + sort.SliceStable(sources, func(i, j int) bool { return util(sources[i], 0) > util(sources[j], 0) }) + sort.SliceStable(targets, func(i, j int) bool { return util(targets[i], 1) < util(targets[j], 1) }) + for _, source := range sources { + for _, target := range targets { + if target.id == source.id { + continue + } + // Targets are ordered by their fullness after taking the shard, so once + // one would end up fuller than the source, every later one would too. + if util(target, 1) > util(source, -1) { + break + } + pair := [2]*Node{source, target} + if blocked[pair] { + continue + } + if m := pickGlobalShard(spread, rackMachineCount, source, target, diskType, dataShardsByVolume, parityShardsByVolume); m != nil { + return m + } + blocked[pair] = true + } + } + return nil +} + +// pickGlobalShard returns a move of one of source's shards to target that keeps +// each volume spread across machines, or nil. +func pickGlobalShard(spread rackSpread, rackMachineCount int, source, target *Node, diskType string, dataShardsByVolume, parityShardsByVolume map[volKey]int) *move { + sameMachine := source.host == target.host + // Prefer a volume absent from the target's machine (pass 0) before adding to + // one already there (pass 1), to keep volumes spread. + for pass := 0; pass < 2; pass++ { + for _, vk := range sortedVolumeKeys(source.shards) { + info := source.shards[vk] + targetInfo := target.shards[vk] + volumeOnTargetMachine := spread.machine[vk][target.host] > 0 + if pass == 0 && volumeOnTargetMachine { + continue // pass 0: only volumes absent from the target machine + } + if pass == 1 { + if !volumeOnTargetMachine { + continue + } + if sameMachine { + // The machine's count is unchanged; keep the volume on as many + // of the machine's disks as before by not doubling up on one. + if targetInfo != nil && targetInfo.shardBits.Count() > 0 { + continue + } + } else { + // Protect the volume's machine spread only where it's achievable + // (enough machines for this rack's shards to each stay within + // parity); there a cross-machine load move is allowed only if it + // doesn't raise the target machine's count past the source's. + // Where it isn't achievable, capacity rules and any leveling move + // is fine. Feasibility uses the rack's shards, not the whole volume. + parity := parityShardsByVolume[vk] + spreadFeasible := parity > 0 && rackMachineCount >= ceilDivide(spread.rack[vk], parity) + if spreadFeasible && spread.machine[vk][target.host] >= spread.machine[vk][source.host] { + continue + } + } + } + // Walk the volume's actual shard bitmap so custom ratios with more + // than the standard total (ids 14..MaxShardCount-1) are candidates too. + for sid := range info.shardBits.All() { + if targetInfo != nil && targetInfo.shardBits.Has(sid) { + continue + } + shardID := int(sid) + dataShards := dataShardsByVolume[vk] + if dataShards <= 0 { + dataShards = erasure_coding.DataShardsCount + } + return &move{ + volumeID: vk.vid, + shardID: shardID, + collection: vk.collection, + source: source, + sourceDisk: shardDiskID(source, vk, shardID), + target: target, + targetDisk: pickBestDiskOnNode(target, vk, diskType, shardID, dataShards), + phase: "global", + } + } + } + } + return nil +} + // shardsByGroup classifies a volume's shards into data (id < dataShards) and // parity buckets, grouped by key(node). func shardsByGroup(vk volKey, nodes map[string]*Node, dataShards int, key func(*Node) string) (dataPer, parityPer map[string][]int) { diff --git a/weed/storage/erasure_coding/ecbalancer/balancer_test.go b/weed/storage/erasure_coding/ecbalancer/balancer_test.go index 0bb2ddc95f..eebdbc0b75 100644 --- a/weed/storage/erasure_coding/ecbalancer/balancer_test.go +++ b/weed/storage/erasure_coding/ecbalancer/balancer_test.go @@ -990,3 +990,126 @@ func TestNoReceiveDiskLeavesSiblingsEligible(t *testing.T) { t.Fatal("a node with one refused disk should still receive on its other disks") } } + +// TestGlobalDrainsToSameMachineWhenOtherMachinesAreBlocked: hosts run one server +// per disk, and every volume has one shard on each machine, all on the first +// server. The emptiest server is on another machine, where any shard would +// concentrate its volume, so the fullest server must drain onto its own +// machine's empty server instead of the phase giving up for the whole rack. +func TestGlobalDrainsToSameMachineWhenOtherMachinesAreBlocked(t *testing.T) { + topo := NewTopology() + mk := func(id, host string, free int) *Node { + n := topo.AddNode(id, "dc1", "dc1:rack1", free) + n.SetHost(host) + return n + } + a1, b1, c1 := mk("a1", "boxA", 2), mk("b1", "boxB", 4), mk("c1", "boxC", 4) + a2, c2 := mk("a2", "boxA", 40), mk("c2", "boxC", 40) + mk("b2", "boxB", 40) // the emptiest server, on another machine than a1 + a2.AddShards(99, "col1", 0, bits(0)) + c2.AddShards(99, "col1", 0, bits(1)) + data, parity := map[volKey]int{{"col1", 99}: 2}, map[volKey]int{{"col1", 99}: 1} + for vid := uint32(1); vid <= 6; vid++ { + a1.AddShards(vid, "col1", 0, bits(0)) + b1.AddShards(vid, "col1", 0, bits(1)) + c1.AddShards(vid, "col1", 0, bits(2)) + data[volKey{"col1", vid}], parity[volKey{"col1", vid}] = 2, 1 + } + + moves := detectGlobalImbalance(topo.nodes, buildRacks(topo.nodes), "", 0.2, data, parity, 10, true) + if len(moves) == 0 { + t.Fatal("no global moves: the fullest server never drained") + } + if moves[0].source != a1 || moves[0].target != a2 { + t.Errorf("first move %s -> %s, want a1 -> a2", moves[0].source.id, moves[0].target.id) + } + held := map[string]map[uint32]int{} + for _, m := range moves { + if m.source.host != m.target.host { + t.Errorf("cross-machine move %d.%d %s -> %s concentrates the volume", m.volumeID, m.shardID, m.source.id, m.target.id) + } + if held[m.target.id] == nil { + held[m.target.id] = map[uint32]int{} + } + if held[m.target.id][m.volumeID]++; held[m.target.id][m.volumeID] > 1 { + t.Errorf("%s received two shards of volume %d", m.target.id, m.volumeID) + } + } +} + +// TestGlobalTriesTargetsByFullnessAfterTheMove: a one-slot server is the +// emptiest but would be full after one shard, which must not stop the search +// before a larger server that can take the shard. +func TestGlobalTriesTargetsByFullnessAfterTheMove(t *testing.T) { + topo := NewTopology() + src := topo.AddNode("src", "dc1", "dc1:rack1", 2) + topo.AddNode("tiny", "dc1", "dc1:rack1", 1) + big := topo.AddNode("big", "dc1", "dc1:rack1", 9) + data, parity := map[volKey]int{}, map[volKey]int{} + for vid := uint32(1); vid <= 9; vid++ { + n := src + if vid == 9 { + n = big + } + n.AddShards(vid, "col1", 0, bits(0)) + data[volKey{"col1", vid}], parity[volKey{"col1", vid}] = 2, 1 + } + + moves := detectGlobalImbalance(topo.nodes, buildRacks(topo.nodes), "", 0.2, data, parity, 10, true) + if len(moves) == 0 { + t.Fatal("no global moves: the one-slot server stopped the search") + } + for _, m := range moves { + if m.source != src || m.target != big { + t.Errorf("move %d.%d %s -> %s, want src -> big", m.volumeID, m.shardID, m.source.id, m.target.id) + } + } +} + +// TestGlobalMoveFreesTheSourceDisk: a shard moved off a server frees its disk +// slot, so a later pick of that server can choose the disk again. +func TestGlobalMoveFreesTheSourceDisk(t *testing.T) { + topo := NewTopology() + src := topo.AddNode("src", "dc1", "dc1:rack1", 0) + src.AddDisk(1, "", 0, 4) + topo.AddNode("dst", "dc1", "dc1:rack1", 10) + data, parity := map[volKey]int{}, map[volKey]int{} + for vid := uint32(1); vid <= 4; vid++ { + src.AddShards(vid, "col1", 1, bits(0)) + data[volKey{"col1", vid}], parity[volKey{"col1", vid}] = 2, 1 + } + + moves := detectGlobalImbalance(topo.nodes, buildRacks(topo.nodes), "", 0.2, data, parity, 1, true) + if len(moves) != 1 { + t.Fatalf("got %d moves, want 1", len(moves)) + } + if d := src.disks[1]; d.freeSlots != 1 || d.shardCount != 3 { + t.Errorf("source disk after the move: %d free, %d shards; want 1 free, 3 shards", d.freeSlots, d.shardCount) + } +} + +// TestGlobalSameMachineMoveSkipsASiblingHoldingTheVolume: a same-machine move +// must not put a second shard of a volume on a server that already holds one. +func TestGlobalSameMachineMoveSkipsASiblingHoldingTheVolume(t *testing.T) { + topo := NewTopology() + a1 := topo.AddNode("a1", "dc1", "dc1:rack1", 1) + a2 := topo.AddNode("a2", "dc1", "dc1:rack1", 10) + a1.SetHost("boxA") + a2.SetHost("boxA") + data, parity := map[volKey]int{}, map[volKey]int{} + for vid := uint32(1); vid <= 3; vid++ { + a1.AddShards(vid, "col1", 0, bits(0)) + data[volKey{"col1", vid}], parity[volKey{"col1", vid}] = 2, 1 + } + a2.AddShards(1, "col1", 0, bits(1)) + + moves := detectGlobalImbalance(topo.nodes, buildRacks(topo.nodes), "", 0.2, data, parity, 10, true) + if len(moves) == 0 { + t.Fatal("no global moves") + } + for _, m := range moves { + if m.volumeID == 1 { + t.Errorf("moved shard 1.%d %s -> %s onto a server holding the volume", m.shardID, m.source.id, m.target.id) + } + } +} diff --git a/weed/worker/tasks/ec_balance/detection.go b/weed/worker/tasks/ec_balance/detection.go index b74f2bf289..661c2976b7 100644 --- a/weed/worker/tasks/ec_balance/detection.go +++ b/weed/worker/tasks/ec_balance/detection.go @@ -97,6 +97,12 @@ func Detection( return nil, false, nil } + // The global phase is capped per rack while the per-volume phases are not, + // so order it ahead of the within-rack moves: otherwise a long within-rack + // backlog pushes every global move past the detection cap, and a server that + // holds far more shards than its peers never drains. + sortMovesByPhase(moves) + hasMore := false if maxResults > 0 && len(moves) > maxResults { moves = moves[:maxResults] @@ -339,6 +345,48 @@ func countEcShards(ecShardInfos []*master_pb.VolumeEcShardInformationMessage) in return count } +// sortMovesByPhase orders a plan for truncation, keeping the planner's order +// within a phase: dedup and cross-rack first, then the capped global phase, +// then within-rack. A volume's moves keep their plan order: a global move can +// take a shard that an earlier within-rack move placed, and the scheduler runs +// only a volume's first move per cycle. +func sortMovesByPhase(moves []ecbalancer.Move) { + type volumeKey struct { + collection string + vid uint32 + } + volumeOrder := make(map[volumeKey]int) + order := make([]int, len(moves)) + for i, m := range moves { + vk := volumeKey{m.Collection, m.VolumeID} + volumeOrder[vk] = max(volumeOrder[vk], movePhaseOrder(m.Phase)) + order[i] = volumeOrder[vk] + } + idx := make([]int, len(moves)) + for i := range idx { + idx[i] = i + } + slices.SortStableFunc(idx, func(a, b int) int { return order[a] - order[b] }) + sorted := make([]ecbalancer.Move, len(moves)) + for i, j := range idx { + sorted[i] = moves[j] + } + copy(moves, sorted) +} + +func movePhaseOrder(phase string) int { + switch phase { + case "dedup": + return 0 + case "cross_rack": + return 1 + case "global": + return 2 + default: + return 3 + } +} + func movePhasePriority(phase string) types.TaskPriority { switch phase { case "dedup": diff --git a/weed/worker/tasks/ec_balance/detection_test.go b/weed/worker/tasks/ec_balance/detection_test.go index 2770fdcc8a..80374e5b4d 100644 --- a/weed/worker/tasks/ec_balance/detection_test.go +++ b/weed/worker/tasks/ec_balance/detection_test.go @@ -3,8 +3,11 @@ package ec_balance import ( "context" "net" + "slices" + "strings" "testing" + "github.com/seaweedfs/seaweedfs/weed/admin/topology" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer" @@ -154,3 +157,99 @@ func TestMovePhasePriority(t *testing.T) { // keep the erasure_coding import meaningful for future adapter tests var _ = erasure_coding.DataShardsCount + +// TestSortMovesByPhaseKeepsGlobalAheadOfWithinRack: a long within-rack backlog +// must not push the capped global moves past the detection cap. +func TestSortMovesByPhaseKeepsGlobalAheadOfWithinRack(t *testing.T) { + moves := []ecbalancer.Move{ + {VolumeID: 1, Phase: "within_rack"}, + {VolumeID: 2, Phase: "dedup"}, + {VolumeID: 3, Phase: "within_rack"}, + {VolumeID: 4, Phase: "global"}, + {VolumeID: 5, Phase: "cross_rack"}, + {VolumeID: 6, Phase: "global"}, + } + sortMovesByPhase(moves) + var got []uint32 + for _, m := range moves { + got = append(got, m.VolumeID) + } + if want := []uint32{2, 5, 4, 6, 1, 3}; !slices.Equal(got, want) { + t.Errorf("order = %v, want %v", got, want) + } +} + +// TestSortMovesByPhaseKeepsAVolumesMovesInPlanOrder: a global move of a shard +// that a within-rack move placed must not run ahead of that move. +func TestSortMovesByPhaseKeepsAVolumesMovesInPlanOrder(t *testing.T) { + moves := []ecbalancer.Move{ + {VolumeID: 1, ShardID: 3, SourceNode: "within 1", Phase: "within_rack"}, + {VolumeID: 2, SourceNode: "within 2", Phase: "within_rack"}, + {VolumeID: 1, ShardID: 3, SourceNode: "global 1", Phase: "global"}, + {VolumeID: 3, SourceNode: "global 3", Phase: "global"}, + {VolumeID: 2, Collection: "other", SourceNode: "global other 2", Phase: "global"}, + } + sortMovesByPhase(moves) + var got []string + for _, m := range moves { + got = append(got, m.SourceNode) + } + if want := []string{"global 3", "global other 2", "within 1", "within 2", "global 1"}; !slices.Equal(got, want) { + t.Errorf("order = %q, want %q", got, want) + } +} + +// TestDetectionKeepsGlobalMovesUnderTheCap: the within-rack moves that come +// first in plan order must not push every global move past maxResults. +func TestDetectionKeepsGlobalMovesUnderTheCap(t *testing.T) { + disk := func(shards ...*master_pb.VolumeEcShardInformationMessage) map[string]*master_pb.DiskInfo { + return map[string]*master_pb.DiskInfo{"": {Type: "", MaxVolumeCount: 100, EcShardInfos: shards}} + } + // Volume 100 sits on one machine (within-rack moves); volumes 1-20 load one + // server (global moves). + var loaded []*master_pb.VolumeEcShardInformationMessage + for vid := uint32(1); vid <= 20; vid++ { + loaded = append(loaded, &master_pb.VolumeEcShardInformationMessage{Id: vid, Collection: "col1", EcIndexBits: 1}) + } + topoInfo := &master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{{Id: "rack1", DataNodeInfos: []*master_pb.DataNodeInfo{ + {Id: "10.0.0.1:8080", DiskInfos: disk(&master_pb.VolumeEcShardInformationMessage{Id: 100, Collection: "col1", EcIndexBits: 0x3FFF})}, + {Id: "10.0.0.2:8080", DiskInfos: disk(loaded...)}, + {Id: "10.0.0.3:8080", DiskInfos: disk()}, + {Id: "10.0.0.4:8080", DiskInfos: disk()}, + }}}, + }}, + } + at := topology.NewActiveTopology(0) + if err := at.UpdateTopology(topoInfo); err != nil { + t.Fatalf("UpdateTopology: %v", err) + } + clusterInfo := &types.ClusterInfo{ActiveTopology: at} + + all, _, err := Detection(context.Background(), nil, clusterInfo, NewDefaultConfig(), 0) + if err != nil { + t.Fatalf("Detection: %v", err) + } + withinRack := 0 + for _, r := range all { + if strings.Contains(r.Reason, " within_rack:") { + withinRack++ + } + } + if withinRack < 2 { + t.Fatalf("fixture plans %d within-rack moves, want at least 2", withinRack) + } + + results, hasMore, err := Detection(context.Background(), nil, clusterInfo, NewDefaultConfig(), 1) + if err != nil { + t.Fatalf("Detection: %v", err) + } + if len(results) != 1 || !hasMore { + t.Fatalf("got %d results (hasMore %v), want 1 and more", len(results), hasMore) + } + if !strings.Contains(results[0].Reason, " global:") { + t.Errorf("kept %q, want a global move", results[0].Reason) + } +}