diff --git a/weed/shell/command_ec_balance.go b/weed/shell/command_ec_balance.go index 71169bb88..695ec676b 100644 --- a/weed/shell/command_ec_balance.go +++ b/weed/shell/command_ec_balance.go @@ -5,6 +5,7 @@ import ( "io" "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" ) @@ -22,10 +23,14 @@ func (c *commandEcBalance) Name() string { func (c *commandEcBalance) Help() string { return `balance all ec shards among all racks and volume servers - ec.balance [-c EACH_COLLECTION|] [-apply] [-dataCenter ] [-shardReplicaPlacement ] [-diskType ] + ec.balance [-c EACH_COLLECTION|] [-apply] [-dataCenter ] [-shardReplicaPlacement ] [-diskType ] [-volumeIds [,...]] Options: -diskType: the disk type for EC shards (hdd, ssd, or empty for default hdd) + -volumeIds: only balance these ec volume ids, e.g. -volumeIds 123,456. The plan is + built as if no other ec volume existed, so nothing else is moved or deduplicated. + Ids without any ec shard in the selected collection, dataCenter and disk type are + an error. Algorithm: ` + ecBalanceAlgorithmDescription @@ -41,6 +46,7 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W dc := balanceCommand.String("dataCenter", "", "only apply the balancing for this dataCenter") shardReplicaPlacement := balanceCommand.String("shardReplicaPlacement", "", "replica placement for EC shards, or master default if empty") diskTypeStr := balanceCommand.String("diskType", "", "the disk type for EC shards (hdd, ssd, or empty for default hdd)") + volumeIdsStr := balanceCommand.String("volumeIds", "", "optional comma-separated list of ec volume ids to balance; defaults to all") maxParallelization := balanceCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") applyBalancing := balanceCommand.Bool("apply", false, "apply the balancing plan") // TODO: remove this alias @@ -53,6 +59,13 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W handleDeprecatedForceFlag(writer, balanceCommand, applyBalancingAlias, applyBalancing) infoAboutSimulationMode(writer, *applyBalancing, "-apply") + var volumeIds []needle.VolumeId + if *volumeIdsStr != "" { + if volumeIds, err = parseVolumeIdsFlag(*volumeIdsStr); err != nil { + return err + } + } + if err = commandEnv.confirmIsLocked(args); err != nil { return } @@ -75,5 +88,5 @@ func (c *commandEcBalance) Do(args []string, commandEnv *CommandEnv, writer io.W diskType := types.ToDiskType(*diskTypeStr) - return EcBalance(commandEnv, collections, *dc, rp, diskType, *maxParallelization, *applyBalancing, nil) + return EcBalance(commandEnv, collections, *dc, rp, diskType, *maxParallelization, *applyBalancing, nil, volumeIds) } diff --git a/weed/shell/command_ec_common.go b/weed/shell/command_ec_common.go index fefaba3cb..cc93d7cff 100644 --- a/weed/shell/command_ec_common.go +++ b/weed/shell/command_ec_common.go @@ -7,6 +7,8 @@ import ( "regexp" "slices" "sort" + "strconv" + "strings" "time" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -848,6 +850,9 @@ type ecBalancer struct { applyBalancing bool maxParallelization int diskType types.DiskType + // volumeIds narrows the plan to these ec volume ids; nil balances every volume + // of the selected collections. + volumeIds map[uint32]bool } // excludeNodes is a set of server addresses kept out of the balance as copy/move @@ -855,7 +860,10 @@ type ecBalancer struct { // reach: such a node may still hold a stale-generation shard orphan, and pairing // it with a new-generation shard from a balance copy would mix generations on one // node. The standalone ec.balance command passes nil. -func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, applyBalancing bool, excludeNodes map[pb.ServerAddress]struct{}) (err error) { +// +// volumeIds, when non-empty, restricts the plan to those ec volume ids; empty +// balances every volume of the given collections. +func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplicaPlacement *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, applyBalancing bool, excludeNodes map[pb.ServerAddress]struct{}, volumeIds []needle.VolumeId) (err error) { // collect all ec nodes allEcNodes, totalFreeEcSlots, err := collectEcNodesForDC(commandEnv, dc, diskType) if err != nil { @@ -883,6 +891,14 @@ func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplic return fmt.Errorf("no free ec shard slots. only %d left", totalFreeEcSlots) } + var volumeIdFilter map[uint32]bool + if len(volumeIds) > 0 { + volumeIdFilter = make(map[uint32]bool, len(volumeIds)) + for _, vid := range volumeIds { + volumeIdFilter[uint32(vid)] = true + } + } + ecb := &ecBalancer{ commandEnv: commandEnv, ecNodes: allEcNodes, @@ -890,6 +906,7 @@ func EcBalance(commandEnv *CommandEnv, collections []string, dc string, ecReplic applyBalancing: applyBalancing, maxParallelization: maxParallelization, diskType: diskType, + volumeIds: volumeIdFilter, } if len(collections) == 0 { @@ -908,7 +925,24 @@ func shellECRatio(_ string) (int, int) { // balance plans EC shard moves with the shared planner and executes them. When // collections is empty all collections present are balanced. func (ecb *ecBalancer) balance(collections []string) error { - topo, volumeRatio := toBalancerTopology(ecb.ecNodes, collections, ecb.diskType) + topo, volumeRatio, selected := toBalancerTopology(ecb.ecNodes, collections, ecb.diskType, ecb.volumeIds) + if len(ecb.volumeIds) > 0 { + requested := make([]uint32, 0, len(ecb.volumeIds)) + for vid := range ecb.volumeIds { + requested = append(requested, vid) + } + slices.Sort(requested) + var missing []uint32 + for _, vid := range requested { + if !selected[vid] { + missing = append(missing, vid) + } + } + if len(missing) > 0 { + return fmt.Errorf("no ec shards found for volume(s) %v: not an ec volume, or outside the selected collection, dataCenter or diskType", missing) + } + fmt.Printf("balancing ec volume(s) %v\n", requested) + } moves := ecbalancer.Plan(topo, ecbalancer.Options{ DiskType: string(ecb.diskType), ImbalanceThreshold: 0, // the shell balances to an even distribution @@ -923,15 +957,27 @@ func (ecb *ecBalancer) balance(collections []string) error { // shard count when capacities are uniform. GlobalUtilizationBased: true, }) + if len(ecb.volumeIds) > 0 { + var deletions int + for _, m := range moves { + if m.Phase == "dedup" { + deletions++ + } + } + fmt.Printf("planned %d ec shard move(s) and %d ec shard deletion(s)\n", len(moves)-deletions, deletions) + } return ecb.executeMoves(moves) } // toBalancerTopology builds an ecbalancer.Topology from the shell's EcNode model, -// including the shards of the requested collections (all collections when empty). +// including the shards of the requested collections (all collections when empty) +// and, when volumeIds is non-nil, only those volume ids. Volumes left out here are +// invisible to the planner, so no phase - dedup included - can plan against them. // It also returns a per-volume ratio lookup built from each shard's heartbeat // (0,0 when unreported, e.g. always in OSS), which Plan prefers over the -// collection ratio for mixed-ratio clusters. -func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.DiskType) (*ecbalancer.Topology, func(collection string, vid uint32) (int, int)) { +// collection ratio for mixed-ratio clusters, and the set of volume ids that made +// it into the topology. +func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types.DiskType, volumeIds map[uint32]bool) (*ecbalancer.Topology, func(collection string, vid uint32) (int, int), map[uint32]bool) { allowed := make(map[string]bool, len(collections)) for _, c := range collections { allowed[c] = true @@ -942,6 +988,7 @@ func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types. vid uint32 } volRatios := make(map[volRatioKey][2]int) + selected := make(map[uint32]bool) topo := ecbalancer.NewTopology() for _, en := range ecNodes { @@ -961,6 +1008,10 @@ func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types. if len(allowed) > 0 && !allowed[eci.Collection] { continue } + if volumeIds != nil && !volumeIds[eci.Id] { + continue + } + selected[eci.Id] = true node.AddShards(eci.Id, eci.Collection, eci.DiskId, erasure_coding.ShardBits(eci.EcIndexBits)) if d, p := ecbalancer.VolumeShardRatio(eci); d > 0 || p > 0 { volRatios[volRatioKey{eci.Collection, eci.Id}] = [2]int{d, p} @@ -972,7 +1023,7 @@ func toBalancerTopology(ecNodes []*EcNode, collections []string, diskType types. r := volRatios[volRatioKey{collection, vid}] return r[0], r[1] } - return topo, volumeRatio + return topo, volumeRatio, selected } // executeMoves carries out the planned moves. Phases run in order (a within-rack @@ -1096,6 +1147,35 @@ func (ecb *ecBalancer) applyShardMoveRPC(src, dst *EcNode, collection string, vi return sourceServerDeleteEcShards(grpcDialOption, collection, vid, srcAddr, copiedShardIds) } +// parseVolumeIdsFlag parses a comma-separated -volumeIds flag value, dropping +// duplicates and keeping the given order. +func parseVolumeIdsFlag(volumeIdsStr string) ([]needle.VolumeId, error) { + var volumeIds []needle.VolumeId + seen := make(map[needle.VolumeId]bool) + for _, part := range strings.Split(volumeIdsStr, ",") { + part = strings.TrimSpace(part) + if part == "" { + continue + } + vidValue, err := strconv.ParseUint(part, 10, 32) + if err != nil || vidValue == 0 { + return nil, fmt.Errorf("invalid volume id %q in -volumeIds", part) + } + // ParseUint with bitSize 32 bounds the value; convert through uint32 + // (matching the rest of the codebase) so the narrowing is provably safe. + vid := needle.VolumeId(uint32(vidValue)) + if seen[vid] { + continue + } + seen[vid] = true + volumeIds = append(volumeIds, vid) + } + if len(volumeIds) == 0 { + return nil, fmt.Errorf("-volumeIds does not contain any valid volume id") + } + return volumeIds, nil +} + // compileCollectionPattern compiles a regex pattern for collection matching. // Empty patterns match empty collections only. // The special keyword CollectionDefault ("_default") matches empty collections. diff --git a/weed/shell/command_ec_common_test.go b/weed/shell/command_ec_common_test.go index 2ec0600e5..c7a81ec7f 100644 --- a/weed/shell/command_ec_common_test.go +++ b/weed/shell/command_ec_common_test.go @@ -4,7 +4,9 @@ import ( "testing" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/stretchr/testify/assert" ) // countFreeShardSlots must report zero for a physically near-full disk even @@ -35,3 +37,15 @@ func TestCountFreeShardSlotsPhysicalDiskGate(t *testing.T) { t.Errorf("unreported-bytes disk free shard slots = %d, want > 0 (slot fallback)", got) } } + +func TestParseVolumeIdsFlag(t *testing.T) { + vids, err := parseVolumeIdsFlag("101, 102,101, 103") + assert.NoError(t, err) + assert.Equal(t, []needle.VolumeId{101, 102, 103}, vids) + + _, err = parseVolumeIdsFlag("101,abc") + assert.Error(t, err) + + _, err = parseVolumeIdsFlag(" , ") + assert.Error(t, err) +} diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go index c9dcce0ea..a22c1f1f3 100644 --- a/weed/shell/command_ec_encode.go +++ b/weed/shell/command_ec_encode.go @@ -163,7 +163,7 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr if *volumeId != 0 { volumeIds = append(volumeIds, needle.VolumeId(*volumeId)) } else { - volumeIds, err = parseEcEncodeVolumeIds(*volumeIdsStr) + volumeIds, err = parseVolumeIdsFlag(*volumeIdsStr) if err != nil { return err } @@ -202,33 +202,6 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr return nil } -func parseEcEncodeVolumeIds(volumeIdsStr string) ([]needle.VolumeId, error) { - var volumeIds []needle.VolumeId - seen := make(map[needle.VolumeId]bool) - for _, part := range strings.Split(volumeIdsStr, ",") { - part = strings.TrimSpace(part) - if part == "" { - continue - } - vidValue, err := strconv.ParseUint(part, 10, 32) - if err != nil || vidValue == 0 { - return nil, fmt.Errorf("invalid volume id %q in -volumeIds", part) - } - // ParseUint with bitSize 32 bounds the value; convert through uint32 - // (matching the rest of the codebase) so the narrowing is provably safe. - vid := needle.VolumeId(uint32(vidValue)) - if seen[vid] { - continue - } - seen[vid] = true - volumeIds = append(volumeIds, vid) - } - if len(volumeIds) == 0 { - return nil, fmt.Errorf("-volumeIds does not contain any valid volume id") - } - return volumeIds, nil -} - func chunkVolumeIds(volumeIds []needle.VolumeId, batchSize int) [][]needle.VolumeId { if batchSize <= 0 || len(volumeIds) == 0 { return [][]needle.VolumeId{volumeIds} @@ -288,7 +261,7 @@ func processEcEncodeBatch(commandEnv *CommandEnv, writer io.Writer, volumeIds [] // safely verified and deleted without waiting for all batches to finish. // skippedNodes are excluded so a recovered node's stale orphan is never // paired with a new-generation shard. - if err := EcBalance(commandEnv, balanceCollections, "", rp, diskType, maxParallelization, applyBalancing, skippedNodes); err != nil { + if err := EcBalance(commandEnv, balanceCollections, "", rp, diskType, maxParallelization, applyBalancing, skippedNodes, nil); err != nil { return fmt.Errorf("re-balance ec shards for collection(s) %v: %w", balanceCollections, err) } if err := verifyEcShardsBeforeDelete(commandEnv, volumeIds, diskType, applyBalancing); err != nil { @@ -518,7 +491,7 @@ func clearPreexistingEcShards(commandEnv *CommandEnv, topologyInfo *master_pb.To // it. Always delete the full shard-id range so a wider custom ratio's // leftovers are covered too. reportedKey := func(addr pb.ServerAddress, vid uint32) string { - return string(addr) + "\x00" + strconv.Itoa(int(vid)) + return string(addr) + "\x00" + strconv.FormatUint(uint64(vid), 10) } reported := make(map[string]struct{}) var nodes []pb.ServerAddress diff --git a/weed/shell/command_ec_encode_test.go b/weed/shell/command_ec_encode_test.go index ad8abe9e6..5ac95a1b1 100644 --- a/weed/shell/command_ec_encode_test.go +++ b/weed/shell/command_ec_encode_test.go @@ -250,18 +250,6 @@ func TestEcEncodeNodeCountCheck(t *testing.T) { assert.True(t, willProceed, "Should proceed with -force even with %d nodes", nodeCount) } -func TestParseEcEncodeVolumeIds(t *testing.T) { - vids, err := parseEcEncodeVolumeIds("101, 102,101, 103") - assert.NoError(t, err) - assert.Equal(t, []needle.VolumeId{101, 102, 103}, vids) - - _, err = parseEcEncodeVolumeIds("101,abc") - assert.Error(t, err) - - _, err = parseEcEncodeVolumeIds(" , ") - assert.Error(t, err) -} - func TestChunkVolumeIds(t *testing.T) { vids := []needle.VolumeId{101, 102, 103, 104, 105} diff --git a/weed/shell/command_ec_test.go b/weed/shell/command_ec_test.go index 63a307446..24f6f04c0 100644 --- a/weed/shell/command_ec_test.go +++ b/weed/shell/command_ec_test.go @@ -1,6 +1,7 @@ package shell import ( + "strings" "testing" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" @@ -458,3 +459,61 @@ func TestCommandEcBalanceIssue8793Topology(t *testing.T) { } } } + +// TestCommandEcBalanceVolumeIdsFilter checks that -volumeIds keeps every phase, +// dedup included, off the volumes that were not asked for. +func TestCommandEcBalanceVolumeIdsFilter(t *testing.T) { + ecb := &ecBalancer{ + ecNodes: []*EcNode{ + // Volume 1: all shards on one node, so balancing has plenty to move. + newEcNode("dc1", "rack1", "dn1", 100).addEcVolumeAndShardsForTest(1, "c1", []erasure_coding.ShardId{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13}), + // Volume 2 is equally lopsided, and shard 0 is duplicated on dn3. + newEcNode("dc1", "rack2", "dn2", 100).addEcVolumeAndShardsForTest(2, "c1", []erasure_coding.ShardId{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13}), + newEcNode("dc1", "rack3", "dn3", 100).addEcVolumeAndShardsForTest(2, "c1", []erasure_coding.ShardId{0}), + newEcNode("dc1", "rack4", "dn4", 100), + newEcNode("dc1", "rack5", "dn5", 100), + newEcNode("dc1", "rack6", "dn6", 100), + }, + applyBalancing: false, + diskType: types.HardDriveType, + volumeIds: map[uint32]bool{1: true}, + } + + if err := ecb.balance([]string{"c1"}); err != nil { + t.Fatalf("balance: %v", err) + } + + // Volume 1 spreads out. + if count := ecb.ecNodes[0].localShardIdCount(1); count == 14 { + t.Errorf("volume 1 was not balanced: dn1 still holds all %d shards", count) + } + + // Volume 2 keeps every shard where it was, duplicate included. + if count := ecb.ecNodes[1].localShardIdCount(2); count != 14 { + t.Errorf("dn2 holds %d shards of the unselected volume 2, want 14", count) + } + if count := ecb.ecNodes[2].localShardIdCount(2); count != 1 { + t.Errorf("dn3 holds %d shards of the unselected volume 2, want the duplicate to survive", count) + } +} + +func TestCommandEcBalanceVolumeIdsNotFound(t *testing.T) { + ecb := &ecBalancer{ + ecNodes: []*EcNode{ + newEcNode("dc1", "rack1", "dn1", 100).addEcVolumeAndShardsForTest(1, "c1", []erasure_coding.ShardId{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13}), + newEcNode("dc1", "rack2", "dn2", 100), + }, + applyBalancing: false, + diskType: types.HardDriveType, + volumeIds: map[uint32]bool{1: true, 99: true}, + } + + err := ecb.balance([]string{"c1"}) + if err == nil || !strings.Contains(err.Error(), "[99]") { + t.Fatalf("want an error naming volume 99, got %v", err) + } + // The valid id must not be balanced either: the plan is all-or-nothing. + if count := ecb.ecNodes[0].localShardIdCount(1); count != 14 { + t.Errorf("dn1 holds %d shards, want the rejected plan to leave volume 1 untouched", count) + } +}