diff --git a/weed/storage/erasure_coding/distribution/README.md b/weed/storage/erasure_coding/distribution/README.md index 91cfd8fe8a..dbd2dfe096 100644 --- a/weed/storage/erasure_coding/distribution/README.md +++ b/weed/storage/erasure_coding/distribution/README.md @@ -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) diff --git a/weed/storage/erasure_coding/distribution/analysis.go b/weed/storage/erasure_coding/distribution/analysis.go index b939df53ee..e19d4624a0 100644 --- a/weed/storage/erasure_coding/distribution/analysis.go +++ b/weed/storage/erasure_coding/distribution/analysis.go @@ -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 diff --git a/weed/storage/erasure_coding/distribution/config.go b/weed/storage/erasure_coding/distribution/config.go index b4935b0c71..4a0402d754 100644 --- a/weed/storage/erasure_coding/distribution/config.go +++ b/weed/storage/erasure_coding/distribution/config.go @@ -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 +} diff --git a/weed/storage/erasure_coding/distribution/distribution.go b/weed/storage/erasure_coding/distribution/distribution.go index 1ef05c55d6..3cf38c5bdc 100644 --- a/weed/storage/erasure_coding/distribution/distribution.go +++ b/weed/storage/erasure_coding/distribution/distribution.go @@ -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() +} diff --git a/weed/storage/erasure_coding/distribution/distribution_test.go b/weed/storage/erasure_coding/distribution/distribution_test.go new file mode 100644 index 0000000000..fe6962dccf --- /dev/null +++ b/weed/storage/erasure_coding/distribution/distribution_test.go @@ -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() }) +} diff --git a/weed/storage/erasure_coding/distribution/rebalancer.go b/weed/storage/erasure_coding/distribution/rebalancer.go index 2442e59a97..7ee178783b 100644 --- a/weed/storage/erasure_coding/distribution/rebalancer.go +++ b/weed/storage/erasure_coding/distribution/rebalancer.go @@ -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 +}