mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-22 22:11:39 +02:00
ec distribution: implement the documented shard distribution API (#11915)
* ec distribution: parse EC and replication configs into a shard plan NewECConfig/DefaultECConfig and NewReplicationConfig(FromString) turn an EC ratio and a replica placement into ECConfig/ReplicationConfig values, and CalculateDistribution derives the target and maximum shard counts per data center, rack, and node. ECDistribution.Summary and FaultToleranceAnalysis describe, per topology level, whether losing one failure domain leaves enough shards to reconstruct the volume, so the EC durability guarantees of a given setup can be answered directly instead of only documented informally. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: analyze topology and plan shard moves TopologyAnalysis records nodes and shard locations per DC/rack/node, and Rebalancer.PlanRebalance emits the moves that level the layout: data centers first, then racks within each DC, then nodes within each rack, moving parity shards before data shards and honoring per-node free slots. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: cover the plan math and rebalancing with tests Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: keep AddShardLocation and node ShardIDs in sync AddShardLocation updated the count maps but never the node's ShardIDs — yet every rebalancer count reads len(node.ShardIDs), so the documented AddNode-then-AddShardLocation flow produced an empty rebalance plan. It now appends to the node's shard list and skips shards the node already has, making the two registration styles equivalent and idempotent. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: key rack-scoped maps by dc/rack Rack names are only unique within a data center — two DCs can both hold a "rack1" — so ShardsByRack, RackToShards, and RackToNodes keyed by the bare rack name merged distinct domains and corrupted both the DC-level shard counts and rack-level balancing. Rack lookups now use the dc/rack pair; DCToRacks still lists bare rack names. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: never move a shard onto a node already holding it The move picker selected the destination node before the shard, so a node already holding the chosen shard could be targeted. Pick the shard first (parity-first) and choose among destination nodes that lack it. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: tolerate zero-value replication configs A ReplicationConfig{} with zeroed minimums made maxShare divide by zero; clamp each level's unit count to at least one. Covers the new behaviors with tests. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: keep rack keys distinct when names contain "/" rackKey joined data center and rack with "/", so dc "a"/rack "b/c" and dc "a/b"/rack "c" shared one key and were counted as the same failure domain. Separate the parts with a NUL byte, which cannot appear in a name. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: plan rebalances on a copy of the topology recordMove mutated the caller's TopologyAnalysis while planning, so an interrupted or partial execution left the input describing moves that never happened, and replanning on it returned nothing. PlanRebalance now simulates on a cloned analysis; plan.Analysis reports the layout the moves would produce and the caller's input is left untouched. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: balance nodes against the rack's current shard count The node-level target still read the original analysis, so a rack that had just sent or received shards during DC/rack balancing aimed at a stale total and left its nodes unbalanced. * ec distribution: register each node id once Adding the same node twice appended it to RackToNodes twice, so every count that walks rack nodes double-counted its shards. * ec distribution: normalize replication counts in the stored plan A zero-value ReplicationConfig computed targets against one domain per level while Summary still reported zero DCs, racks, and nodes. * ec distribution: cap a survivable level's max at the loss budget * ec distribution: skip shards already inside the destination domain * ec distribution: register nodes implied by shard locations * ec distribution: merge repeat node registrations A shard location now auto-registers an unknown node with no free slots; a later AddNode carrying the real capacity was then dropped entirely. Repeat registration now updates the node's capacity and adopts any new shards, while the first registration keeps the topology placement. * ec distribution: try other overloaded sources before stopping When the most overloaded domain could not shed any shard, balance returned early and skipped every other source that still had movable shards. A source that cannot move now gets set aside for the rest of the pass. * ec distribution: collapse chained shard moves to the final destination Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * ec distribution: schedule collapsed moves behind the departures that free slots Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Chris LuandDevin authored and GitHub committed
1 parent
e645449872
commit
137f23a229
6 files changed
+1088
-3
No files matched your search
@@ -30,8 +30,8 @@ fmt.Println(dist.Summary())
|
||||
// Replication: replication=110 (DCs:2, Racks/DC:2, Nodes/Rack:1)
|
||||
// Distribution Plan:
|
||||
// Data Centers: 2 (target 7 shards each, max 9)
|
||||
// Racks per DC: 2 (target 4 shards each, max 6)
|
||||
// Nodes per Rack: 1 (target 4 shards each, max 6)
|
||||
// Racks per DC: 2 (target 4 shards each, max 4)
|
||||
// Nodes per Rack: 1 (target 4 shards each, max 4)
|
||||
```
|
||||
|
||||
### Custom EC Ratios (seaweed-enterprise)
|
||||
|
||||
@@ -1,5 +1,103 @@
|
||||
package distribution
|
||||
|
||||
// NewTopologyAnalysis creates an empty topology analysis.
|
||||
func NewTopologyAnalysis() *TopologyAnalysis {
|
||||
return &TopologyAnalysis{
|
||||
ShardsByDC: make(map[string]int),
|
||||
ShardsByRack: make(map[string]int),
|
||||
ShardsByNode: make(map[string]int),
|
||||
DCToShards: make(map[string][]int),
|
||||
RackToShards: make(map[string][]int),
|
||||
NodeToShards: make(map[string][]int),
|
||||
DCToRacks: make(map[string][]string),
|
||||
RackToNodes: make(map[string][]*TopologyNode),
|
||||
AllNodes: make(map[string]*TopologyNode),
|
||||
}
|
||||
}
|
||||
|
||||
// AddNode registers a node in the topology, including any shards it already
|
||||
// holds for the volume being analyzed. A node id may only register once;
|
||||
// registering it again updates its capacity and adds any new shards. The
|
||||
// first registration wins for the node's place in the topology.
|
||||
func (a *TopologyAnalysis) AddNode(node *TopologyNode) {
|
||||
registered, ok := a.AllNodes[node.NodeID]
|
||||
if !ok {
|
||||
registered = node
|
||||
a.AllNodes[node.NodeID] = node
|
||||
a.RackToNodes[rackKey(node.DataCenter, node.Rack)] = append(a.RackToNodes[rackKey(node.DataCenter, node.Rack)], node)
|
||||
if !contains(a.DCToRacks[node.DataCenter], node.Rack) {
|
||||
a.DCToRacks[node.DataCenter] = append(a.DCToRacks[node.DataCenter], node.Rack)
|
||||
}
|
||||
} else {
|
||||
registered.FreeSlots = node.FreeSlots
|
||||
registered.TotalShards = node.TotalShards
|
||||
}
|
||||
seen := node.ShardIDs[:0]
|
||||
for _, shardID := range node.ShardIDs {
|
||||
if !contains(seen, shardID) {
|
||||
seen = append(seen, shardID)
|
||||
}
|
||||
}
|
||||
node.ShardIDs = seen
|
||||
for _, shardID := range node.ShardIDs {
|
||||
a.AddShardLocation(ShardLocation{
|
||||
ShardID: shardID,
|
||||
NodeID: registered.NodeID,
|
||||
DataCenter: registered.DataCenter,
|
||||
Rack: registered.Rack,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// AddShardLocation records one shard's placement in the topology. Repeated
|
||||
// calls for a shard already known to its node are ignored, so callers may
|
||||
// either set ShardIDs on the TopologyNode or register each shard here. A
|
||||
// location naming an unregistered node registers that node so the rebalancer
|
||||
// can see the shard; for a registered node the node's own dc/rack win.
|
||||
func (a *TopologyAnalysis) AddShardLocation(loc ShardLocation) {
|
||||
node, ok := a.AllNodes[loc.NodeID]
|
||||
if !ok {
|
||||
node = &TopologyNode{NodeID: loc.NodeID, DataCenter: loc.DataCenter, Rack: loc.Rack}
|
||||
a.AddNode(node)
|
||||
}
|
||||
if contains(a.NodeToShards[loc.NodeID], loc.ShardID) {
|
||||
return
|
||||
}
|
||||
a.ShardsByDC[node.DataCenter]++
|
||||
a.ShardsByRack[rackKey(node.DataCenter, node.Rack)]++
|
||||
a.ShardsByNode[loc.NodeID]++
|
||||
a.DCToShards[node.DataCenter] = append(a.DCToShards[node.DataCenter], loc.ShardID)
|
||||
a.RackToShards[rackKey(node.DataCenter, node.Rack)] = append(a.RackToShards[rackKey(node.DataCenter, node.Rack)], loc.ShardID)
|
||||
a.NodeToShards[loc.NodeID] = append(a.NodeToShards[loc.NodeID], loc.ShardID)
|
||||
a.TotalShards++
|
||||
if !contains(node.ShardIDs, loc.ShardID) {
|
||||
node.ShardIDs = append(node.ShardIDs, loc.ShardID)
|
||||
}
|
||||
}
|
||||
|
||||
// rackKey qualifies a rack name with its data center: rack names are only
|
||||
// unique within a DC, so two DCs can both hold a "rack1". NUL separates the
|
||||
// parts so a "/" inside either name cannot merge distinct racks.
|
||||
func rackKey(dc, rack string) string {
|
||||
return dc + "\x00" + rack
|
||||
}
|
||||
|
||||
// Finalize computes topology statistics after all nodes and shards are added.
|
||||
func (a *TopologyAnalysis) Finalize() {
|
||||
a.TotalNodes = len(a.AllNodes)
|
||||
a.TotalRacks = len(a.RackToNodes)
|
||||
a.TotalDCs = len(a.DCToRacks)
|
||||
}
|
||||
|
||||
func contains[T comparable](list []T, s T) bool {
|
||||
for _, item := range list {
|
||||
if item == s {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// ShardLocation represents where a shard is located in the topology
|
||||
type ShardLocation struct {
|
||||
ShardID int
|
||||
|
||||
@@ -1,6 +1,14 @@
|
||||
// Package distribution provides EC shard distribution algorithms with configurable EC ratios.
|
||||
package distribution
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
||||
)
|
||||
|
||||
// ECConfig holds erasure coding configuration parameters.
|
||||
// This replaces hard-coded constants like DataShardsCount=10, ParityShardsCount=4.
|
||||
type ECConfig struct {
|
||||
@@ -17,3 +25,76 @@ type ReplicationConfig struct {
|
||||
// Original replication string (for logging/debugging)
|
||||
Original string
|
||||
}
|
||||
|
||||
// NewECConfig creates an erasure coding configuration with the given shard counts.
|
||||
func NewECConfig(dataShards, parityShards int) (ECConfig, error) {
|
||||
if dataShards < 1 || parityShards < 1 {
|
||||
return ECConfig{}, fmt.Errorf("data and parity shard counts must be positive, got %d+%d", dataShards, parityShards)
|
||||
}
|
||||
return ECConfig{DataShards: dataShards, ParityShards: parityShards}, nil
|
||||
}
|
||||
|
||||
// DefaultECConfig returns the standard 10+4 erasure coding configuration.
|
||||
func DefaultECConfig() ECConfig {
|
||||
return ECConfig{DataShards: erasure_coding.DataShardsCount, ParityShards: erasure_coding.ParityShardsCount}
|
||||
}
|
||||
|
||||
// NewReplicationConfig parses a replication policy string like "110".
|
||||
func NewReplicationConfigFromString(s string) (ReplicationConfig, error) {
|
||||
rp, err := super_block.NewReplicaPlacementFromString(s)
|
||||
if err != nil {
|
||||
return ReplicationConfig{}, err
|
||||
}
|
||||
return NewReplicationConfig(rp), nil
|
||||
}
|
||||
|
||||
// NewReplicationConfig converts a replica placement into a replication config.
|
||||
func NewReplicationConfig(rp *super_block.ReplicaPlacement) ReplicationConfig {
|
||||
if rp == nil {
|
||||
rp = &super_block.ReplicaPlacement{}
|
||||
}
|
||||
return ReplicationConfig{
|
||||
MinDataCenters: rp.DiffDataCenterCount + 1,
|
||||
MinRacksPerDC: rp.DiffRackCount + 1,
|
||||
MinNodesPerRack: rp.SameRackCount + 1,
|
||||
Original: rp.String(),
|
||||
}
|
||||
}
|
||||
|
||||
// TotalShards returns the number of shards each volume splits into.
|
||||
func (ec ECConfig) TotalShards() int {
|
||||
return ec.DataShards + ec.ParityShards
|
||||
}
|
||||
|
||||
// CanLoseShards returns how many shards of a volume can be missing before the
|
||||
// volume is no longer readable.
|
||||
func (ec ECConfig) CanLoseShards() int {
|
||||
return ec.ParityShards
|
||||
}
|
||||
|
||||
// IsDataShard reports whether the shard id is a data shard, which serves reads.
|
||||
func (ec ECConfig) IsDataShard(shardID int) bool {
|
||||
return shardID >= 0 && shardID < ec.DataShards
|
||||
}
|
||||
|
||||
// IsParityShard reports whether the shard id is a parity shard, which is only
|
||||
// used for reconstruction.
|
||||
func (ec ECConfig) IsParityShard(shardID int) bool {
|
||||
return shardID >= ec.DataShards && shardID < ec.TotalShards()
|
||||
}
|
||||
|
||||
// SortShardsDataFirst orders shard ids so data shards come before parity
|
||||
// shards, for initial placement.
|
||||
func (ec ECConfig) SortShardsDataFirst(shards []int) []int {
|
||||
sorted := append([]int(nil), shards...)
|
||||
sort.Ints(sorted)
|
||||
return sorted
|
||||
}
|
||||
|
||||
// SortShardsParityFirst orders shard ids so parity shards come before data
|
||||
// shards, for picking which shards to move away first when rebalancing.
|
||||
func (ec ECConfig) SortShardsParityFirst(shards []int) []int {
|
||||
sorted := append([]int(nil), shards...)
|
||||
sort.Sort(sort.Reverse(sort.IntSlice(sorted)))
|
||||
return sorted
|
||||
}
|
||||
@@ -1,5 +1,10 @@
|
||||
package distribution
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// ECDistribution represents the target distribution of EC shards
|
||||
// based on EC configuration and replication policy.
|
||||
type ECDistribution struct {
|
||||
@@ -20,3 +25,97 @@ type ECDistribution struct {
|
||||
MaxShardsPerRack int
|
||||
MaxShardsPerNode int
|
||||
}
|
||||
|
||||
// CalculateDistribution computes the balanced shard counts each topology level
|
||||
// should hold for the given EC ratio spread over the minimum topology the
|
||||
// replication policy describes.
|
||||
func CalculateDistribution(ec ECConfig, rep ReplicationConfig) *ECDistribution {
|
||||
total := ec.TotalShards()
|
||||
rep.MinDataCenters = max(rep.MinDataCenters, 1)
|
||||
rep.MinRacksPerDC = max(rep.MinRacksPerDC, 1)
|
||||
rep.MinNodesPerRack = max(rep.MinNodesPerRack, 1)
|
||||
dcUnits := rep.MinDataCenters
|
||||
rackUnits := dcUnits * rep.MinRacksPerDC
|
||||
nodeUnits := rackUnits * rep.MinNodesPerRack
|
||||
return &ECDistribution{
|
||||
ECConfig: ec,
|
||||
ReplicationConfig: rep,
|
||||
TargetShardsPerDC: evenShare(total, dcUnits),
|
||||
TargetShardsPerRack: evenShare(total, rackUnits),
|
||||
TargetShardsPerNode: evenShare(total, nodeUnits),
|
||||
MaxShardsPerDC: maxShare(total, dcUnits, ec.CanLoseShards()),
|
||||
MaxShardsPerRack: maxShare(total, rackUnits, ec.CanLoseShards()),
|
||||
MaxShardsPerNode: maxShare(total, nodeUnits, ec.CanLoseShards()),
|
||||
}
|
||||
}
|
||||
|
||||
// evenShare is the balanced shard count per unit at one topology level.
|
||||
func evenShare(total, units int) int {
|
||||
return ceilDivide(total, units)
|
||||
}
|
||||
|
||||
// maxShare bounds how many shards a single failure domain may hold: the even
|
||||
// share plus the slack that would pad the total to the next multiple of the
|
||||
// unit count. When the even share is survivable on its own, the loss budget
|
||||
// caps the bound so the fault tolerance report stays true even at the max.
|
||||
func maxShare(total, units, lossBudget int) int {
|
||||
if units < 1 {
|
||||
units = 1
|
||||
}
|
||||
m := ceilDivide(total, units) + units - total%units
|
||||
if m > total {
|
||||
m = total
|
||||
}
|
||||
if evenShare(total, units) <= lossBudget && m > lossBudget {
|
||||
m = lossBudget
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
func ceilDivide(a, b int) int {
|
||||
if b < 1 {
|
||||
return a
|
||||
}
|
||||
return (a + b - 1) / b
|
||||
}
|
||||
|
||||
// Summary describes the distribution plan in human-readable form.
|
||||
func (d *ECDistribution) Summary() string {
|
||||
var sb strings.Builder
|
||||
ec := d.ECConfig
|
||||
rep := d.ReplicationConfig
|
||||
fmt.Fprintf(&sb, "EC Configuration: %d+%d (total: %d, can lose: %d)\n",
|
||||
ec.DataShards, ec.ParityShards, ec.TotalShards(), ec.CanLoseShards())
|
||||
fmt.Fprintf(&sb, "Replication: replication=%s (DCs:%d, Racks/DC:%d, Nodes/Rack:%d)\n",
|
||||
rep.Original, rep.MinDataCenters, rep.MinRacksPerDC, rep.MinNodesPerRack)
|
||||
sb.WriteString("Distribution Plan:\n")
|
||||
fmt.Fprintf(&sb, " Data Centers: %d (target %d shards each, max %d)\n",
|
||||
rep.MinDataCenters, d.TargetShardsPerDC, d.MaxShardsPerDC)
|
||||
fmt.Fprintf(&sb, " Racks per DC: %d (target %d shards each, max %d)\n",
|
||||
rep.MinRacksPerDC, d.TargetShardsPerRack, d.MaxShardsPerRack)
|
||||
fmt.Fprintf(&sb, " Nodes per Rack: %d (target %d shards each, max %d)\n",
|
||||
rep.MinNodesPerRack, d.TargetShardsPerNode, d.MaxShardsPerNode)
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
// FaultToleranceAnalysis reports whether losing one unit at each topology
|
||||
// level still leaves enough shards to reconstruct the volume.
|
||||
func (d *ECDistribution) FaultToleranceAnalysis() string {
|
||||
var sb strings.Builder
|
||||
ec := d.ECConfig
|
||||
fmt.Fprintf(&sb, "Fault Tolerance Analysis for %d+%d:\n", ec.DataShards, ec.ParityShards)
|
||||
writeLevel := func(name string, target int) {
|
||||
remaining := ec.TotalShards() - target
|
||||
status := "SURVIVABLE ✓"
|
||||
if remaining < ec.DataShards {
|
||||
status = "NOT SURVIVABLE ✗"
|
||||
}
|
||||
fmt.Fprintf(&sb, " %s Failure: %s\n", name, status)
|
||||
fmt.Fprintf(&sb, " - Losing one %s loses ~%d shards\n", name, target)
|
||||
fmt.Fprintf(&sb, " - Remaining: %d shards (need %d)\n", remaining, ec.DataShards)
|
||||
}
|
||||
writeLevel("DC", d.TargetShardsPerDC)
|
||||
writeLevel("Rack", d.TargetShardsPerRack)
|
||||
writeLevel("Node", d.TargetShardsPerNode)
|
||||
return sb.String()
|
||||
}
|
||||
@@ -0,0 +1,500 @@
|
||||
package distribution
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestDefaultECConfig(t *testing.T) {
|
||||
ec := DefaultECConfig()
|
||||
assert.Equal(t, 10, ec.DataShards)
|
||||
assert.Equal(t, 4, ec.ParityShards)
|
||||
assert.Equal(t, 14, ec.TotalShards())
|
||||
assert.Equal(t, 4, ec.CanLoseShards())
|
||||
assert.True(t, ec.IsDataShard(0))
|
||||
assert.True(t, ec.IsDataShard(9))
|
||||
assert.False(t, ec.IsDataShard(10))
|
||||
assert.True(t, ec.IsParityShard(13))
|
||||
assert.False(t, ec.IsParityShard(14))
|
||||
assert.Equal(t, []int{0, 1, 10, 11}, ec.SortShardsDataFirst([]int{11, 1, 10, 0}))
|
||||
assert.Equal(t, []int{13, 12, 2, 1}, ec.SortShardsParityFirst([]int{2, 13, 1, 12}))
|
||||
}
|
||||
|
||||
func TestNewECConfig(t *testing.T) {
|
||||
ec, err := NewECConfig(8, 4)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 12, ec.TotalShards())
|
||||
|
||||
for _, bad := range [][2]int{{0, 4}, {8, 0}, {-1, 4}} {
|
||||
_, err := NewECConfig(bad[0], bad[1])
|
||||
assert.Error(t, err, "config %v must be rejected", bad)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewReplicationConfigFromString(t *testing.T) {
|
||||
rep, err := NewReplicationConfigFromString("110")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 2, rep.MinDataCenters)
|
||||
assert.Equal(t, 2, rep.MinRacksPerDC)
|
||||
assert.Equal(t, 1, rep.MinNodesPerRack)
|
||||
assert.Equal(t, "110", rep.Original)
|
||||
|
||||
rep, err = NewReplicationConfigFromString("000")
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, rep.MinDataCenters)
|
||||
assert.Equal(t, 1, rep.MinRacksPerDC)
|
||||
assert.Equal(t, 1, rep.MinNodesPerRack)
|
||||
|
||||
_, err = NewReplicationConfigFromString("abc")
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func TestCalculateDistribution(t *testing.T) {
|
||||
rep, err := NewReplicationConfigFromString("110")
|
||||
require.NoError(t, err)
|
||||
dist := CalculateDistribution(DefaultECConfig(), rep)
|
||||
|
||||
assert.Equal(t, 7, dist.TargetShardsPerDC)
|
||||
assert.Equal(t, 4, dist.TargetShardsPerRack)
|
||||
assert.Equal(t, 4, dist.TargetShardsPerNode)
|
||||
assert.Equal(t, 9, dist.MaxShardsPerDC)
|
||||
assert.Equal(t, 4, dist.MaxShardsPerRack)
|
||||
assert.Equal(t, 4, dist.MaxShardsPerNode)
|
||||
|
||||
expected := "EC Configuration: 10+4 (total: 14, can lose: 4)\n" +
|
||||
"Replication: replication=110 (DCs:2, Racks/DC:2, Nodes/Rack:1)\n" +
|
||||
"Distribution Plan:\n" +
|
||||
" Data Centers: 2 (target 7 shards each, max 9)\n" +
|
||||
" Racks per DC: 2 (target 4 shards each, max 4)\n" +
|
||||
" Nodes per Rack: 1 (target 4 shards each, max 4)\n"
|
||||
assert.Equal(t, expected, dist.Summary())
|
||||
}
|
||||
|
||||
// The reported max may not exceed the loss budget at a survivable level:
|
||||
// 8+4 over "200" lets a domain drift to 7 of 12 shards, but losing that domain
|
||||
// would leave 5 < 8 — capping the max at the budget keeps SURVIVABLE true.
|
||||
func TestCalculateDistributionMaxWithinLossBudget(t *testing.T) {
|
||||
ec, err := NewECConfig(8, 4)
|
||||
require.NoError(t, err)
|
||||
rep, err := NewReplicationConfigFromString("200")
|
||||
require.NoError(t, err)
|
||||
dist := CalculateDistribution(ec, rep)
|
||||
|
||||
assert.Equal(t, 4, dist.TargetShardsPerDC)
|
||||
assert.Equal(t, 4, dist.MaxShardsPerDC)
|
||||
assert.Equal(t, 4, dist.MaxShardsPerRack)
|
||||
assert.Equal(t, 4, dist.MaxShardsPerNode)
|
||||
|
||||
// a level whose even share already exceeds the budget keeps the slack bound
|
||||
dist = CalculateDistribution(DefaultECConfig(), rep)
|
||||
assert.Equal(t, 6, dist.MaxShardsPerDC) // even 5 > 4: DC loss is not survivable anyway
|
||||
}
|
||||
|
||||
func TestCalculateDistributionSingleDomain(t *testing.T) {
|
||||
rep, err := NewReplicationConfigFromString("000")
|
||||
require.NoError(t, err)
|
||||
dist := CalculateDistribution(DefaultECConfig(), rep)
|
||||
assert.Equal(t, 14, dist.TargetShardsPerDC)
|
||||
assert.Equal(t, 14, dist.MaxShardsPerDC)
|
||||
}
|
||||
|
||||
func TestFaultToleranceAnalysis(t *testing.T) {
|
||||
ec, err := NewECConfig(8, 4)
|
||||
require.NoError(t, err)
|
||||
rep, err := NewReplicationConfigFromString("200")
|
||||
require.NoError(t, err)
|
||||
dist := CalculateDistribution(ec, rep)
|
||||
|
||||
analysis := dist.FaultToleranceAnalysis()
|
||||
assert.Contains(t, analysis, "Fault Tolerance Analysis for 8+4:")
|
||||
assert.Contains(t, analysis, "DC Failure: SURVIVABLE")
|
||||
assert.Contains(t, analysis, "- Losing one DC loses ~4 shards")
|
||||
assert.Contains(t, analysis, "- Remaining: 8 shards (need 8)")
|
||||
}
|
||||
|
||||
func TestFaultToleranceAnalysisNotSurvivable(t *testing.T) {
|
||||
rep, err := NewReplicationConfigFromString("000")
|
||||
require.NoError(t, err)
|
||||
dist := CalculateDistribution(DefaultECConfig(), rep)
|
||||
analysis := dist.FaultToleranceAnalysis()
|
||||
assert.Contains(t, analysis, "DC Failure: NOT SURVIVABLE")
|
||||
assert.Contains(t, analysis, "Rack Failure: NOT SURVIVABLE")
|
||||
assert.Contains(t, analysis, "Node Failure: NOT SURVIVABLE")
|
||||
}
|
||||
|
||||
func makeNode(id, dc, rack string, freeSlots int, shards ...int) *TopologyNode {
|
||||
return &TopologyNode{NodeID: id, DataCenter: dc, Rack: rack, FreeSlots: freeSlots, ShardIDs: shards}
|
||||
}
|
||||
|
||||
func TestTopologyAnalysis(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 5, 0, 1))
|
||||
a.AddNode(makeNode("n2", "dc1", "r1", 5, 2))
|
||||
a.AddNode(makeNode("n3", "dc2", "r2", 5))
|
||||
a.Finalize()
|
||||
|
||||
assert.Equal(t, 3, a.TotalShards)
|
||||
assert.Equal(t, 3, a.TotalNodes)
|
||||
assert.Equal(t, 2, a.TotalRacks)
|
||||
assert.Equal(t, 2, a.TotalDCs)
|
||||
assert.Equal(t, 3, a.ShardsByDC["dc1"])
|
||||
assert.Equal(t, 0, a.ShardsByDC["dc2"])
|
||||
assert.Equal(t, []string{"r1"}, a.DCToRacks["dc1"])
|
||||
assert.Equal(t, []int{0, 1, 2}, a.RackToShards[rackKey("dc1", "r1")])
|
||||
}
|
||||
|
||||
// A node registered twice must still count once in rack and shard totals.
|
||||
func TestTopologyAnalysisDuplicateNode(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 5, 0, 1))
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 5, 0, 1))
|
||||
a.Finalize()
|
||||
|
||||
assert.Equal(t, 1, a.TotalNodes)
|
||||
assert.Len(t, a.RackToNodes[rackKey("dc1", "r1")], 1)
|
||||
assert.Equal(t, 2, a.ShardsByNode["n1"])
|
||||
}
|
||||
|
||||
// A node auto-registered by a shard location gets its real capacity when the
|
||||
// node itself registers later, plus any shards the location did not mention.
|
||||
func TestTopologyAnalysisNodeCapacityLateRegistration(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddShardLocation(ShardLocation{ShardID: 0, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 10, 1))
|
||||
a.Finalize()
|
||||
|
||||
assert.Equal(t, 10, a.AllNodes["n1"].FreeSlots)
|
||||
assert.Equal(t, []int{0, 1}, a.AllNodes["n1"].ShardIDs)
|
||||
assert.Equal(t, 2, a.ShardsByNode["n1"])
|
||||
assert.Len(t, a.RackToNodes[rackKey("dc1", "r1")], 1)
|
||||
}
|
||||
|
||||
// Shards may be registered either through TopologyNode.ShardIDs or through
|
||||
// AddShardLocation — both must leave the node holding the shards.
|
||||
func TestTopologyAnalysisAddShardLocation(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 5))
|
||||
a.AddShardLocation(ShardLocation{ShardID: 0, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
|
||||
a.AddShardLocation(ShardLocation{ShardID: 1, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
|
||||
a.AddShardLocation(ShardLocation{ShardID: 1, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
|
||||
a.Finalize()
|
||||
|
||||
assert.Equal(t, 2, a.TotalShards)
|
||||
assert.Equal(t, 2, a.ShardsByNode["n1"])
|
||||
assert.Equal(t, []int{0, 1}, a.AllNodes["n1"].ShardIDs)
|
||||
}
|
||||
|
||||
// Rack names are only unique within a DC, so dc1/rack1 and dc2/rack1 are
|
||||
// distinct domains.
|
||||
func TestTopologyAnalysisSameRackNameAcrossDCs(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "rack1", 5, 0, 1))
|
||||
a.AddNode(makeNode("n2", "dc2", "rack1", 5, 2))
|
||||
a.Finalize()
|
||||
|
||||
assert.Equal(t, 2, a.TotalDCs)
|
||||
assert.Equal(t, 2, a.TotalRacks)
|
||||
assert.Equal(t, 2, a.ShardsByRack[rackKey("dc1", "rack1")])
|
||||
assert.Equal(t, 1, a.ShardsByRack[rackKey("dc2", "rack1")])
|
||||
}
|
||||
|
||||
// A "/" inside a data-center or rack name must not merge distinct racks.
|
||||
func TestTopologyAnalysisSlashInNames(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "a", "b/c", 5, 0))
|
||||
a.AddNode(makeNode("n2", "a/b", "c", 5, 1))
|
||||
a.Finalize()
|
||||
|
||||
assert.Equal(t, 2, a.TotalRacks)
|
||||
assert.Equal(t, 1, a.ShardsByRack[rackKey("a", "b/c")])
|
||||
assert.Equal(t, 1, a.ShardsByRack[rackKey("a/b", "c")])
|
||||
}
|
||||
|
||||
func TestPlanRebalanceAcrossDCs(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("a1", "dcA", "rA1", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("a2", "dcA", "rA2", 10, 4, 5, 6, 7))
|
||||
a.AddNode(makeNode("a3", "dcA", "rA3", 10, 8, 9, 10, 11))
|
||||
a.AddNode(makeNode("a4", "dcA", "rA4", 10, 12, 13))
|
||||
a.AddNode(makeNode("b1", "dcB", "rB1", 10))
|
||||
a.AddNode(makeNode("b2", "dcB", "rB2", 10))
|
||||
a.Finalize()
|
||||
|
||||
rep, err := NewReplicationConfigFromString("110")
|
||||
require.NoError(t, err)
|
||||
plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, 7, plan.MovesAcrossDC)
|
||||
assert.Equal(t, 7, plan.Analysis.ShardsByDC["dcA"])
|
||||
assert.Equal(t, 7, plan.Analysis.ShardsByDC["dcB"])
|
||||
|
||||
// the given analysis still describes the un-moved layout
|
||||
assert.Equal(t, 14, a.ShardsByDC["dcA"])
|
||||
assert.Equal(t, 0, a.ShardsByDC["dcB"])
|
||||
assert.Equal(t, []int{0, 1, 2, 3}, a.AllNodes["a1"].ShardIDs)
|
||||
|
||||
// parity shards (10-13) move before data shards
|
||||
assert.True(t, plan.Moves[0].ShardID >= 10)
|
||||
assert.Equal(t, "b1", plan.Moves[0].DestNode.NodeID)
|
||||
}
|
||||
|
||||
func TestPlanRebalanceAcrossRacks(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("n2", "dc1", "r2", 10))
|
||||
a.Finalize()
|
||||
|
||||
rep, err := NewReplicationConfigFromString("110")
|
||||
require.NoError(t, err)
|
||||
plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, 2, plan.TotalMoves)
|
||||
assert.Equal(t, 2, plan.MovesAcrossRack)
|
||||
assert.Equal(t, 2, plan.Analysis.ShardsByRack[rackKey("dc1", "r1")])
|
||||
assert.Equal(t, 2, plan.Analysis.ShardsByRack[rackKey("dc1", "r2")])
|
||||
// parity-first ordering: shards 3 then 2 move out of r1
|
||||
assert.Equal(t, 3, plan.Moves[0].ShardID)
|
||||
assert.Equal(t, 2, plan.Moves[1].ShardID)
|
||||
}
|
||||
|
||||
func TestPlanRebalanceWithinRack(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("n2", "dc1", "r1", 10))
|
||||
a.Finalize()
|
||||
|
||||
rep, err := NewReplicationConfigFromString("001")
|
||||
require.NoError(t, err)
|
||||
plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, 2, plan.TotalMoves)
|
||||
assert.Equal(t, 2, plan.MovesWithinRack)
|
||||
assert.Equal(t, 2, plan.Analysis.ShardsByNode["n1"])
|
||||
assert.Equal(t, 2, plan.Analysis.ShardsByNode["n2"])
|
||||
assert.Equal(t, []int{3, 2}, plan.Analysis.AllNodes["n2"].ShardIDs)
|
||||
assert.Equal(t, []int{0, 1, 2, 3}, a.AllNodes["n1"].ShardIDs)
|
||||
}
|
||||
|
||||
func TestPlanRebalanceRespectsFreeSlots(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("n2", "dc1", "r1", 1))
|
||||
a.Finalize()
|
||||
|
||||
plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 1, MinRacksPerDC: 1, MinNodesPerRack: 2}).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, 1, plan.TotalMoves, "only one slot is free on the destination")
|
||||
assert.Equal(t, 3, plan.Analysis.ShardsByNode["n1"])
|
||||
assert.Equal(t, 1, plan.Analysis.ShardsByNode["n2"])
|
||||
assert.Equal(t, 0, plan.Analysis.AllNodes["n2"].FreeSlots)
|
||||
assert.Equal(t, 1, a.AllNodes["n2"].FreeSlots)
|
||||
}
|
||||
|
||||
func TestPlanRebalanceNoopWhenBalanced(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1))
|
||||
a.AddNode(makeNode("n2", "dc1", "r1", 10, 2, 3))
|
||||
a.Finalize()
|
||||
|
||||
plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 1, MinRacksPerDC: 1, MinNodesPerRack: 2}).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 0, plan.TotalMoves)
|
||||
}
|
||||
|
||||
func TestPlanRebalanceNilAnalysis(t *testing.T) {
|
||||
_, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{}).PlanRebalance(nil)
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
// Node balancing inside a rack must aim for the rack's count after
|
||||
// DC- and rack-level moves, not the count the rack started with.
|
||||
func TestPlanRebalanceNodeTargetAfterMoves(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13))
|
||||
a.AddNode(makeNode("n2", "dc1", "r1", 10))
|
||||
a.AddNode(makeNode("n3", "dc1", "r2", 10))
|
||||
a.Finalize()
|
||||
|
||||
rep, err := NewReplicationConfigFromString("010")
|
||||
require.NoError(t, err)
|
||||
plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, 7, plan.Analysis.ShardsByRack[rackKey("dc1", "r1")])
|
||||
assert.Equal(t, 7, plan.Analysis.ShardsByRack[rackKey("dc1", "r2")])
|
||||
assert.Equal(t, 4, plan.Analysis.ShardsByNode["n1"])
|
||||
assert.Equal(t, 3, plan.Analysis.ShardsByNode["n2"])
|
||||
assert.Equal(t, []int{0, 1, 2, 3}, plan.Analysis.AllNodes["n1"].ShardIDs)
|
||||
assert.Equal(t, []int{6, 5, 4}, plan.Analysis.AllNodes["n2"].ShardIDs)
|
||||
}
|
||||
|
||||
// A shard that already has a copy anywhere in the destination domain must not
|
||||
// be moved there again — the second copy adds no redundancy.
|
||||
func TestPlanRebalanceNoDuplicateShardInDomain(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("a1", "dcA", "rA1", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("b1", "dcB", "rB1", 10, 3))
|
||||
a.AddNode(makeNode("b2", "dcB", "rB2", 10))
|
||||
a.Finalize()
|
||||
|
||||
rep, err := NewReplicationConfigFromString("100")
|
||||
require.NoError(t, err)
|
||||
plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
// dcA holds 4, dcB holds 1, target 3 each: one move, and it cannot be shard 3
|
||||
require.Len(t, plan.Moves, 1)
|
||||
assert.Equal(t, 2, plan.Moves[0].ShardID)
|
||||
assert.Equal(t, []int{2}, plan.Analysis.AllNodes["b2"].ShardIDs)
|
||||
assert.Equal(t, []int{3}, plan.Analysis.AllNodes["b1"].ShardIDs)
|
||||
}
|
||||
|
||||
// Shard locations may be registered without registering the node first; the
|
||||
// rebalancer must still see the shard, counted under the node's own dc/rack.
|
||||
func TestPlanRebalanceShardForUnregisteredNode(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddShardLocation(ShardLocation{ShardID: 0, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
|
||||
a.AddShardLocation(ShardLocation{ShardID: 1, NodeID: "n1", DataCenter: "dc1", Rack: "r1"})
|
||||
a.AddNode(makeNode("n2", "dc1", "r1", 10))
|
||||
a.Finalize()
|
||||
|
||||
assert.Equal(t, 1, a.TotalDCs)
|
||||
assert.Equal(t, 2, a.ShardsByNode["n1"])
|
||||
assert.Equal(t, []int{0, 1}, a.AllNodes["n1"].ShardIDs)
|
||||
|
||||
plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 1, MinRacksPerDC: 1, MinNodesPerRack: 2}).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, 1, plan.TotalMoves)
|
||||
assert.Equal(t, 1, plan.Analysis.ShardsByNode["n1"])
|
||||
assert.Equal(t, 1, plan.Analysis.ShardsByNode["n2"])
|
||||
}
|
||||
|
||||
// When the most overloaded domain can only offer duplicates, balancing must
|
||||
// fall through to the next overloaded source rather than stop.
|
||||
func TestPlanRebalanceSkipsBlockedSource(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("a1", "dcA", "rA", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("a2", "dcA", "rA", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("a3", "dcA", "rA", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("b1", "dcB", "rB", 10, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13))
|
||||
a.AddNode(makeNode("c1", "dcC", "rC", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("c2", "dcC", "rC", 10))
|
||||
a.AddNode(makeNode("d1", "dcD", "rD", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("d2", "dcD", "rD", 10))
|
||||
a.Finalize()
|
||||
|
||||
plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 4, MinRacksPerDC: 1, MinNodesPerRack: 1}).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
// dcA is stuck on duplicates (target 8), but dcB still sheds two shards
|
||||
assert.Equal(t, 12, plan.Analysis.ShardsByDC["dcA"])
|
||||
assert.Equal(t, 8, plan.Analysis.ShardsByDC["dcB"])
|
||||
assert.Equal(t, 5, plan.Analysis.ShardsByDC["dcC"])
|
||||
assert.Equal(t, 5, plan.Analysis.ShardsByDC["dcD"])
|
||||
}
|
||||
|
||||
// A shard that dc-level balancing lands on a node can be shed again at rack or
|
||||
// node level; its moves must collapse to one transfer to the final holder.
|
||||
func TestPlanRebalanceCollapsesChainedMoves(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("a1", "dcA", "rA", 20, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13))
|
||||
a.AddNode(makeNode("b1", "dcB", "rB1", 20))
|
||||
a.AddNode(makeNode("c1", "dcB", "rB2", 20))
|
||||
a.AddNode(makeNode("c2", "dcB", "rB2", 20))
|
||||
a.AddNode(makeNode("c3", "dcB", "rB2", 20))
|
||||
a.Finalize()
|
||||
|
||||
rep, err := NewReplicationConfigFromString("100")
|
||||
require.NoError(t, err)
|
||||
plan, err := NewRebalancer(DefaultECConfig(), rep).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
moved := make(map[int]string)
|
||||
for _, m := range plan.Moves {
|
||||
_, twice := moved[m.ShardID]
|
||||
assert.False(t, twice, "shard %d moves more than once", m.ShardID)
|
||||
moved[m.ShardID] = m.DestNode.NodeID
|
||||
assert.Contains(t, m.DestNode.ShardIDs, m.ShardID)
|
||||
}
|
||||
assert.Equal(t, 7, plan.TotalMoves)
|
||||
assert.Equal(t, 7, plan.MovesAcrossDC)
|
||||
// shard 12 reached b1 through c1 in simulation; the plan moves it directly
|
||||
assert.Equal(t, "b1", moved[12])
|
||||
assert.Equal(t, []int{13, 9, 12}, plan.Analysis.AllNodes["b1"].ShardIDs)
|
||||
}
|
||||
|
||||
// A collapsed move must not be scheduled before the departures that free its
|
||||
// destination: applying the plan in order may never target a full node.
|
||||
func TestPlanRebalanceMovesRespectSlots(t *testing.T) {
|
||||
ec, err := NewECConfig(20, 4)
|
||||
require.NoError(t, err)
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("a1", "dcA", "rA", 20, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16))
|
||||
a.AddNode(makeNode("b1", "dcB", "rB1", 10))
|
||||
a.AddNode(makeNode("b2", "dcB", "rB1", 0, 20, 21, 22, 23))
|
||||
a.AddNode(makeNode("c1", "dcB", "rB2", 10, 17, 18, 19))
|
||||
a.Finalize()
|
||||
|
||||
rep, err := NewReplicationConfigFromString("100")
|
||||
require.NoError(t, err)
|
||||
plan, err := NewRebalancer(ec, rep).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
free := make(map[string]int)
|
||||
for _, n := range a.AllNodes {
|
||||
free[n.NodeID] = n.FreeSlots
|
||||
}
|
||||
firstDeparture, firstArrival := -1, -1
|
||||
for i, m := range plan.Moves {
|
||||
assert.Greater(t, free[m.DestNode.NodeID], 0, "move %d targets a full node", i)
|
||||
free[m.DestNode.NodeID]--
|
||||
free[m.SourceNode.NodeID]++
|
||||
if m.DestNode.NodeID == "b2" && firstArrival < 0 {
|
||||
firstArrival = i
|
||||
}
|
||||
if m.SourceNode.NodeID == "b2" && firstDeparture < 0 {
|
||||
firstDeparture = i
|
||||
}
|
||||
}
|
||||
moved := make(map[int]string)
|
||||
for _, m := range plan.Moves {
|
||||
moved[m.ShardID] = m.DestNode.NodeID
|
||||
}
|
||||
// shard 16 reaches full-at-start b2 through b1; the plan moves it directly,
|
||||
// but only after the departures that free b2's slots
|
||||
assert.Equal(t, "b2", moved[16])
|
||||
assert.Equal(t, "c1", moved[23])
|
||||
assert.Greater(t, firstArrival, firstDeparture)
|
||||
}
|
||||
|
||||
func TestPlanRebalanceNoDuplicateShardOnNode(t *testing.T) {
|
||||
a := NewTopologyAnalysis()
|
||||
a.AddNode(makeNode("n1", "dc1", "r1", 10, 0, 1, 2, 3))
|
||||
a.AddNode(makeNode("n2", "dc1", "r1", 10, 3))
|
||||
a.Finalize()
|
||||
|
||||
plan, err := NewRebalancer(DefaultECConfig(), ReplicationConfig{MinDataCenters: 1, MinRacksPerDC: 1, MinNodesPerRack: 2}).PlanRebalance(a)
|
||||
require.NoError(t, err)
|
||||
|
||||
// n2 already holds 3, so the balancing move brings a different shard
|
||||
assert.Equal(t, 3, plan.Analysis.ShardsByNode["n1"])
|
||||
assert.Equal(t, 2, plan.Analysis.ShardsByNode["n2"])
|
||||
assert.Equal(t, []int{3, 2}, plan.Analysis.AllNodes["n2"].ShardIDs)
|
||||
for _, move := range plan.Moves {
|
||||
assert.NotContains(t, move.DestNode.ShardIDs[:len(move.DestNode.ShardIDs)-1], move.ShardID,
|
||||
"shard %d must not be moved onto a node already holding it", move.ShardID)
|
||||
}
|
||||
}
|
||||
|
||||
func TestCalculateDistributionZeroReplication(t *testing.T) {
|
||||
dist := CalculateDistribution(DefaultECConfig(), ReplicationConfig{})
|
||||
assert.Equal(t, 14, dist.TargetShardsPerDC)
|
||||
assert.Contains(t, dist.Summary(), "Data Centers: 1")
|
||||
assert.NotPanics(t, func() { dist.FaultToleranceAnalysis() })
|
||||
}
|
||||
@@ -1,5 +1,10 @@
|
||||
package distribution
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sort"
|
||||
)
|
||||
|
||||
// ShardMove represents a planned shard move
|
||||
type ShardMove struct {
|
||||
ShardID int
|
||||
@@ -12,7 +17,7 @@ type ShardMove struct {
|
||||
type RebalancePlan struct {
|
||||
Moves []ShardMove
|
||||
Distribution *ECDistribution
|
||||
Analysis *TopologyAnalysis
|
||||
Analysis *TopologyAnalysis // simulated layout once every move applies; move endpoints reference its nodes
|
||||
|
||||
// Statistics
|
||||
TotalMoves int
|
||||
@@ -26,3 +31,305 @@ type Rebalancer struct {
|
||||
ecConfig ECConfig
|
||||
repConfig ReplicationConfig
|
||||
}
|
||||
|
||||
// NewRebalancer creates a rebalancer for the given EC ratio and replication policy.
|
||||
func NewRebalancer(ec ECConfig, rep ReplicationConfig) *Rebalancer {
|
||||
return &Rebalancer{ecConfig: ec, repConfig: rep}
|
||||
}
|
||||
|
||||
// PlanRebalance generates shard moves that bring the volume's shard layout to
|
||||
// the balanced distribution: first across data centers, then racks within each
|
||||
// data center, then nodes within each rack. Moves stop when source and
|
||||
// destination both reach their even share, or when destination nodes have no
|
||||
// free slots. The moves are planned on a copy of the topology, so the given
|
||||
// analysis still describes the cluster as it is; plan.Analysis shows the
|
||||
// layout the moves would produce.
|
||||
func (r *Rebalancer) PlanRebalance(analysis *TopologyAnalysis) (*RebalancePlan, error) {
|
||||
if analysis == nil {
|
||||
return nil, fmt.Errorf("cannot plan rebalancing without a topology analysis")
|
||||
}
|
||||
sim := NewTopologyAnalysis()
|
||||
for _, id := range sortedKeys(analysis.AllNodes) {
|
||||
node := *analysis.AllNodes[id]
|
||||
node.ShardIDs = append([]int(nil), node.ShardIDs...)
|
||||
sim.AddNode(&node)
|
||||
}
|
||||
sim.Finalize()
|
||||
plan := &RebalancePlan{
|
||||
Distribution: CalculateDistribution(r.ecConfig, r.repConfig),
|
||||
Analysis: sim,
|
||||
}
|
||||
|
||||
recordMove := func(shardID int, src, dst *TopologyNode, level, domain string) {
|
||||
src.ShardIDs = removeShard(src.ShardIDs, shardID)
|
||||
dst.ShardIDs = append(dst.ShardIDs, shardID)
|
||||
src.FreeSlots++
|
||||
dst.FreeSlots--
|
||||
sim.ShardsByDC[src.DataCenter]--
|
||||
sim.ShardsByDC[dst.DataCenter]++
|
||||
sim.ShardsByRack[rackKey(src.DataCenter, src.Rack)]--
|
||||
sim.ShardsByRack[rackKey(dst.DataCenter, dst.Rack)]++
|
||||
sim.ShardsByNode[src.NodeID]--
|
||||
sim.ShardsByNode[dst.NodeID]++
|
||||
sim.DCToShards[src.DataCenter] = removeShard(sim.DCToShards[src.DataCenter], shardID)
|
||||
sim.DCToShards[dst.DataCenter] = append(sim.DCToShards[dst.DataCenter], shardID)
|
||||
sim.RackToShards[rackKey(src.DataCenter, src.Rack)] = removeShard(sim.RackToShards[rackKey(src.DataCenter, src.Rack)], shardID)
|
||||
sim.RackToShards[rackKey(dst.DataCenter, dst.Rack)] = append(sim.RackToShards[rackKey(dst.DataCenter, dst.Rack)], shardID)
|
||||
sim.NodeToShards[src.NodeID] = removeShard(sim.NodeToShards[src.NodeID], shardID)
|
||||
sim.NodeToShards[dst.NodeID] = append(sim.NodeToShards[dst.NodeID], shardID)
|
||||
plan.Moves = append(plan.Moves, ShardMove{
|
||||
ShardID: shardID,
|
||||
SourceNode: src,
|
||||
DestNode: dst,
|
||||
Reason: fmt.Sprintf("%s balance: %s above target", level, domain),
|
||||
})
|
||||
}
|
||||
|
||||
moveOut := func(level, srcDomain string, srcNodes, dstNodes []*TopologyNode) bool {
|
||||
var shardIDs []int
|
||||
for _, n := range srcNodes {
|
||||
shardIDs = append(shardIDs, n.ShardIDs...)
|
||||
}
|
||||
for _, shardID := range r.ecConfig.SortShardsParityFirst(shardIDs) {
|
||||
var src *TopologyNode
|
||||
for _, n := range srcNodes {
|
||||
if contains(n.ShardIDs, shardID) {
|
||||
src = n
|
||||
break
|
||||
}
|
||||
}
|
||||
if src == nil {
|
||||
continue
|
||||
}
|
||||
// a shard already inside the destination domain gains nothing there
|
||||
duplicate := false
|
||||
for _, n := range dstNodes {
|
||||
if contains(n.ShardIDs, shardID) {
|
||||
duplicate = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if duplicate {
|
||||
continue
|
||||
}
|
||||
var dst *TopologyNode
|
||||
for _, n := range dstNodes {
|
||||
if n.FreeSlots > 0 && (dst == nil || nodeLess(n, dst)) {
|
||||
dst = n
|
||||
}
|
||||
}
|
||||
if dst == nil {
|
||||
continue
|
||||
}
|
||||
recordMove(shardID, src, dst, level, srcDomain)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// balance moves shards between domains of one level until every domain
|
||||
// holds at most its even share or no destination can take a shard. A
|
||||
// source that cannot shed any shard is set aside rather than stopping
|
||||
// the level, so other overloaded sources still get to move.
|
||||
balance := func(level string, domains []string, countOf func(string) int, target int, nodesOf func(string) []*TopologyNode, dstDomains func(string) []string) {
|
||||
blocked := make(map[string]bool)
|
||||
for {
|
||||
src := ""
|
||||
srcFound := false
|
||||
for _, d := range domains {
|
||||
if blocked[d] {
|
||||
continue
|
||||
}
|
||||
if countOf(d) > target && (!srcFound || countOf(d) > countOf(src)) {
|
||||
src, srcFound = d, true
|
||||
}
|
||||
}
|
||||
if !srcFound {
|
||||
return
|
||||
}
|
||||
dsts := append([]string(nil), dstDomains(src)...)
|
||||
sort.Slice(dsts, func(i, j int) bool {
|
||||
if countOf(dsts[i]) != countOf(dsts[j]) {
|
||||
return countOf(dsts[i]) < countOf(dsts[j])
|
||||
}
|
||||
return dsts[i] < dsts[j]
|
||||
})
|
||||
moved := false
|
||||
for _, dst := range dsts {
|
||||
if countOf(dst) < target && moveOut(level, src, nodesOf(src), nodesOf(dst)) {
|
||||
moved = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !moved {
|
||||
blocked[src] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
dcList := sortedKeys(sim.DCToRacks)
|
||||
dcRackKeys := func(dc string) []string {
|
||||
racks := sim.DCToRacks[dc]
|
||||
keys := make([]string, 0, len(racks))
|
||||
for _, rack := range racks {
|
||||
keys = append(keys, rackKey(dc, rack))
|
||||
}
|
||||
return keys
|
||||
}
|
||||
totalShards := 0
|
||||
for _, dc := range dcList {
|
||||
totalShards += shardCount(dcRackKeys(dc), sim.RackToNodes)
|
||||
}
|
||||
dcCount := func(dc string) int { return shardCount(dcRackKeys(dc), sim.RackToNodes) }
|
||||
balance("dc", dcList, dcCount, ceilDivide(totalShards, len(dcList)),
|
||||
func(dc string) []*TopologyNode { return nodesOfRacks(dcRackKeys(dc), sim.RackToNodes) },
|
||||
func(string) []string { return dcList })
|
||||
|
||||
for _, dc := range dcList {
|
||||
racks := sim.DCToRacks[dc]
|
||||
rackCount := func(rack string) int { return shardCount([]string{rackKey(dc, rack)}, sim.RackToNodes) }
|
||||
balance("rack", racks, rackCount, ceilDivide(dcCount(dc), len(racks)),
|
||||
func(rack string) []*TopologyNode { return sim.RackToNodes[rackKey(dc, rack)] },
|
||||
func(string) []string { return racks })
|
||||
}
|
||||
|
||||
for _, rack := range sortedKeys(sim.RackToNodes) {
|
||||
nodes := sim.RackToNodes[rack]
|
||||
nodeCount := func(nodeID string) int {
|
||||
for _, n := range nodes {
|
||||
if n.NodeID == nodeID {
|
||||
return len(n.ShardIDs)
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
nodeIDs := make([]string, 0, len(nodes))
|
||||
for _, n := range nodes {
|
||||
nodeIDs = append(nodeIDs, n.NodeID)
|
||||
}
|
||||
sort.Strings(nodeIDs)
|
||||
balance("node", nodeIDs, nodeCount, ceilDivide(shardCount([]string{rack}, sim.RackToNodes), len(nodes)),
|
||||
func(nodeID string) []*TopologyNode {
|
||||
for _, n := range nodes {
|
||||
if n.NodeID == nodeID {
|
||||
return []*TopologyNode{n}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
},
|
||||
func(string) []string { return nodeIDs })
|
||||
}
|
||||
|
||||
// A copy of a shard moves at most once per level, and each finer level only
|
||||
// moves it deeper inside the domain the previous level sent it to, so its
|
||||
// moves form a chain from the first source to the final destination.
|
||||
// Collapse each chain into one transfer — intermediate hops would copy the
|
||||
// shard over the network only to move it again. Chains are keyed by where
|
||||
// the copy sits, so duplicate copies of a shard still move independently.
|
||||
pending := make(map[ShardLocation]int)
|
||||
moves := plan.Moves[:0]
|
||||
for _, m := range plan.Moves {
|
||||
key := ShardLocation{ShardID: m.ShardID, NodeID: m.SourceNode.NodeID}
|
||||
if i, ok := pending[key]; ok {
|
||||
delete(pending, key)
|
||||
moves[i].DestNode = m.DestNode
|
||||
pending[ShardLocation{ShardID: m.ShardID, NodeID: m.DestNode.NodeID}] = i
|
||||
continue
|
||||
}
|
||||
pending[ShardLocation{ShardID: m.ShardID, NodeID: m.DestNode.NodeID}] = len(moves)
|
||||
moves = append(moves, m)
|
||||
}
|
||||
// Order the collapsed moves so each destination has a free slot when its
|
||||
// move is applied: a chain's final destination may only gain a slot from a
|
||||
// later move's departure, so the early position the collapse kept can be
|
||||
// wrong. Emit every move whose destination has room first; a source gains
|
||||
// its slot back as soon as its move is emitted.
|
||||
free := make(map[string]int)
|
||||
for _, n := range sim.AllNodes {
|
||||
free[n.NodeID] = n.FreeSlots
|
||||
}
|
||||
for _, m := range moves {
|
||||
free[m.DestNode.NodeID]++
|
||||
free[m.SourceNode.NodeID]--
|
||||
}
|
||||
ordered := make([]ShardMove, 0, len(moves))
|
||||
for len(moves) > 0 {
|
||||
kept := moves[:0]
|
||||
for _, m := range moves {
|
||||
if free[m.DestNode.NodeID] > 0 {
|
||||
free[m.DestNode.NodeID]--
|
||||
free[m.SourceNode.NodeID]++
|
||||
ordered = append(ordered, m)
|
||||
} else {
|
||||
kept = append(kept, m)
|
||||
}
|
||||
}
|
||||
if len(kept) == len(moves) {
|
||||
ordered = append(ordered, kept...)
|
||||
break
|
||||
}
|
||||
moves = kept
|
||||
}
|
||||
plan.Moves = ordered
|
||||
for _, m := range ordered {
|
||||
switch {
|
||||
case m.SourceNode.DataCenter != m.DestNode.DataCenter:
|
||||
plan.MovesAcrossDC++
|
||||
case m.SourceNode.Rack != m.DestNode.Rack:
|
||||
plan.MovesAcrossRack++
|
||||
default:
|
||||
plan.MovesWithinRack++
|
||||
}
|
||||
}
|
||||
plan.TotalMoves = len(ordered)
|
||||
return plan, nil
|
||||
}
|
||||
|
||||
// nodeLess prefers the node holding fewer of the volume's shards, then the one
|
||||
// with more free slots, for a stable destination choice.
|
||||
func nodeLess(a, b *TopologyNode) bool {
|
||||
if len(a.ShardIDs) != len(b.ShardIDs) {
|
||||
return len(a.ShardIDs) < len(b.ShardIDs)
|
||||
}
|
||||
if a.FreeSlots != b.FreeSlots {
|
||||
return a.FreeSlots > b.FreeSlots
|
||||
}
|
||||
return a.NodeID < b.NodeID
|
||||
}
|
||||
|
||||
// shardCount sums the shards the given racks currently hold for the volume.
|
||||
func shardCount(racks []string, rackToNodes map[string][]*TopologyNode) int {
|
||||
total := 0
|
||||
for _, rack := range racks {
|
||||
for _, n := range rackToNodes[rack] {
|
||||
total += len(n.ShardIDs)
|
||||
}
|
||||
}
|
||||
return total
|
||||
}
|
||||
|
||||
func nodesOfRacks(racks []string, rackToNodes map[string][]*TopologyNode) []*TopologyNode {
|
||||
var nodes []*TopologyNode
|
||||
for _, rack := range racks {
|
||||
nodes = append(nodes, rackToNodes[rack]...)
|
||||
}
|
||||
return nodes
|
||||
}
|
||||
|
||||
func removeShard(shards []int, shardID int) []int {
|
||||
for i, s := range shards {
|
||||
if s == shardID {
|
||||
return append(shards[:i], shards[i+1:]...)
|
||||
}
|
||||
}
|
||||
return shards
|
||||
}
|
||||
|
||||
func sortedKeys[V any](m map[string]V) []string {
|
||||
keys := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
return keys
|
||||
}
|
||||
Reference in new issue
Block a user