diff --git a/weed/worker/tasks/ec_balance/config.go b/weed/worker/tasks/ec_balance/config.go index 33780cdbed..2f54d8492a 100644 --- a/weed/worker/tasks/ec_balance/config.go +++ b/weed/worker/tasks/ec_balance/config.go @@ -9,6 +9,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/worker/tasks/base" ) +// DefaultGlobalMaxMovesPerRack moves the global phase incrementally across detection cycles +// rather than draining a rack in one batch; raise it to work off a large skew faster. +const DefaultGlobalMaxMovesPerRack = 10 + // Config extends BaseConfig with EC balance specific settings type Config struct { base.BaseConfig @@ -19,7 +23,10 @@ type Config struct { PreferredTags []string `json:"preferred_tags"` ReplicaPlacement string `json:"replica_placement"` // e.g. "020"; empty falls back to the master default replication (even spread only when that default is empty or zero) IoBytePerSecond int64 `json:"io_byte_per_second"` // limit each shard copy's rate; 0 falls back to the volume server's maintenance rate - DataCenterFilter string `json:"-"` // per-detection-run, not persisted + // GlobalMaxMovesPerRack caps the cross-volume load-balancing moves planned per rack in one + // detection run. The plugin worker form sets it; the legacy task policy has no field for it. + GlobalMaxMovesPerRack int `json:"global_max_moves_per_rack"` + DataCenterFilter string `json:"-"` // per-detection-run, not persisted } // NewDefaultConfig creates a new default EC balance configuration @@ -30,11 +37,12 @@ func NewDefaultConfig() *Config { ScanIntervalSeconds: 60 * 60, // 1 hour MaxConcurrent: 1, }, - ImbalanceThreshold: 0.2, // 20% - MinServerCount: 3, - CollectionFilter: "", - DiskType: "", - PreferredTags: nil, + ImbalanceThreshold: 0.2, // 20% + MinServerCount: 3, + GlobalMaxMovesPerRack: DefaultGlobalMaxMovesPerRack, + CollectionFilter: "", + DiskType: "", + PreferredTags: nil, } } diff --git a/weed/worker/tasks/ec_balance/detection.go b/weed/worker/tasks/ec_balance/detection.go index 661c2976b7..b29c83ee2a 100644 --- a/weed/worker/tasks/ec_balance/detection.go +++ b/weed/worker/tasks/ec_balance/detection.go @@ -86,7 +86,7 @@ func Detection( VolumeRatio: volumeRatio, // Move incrementally across detection cycles rather than draining a rack // in one batch; the scheduler re-evaluates each cycle. - GlobalMaxMovesPerRack: 10, + GlobalMaxMovesPerRack: globalMaxMovesPerRack(ecConfig, maxResults), // Balance heterogeneous-capacity racks by fractional fullness. GlobalUtilizationBased: true, RackTotalCapRaised: func(collection string, vid uint32, rackCap, evenCap int) { @@ -387,6 +387,21 @@ func movePhaseOrder(phase string) int { } } +// globalMaxMovesPerRack is the configured per-rack cap on global moves, or the default +// when unset (0 would otherwise mean unlimited to the planner). It never exceeds +// maxResults: the detection keeps at most that many moves, and each planned global move +// costs a scan of the source node's volumes, so a larger cap only adds planning time. +func globalMaxMovesPerRack(c *Config, maxResults int) int { + moves := c.GlobalMaxMovesPerRack + if moves <= 0 { + moves = DefaultGlobalMaxMovesPerRack + } + if maxResults > 0 { + moves = min(moves, maxResults) + } + return moves +} + func movePhasePriority(phase string) types.TaskPriority { switch phase { case "dedup": diff --git a/weed/worker/tasks/ec_balance/detection_test.go b/weed/worker/tasks/ec_balance/detection_test.go index 80374e5b4d..a80ec0c3e2 100644 --- a/weed/worker/tasks/ec_balance/detection_test.go +++ b/weed/worker/tasks/ec_balance/detection_test.go @@ -253,3 +253,65 @@ func TestDetectionKeepsGlobalMovesUnderTheCap(t *testing.T) { t.Errorf("kept %q, want a global move", results[0].Reason) } } + +// TestDetectionUsesConfiguredGlobalCap: the configured per-rack cap reaches the planner, and +// an unset one plans the default rather than the planner's unlimited 0. +func TestDetectionUsesConfiguredGlobalCap(t *testing.T) { + disk := func(shards ...*master_pb.VolumeEcShardInformationMessage) map[string]*master_pb.DiskInfo { + return map[string]*master_pb.DiskInfo{"": {Type: "", MaxVolumeCount: 100, EcShardInfos: shards}} + } + var loaded []*master_pb.VolumeEcShardInformationMessage + for vid := uint32(1); vid <= 40; vid++ { + loaded = append(loaded, &master_pb.VolumeEcShardInformationMessage{Id: vid, Collection: "col1", EcIndexBits: 1}) + } + topoInfo := &master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + Id: "dc1", + RackInfos: []*master_pb.RackInfo{{Id: "rack1", DataNodeInfos: []*master_pb.DataNodeInfo{ + {Id: "10.0.0.1:8080", DiskInfos: disk(loaded...)}, + {Id: "10.0.0.2:8080", DiskInfos: disk()}, + {Id: "10.0.0.3:8080", DiskInfos: disk()}, + {Id: "10.0.0.4:8080", DiskInfos: disk()}, + }}}, + }}, + } + at := topology.NewActiveTopology(0) + if err := at.UpdateTopology(topoInfo); err != nil { + t.Fatalf("UpdateTopology: %v", err) + } + clusterInfo := &types.ClusterInfo{ActiveTopology: at} + + for _, tc := range []struct{ configured, maxResults, want int }{ + {0, 0, DefaultGlobalMaxMovesPerRack}, + {25, 0, 25}, + {25, 20, 20}, + } { + config := NewDefaultConfig() + config.GlobalMaxMovesPerRack = tc.configured + results, _, err := Detection(context.Background(), nil, clusterInfo, config, tc.maxResults) + if err != nil { + t.Fatalf("Detection: %v", err) + } + global := 0 + for _, r := range results { + if strings.Contains(r.Reason, " global:") { + global++ + } + } + if global != tc.want { + t.Errorf("cap %d with %d results planned %d global moves, want %d", tc.configured, tc.maxResults, global, tc.want) + } + } +} + +// TestGlobalMaxMovesPerRackBoundedByMaxResults: a rack plans no more global moves than the +// detection can keep. +func TestGlobalMaxMovesPerRackBoundedByMaxResults(t *testing.T) { + config := &Config{GlobalMaxMovesPerRack: 5000} + if got := globalMaxMovesPerRack(config, 500); got != 500 { + t.Errorf("cap with 500 results = %d, want 500", got) + } + if got := globalMaxMovesPerRack(config, 0); got != 5000 { + t.Errorf("cap with unlimited results = %d, want 5000", got) + } +} diff --git a/weed/worker/tasks/ec_balance/plugin_handler.go b/weed/worker/tasks/ec_balance/plugin_handler.go index d545dc5a47..c0588f8a03 100644 --- a/weed/worker/tasks/ec_balance/plugin_handler.go +++ b/weed/worker/tasks/ec_balance/plugin_handler.go @@ -21,7 +21,10 @@ import ( const ( ecBalanceMinImbalanceThreshold = 0.05 ecBalanceMaxImbalanceThreshold = 0.5 - ecBalanceMinServerCount = 2 + // The planner treats 0 as unlimited, which would plan a whole drain in one detection run. + ecBalanceMinGlobalMaxMovesPerRack = 1 + ecBalanceMaxGlobalMaxMovesPerRack = 5000 + ecBalanceMinServerCount = 2 ) func init() { @@ -149,6 +152,16 @@ func (h *ECBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { Required: true, MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: ecBalanceMinServerCount}}, }, + { + Name: "global_max_moves_per_rack", + Label: "Global Moves Per Rack Per Detection", + Description: "Most cross-volume load-balancing moves planned per rack in one detection run. The default moves slowly; raise it to work off a server that holds far more shards than its peers.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + Required: true, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: ecBalanceMinGlobalMaxMovesPerRack}}, + MaxValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: ecBalanceMaxGlobalMaxMovesPerRack}}, + }, { Name: "io_byte_per_second", Label: "Shard Copy IO Limit (bytes/sec)", @@ -169,10 +182,11 @@ func (h *ECBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { }, }, DefaultValues: map[string]*plugin_pb.ConfigValue{ - "imbalance_threshold": {Kind: &plugin_pb.ConfigValue_DoubleValue{DoubleValue: 0.2}}, - "min_server_count": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 3}}, - "io_byte_per_second": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, - "preferred_tags": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}}, + "imbalance_threshold": {Kind: &plugin_pb.ConfigValue_DoubleValue{DoubleValue: 0.2}}, + "min_server_count": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 3}}, + "global_max_moves_per_rack": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: DefaultGlobalMaxMovesPerRack}}, + "io_byte_per_second": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "preferred_tags": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}}, }, }, AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{ @@ -188,10 +202,11 @@ func (h *ECBalanceHandler) Descriptor() *plugin_pb.JobTypeDescriptor { ExecutionTimeoutSeconds: 1800, }, WorkerDefaultValues: map[string]*plugin_pb.ConfigValue{ - "imbalance_threshold": {Kind: &plugin_pb.ConfigValue_DoubleValue{DoubleValue: 0.2}}, - "min_server_count": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 3}}, - "io_byte_per_second": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, - "preferred_tags": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}}, + "imbalance_threshold": {Kind: &plugin_pb.ConfigValue_DoubleValue{DoubleValue: 0.2}}, + "min_server_count": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 3}}, + "global_max_moves_per_rack": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: DefaultGlobalMaxMovesPerRack}}, + "io_byte_per_second": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "preferred_tags": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}}, }, } } @@ -428,6 +443,9 @@ func deriveECBalanceWorkerConfig(values map[string]*plugin_pb.ConfigValue) *ecBa } taskConfig.MinServerCount = minServerCount + globalMaxMoves := pluginworker.ReadInt64Config(values, "global_max_moves_per_rack", int64(taskConfig.GlobalMaxMovesPerRack)) + taskConfig.GlobalMaxMovesPerRack = int(min(max(globalMaxMoves, ecBalanceMinGlobalMaxMovesPerRack), ecBalanceMaxGlobalMaxMovesPerRack)) + ioBytePerSecond := pluginworker.ReadInt64Config(values, "io_byte_per_second", taskConfig.IoBytePerSecond) if ioBytePerSecond < 0 { ioBytePerSecond = 0 diff --git a/weed/worker/tasks/ec_balance/plugin_handler_test.go b/weed/worker/tasks/ec_balance/plugin_handler_test.go index bfabacc936..68243a6535 100644 --- a/weed/worker/tasks/ec_balance/plugin_handler_test.go +++ b/weed/worker/tasks/ec_balance/plugin_handler_test.go @@ -42,6 +42,48 @@ func TestDeriveECBalanceWorkerConfig(t *testing.T) { } }, }, + { + name: "global moves per rack defaults to 10", + values: nil, + check: func(t *testing.T, config *ecBalanceWorkerConfig) { + if got := config.TaskConfig.GlobalMaxMovesPerRack; got != 10 { + t.Errorf("expected default global_max_moves_per_rack 10, got %d", got) + } + }, + }, + { + name: "global moves per rack override", + values: map[string]*plugin_pb.ConfigValue{ + "global_max_moves_per_rack": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 200}}, + }, + check: func(t *testing.T, config *ecBalanceWorkerConfig) { + if got := config.TaskConfig.GlobalMaxMovesPerRack; got != 200 { + t.Errorf("expected global_max_moves_per_rack 200, got %d", got) + } + }, + }, + { + name: "global moves per rack never 0, which the planner reads as unlimited", + values: map[string]*plugin_pb.ConfigValue{ + "global_max_moves_per_rack": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + }, + check: func(t *testing.T, config *ecBalanceWorkerConfig) { + if got := config.TaskConfig.GlobalMaxMovesPerRack; got != 1 { + t.Errorf("expected clamp to 1, got %d", got) + } + }, + }, + { + name: "global moves per rack clamped to max", + values: map[string]*plugin_pb.ConfigValue{ + "global_max_moves_per_rack": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1 << 31}}, + }, + check: func(t *testing.T, config *ecBalanceWorkerConfig) { + if got := config.TaskConfig.GlobalMaxMovesPerRack; got != 5000 { + t.Errorf("expected clamp to 5000, got %d", got) + } + }, + }, { name: "threshold clamped to min", values: map[string]*plugin_pb.ConfigValue{