diff --git a/weed/storage/erasure_coding/ecbalancer/balancer.go b/weed/storage/erasure_coding/ecbalancer/balancer.go index feac3756d..47569e0e9 100644 --- a/weed/storage/erasure_coding/ecbalancer/balancer.go +++ b/weed/storage/erasure_coding/ecbalancer/balancer.go @@ -504,10 +504,11 @@ func balanceShardTypeAcrossRacks(vk volKey, nodes map[string]*Node, racks map[st var moves []*move for _, pm := range toMove { hasRoom := func(r string) bool { - return r != pm.src.rack && racks[r].freeSlots > 0 && + return r != pm.src.rack && rackFreeForType(racks[r], diskType) > 0 && (!pm.avoidDataRack || !antiAffinity[r]) } - destRack, ok := pickTarget(rackKeys, shardsPerRack, maxPerRack, antiAffinity, hasRoom, withinLimit) + rackFree := func(r string) int { return rackFreeForType(racks[r], diskType) } + destRack, ok := pickTarget(rackKeys, shardsPerRack, maxPerRack, antiAffinity, hasRoom, withinLimit, rackFree) if !ok && totalCap > 0 && rackShardCount[pm.src.rack] > totalCap { // No rack is under the per-type cap. A shard must not stay on a rack // above the total cap while another rack is under it, so relax the @@ -515,12 +516,12 @@ func balanceShardTypeAcrossRacks(vk volKey, nodes map[string]*Node, racks map[st // inside pickTarget. Only a source above the total cap may relax: // otherwise the destination just takes over the per-type overflow // and the next Plan moves it back. - destRack, ok = pickTarget(rackKeys, shardsPerRack, totalCap, antiAffinity, hasRoom, withinLimit) + destRack, ok = pickTarget(rackKeys, shardsPerRack, totalCap, antiAffinity, hasRoom, withinLimit, rackFree) } if !ok { continue } - destNode := pickNodeInRack(racks[destRack], vk, rp) + destNode := pickNodeInRack(racks[destRack], vk, rp, diskType) if destNode == nil { continue } @@ -553,15 +554,55 @@ func balanceShardTypeAcrossRacks(vk volKey, nodes map[string]*Node, racks map[st return moves } -func pickNodeInRack(r *rack, vk volKey, rp *super_block.ReplicaPlacement) *Node { - return pickBestNodeForVolume(sortedNodeSlice(r.nodes), vk, rp) +func pickNodeInRack(r *rack, vk volKey, rp *super_block.ReplicaPlacement, diskType string) *Node { + return pickBestNodeForVolumeOnDisk(sortedNodeSlice(r.nodes), vk, rp, diskType) } -// pickBestNodeForVolume returns the node with the fewest shards of the volume that -// has a free slot and is under the SameRackCount cap, or nil. +// pickBestNodeForVolumeOnDisk is pickBestNodeForVolume restricted to nodes with +// free capacity on a disk of diskType, so the pick cannot land on a node whose +// room sits on another type. +func pickBestNodeForVolumeOnDisk(nodes []*Node, vk volKey, rp *super_block.ReplicaPlacement, diskType string) *Node { + eligible := make([]*Node, 0, len(nodes)) + for _, node := range nodes { + if freeSlotsForType(node, diskType) > 0 { + eligible = append(eligible, node) + } + } + return pickBestNodeForVolume(eligible, vk, rp) +} + +// freeSlotsForType is the node's free EC slots usable by diskType. Per-disk +// metadata is authoritative when present; without it the node's aggregate +// stands in so the volume server's own disk selection can place the shard. +func freeSlotsForType(n *Node, diskType string) int { + if len(n.disks) == 0 { + return n.freeSlots + } + free := 0 + for _, d := range n.disks { + if d.freeSlots > 0 && (diskType == "" || d.diskType == diskType) { + free += d.freeSlots + } + } + return free +} + +func rackFreeForType(r *rack, diskType string) int { + free := 0 + for _, n := range r.nodes { + free += freeSlotsForType(n, diskType) + } + return free +} + +// pickBestNodeForVolume spreads this volume evenly first. Remainders: higher +// freeSlots, then lower liveShards+assignedThisPlace. SameRackCount is a hard +// cap (no +1 for debt). func pickBestNodeForVolume(nodes []*Node, vk volKey, rp *super_block.ReplicaPlacement) *Node { var best *Node bestCount := -1 + bestFree := -1 + bestScore := -1 for _, node := range nodes { if node.freeSlots <= 0 { continue @@ -570,8 +611,20 @@ func pickBestNodeForVolume(nodes []*Node, vk volKey, rp *super_block.ReplicaPlac if rp != nil && rp.SameRackCount > 0 && count >= rp.SameRackCount { continue } - if best == nil || count < bestCount { - best, bestCount = node, count + score := nodeLiveShardCount(node) + better := false + switch { + case best == nil: + better = true + case count != bestCount: + better = count < bestCount + case node.freeSlots != bestFree: + better = node.freeSlots > bestFree + default: + better = score < bestScore + } + if better { + best, bestCount, bestFree, bestScore = node, count, node.freeSlots, score } } return best @@ -662,12 +715,21 @@ func withinRackMachineSpread(vk volKey, r *rack, machines map[string][]*Node, di // shard (free slot, under SameRackCount), so a capped machine is skipped // rather than settled on and the move dropped. destHost, ok := pickTarget(machineKeys, shardsPerMachine, maxPerMachine, nil, - func(h string) bool { return h != pm.src.host && pickBestNodeForVolume(machines[h], vk, rp) != nil }, - func(string) bool { return true }) + func(h string) bool { + return h != pm.src.host && pickBestNodeForVolumeOnDisk(machines[h], vk, rp, diskType) != nil + }, + func(string) bool { return true }, + func(h string) int { + n := 0 + for _, node := range machines[h] { + n += freeSlotsForType(node, diskType) + } + return n + }) if !ok { continue } - destNode := pickBestNodeForVolume(machines[destHost], vk, rp) + destNode := pickBestNodeForVolumeOnDisk(machines[destHost], vk, rp, diskType) if destNode == nil { continue } @@ -749,13 +811,14 @@ func balanceShardTypeAcrossNodes(vk volKey, r *rack, diskType string, dataShards var moves []*move for _, pm := range toMove { destID, ok := pickTarget(nodeKeys, shardsPerNode, maxPerNode, antiAffinity, - func(n string) bool { return n != pm.src.id && r.nodes[n].freeSlots > 0 }, + func(n string) bool { return n != pm.src.id && freeSlotsForType(r.nodes[n], diskType) > 0 }, func(n string) bool { if rp != nil && rp.SameRackCount > 0 { return nodeShardCount[n] < rp.SameRackCount } return true - }) + }, + func(n string) int { return freeSlotsForType(r.nodes[n], diskType) }) if !ok { continue } @@ -967,12 +1030,17 @@ func shardsByGroup(vk volKey, nodes map[string]*Node, dataShards int, key func(* // pickTarget selects a destination key with room for another shard of a type, in // two passes: first excluding anti-affinity targets, then any valid target. Among -// valid targets it prefers the fewest shards of this type; ties break on sorted -// key order, so selection is deterministic. -func pickTarget(candidates []string, shardsPerTarget map[string][]int, maxPerTarget int, antiAffinity map[string]bool, hasFreeSlots, withinLimit func(string) bool) (string, bool) { +// valid targets it prefers the fewest shards of this type (even spread). Count +// ties break on remaining freeSlots, not sorted key order: otherwise 14/4 +// leftovers always land on the first rack/machine/node ids and the extra shards +// of 4+4+3+3 never walk. Equal freeSlots still keep the first seen key so a +// single run stays deterministic. Node-in-rack Place remainders (after even) are +// freeSlots then liveShards — see pickNodeInRackEligible. +func pickTarget(candidates []string, shardsPerTarget map[string][]int, maxPerTarget int, antiAffinity map[string]bool, hasFreeSlots, withinLimit func(string) bool, freeSlots func(string) int) (string, bool) { try := func(skipAnti bool) (string, bool) { best := "" bestCount := maxPerTarget + 1 + bestFree := -1 for _, c := range candidates { if skipAnti && antiAffinity[c] { continue @@ -986,8 +1054,13 @@ func pickTarget(candidates []string, shardsPerTarget map[string][]int, maxPerTar if !withinLimit(c) { continue } - if cnt := len(shardsPerTarget[c]); cnt < bestCount { - best, bestCount = c, cnt + cnt := len(shardsPerTarget[c]) + free := 0 + if freeSlots != nil { + free = freeSlots(c) + } + if best == "" || cnt < bestCount || (cnt == bestCount && free > bestFree) { + best, bestCount, bestFree = c, cnt, free } } return best, best != "" @@ -1169,6 +1242,19 @@ func volumeShardCount(node *Node, vk volKey) int { return 0 } +// nodeLiveShardCount is the disk-level EC shard cardinality (master EcShards), +// not the UI "EC volumes" count (len of the volume map). +func nodeLiveShardCount(n *Node) int { + if n == nil { + return 0 + } + total := 0 + for _, d := range n.disks { + total += d.shardCount + } + return total +} + func nodeInRackHoldingShard(nodes map[string]*Node, rackID string, vk volKey, shardID int) *Node { var inRack []*Node for _, id := range sortedNodeKeys(nodes) { diff --git a/weed/storage/erasure_coding/ecbalancer/balancer_test.go b/weed/storage/erasure_coding/ecbalancer/balancer_test.go index a7ffee955..0748452b3 100644 --- a/weed/storage/erasure_coding/ecbalancer/balancer_test.go +++ b/weed/storage/erasure_coding/ecbalancer/balancer_test.go @@ -90,6 +90,71 @@ func TestPickBestDiskOnNode(t *testing.T) { }) } +func TestPickBestNodeForVolume(t *testing.T) { + const vid = uint32(100) + vk := volKey{collection: "c", vid: vid} + + t.Run("more free slots wins the count tie", func(t *testing.T) { + topo := NewTopology() + thin := topo.AddNode("n1", "dc1", "dc1:r1", 5) + thin.AddDisk(1, "", 5, 0) + fat := topo.AddNode("n2", "dc1", "dc1:r1", 50) + fat.AddDisk(1, "", 50, 0) + if got := pickBestNodeForVolume([]*Node{thin, fat}, vk, nil); got != fat { + t.Errorf("got %v, want n2 (more free slots)", got.id) + } + }) + + t.Run("fewer live shards wins the free tie", func(t *testing.T) { + topo := NewTopology() + idle := topo.AddNode("n1", "dc1", "dc1:r1", 10) + idle.AddDisk(1, "", 10, 0) + busy := topo.AddNode("n2", "dc1", "dc1:r1", 10) + busy.AddDisk(1, "", 10, 9) + if got := pickBestNodeForVolume([]*Node{busy, idle}, vk, nil); got != idle { + t.Errorf("got %v, want n1 (fewer live shards)", got.id) + } + }) + + t.Run("volume spread beats free slots", func(t *testing.T) { + topo := NewTopology() + holding := topo.AddNode("n1", "dc1", "dc1:r1", 100) + holding.AddDisk(1, "", 100, 1) + holding.AddShards(vid, "c", 1, bits(0)) + empty := topo.AddNode("n2", "dc1", "dc1:r1", 1) + empty.AddDisk(1, "", 1, 0) + if got := pickBestNodeForVolume([]*Node{holding, empty}, vk, nil); got != empty { + t.Errorf("got %v, want n2 (volume spread first)", got.id) + } + }) +} + +func TestPlanSkipsDestinationsWithoutMatchingDiskType(t *testing.T) { + topo := NewTopology() + src := topo.AddNode("node1", "dc1", "dc1:rack1", 0) + src.AddDisk(0, "ssd", 0, 14) + src.AddShards(100, "col1", 0, bits(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13)) + // Aggregate capacity alone would make the hdd-only node look like the + // roomiest destination; none of its slots can take an ssd shard. + wrong := topo.AddNode("node2", "dc1", "dc1:rack2", 600) + wrong.AddDisk(1, "hdd", 600, 0) + right := topo.AddNode("node3", "dc1", "dc1:rack3", 600) + right.AddDisk(1, "ssd", 600, 0) + + moves := Plan(topo, Options{ImbalanceThreshold: 0.01, DiskType: "ssd", Ratio: ratio(10, 4)}) + if len(moves) == 0 { + t.Fatal("expected moves") + } + for _, m := range moves { + if m.TargetNode == wrong.id { + t.Errorf("shard %d targeted the hdd-only node", m.ShardID) + } + if m.TargetNode == right.id && m.TargetDisk != 1 { + t.Errorf("shard %d landed on disk %d, want 1", m.ShardID, m.TargetDisk) + } + } +} + func TestPlanSourceDiskAttribution(t *testing.T) { shardsByDisk := map[uint32][]int{0: {0, 1, 2}, 1: {3, 4, 5}, 2: {6, 7}, 3: {8, 9}, 4: {10, 11}, 5: {12, 13}} shardToDisk := map[int]uint32{} diff --git a/weed/storage/erasure_coding/ecbalancer/place.go b/weed/storage/erasure_coding/ecbalancer/place.go index d42d96f24..136d1028c 100644 --- a/weed/storage/erasure_coding/ecbalancer/place.go +++ b/weed/storage/erasure_coding/ecbalancer/place.go @@ -389,7 +389,8 @@ func chooseShardDest(vk volKey, sid int, isData bool, dataShards, typeTotal, num return !tried[r] && racks[r].freeSlots > 0 && rackShardRoom(racks[r], vk, eligible, maxPerDisk) > 0 } for { - destRack, ok := pickTarget(rackKeys, shardsPerRackType, maxPerRack, anti, hasRoom, withinLimit) + destRack, ok := pickTarget(rackKeys, shardsPerRackType, maxPerRack, anti, hasRoom, withinLimit, + func(r string) int { return racks[r].freeSlots }) if !ok { return nil, 0, false, false } @@ -480,15 +481,13 @@ func rackShardRoomUnderRP(r *rack, vk volKey, eligible func(*disk) bool, maxPerD // FromActiveTopology keeps all disk types/tags in the snapshot, so without this a // node with free volume slots but no eligible disk could be chosen. // -// Among eligible nodes it ranks by fewest shards of the volume per machine, then per -// node, with free capacity breaking ties. The free-capacity tie-break (not sorted id) -// keeps the lowest-id machine from winning every volume's first shard against the -// shared encode snapshot and piling up load. +// Even first (this volume, machine then node). Machine freeSlots remainder +// (mFree) was dropped: keep host even (mCount); remainder is disk freeSlots +// then liveShards so thin disks fill first. SameRackCount / 2+2 unchanged. func pickNodeInRackEligible(r *rack, vk volKey, rp *super_block.ReplicaPlacement, eligible func(*disk) bool, maxPerDisk int) *Node { machineShards := countShardsByHost(vk, r.nodes) - machineFree := freeSlotsByHost(r.nodes) var best *Node - var bestMCount, bestMFree, bestNCount, bestNFree int + var bestMCount, bestNCount, bestNFree, bestNScore int for _, id := range sortedNodeKeys(r.nodes) { node := r.nodes[id] if nodeShardRoom(node, vk, eligible, maxPerDisk) <= 0 { @@ -498,22 +497,23 @@ func pickNodeInRackEligible(r *rack, vk volKey, rp *super_block.ReplicaPlacement if rp != nil && rp.SameRackCount > 0 && count >= rp.SameRackCount { continue } - mCount, mFree := machineShards[node.host], machineFree[node.host] + mCount := machineShards[node.host] + nScore := nodeLiveShardCount(node) better := false switch { case best == nil: better = true case mCount != bestMCount: better = mCount < bestMCount - case mFree != bestMFree: - better = mFree > bestMFree case count != bestNCount: better = count < bestNCount - default: + case node.freeSlots != bestNFree: better = node.freeSlots > bestNFree + default: + better = nScore < bestNScore } if better { - best, bestMCount, bestMFree, bestNCount, bestNFree = node, mCount, mFree, count, node.freeSlots + best, bestMCount, bestNCount, bestNFree, bestNScore = node, mCount, count, node.freeSlots, nScore } } return best @@ -626,6 +626,9 @@ func (t *Topology) ClearShardAccounting(nodeID, collection string, vid uint32, s // distributing the new shards) neither occupy capacity nor skew anti-affinity / // per-disk caps during planning. Unlike repair's ClearShardAccounting, it credits // freeSlots because the deletes run before the new writes. +// +// Other volumes' disk.shardCount stay put: zeroing the snapshot here is what +// made within-rack Place see a green field and 50/50 the current volume only. func (t *Topology) ReleaseVolumeShards(collection string, vid uint32) { vk := volKey{collection: collection, vid: vid} for _, n := range t.nodes { diff --git a/weed/storage/erasure_coding/ecbalancer/place_test.go b/weed/storage/erasure_coding/ecbalancer/place_test.go index e6f625a94..cf8f47f2e 100644 --- a/weed/storage/erasure_coding/ecbalancer/place_test.go +++ b/weed/storage/erasure_coding/ecbalancer/place_test.go @@ -2,6 +2,7 @@ package ecbalancer import ( "fmt" + "sort" "testing" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" @@ -500,3 +501,416 @@ func TestPlaceStrictCapsCountEligibleRacks(t *testing.T) { } } } + +// TestPlaceRotatesExtraRackShardsAcrossVolumes: 10+4 onto 4 equal racks is +// always 4+4+3+3. Sorted-id tie-breaks pinned the two fours on the first rack +// ids every volume. Free-slot tie-breaks must walk the extra shards so every +// rack is in the "4" set across a short series. +func TestPlaceRotatesExtraRackShardsAcrossVolumes(t *testing.T) { + topo := buildPlaceTopo(4, 2, 200) + const volumes = 8 + gotFour := map[string]int{} + for v := 0; v < volumes; v++ { + res, err := topo.Place(uint32(v+1), "c1", allShards(), Constraints{}, PlaceStrict) + if err != nil { + t.Fatalf("Place volume %d: %v", v+1, err) + } + perRack := map[string]int{} + for _, d := range res.Destinations { + perRack[d.Rack]++ + } + if len(perRack) != 4 { + t.Fatalf("volume %d racks %v, want 4", v+1, perRack) + } + counts := make([]int, 0, 4) + for rack, n := range perRack { + counts = append(counts, n) + if n != 3 && n != 4 { + t.Errorf("volume %d rack %s has %d shards, want 3 or 4", v+1, rack, n) + } + if n == 4 { + gotFour[rack]++ + } + } + sort.Ints(counts) + if counts[0] != 3 || counts[3] != 4 { + t.Errorf("volume %d rack counts %v, want 3,3,4,4", v+1, perRack) + } + } + if len(gotFour) < 4 { + t.Errorf("extra (4-shard) racks did not rotate; hit %v over %d volumes", gotFour, volumes) + } +} + +func twoDiskRackRP() *super_block.ReplicaPlacement { + return &super_block.ReplicaPlacement{DiffRackCount: 4, SameRackCount: 2} +} + +func placeNeed(t *testing.T, topo *Topology, vid uint32, n int, rp *super_block.ReplicaPlacement) map[string]int { + t.Helper() + need := make([]int, n) + for i := range need { + need[i] = i + } + res, err := topo.Place(vid, "c1", need, Constraints{ + ReplicaPlacement: rp, + Ratio: func(string) (int, int) { return 10, 4 }, + }, PlaceDurabilityFirst) + if err != nil { + t.Fatalf("Place vid=%d: %v", vid, err) + } + got := map[string]int{} + for _, d := range res.Destinations { + got[d.Node]++ + } + return got +} + +// TestWithinRackPaysDownShardDebt3: 2 VS, same disk cap, live 100 vs 135, +// 3 shards, cap 2. freeSlots = cap−shards so the thinner also has more +// remaining slots; extras pay the debt — not 50/50 from vol 1. +// Equal remaining slots (same free, different shards) are the liveShards +// tie-break; see TestEcVolumeCountIsNotScore. +func TestWithinRackPaysDownShardDebt3(t *testing.T) { + topo := NewTopology() + a := topo.AddNode("a:8080", "dc1", "dc1:rack0", 4900) + a.AddDisk(0, "", 4900, 100) + b := topo.AddNode("b:8081", "dc1", "dc1:rack0", 4865) + b.AddDisk(0, "", 4865, 135) + rp := twoDiskRackRP() + + got := placeNeed(t, topo, 1, 3, rp) + if got["a:8080"] != 2 || got["b:8081"] != 1 { + t.Fatalf("first volume %v, want a=2 b=1 (pair on thinner)", got) + } + + const rounds = 35 + for v := uint32(2); v <= rounds; v++ { + beforeA, beforeB := nodeLiveShardCount(a), nodeLiveShardCount(b) + got = placeNeed(t, topo, v, 3, rp) + if beforeA+2 <= beforeB && (got["a:8080"] != 2 || got["b:8081"] != 1) { + t.Fatalf("volume %d %v before A=%d B=%d; debt≥2 must stay 2+1 on A", + v, got, beforeA, beforeB) + } + } + if ga, gb := nodeLiveShardCount(a), nodeLiveShardCount(b); ga != 170 || gb != 170 { + t.Errorf("after %d×(2+1) live A=%d B=%d, want 170/170", rounds, ga, gb) + } +} + +// TestWithinRackFourShardsStayEven: 4 shards / 2 disks is always 2+2. +// Remainder is 0 — packing 3+1 would put 3 shards on one disk. +func TestWithinRackFourShardsStayEven(t *testing.T) { + topo := NewTopology() + a := topo.AddNode("a:8080", "dc1", "dc1:rack0", 5000) + a.AddDisk(0, "", 5000, 100) + b := topo.AddNode("b:8081", "dc1", "dc1:rack0", 5000) + b.AddDisk(0, "", 5000, 130) + rp := twoDiskRackRP() + + got := placeNeed(t, topo, 1, 4, rp) + if got["a:8080"] != 2 || got["b:8081"] != 2 { + t.Fatalf("first volume %v, want 2+2 (no 3+1)", got) + } + if ga, gb := nodeLiveShardCount(a), nodeLiveShardCount(b); ga != 102 || gb != 132 { + t.Fatalf("after 2+2 live A=%d B=%d, want 102/132 (gap frozen)", ga, gb) + } + got = placeNeed(t, topo, 2, 4, rp) + if got["a:8080"] != 2 || got["b:8081"] != 2 { + t.Fatalf("second volume %v, want 2+2", got) + } +} + +// TestEcVolumeCountIsNotScore: both VS show the same "EC volumes" cardinality; +// the pair still goes to the lower shard count. +func TestEcVolumeCountIsNotScore(t *testing.T) { + topo := NewTopology() + a := topo.AddNode("a:8080", "dc1", "dc1:rack0", 5000) + a.AddDisk(0, "", 5000, 2528) + b := topo.AddNode("b:8081", "dc1", "dc1:rack0", 5000) + b.AddDisk(0, "", 5000, 2563) + for i := uint32(1); i <= 1000; i++ { + a.AddShards(i, "other", 0, erasure_coding.ShardBits(1)) + b.AddShards(i, "other", 0, erasure_coding.ShardBits(1)) + } + got := placeNeed(t, topo, 2000, 3, twoDiskRackRP()) + if got["a:8080"] != 2 || got["b:8081"] != 1 { + t.Fatalf("got %v; score must be shard count (2528 vs 2563), not 1000 EC volumes", got) + } +} + +// TestFreeSlotsWinsAsymmetricExtras: more freeSlots on the fatter disk takes +// the remainder even if that disk already has more live shards. 4 shards stay +// 2+2 (remainder 0; no 3+1). +func TestFreeSlotsWinsAsymmetricExtras(t *testing.T) { + topo := NewTopology() + a := topo.AddNode("a:8080", "dc1", "dc1:rack0", 100) + a.AddDisk(0, "", 100, 100) + b := topo.AddNode("b:8081", "dc1", "dc1:rack0", 200) + b.AddDisk(0, "", 200, 135) + rp := twoDiskRackRP() + got := placeNeed(t, topo, 1, 3, rp) + if got["a:8080"] != 1 || got["b:8081"] != 2 { + t.Fatalf("3 shards %v, want 2+1 on B (more freeSlots) even though B has more shards", got) + } + + topo = NewTopology() + a = topo.AddNode("a:8080", "dc1", "dc1:rack0", 100) + a.AddDisk(0, "", 100, 100) + b = topo.AddNode("b:8081", "dc1", "dc1:rack0", 200) + b.AddDisk(0, "", 200, 135) + got = placeNeed(t, topo, 1, 4, rp) + if got["a:8080"] != 2 || got["b:8081"] != 2 { + t.Fatalf("4 shards %v, want 2+2 (remainder 0; freeSlots must not pack 3+1)", got) + } +} + +// TestReleaseVolumeShardsKeepsOtherLiveCounts: encode's ReleaseVolumeShards +// must not zero disk.shardCount for other volumes (that was the green-field bug). +func TestReleaseVolumeShardsKeepsOtherLiveCounts(t *testing.T) { + topo := NewTopology() + n := topo.AddNode("n0:8080", "dc1", "dc1:rack0", 100) + n.AddDisk(0, "", 90, 10) + n.AddShards(1, "c1", 0, erasure_coding.ShardBits(1<<0)) + n.AddShards(2, "c1", 0, erasure_coding.ShardBits(1<<1|1<<2)) + n.disks[0].shardCount = 10 + n.disks[0].freeSlots = 90 + + topo.ReleaseVolumeShards("c1", 1) + + if nodeLiveShardCount(n) != 9 { + t.Errorf("live shardCount = %d after releasing 1 of 10, want 9 (must not zero the snapshot)", nodeLiveShardCount(n)) + } + if _, ok := n.shards[volKey{collection: "c1", vid: 2}]; !ok { + t.Error("other volume shards were dropped") + } +} + +const imbalanceLiveBase = 200 +const imbalanceDiskCap = 5000 + +// buildImbalancedTopo is racks × machines × disks: one VS per machine, disk +// ids 0..D-1. Earlier (rack, machine, disk) ids are fatter — more liveShards, +// fewer freeSlots (cap−live). Sorted-id extras would pile on those first ids; +// freeSlots-then-liveShards extras must walk the other way. +func buildImbalancedTopo(racks, machines, disks int) *Topology { + topo := NewTopology() + idx := 0 + for r := 0; r < racks; r++ { + rackKey := fmt.Sprintf("dc1:rack%d", r) + for m := 0; m < machines; m++ { + host := fmt.Sprintf("10.0.%d.%d", r, m) + lives := make([]int, disks) + frees := make([]int, disks) + nodeFree := 0 + for d := 0; d < disks; d++ { + lives[d] = imbalanceLiveBase - idx + idx++ + frees[d] = imbalanceDiskCap - lives[d] + nodeFree += frees[d] + } + n := topo.AddNode(host+":8080", "dc1", rackKey, nodeFree) + n.SetHost(host) + for d := 0; d < disks; d++ { + n.AddDisk(uint32(d), "", frees[d], lives[d]) + } + } + } + return topo +} + +type diskKey struct { + node string + diskID uint32 +} + +func snapshotDiskLive(topo *Topology) map[diskKey]int { + out := make(map[diskKey]int) + for _, n := range topo.nodes { + for id, d := range n.disks { + out[diskKey{n.id, id}] = d.shardCount + } + } + return out +} + +func snapshotRackLive(topo *Topology) map[string]int { + out := make(map[string]int) + for _, n := range topo.nodes { + out[n.rack] += nodeLiveShardCount(n) + } + return out +} + +func place104Counts(t *testing.T, topo *Topology) (perDisk map[diskKey]int, perRack map[string]int) { + t.Helper() + res, err := topo.Place(1, "c1", allShards(), Constraints{}, PlaceStrict) + if err != nil { + t.Fatalf("Place: %v", err) + } + if len(res.Destinations) != erasure_coding.TotalShardsCount { + t.Fatalf("placed %d shards, want %d", len(res.Destinations), erasure_coding.TotalShardsCount) + } + perDisk = map[diskKey]int{} + perRack = map[string]int{} + for _, d := range res.Destinations { + perDisk[diskKey{d.Node, d.DiskID}]++ + perRack[topo.nodes[d.Node].rack]++ + } + return perDisk, perRack +} + +func assertTailsPreferThinDisks(t *testing.T, topo *Topology, before map[diskKey]int, got map[diskKey]int) { + t.Helper() + for _, n := range topo.nodes { + type row struct { + live int + got int + id uint32 + } + rows := make([]row, 0, len(n.disks)) + for id := range n.disks { + k := diskKey{n.id, id} + rows = append(rows, row{live: before[k], got: got[k], id: id}) + } + sort.Slice(rows, func(i, j int) bool { + if rows[i].live != rows[j].live { + return rows[i].live > rows[j].live + } + return rows[i].id < rows[j].id + }) + for i := 1; i < len(rows); i++ { + if rows[i-1].got > rows[i].got { + t.Errorf("node %s: fatter disk %d (live=%d) got %d of this volume, thinner disk %d (live=%d) got %d; remainder tails must prefer thin disks, not first ids", + n.id, rows[i-1].id, rows[i-1].live, rows[i-1].got, rows[i].id, rows[i].live, rows[i].got) + } + } + } +} + +func assertRackExtrasWalkThin(t *testing.T, before map[string]int, got map[string]int) { + t.Helper() + type row struct { + rack string + live int + got int + } + rows := make([]row, 0, len(before)) + for rk, live := range before { + rows = append(rows, row{rack: rk, live: live, got: got[rk]}) + } + sort.Slice(rows, func(i, j int) bool { + if rows[i].live != rows[j].live { + return rows[i].live > rows[j].live + } + return rows[i].rack < rows[j].rack + }) + for i := 1; i < len(rows); i++ { + if rows[i-1].got > rows[i].got { + t.Errorf("fatter rack %s (live=%d) got %d, thinner rack %s (live=%d) got %d; extras must walk to thinner racks", + rows[i-1].rack, rows[i-1].live, rows[i-1].got, rows[i].rack, rows[i].live, rows[i].got) + } + } +} + +func assertLiveGapFrozen(t *testing.T, before, after map[diskKey]int, placed map[diskKey]int) { + t.Helper() + keys := make([]diskKey, 0, len(before)) + for k := range before { + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { + if keys[i].node != keys[j].node { + return keys[i].node < keys[j].node + } + return keys[i].diskID < keys[j].diskID + }) + for i := 0; i < len(keys); i++ { + for j := i + 1; j < len(keys); j++ { + a, b := keys[i], keys[j] + if before[a]-before[b] != after[a]-after[b] { + t.Errorf("live gap %s/%d vs %s/%d changed %d → %d (placed %d vs %d); remainder 0 must not close the gap", + a.node, a.diskID, b.node, b.diskID, before[a]-before[b], after[a]-after[b], placed[a], placed[b]) + } + } + } +} + +// TestPlace10Plus4ImbalanceTiebreak: 10+4 onto seeded liveShards/freeSlots +// skew. Remainder tails pay thin disks/racks, not sorted-first ids. Remainder +// 0 (1-1-7, 7-1-2) is a no-op on the live gap — that is the contract. +// Existing 4-1-2 / within-rack 2-disk cases stay in the tests above. +func TestPlace10Plus4ImbalanceTiebreak(t *testing.T) { + cases := []struct { + name string + racks, machines, disks int + remainderZero bool + fatFirstGetsZero bool + rackExtras bool + fatRacksTwoPlusOne bool + }{ + {name: "1-1-4", racks: 1, machines: 1, disks: 4}, + {name: "1-1-6", racks: 1, machines: 1, disks: 6}, + {name: "1-1-8", racks: 1, machines: 1, disks: 8}, + {name: "1-1-15", racks: 1, machines: 1, disks: 15, fatFirstGetsZero: true}, + {name: "1-1-7", racks: 1, machines: 1, disks: 7, remainderZero: true}, + {name: "7-1-2", racks: 7, machines: 1, disks: 2, remainderZero: true}, + {name: "4-2-3", racks: 4, machines: 2, disks: 3, rackExtras: true}, + {name: "5-2-2", racks: 5, machines: 2, disks: 2, rackExtras: true}, + {name: "6-1-2", racks: 6, machines: 1, disks: 2, rackExtras: true, fatRacksTwoPlusOne: true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + topo := buildImbalancedTopo(tc.racks, tc.machines, tc.disks) + beforeDisk := snapshotDiskLive(topo) + beforeRack := snapshotRackLive(topo) + + gotDisk, gotRack := place104Counts(t, topo) + + if tc.fatFirstGetsZero { + var zero []diskKey + for _, n := range topo.nodes { + for id := range n.disks { + k := diskKey{n.id, id} + if gotDisk[k] == 0 { + zero = append(zero, k) + } + } + } + if len(zero) != 1 { + t.Fatalf("want exactly one disk with 0 shards of this volume, got %v (placed %v)", zero, gotDisk) + } + if zero[0].diskID != 0 { + t.Fatalf("zero-of-this-volume disk is %s/%d; must be the fat first-id disk, not the last sorted id", zero[0].node, zero[0].diskID) + } + } + + assertTailsPreferThinDisks(t, topo, beforeDisk, gotDisk) + + if tc.rackExtras || tc.remainderZero && tc.racks > 1 { + assertRackExtrasWalkThin(t, beforeRack, gotRack) + } + + if tc.fatRacksTwoPlusOne { + for _, n := range topo.nodes { + if gotRack[n.rack] != 3 { + continue + } + counts := make([]int, 0, len(n.disks)) + for id := range n.disks { + counts = append(counts, gotDisk[diskKey{n.id, id}]) + } + sort.Ints(counts) + if len(counts) != 2 || counts[0] != 1 || counts[1] != 2 { + t.Errorf("rack %s has 3 shards on disks %v, want 2+1", n.rack, counts) + } + } + } + + if tc.remainderZero { + after := snapshotDiskLive(topo) + assertLiveGapFrozen(t, beforeDisk, after, gotDisk) + } + }) + } +} diff --git a/weed/worker/tasks/ec_balance/detection.go b/weed/worker/tasks/ec_balance/detection.go index 9d147ae3c..253efb858 100644 --- a/weed/worker/tasks/ec_balance/detection.go +++ b/weed/worker/tasks/ec_balance/detection.go @@ -3,6 +3,7 @@ package ec_balance import ( "context" "fmt" + "slices" "time" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -186,6 +187,7 @@ func buildBalancerTopology(topoInfo *master_pb.TopologyInfo, config *Config, all for _, dn := range rack.DataNodeInfos { freeSlots := 0 + freeByType := make(map[string]int) // disk type -> free EC slots diskTypeOf := make(map[uint32]string) // physical disk_id -> disk type diskShardCount := make(map[uint32]int) fullDiskTypes := make(map[string]bool) // physically near-full, ineligible as a target @@ -207,6 +209,7 @@ func buildBalancerTopology(topoInfo *master_pb.TopologyInfo, config *Config, all fs := int(diskInfo.MaxVolumeCount-diskInfo.VolumeCount)*erasure_coding.DataShardsCount - countEcShards(diskInfo.EcShardInfos) if fs > 0 && !fullDiskTypes[diskType] { freeSlots += fs + freeByType[diskType] += fs } // Discover physical disks from regular volumes too, so an // EC-empty disk is still a candidate destination. @@ -234,23 +237,33 @@ func buildBalancerTopology(topoInfo *master_pb.TopologyInfo, config *Config, all // machines, not just nodes (servers on one host are one fault domain). node.SetHost(pb.NewServerAddressFromDataNode(dn).ToHost()) - // Spread the node's free slots only over its non-full disks, so a - // physically full disk type doesn't dilute the share the usable disks - // advertise (its own disks get 0 below). The total is preserved. - nonFullDiskCount := 0 + // Spread each disk type's free slots over that type's non-full + // disks, so a physically full disk type doesn't dilute the share + // the usable disks advertise (its own disks get 0 below) and other + // types don't inflate it. Ceil division keeps the remainder on the + // first disks instead of dropping it. + nonFullByType := make(map[string]int) for _, dt := range diskTypeOf { if !fullDiskTypes[dt] { - nonFullDiskCount++ + nonFullByType[dt]++ } } - perDiskFree := 0 - if nonFullDiskCount > 0 && freeSlots > 0 { - perDiskFree = freeSlots / nonFullDiskCount + var diskIDs []uint32 + for diskID := range diskTypeOf { + diskIDs = append(diskIDs, diskID) } - for diskID, diskType := range diskTypeOf { - diskFree := perDiskFree - if fullDiskTypes[diskType] { - diskFree = 0 + slices.Sort(diskIDs) + remainingByType := make(map[string]int, len(freeByType)) + for dt, free := range freeByType { + remainingByType[dt] = free + } + for _, diskID := range diskIDs { + diskType := diskTypeOf[diskID] + diskFree := 0 + if count := nonFullByType[diskType]; !fullDiskTypes[diskType] && count > 0 { + diskFree = (remainingByType[diskType] + count - 1) / count + remainingByType[diskType] -= diskFree + nonFullByType[diskType]-- } node.AddDisk(diskID, diskType, diskFree, diskShardCount[diskID]) } diff --git a/weed/worker/tasks/ec_balance/multidisk_detection_test.go b/weed/worker/tasks/ec_balance/multidisk_detection_test.go index eb7e6f886..37b70f6d7 100644 --- a/weed/worker/tasks/ec_balance/multidisk_detection_test.go +++ b/weed/worker/tasks/ec_balance/multidisk_detection_test.go @@ -8,6 +8,8 @@ import ( "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" + "github.com/seaweedfs/seaweedfs/weed/util/wildcard" "github.com/seaweedfs/seaweedfs/weed/worker/types" ) @@ -333,3 +335,68 @@ func TestResolveReplicaPlacementFallsBackToMasterDefault(t *testing.T) { }) } } + +// TestBuildBalancerTopologyKeepsSlotRemainder: a node whose single free EC slot +// spans two physical disks must still offer it. Splitting the slot evenly would +// floor each disk to zero and exclude the node despite usable capacity. +func TestBuildBalancerTopologyKeepsSlotRemainder(t *testing.T) { + var all uint32 + for i := 0; i < 14; i++ { + all |= 1 << uint(i) + } + // n2 has one free shard slot: (7-6)*DataShardsCount - 9 = 1, spread over + // disks 0 and 1. Volume 200 is collection-filtered out of balancing so only + // volume 100 moves. + n1 := &master_pb.DataNodeInfo{ + Id: "n1", + DiskInfos: map[string]*master_pb.DiskInfo{"": { + Type: "", + MaxVolumeCount: 10, + VolumeCount: 1, + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + {Id: 100, Collection: "col1", DiskId: 0, EcIndexBits: all}, + }, + VolumeInfos: []*master_pb.VolumeInformationMessage{{Id: 90000, DiskId: 0}}, + }}, + } + n2 := &master_pb.DataNodeInfo{ + Id: "n2", + DiskInfos: map[string]*master_pb.DiskInfo{"": { + Type: "", + MaxVolumeCount: 7, + VolumeCount: 6, + EcShardInfos: []*master_pb.VolumeEcShardInformationMessage{ + {Id: 200, Collection: "other", DiskId: 0, EcIndexBits: 0x1F}, + {Id: 200, Collection: "other", DiskId: 1, EcIndexBits: 0x1E0}, + }, + VolumeInfos: []*master_pb.VolumeInformationMessage{ + {Id: 90001, DiskId: 0}, {Id: 90002, DiskId: 0}, {Id: 90003, DiskId: 0}, + {Id: 90004, DiskId: 1}, {Id: 90005, DiskId: 1}, {Id: 90006, DiskId: 1}, + }, + }}, + } + topoInfo := &master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{ + {Id: "rack1", DataNodeInfos: []*master_pb.DataNodeInfo{n1}}, + {Id: "rack2", DataNodeInfos: []*master_pb.DataNodeInfo{n2}}, + }, + }}, + } + allowed, err := wildcard.CompileCollectionMatcher("col1") + if err != nil { + t.Fatalf("CompileCollectionMatcher: %v", err) + } + topo, _, _ := buildBalancerTopology(topoInfo, NewDefaultConfig(), allowed) + moves := ecbalancer.Plan(topo, ecbalancer.Options{ImbalanceThreshold: 0.01}) + toN2 := 0 + for _, m := range moves { + if m.VolumeID == 100 && m.TargetNode == "n2" { + toN2++ + } + } + if toN2 != 1 { + t.Errorf("volume 100 moves to n2 = %d, want 1 (the node's one free slot divided across two disks must not floor to zero)", toN2) + } +}