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"