diff --git a/weed/storage/erasure_coding/ecbalancer/balancer.go b/weed/storage/erasure_coding/ecbalancer/balancer.go index 7679dada8..c53b5335e 100644 --- a/weed/storage/erasure_coding/ecbalancer/balancer.go +++ b/weed/storage/erasure_coding/ecbalancer/balancer.go @@ -1120,6 +1120,15 @@ func countShardsByHost(vk volKey, nodes map[string]*Node) map[string]int { return m } +// freeSlotsByHost sums each machine's free EC shard slots. +func freeSlotsByHost(nodes map[string]*Node) map[string]int { + m := make(map[string]int) + for _, node := range nodes { + m[node.host] += node.freeSlots + } + return m +} + func countShardsByNode(vk volKey, nodes map[string]*Node) map[string]int { m := make(map[string]int) for id, node := range nodes { diff --git a/weed/storage/erasure_coding/ecbalancer/balancer_test.go b/weed/storage/erasure_coding/ecbalancer/balancer_test.go index 677c4b193..28c596e6e 100644 --- a/weed/storage/erasure_coding/ecbalancer/balancer_test.go +++ b/weed/storage/erasure_coding/ecbalancer/balancer_test.go @@ -260,6 +260,87 @@ func allBits(n int) erasure_coding.ShardBits { return b } +// TestPlanConvergesToBalancedLoadOverRounds starts skewed -- machine 10.0.0.1 holds +// 4 shards of every volume, the others 2 -- and runs repeated worker-style balance +// passes (capped moves per pass). It must converge to an even per-machine shard load. +func TestPlanConvergesToBalancedLoadOverRounds(t *testing.T) { + topo := NewTopology() + hosts := []string{"10.0.0.1", "10.0.0.2", "10.0.0.3", "10.0.0.4", "10.0.0.5", "10.0.0.6"} + for _, h := range hosts { + for _, p := range []string{":8080", ":8081"} { + n := topo.AddNode(h+p, "dc1", "dc1:rack1", 4000) + n.SetHost(h) + n.AddDisk(0, "", 4000, 0) + } + } + add := func(id string, vid uint32, ids ...int) { + n := topo.nodes[id] + n.AddShards(vid, "c1", 0, bits(ids...)) + n.disks[0].shardCount += len(ids) + n.disks[0].freeSlots -= len(ids) + n.freeSlots -= len(ids) + } + const volumes = 60 + for v := 0; v < volumes; v++ { + vid := uint32(v) + add("10.0.0.1:8080", vid, 0, 1) // machine 10.0.0.1 gets 4 shards/volume + add("10.0.0.1:8081", vid, 2, 3) + add("10.0.0.2:8080", vid, 4) + add("10.0.0.2:8081", vid, 5) + add("10.0.0.3:8080", vid, 6) + add("10.0.0.3:8081", vid, 7) + add("10.0.0.4:8080", vid, 8) + add("10.0.0.4:8081", vid, 9) + add("10.0.0.5:8080", vid, 10) + add("10.0.0.5:8081", vid, 11) + add("10.0.0.6:8080", vid, 12) + add("10.0.0.6:8081", vid, 13) + } + + perMachine := func() map[string]int { + m := map[string]int{} + for _, h := range hosts { + m[h] = 0 // seed every host so a fully drained one still counts + } + for _, n := range topo.nodes { + for _, info := range n.shards { + m[n.host] += info.shardBits.Count() + } + } + return m + } + opts := Options{Ratio: ratio(10, 4), GlobalUtilizationBased: true, GlobalMaxMovesPerRack: 8} + + rounds, converged := 0, false + for ; rounds < 200; rounds++ { + if len(Plan(topo, opts)) == 0 { + converged = true + break + } + } + if !converged { + t.Fatalf("balance did not converge within %d rounds", rounds) + } + + pm := perMachine() + min, max := 1<<30, 0 + for _, c := range pm { + if c < min { + min = c + } + if c > max { + max = c + } + } + t.Logf("converged after %d rounds: %v", rounds, pm) + if rounds < 2 { + t.Errorf("expected convergence to take multiple rounds (capped moves), took %d", rounds) + } + if float64(max) > 1.1*float64(min) { + t.Errorf("did not converge to balanced load: min=%d max=%d %v", min, max, pm) + } +} + func TestGlobalImbalanceMovesFromFullToEmpty(t *testing.T) { topo := NewTopology() n1 := topo.AddNode("node1", "dc1", "dc1:rack1", 5) diff --git a/weed/storage/erasure_coding/ecbalancer/place.go b/weed/storage/erasure_coding/ecbalancer/place.go index c754995e2..84a43c632 100644 --- a/weed/storage/erasure_coding/ecbalancer/place.go +++ b/weed/storage/erasure_coding/ecbalancer/place.go @@ -389,12 +389,15 @@ func rackHasFreeDisk(r *rack, eligible func(*disk) bool) bool { // eligible disk. 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 prefers the one whose machine holds the fewest shards of -// the volume (tie-broken by the node's own count), spreading shards across machines. +// 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. func pickNodeInRackEligible(r *rack, vk volKey, rp *super_block.ReplicaPlacement, eligible func(*disk) bool) *Node { machineShards := countShardsByHost(vk, r.nodes) + machineFree := freeSlotsByHost(r.nodes) var best *Node - bestMachineCount, bestNodeCount := -1, -1 + var bestMCount, bestMFree, bestNCount, bestNFree int for _, id := range sortedNodeKeys(r.nodes) { node := r.nodes[id] if node.freeSlots <= 0 { @@ -407,9 +410,22 @@ func pickNodeInRackEligible(r *rack, vk volKey, rp *super_block.ReplicaPlacement if rp != nil && rp.SameRackCount > 0 && count >= rp.SameRackCount { continue } - mCount := machineShards[node.host] - if best == nil || mCount < bestMachineCount || (mCount == bestMachineCount && count < bestNodeCount) { - best, bestMachineCount, bestNodeCount = node, mCount, count + mCount, mFree := machineShards[node.host], machineFree[node.host] + 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: + better = node.freeSlots > bestNFree + } + if better { + best, bestMCount, bestMFree, bestNCount, bestNFree = node, mCount, mFree, count, node.freeSlots } } return best diff --git a/weed/storage/erasure_coding/ecbalancer/place_test.go b/weed/storage/erasure_coding/ecbalancer/place_test.go index 4b2814270..e6f625a94 100644 --- a/weed/storage/erasure_coding/ecbalancer/place_test.go +++ b/weed/storage/erasure_coding/ecbalancer/place_test.go @@ -73,6 +73,48 @@ func TestPlaceStrictSpreadAndCaps(t *testing.T) { } } +// TestPlaceDistributesEvenlyAcrossManyVolumes: placing many volumes against the shared +// encode snapshot must not pile shards onto the sorted-first machine. +func TestPlaceDistributesEvenlyAcrossManyVolumes(t *testing.T) { + topo := NewTopology() + hosts := []string{"10.0.0.1", "10.0.0.2", "10.0.0.3", "10.0.0.4", "10.0.0.5", "10.0.0.6"} + for _, h := range hosts { + for _, port := range []string{":8080", ":8081"} { + n := topo.AddNode(h+port, "dc1", "dc1:rack1", 1000) + n.SetHost(h) + n.AddDisk(0, "", 1000, 0) + } + } + + const volumes = 120 + for v := 0; v < volumes; v++ { + if _, err := topo.Place(uint32(v), "c1", allShards(), Constraints{}, PlaceDurabilityFirst); err != nil { + t.Fatalf("Place volume %d: %v", v, err) + } + } + + perMachine := map[string]int{} + for _, n := range topo.nodes { + for _, d := range n.disks { + perMachine[n.host] += d.shardCount + } + } + min, max := 1<<30, 0 + for _, c := range perMachine { + if c < min { + min = c + } + if c > max { + max = c + } + } + // Even distribution is 120*14/6 = 280 shards/machine. Catch a lopsided pileup + // (the production symptom was the first machine at ~1.7x the rest). + if float64(max) > 1.25*float64(min) { + t.Errorf("EC shards piled unevenly across machines: min=%d max=%d %v", min, max, perMachine) + } +} + // TestPlaceSpreadsAcrossMachines: one rack runs two physical machines (each with // four volume servers). Placing a 10+4 volume must spread its shards across both // machines so no single machine holds more than ceil(14/2)=7, otherwise losing one