From 298ab35fd7551defd2ba3b00880196aaacbaa29a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 10 Jul 2026 16:44:46 -0700 Subject: [PATCH] shell: default batch size for ec.encode and ec.decode (#10308) * shell: ec.encode batches 10 volumes by default Encoding a whole collection as one batch means a late failure leaves everything half-converted. Default -batchSize to 10 so each batch is encoded, rebalanced, verified, and its originals deleted before the next starts. -batchSize=0 keeps the all-at-once behavior. Batch progress messages now print only when there is more than one batch, so small runs read as before. * shell: ec.decode decodes in batches, refreshing topology in between ec.decode walked every volume off the single topology snapshot taken at startup, which goes stale as earlier decodes move shards around and create volumes. Decode 10 volumes per batch by default, re-collecting the topology and rebuilding the free-space accounting between batches. -batchSize=0 keeps the single-snapshot behavior. --- weed/shell/command_ec_decode.go | 30 ++++++++++++++++++++++++---- weed/shell/command_ec_encode.go | 22 ++++++++++---------- weed/shell/command_ec_encode_test.go | 6 +++--- weed/shell/common.go | 2 ++ 4 files changed, 42 insertions(+), 18 deletions(-) diff --git a/weed/shell/command_ec_decode.go b/weed/shell/command_ec_decode.go index edb1120b4..a7135efc2 100644 --- a/weed/shell/command_ec_decode.go +++ b/weed/shell/command_ec_decode.go @@ -36,7 +36,7 @@ func (c *commandEcDecode) Name() string { func (c *commandEcDecode) Help() string { return `decode a erasure coded volume into a normal volume - ec.decode [-collection=""] [-volumeId=] [-diskType=] [-checkMinFreeSpace] + ec.decode [-collection=""] [-volumeId=] [-batchSize=10] [-diskType=] [-checkMinFreeSpace] The -collection parameter supports regular expressions for pattern matching: - Use exact match: ec.decode -collection="^mybucket$" @@ -46,6 +46,7 @@ func (c *commandEcDecode) Help() string { Options: -diskType: source disk type where EC shards are stored (hdd, ssd, or empty for default hdd) -checkMinFreeSpace: check min free space when selecting the decode target (default true) + -batchSize: decode this many volumes per topology refresh (default 10; 0 = one snapshot for all volumes) Examples: # Decode EC shards from HDD (default) @@ -67,9 +68,13 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr collection := decodeCommand.String("collection", "", "the collection name") diskTypeStr := decodeCommand.String("diskType", "", "source disk type where EC shards are stored (hdd, ssd, or empty for default hdd)") checkMinFreeSpace := decodeCommand.Bool("checkMinFreeSpace", true, "check min free space when selecting the decode target") + batchSize := decodeCommand.Int("batchSize", DefaultEcBatchSize, "decode up to this many volumes per topology refresh (0 = one snapshot for all volumes)") if err = decodeCommand.Parse(args); err != nil { return nil } + if *batchSize < 0 { + return fmt.Errorf("-batchSize must be >= 0") + } if err = commandEnv.confirmIsLocked(args); err != nil { return @@ -99,9 +104,26 @@ func (c *commandEcDecode) Do(args []string, commandEnv *CommandEnv, writer io.Wr return err } fmt.Printf("ec decode volumes: %v\n", volumeIds) - for _, vid := range volumeIds { - if err = doEcDecode(commandEnv, topologyInfo, *collection, vid, diskType, *checkMinFreeSpace, diskUsageState); err != nil { - return err + batches := chunkVolumeIds(volumeIds, *batchSize) + for i, batch := range batches { + if i > 0 { + // earlier batches moved shards and created volumes; re-snapshot so + // shard locations and free-space accounting stay accurate + topologyInfo, _, err = collectTopologyInfo(commandEnv, 0) + if err != nil { + return err + } + if *checkMinFreeSpace { + diskUsageState = newDecodeDiskUsageState(topologyInfo, diskType) + } + } + if len(batches) > 1 { + fmt.Printf("ec decode batch %d/%d: %v\n", i+1, len(batches), batch) + } + for _, vid := range batch { + if err = doEcDecode(commandEnv, topologyInfo, *collection, vid, diskType, *checkMinFreeSpace, diskUsageState); err != nil { + return err + } } } diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go index 0d48aa06e..596af9eaf 100644 --- a/weed/shell/command_ec_encode.go +++ b/weed/shell/command_ec_encode.go @@ -46,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] [-batchSize=0] [-verbose] [-sourceDiskType=] [-diskType=] - ec.encode [-volumeId=|-volumeIds=,...] [-batchSize=0] [-verbose] [-diskType=] + ec.encode [-collection=""] [-fullPercent=95 -quietFor=1h] [-batchSize=10] [-verbose] [-sourceDiskType=] [-diskType=] + ec.encode [-volumeId=|-volumeIds=,...] [-batchSize=10] [-verbose] [-diskType=] This command will: 1. freeze one volume @@ -72,11 +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 + -batchSize: encode/rebalance/verify/delete this many volumes at a time (default 10; 0 = all in one batch) -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. + Each batch is committed independently. If a later batch fails, earlier batches + are already encoded and their original volumes deleted. Examples: # Encode SSD volumes to SSD EC shards (same tier) @@ -108,7 +108,7 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr 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") + batchSize := encodeCommand.Int("batchSize", DefaultEcBatchSize, "encode/re-balance/verify/delete up to this many volumes at a time (0 = all in one batch)") 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)") @@ -182,19 +182,19 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr return fmt.Errorf("-batchSize must be >= 0") } - batches := chunkEcEncodeVolumeIds(volumeIds, *batchSize) - if *batchSize > 0 { + batches := chunkVolumeIds(volumeIds, *batchSize) + if len(batches) > 1 { 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 { + if len(batches) > 1 { 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 { + if len(batches) > 1 { fmt.Printf("Successfully completed EC encoding for %d volumes in %d batch(es)\n", len(volumeIds), len(batches)) } @@ -228,7 +228,7 @@ func parseEcEncodeVolumeIds(volumeIdsStr string) ([]needle.VolumeId, error) { return volumeIds, nil } -func chunkEcEncodeVolumeIds(volumeIds []needle.VolumeId, batchSize int) [][]needle.VolumeId { +func chunkVolumeIds(volumeIds []needle.VolumeId, batchSize int) [][]needle.VolumeId { if batchSize <= 0 || len(volumeIds) == 0 { return [][]needle.VolumeId{volumeIds} } diff --git a/weed/shell/command_ec_encode_test.go b/weed/shell/command_ec_encode_test.go index 00672cc99..a7760463e 100644 --- a/weed/shell/command_ec_encode_test.go +++ b/weed/shell/command_ec_encode_test.go @@ -262,16 +262,16 @@ func TestParseEcEncodeVolumeIds(t *testing.T) { assert.Error(t, err) } -func TestChunkEcEncodeVolumeIds(t *testing.T) { +func TestChunkVolumeIds(t *testing.T) { vids := []needle.VolumeId{101, 102, 103, 104, 105} assert.Equal(t, [][]needle.VolumeId{ {101, 102}, {103, 104}, {105}, - }, chunkEcEncodeVolumeIds(vids, 2)) + }, chunkVolumeIds(vids, 2)) - assert.Equal(t, [][]needle.VolumeId{vids}, chunkEcEncodeVolumeIds(vids, 0)) + assert.Equal(t, [][]needle.VolumeId{vids}, chunkVolumeIds(vids, 0)) } func ecShardVisibilityTestTopology(nodes ...*master_pb.DataNodeInfo) *master_pb.TopologyInfo { diff --git a/weed/shell/common.go b/weed/shell/common.go index 922cd5316..4357cdcde 100644 --- a/weed/shell/common.go +++ b/weed/shell/common.go @@ -9,6 +9,8 @@ import ( var ( // Default maximum parallelization/concurrency for commands supporting it. DefaultMaxParallelization = 10 + // Default number of volumes EC encode/decode process per batch. + DefaultEcBatchSize = 10 // CollectionDefault is the special keyword to match empty collection names. // Use "_default" to avoid collision with a literal collection named "default". CollectionDefault = "_default"