From e5dc98dcb2b7fe96b0d3fe26e96a0fbee8b1a07b Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 9 Aug 2026 09:37:49 -0700 Subject: [PATCH] ec.balance: add a -volumeIds filter (#10667) * ec.balance: add a -volumeIds filter Collection scope is often too broad for maintenance. -volumeIds narrows the plan to the given ec volume ids by leaving every other volume out of the topology handed to the planner, so no phase, dedup included, can plan against them. Ids with no ec shard in the selected collection, dataCenter and disk type are rejected rather than silently skipped. * ec.encode: key the orphan sweep without narrowing the volume id int is 32-bit on 32-bit builds, so int(vid) wraps for volume ids above MaxInt32. Format the id as the uint32 it is. --- weed/shell/command_ec_balance.go | 17 ++++- weed/shell/command_ec_common.go | 92 ++++++++++++++++++++++++++-- weed/shell/command_ec_common_test.go | 14 +++++ weed/shell/command_ec_encode.go | 33 +--------- weed/shell/command_ec_encode_test.go | 12 ---- weed/shell/command_ec_test.go | 59 ++++++++++++++++++ 6 files changed, 177 insertions(+), 50 deletions(-) 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) + } +}