From 4f9393889c04c0d6211124df6f0449075eb6005a Mon Sep 17 00:00:00 2001 From: qzhello <951012707@qq.com> Date: Mon, 22 Jun 2026 16:22:20 +0800 Subject: [PATCH] feat(shell): Support batched EC encode and multi-volume selection in ec.encode (#10030) * fix(shell): correct volume.list -writable filter unit and comparison * fix(shell): correct volume.list -writable filter unit and comparison * chore(shell): fix typo in EC shard helper param names * fix(shell): use exact match for volume.balance -racks/-nodes filter The old strings.Contains-based filter quietly included any id that was a substring of the user-supplied flag value (e.g. -racks=rack10 also matched rack1). Replace it with an exact-match set parsed from the comma-separated flag value, and add regression tests for both -racks and -nodes paths. Also fix a small typo in the "remote storage" error returned by maybeMoveOneVolume. * fix(shell): use exact match for volume.balance -racks/-nodes filter The old strings.Contains-based filter quietly included any id that was a substring of the user-supplied flag value (e.g. -racks=rack10 also matched rack1). Replace it with an exact-match set parsed from the comma-separated flag value, and add regression tests for both -racks and -nodes paths. Also fix a small typo in the "remote storage" error returned by maybeMoveOneVolume. * refactor(shell): drop nil sentinel in splitCSVSet, use len() in callers * feat(shell): support batched EC encode and multi-volume selection Add -volumeIds (comma-separated) and -batchSize flags to ec.encode. When -batchSize > 0, volumes are processed in independent batches, each committed separately: encode -> rebalance -> verify -> delete originals. This bounds the working set and lets source volumes be reclaimed without waiting for the entire set to finish, at the cost of per-batch rebalancing. Because each batch deletes its originals, a failure in a later batch is unrecoverable for already-completed batches. To let the single-volume, multi-volume, and collection paths share one per-batch routine, the re-balance scope is now always derived from the volumes actually selected for encoding (collectCollectionsForVolumeIds), rather than every collection matching the -collection regex. Practical effect: with -collection, a collection that matches the pattern but contributes no encodable volumes is no longer re-balanced as a side effect. The -volumeId path is unchanged; -batchSize=0 (default) preserves the original single-pass flow. The per-batch routine reuses the existing assertEncodableRegularVolumes guard, doEcEncode skipped-node handling, and verifyEcShardsBeforeDelete retry loop. The capacity pre-flight check takes the already-fetched topology instead of issuing another VolumeList to the master per batch. Also clarify the -collection flag description to note it accepts a regex pattern, matching the existing command help. -volumeId and -volumeIds are mutually exclusive; ids in -volumeIds are validated and de-duplicated. --- weed/shell/command_ec_encode.go | 187 ++++++++++++++++++++------- weed/shell/command_ec_encode_test.go | 24 ++++ 2 files changed, 164 insertions(+), 47 deletions(-) diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go index 29652decb..7296bfdb6 100644 --- a/weed/shell/command_ec_encode.go +++ b/weed/shell/command_ec_encode.go @@ -9,6 +9,7 @@ import ( "regexp" "sort" "strconv" + "strings" "sync" "time" @@ -27,6 +28,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" "github.com/seaweedfs/seaweedfs/weed/storage/volume_replica" ) @@ -44,8 +46,8 @@ func (c *commandEcEncode) Name() string { func (c *commandEcEncode) Help() string { return `apply erasure coding to a volume - ec.encode [-collection=""] [-fullPercent=95 -quietFor=1h] [-verbose] [-sourceDiskType=] [-diskType=] - ec.encode [-collection=""] [-volumeId=] [-verbose] [-diskType=] + ec.encode [-collection=""] [-fullPercent=95 -quietFor=1h] [-batchSize=0] [-verbose] [-sourceDiskType=] [-diskType=] + ec.encode [-volumeId=|-volumeIds=,...] [-batchSize=0] [-verbose] [-diskType=] This command will: 1. freeze one volume @@ -70,6 +72,11 @@ func (c *commandEcEncode) Help() string { -verbose: show detailed reasons why volumes are not selected for encoding -sourceDiskType: filter source volumes by disk type (hdd, ssd, or empty for all) -diskType: target disk type for EC shards (hdd, ssd, or empty for default hdd) + -batchSize: if > 0, encode/rebalance/verify/delete this many volumes at a time + -volumeIds: comma-separated volume IDs to encode + + When -batchSize is set, each batch is committed independently. If a later batch fails, + earlier batches may already be encoded and their original volumes deleted. Examples: # Encode SSD volumes to SSD EC shards (same tier) @@ -81,6 +88,9 @@ func (c *commandEcEncode) Help() string { # Encode all volumes to SSD EC shards ec.encode -collection=mybucket -diskType=ssd + # Encode selected volume IDs and delete originals after each batch + ec.encode -volumeIds=101,102,103 -batchSize=2 + Re-balancing algorithm: ` + ecBalanceAlgorithmDescription } @@ -93,10 +103,12 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr encodeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) volumeId := encodeCommand.Int("volumeId", 0, "the volume id") - collection := encodeCommand.String("collection", "", "the collection name") + volumeIdsStr := encodeCommand.String("volumeIds", "", "comma-separated volume ids") + collection := encodeCommand.String("collection", "", "collection name or regex pattern") fullPercentage := encodeCommand.Float64("fullPercent", 95, "the volume reaches the percentage of max volume size") quietPeriod := encodeCommand.Duration("quietFor", time.Hour, "select volumes without no writes for this period") maxParallelization := encodeCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") + batchSize := encodeCommand.Int("batchSize", 0, "if > 0, encode/re-balance/verify/delete up to this many volumes at a time") forceChanges := encodeCommand.Bool("force", false, "force the encoding even if the cluster has less than recommended 4 nodes") shardReplicaPlacement := encodeCommand.String("shardReplicaPlacement", "", "replica placement for EC shards, or master default if empty") sourceDiskTypeStr := encodeCommand.String("sourceDiskType", "", "filter source volumes by disk type (hdd, ssd, or empty for all)") @@ -143,14 +155,21 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr } var volumeIds []needle.VolumeId - var balanceCollections []string - if vid := needle.VolumeId(*volumeId); vid != 0 { - // volumeId is provided - volumeIds = append(volumeIds, vid) - balanceCollections = collectCollectionsForVolumeIds(topologyInfo, volumeIds) + if *volumeId != 0 || strings.TrimSpace(*volumeIdsStr) != "" { + if *volumeId != 0 && strings.TrimSpace(*volumeIdsStr) != "" { + return fmt.Errorf("-volumeId and -volumeIds are mutually exclusive") + } + if *volumeId != 0 { + volumeIds = append(volumeIds, needle.VolumeId(*volumeId)) + } else { + volumeIds, err = parseEcEncodeVolumeIds(*volumeIdsStr) + if err != nil { + return err + } + } } else { // apply to all volumes for the given collection pattern (regex) - volumeIds, balanceCollections, err = collectVolumeIdsForEcEncode(commandEnv, *collection, sourceDiskType, *fullPercentage, *quietPeriod, *verbose) + volumeIds, _, err = collectVolumeIdsForEcEncode(commandEnv, *collection, sourceDiskType, *fullPercentage, *quietPeriod, *verbose) if err != nil { return err } @@ -159,74 +178,148 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr fmt.Println("No volumes, nothing to do.") return nil } + if *batchSize < 0 { + return fmt.Errorf("-batchSize must be >= 0") + } + + batches := chunkEcEncodeVolumeIds(volumeIds, *batchSize) + if *batchSize > 0 { + fmt.Printf("Processing %d volumes in %d batch(es), batchSize=%d\n", len(volumeIds), len(batches), *batchSize) + } + for i, batchVolumeIds := range batches { + if *batchSize > 0 { + fmt.Printf("Starting EC encoding batch %d/%d with %d volumes: %v\n", i+1, len(batches), len(batchVolumeIds), batchVolumeIds) + } + if err := processEcEncodeBatch(commandEnv, writer, batchVolumeIds, rp, diskType, *maxParallelization, *applyBalancing, *collection); err != nil { + return fmt.Errorf("ec encode batch %d/%d for volumes %v: %w", i+1, len(batches), batchVolumeIds, err) + } + } + if *batchSize > 0 { + fmt.Printf("Successfully completed EC encoding for %d volumes in %d batch(es)\n", len(volumeIds), len(batches)) + } + + 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 chunkEcEncodeVolumeIds(volumeIds []needle.VolumeId, batchSize int) [][]needle.VolumeId { + if batchSize <= 0 || len(volumeIds) == 0 { + return [][]needle.VolumeId{volumeIds} + } + var batches [][]needle.VolumeId + for start := 0; start < len(volumeIds); start += batchSize { + end := start + batchSize + if end > len(volumeIds) { + end = len(volumeIds) + } + batches = append(batches, volumeIds[start:end]) + } + return batches +} + +func processEcEncodeBatch(commandEnv *CommandEnv, writer io.Writer, volumeIds []needle.VolumeId, rp *super_block.ReplicaPlacement, diskType types.DiskType, maxParallelization int, applyBalancing bool, collectionForMessage string) error { + topologyInfo, _, err := collectTopologyInfo(commandEnv, 0) + if err != nil { + return err + } // Refuse to encode a volume that is already EC (present only as shards): // an EC volume has no .dat, so re-encoding it would tear down its only // copy before failing. A regular volume (with a .dat) passes. This closes // the operator-rerun / script-retry path; a worker racing the snapshot is // handled by encode fencing, not here. - if err = assertEncodableRegularVolumes(topologyInfo, volumeIds); err != nil { + if err := assertEncodableRegularVolumes(topologyInfo, volumeIds); err != nil { return err } - // Collect volume ID to collection name mapping for the sync operation volumeIdToCollection := collectVolumeIdToCollection(topologyInfo, volumeIds) + balanceCollections := collectCollectionsForVolumeIds(topologyInfo, volumeIds) - // Collect volume locations BEFORE EC encoding starts to avoid race condition - // where the master metadata is updated after EC encoding but before deletion fmt.Printf("Collecting volume locations for %d volumes before EC encoding...\n", len(volumeIds)) volumeLocationsMap, err := volumeLocations(commandEnv, volumeIds) if err != nil { return fmt.Errorf("failed to collect volume locations before EC encoding: %w", err) } - // Pre-flight check: verify the target disk type has capacity for EC shards - // This prevents encoding shards only to fail during rebalance - _, totalFreeEcSlots, err := collectEcNodesForDC(commandEnv, "", diskType) - if err != nil { - return fmt.Errorf("failed to check EC shard capacity: %w", err) + if err := checkEcEncodeCapacity(topologyInfo, len(volumeIds), diskType, collectionForMessage); err != nil { + return err } - // Calculate required slots: each volume needs TotalShardsCount (14) shards distributed - requiredSlots := len(volumeIds) * erasure_coding.TotalShardsCount + skippedNodes, err := doEcEncode(commandEnv, writer, volumeIdToCollection, volumeIds, maxParallelization, topologyInfo) + if err != nil { + return fmt.Errorf("ec encode for volumes %v: %w", volumeIds, err) + } + // EcBalance works at collection scope. In batch mode this intentionally + // rebalances each collection after every batch so source volumes can be + // 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 { + return fmt.Errorf("re-balance ec shards for collection(s) %v: %w", balanceCollections, err) + } + if err := verifyEcShardsBeforeDelete(commandEnv, volumeIds, diskType); err != nil { + return fmt.Errorf("verify EC shards before deleting originals: %w", err) + } + fmt.Printf("Deleting original volumes after EC encoding...\n") + if err := doDeleteVolumesWithLocations(commandEnv, volumeIds, volumeLocationsMap, maxParallelization); err != nil { + return fmt.Errorf("delete original volumes after EC encoding: %w", err) + } + fmt.Printf("Successfully completed EC encoding for %d volumes\n", len(volumeIds)) + return nil +} + +func checkEcEncodeCapacity(topologyInfo *master_pb.TopologyInfo, volumeCount int, diskType types.DiskType, collectionForMessage string) error { + // Pre-flight check: verify the target disk type has capacity for EC shards. + // This prevents encoding shards only to fail during rebalance. Reuse the + // caller's topology snapshot instead of issuing another VolumeList to the + // master per batch. + _, totalFreeEcSlots := collectEcVolumeServersByDc(topologyInfo, "", diskType) + + // Each volume needs TotalShardsCount (14) shards distributed. + requiredSlots := volumeCount * erasure_coding.TotalShardsCount if totalFreeEcSlots < 1 { - // No capacity at all on the target disk type if diskType != types.HardDriveType { + tryDiskTypeMessage := "Try passing -diskType=hdd, or omit -diskType to use the default (hdd)" + if collectionForMessage != "" { + tryDiskTypeMessage = fmt.Sprintf("Try:\n ec.encode -collection=%s -diskType=hdd\nOr omit -diskType to use the default (hdd)", collectionForMessage) + } return fmt.Errorf("no free ec shard slots on disk type '%s'. The target disk type has no capacity.\n"+ - "Your volumes are likely on a different disk type. Try:\n"+ - " ec.encode -collection=%s -diskType=hdd\n"+ - "Or omit -diskType to use the default (hdd)", diskType, *collection) + "Your volumes are likely on a different disk type. %s", diskType, tryDiskTypeMessage) } return fmt.Errorf("no free ec shard slots. only %d left on disk type '%s'", totalFreeEcSlots, diskType) } if totalFreeEcSlots < requiredSlots { fmt.Printf("Warning: limited EC shard capacity. Need %d slots for %d volumes, but only %d slots available on disk type '%s'.\n", - requiredSlots, len(volumeIds), totalFreeEcSlots, diskType) + requiredSlots, volumeCount, totalFreeEcSlots, diskType) fmt.Printf("Rebalancing may not achieve optimal distribution.\n") } - - // encode all requested volumes... - skippedNodes, err := doEcEncode(commandEnv, writer, volumeIdToCollection, volumeIds, *maxParallelization, topologyInfo) - if err != nil { - return fmt.Errorf("ec encode for volumes %v: %w", volumeIds, err) - } - // ...re-balance ec shards, excluding nodes the orphan sweep could not reach 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 { - return fmt.Errorf("re-balance ec shards for collection(s) %v: %w", balanceCollections, err) - } - // A partial encode followed by source deletion is unrecoverable. - if err := verifyEcShardsBeforeDelete(commandEnv, volumeIds, diskType); err != nil { - return fmt.Errorf("verify EC shards before deleting originals: %w", err) - } - // ...then delete original volumes using pre-collected locations. - fmt.Printf("Deleting original volumes after EC encoding...\n") - if err := doDeleteVolumesWithLocations(commandEnv, volumeIds, volumeLocationsMap, *maxParallelization); err != nil { - return fmt.Errorf("delete original volumes after EC encoding: %w", err) - } - fmt.Printf("Successfully completed EC encoding for %d volumes\n", len(volumeIds)) - return nil } diff --git a/weed/shell/command_ec_encode_test.go b/weed/shell/command_ec_encode_test.go index 76fc6bf43..9c907c52e 100644 --- a/weed/shell/command_ec_encode_test.go +++ b/weed/shell/command_ec_encode_test.go @@ -246,3 +246,27 @@ func TestEcEncodeNodeCountCheck(t *testing.T) { willProceed = forceChanges || nodeCount >= minNodeCount 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 TestChunkEcEncodeVolumeIds(t *testing.T) { + vids := []needle.VolumeId{101, 102, 103, 104, 105} + + assert.Equal(t, [][]needle.VolumeId{ + {101, 102}, + {103, 104}, + {105}, + }, chunkEcEncodeVolumeIds(vids, 2)) + + assert.Equal(t, [][]needle.VolumeId{vids}, chunkEcEncodeVolumeIds(vids, 0)) +}