diff --git a/weed/ec/ec_rebuild.go b/weed/ec/ec_rebuild.go index ddb234995a..e9502cdf99 100644 --- a/weed/ec/ec_rebuild.go +++ b/weed/ec/ec_rebuild.go @@ -2,12 +2,14 @@ package ec import ( "context" + "errors" "fmt" "io" "slices" "sync" "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/operation/volume_move" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" @@ -26,6 +28,11 @@ type ecRebuilder struct { volumeIds []needle.VolumeId diskType types.DiskType + maxParallelization int + // corruptShards holds, per volume and shard, the nodes whose copy failed + // the checksum scrub. It is filled before any rebuild starts. + corruptShards map[needle.VolumeId]map[erasure_coding.ShardId][]*EcNode + ewg *util.ErrorWaitGroup ecNodesMu sync.Mutex } @@ -35,7 +42,9 @@ type ecRebuilder struct { // left unmounted by a missing .ecx index; such shards are invisible to the // master, so recovering them first avoids regenerating data that is actually // present. volumeIds, when non-empty, restricts the rebuild to those volumes. -func RebuildEcVolumes(env *Env, ecNodes []*EcNode, writer io.Writer, collections []string, volumeIds []needle.VolumeId, diskType types.DiskType, maxParallelization int, applyChanges bool) error { +// verifyChecksums scrubs every shard against its checksum sidecar first, and +// rebuilds shards whose copies fail it even in volumes with every shard present. +func RebuildEcVolumes(env *Env, ecNodes []*EcNode, writer io.Writer, collections []string, volumeIds []needle.VolumeId, diskType types.DiskType, maxParallelization int, applyChanges bool, verifyChecksums bool) error { erb := &ecRebuilder{ env: env, ecNodes: ecNodes, @@ -45,6 +54,8 @@ func RebuildEcVolumes(env *Env, ecNodes []*EcNode, writer io.Writer, collections volumeIds: volumeIds, diskType: diskType, + maxParallelization: maxParallelization, + ewg: util.NewErrorWaitGroup(maxParallelization), } @@ -53,6 +64,10 @@ func RebuildEcVolumes(env *Env, ecNodes []*EcNode, writer io.Writer, collections // volume as short or unrepairable even though its data is intact (issue #10104). erb.recoverMissingIndexes() + if verifyChecksums { + erb.findCorruptShards(collections) + } + fmt.Printf("rebuildEcVolumes for %d collection(s)\n", len(collections)) for _, c := range collections { erb.rebuildEcVolumes(c) @@ -162,6 +177,8 @@ func (erb *ecRebuilder) rebuildEcVolumes(collection string) { } erb.ecNodesMu.Unlock() + erb.dropCorruptShards(collection, ecShardMap) + for vid, locations := range ecShardMap { if !erb.matchesVolumeId(vid) { continue @@ -247,6 +264,125 @@ func (erb *ecRebuilder) recoverMissingIndexes() { erb.ecNodesMu.Unlock() } +// findCorruptShards checksum-scrubs each node's shards of the targeted volumes +// and records the copies that fail their checksum sidecar. A node whose scrub +// fails fails the run, but its copies are taken as they are so the rest of the +// cluster is still repaired. +func (erb *ecRebuilder) findCorruptShards(collections []string) { + erb.ecNodesMu.Lock() + nodes := append([]*EcNode(nil), erb.ecNodes...) + erb.ecNodesMu.Unlock() + + erb.corruptShards = make(map[needle.VolumeId]map[erasure_coding.ShardId][]*EcNode) + var mu sync.Mutex + ewg := util.NewErrorWaitGroup(erb.maxParallelization) + for _, node := range nodes { + var vids []uint32 + for _, diskInfo := range node.Info.DiskInfos { + for _, ecShardInfo := range diskInfo.EcShardInfos { + if slices.Contains(collections, ecShardInfo.Collection) && erb.matchesVolumeId(needle.VolumeId(ecShardInfo.Id)) && !slices.Contains(vids, ecShardInfo.Id) { + vids = append(vids, ecShardInfo.Id) + } + } + } + if len(vids) == 0 { + continue + } + ewg.Add(func() error { + var resp *volume_server_pb.ScrubEcVolumeResponse + err := operation.WithVolumeServerClient(false, pb.NewServerAddressFromDataNode(node.Info), erb.env.GrpcDialOption, func(client volume_server_pb.VolumeServerClient) (scrubErr error) { + resp, scrubErr = client.ScrubEcVolume(context.Background(), &volume_server_pb.ScrubEcVolumeRequest{ + Mode: volume_server_pb.VolumeScrubMode_CHECKSUM, + VolumeIds: vids, + }) + return scrubErr + }) + mu.Lock() + defer mu.Unlock() + if err != nil { + erb.write("%s checksum scrub failed: %v\n", node.Info.Id, err) + erb.ewg.AddErrorf("checksum scrub on %s: %w", node.Info.Id, err) + return nil + } + for _, detail := range resp.GetDetails() { + erb.write("%s checksum scrub: %s\n", node.Info.Id, detail) + } + for _, shardInfo := range resp.GetBrokenShardInfos() { + vid, shardId := needle.VolumeId(shardInfo.VolumeId), erasure_coding.ShardId(shardInfo.ShardId) + erb.write("ec volume %d shard %d on %s fails its checksum\n", vid, shardId, node.Info.Id) + if erb.corruptShards[vid] == nil { + erb.corruptShards[vid] = make(map[erasure_coding.ShardId][]*EcNode) + } + erb.corruptShards[vid][shardId] = append(erb.corruptShards[vid][shardId], node) + } + return nil + }) + } + ewg.Wait() +} + +// dropCorruptShards takes the copies that failed the checksum scrub out of the +// rebuild plan: a shard left without a good copy counts as missing, so the +// rebuild regenerates it instead of skipping a volume that only looks complete. +// A failing copy of a shard another holder still serves is removed right away. +func (erb *ecRebuilder) dropCorruptShards(collection string, ecShardMap EcShardMap) { + for vid, byShard := range erb.corruptShards { + locations, found := ecShardMap[vid] + if !found { + continue + } + for shardId, corruptNodes := range byShard { + locations[shardId] = slices.DeleteFunc(locations[shardId], func(node *EcNode) bool { + return slices.Contains(corruptNodes, node) + }) + if len(locations[shardId]) > 0 { + if err := erb.removeCorruptCopies(collection, vid, nil, []erasure_coding.ShardId{shardId}); err != nil { + erb.ewg.AddErrorf("ec volume %d: %w", vid, err) + } + } + } + } +} + +// removeCorruptCopies unmounts and deletes the copies of shardIds that failed +// the checksum scrub on any node other than keep, which holds the good copy. +func (erb *ecRebuilder) removeCorruptCopies(collection string, volumeId needle.VolumeId, keep *EcNode, shardIds []erasure_coding.ShardId) error { + var nodes []*EcNode + byNode := make(map[*EcNode][]erasure_coding.ShardId) + for _, shardId := range shardIds { + for _, node := range erb.corruptShards[volumeId][shardId] { + if node == keep { + continue + } + if _, found := byNode[node]; !found { + nodes = append(nodes, node) + } + byNode[node] = append(byNode[node], shardId) + } + } + + var errs []error + for _, node := range nodes { + shardIds := byNode[node] + if !erb.applyChanges { + erb.write("would remove corrupt shards %d.%v from %s\n", volumeId, shardIds, node.Info.Id) + continue + } + if !erb.isLocked() { + return errors.Join(append(errs, fmt.Errorf("lock is lost"))...) + } + if err := volume_move.NewMover(erb.env.GrpcDialOption).RemoveEcShards(context.Background(), volumeId, collection, pb.NewServerAddressFromDataNode(node.Info), shardIds); err != nil { + errs = append(errs, fmt.Errorf("remove corrupt shards %d.%v from %s: %w", volumeId, shardIds, node.Info.Id, err)) + continue + } + erb.write("removed corrupt shards %d.%v from %s\n", volumeId, shardIds, node.Info.Id) + erb.ecNodesMu.Lock() + node.DeleteEcVolumeShards(volumeId, shardIds) + erb.ecNodesMu.Unlock() + } + return errors.Join(errs...) +} + func (erb *ecRebuilder) rebuildOneEcVolume(collection string, volumeId needle.VolumeId, locations EcShardLocations, rebuilder *EcNode) error { if !erb.isLocked() { return fmt.Errorf("lock is lost") @@ -286,7 +422,14 @@ func (erb *ecRebuilder) rebuildOneEcVolume(collection string, volumeId needle.Vo } if !erb.applyChanges { - return nil + var regenerated []erasure_coding.ShardId + for shardId := range erb.corruptShards[volumeId] { + if len(locations[shardId]) == 0 { + regenerated = append(regenerated, shardId) + } + } + slices.Sort(regenerated) + return erb.removeCorruptCopies(collection, volumeId, rebuilder, regenerated) } // generate ec shards, and maybe ecx file @@ -303,10 +446,10 @@ func (erb *ecRebuilder) rebuildOneEcVolume(collection string, volumeId needle.Vo // ensure ECNode updates are atomic erb.ecNodesMu.Lock() - defer erb.ecNodesMu.Unlock() rebuilder.AddEcVolumeShards(volumeId, collection, generatedShardIds, erb.diskType) + erb.ecNodesMu.Unlock() - return nil + return erb.removeCorruptCopies(collection, volumeId, rebuilder, generatedShardIds) } func (erb *ecRebuilder) generateMissingShards(collection string, volumeId needle.VolumeId, sourceLocation pb.ServerAddress) (rebuiltShardIds []erasure_coding.ShardId, err error) { diff --git a/weed/ec/ec_rebuild_checksum_test.go b/weed/ec/ec_rebuild_checksum_test.go new file mode 100644 index 0000000000..f0161adb59 --- /dev/null +++ b/weed/ec/ec_rebuild_checksum_test.go @@ -0,0 +1,313 @@ +package ec + +import ( + "bytes" + "context" + "fmt" + "net" + "slices" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// ecRebuildFakeServer adds the checksum scrub and the rebuild to the balance +// fake: copied shards sit on disk unmounted, and a rebuild regenerates every +// shard that is neither mounted nor copied, plus the mounted ones that fail +// the checksum. +type ecRebuildFakeServer struct { + *ecBalanceFakeServer + corrupt map[uint32]bool // shard id -> fails the checksum + copied map[uint32]bool + failUnmount bool + failScrub bool + scrubs int + rebuilds int + copiedIds []uint32 + deleted []uint32 +} + +func (f *ecRebuildFakeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb.ScrubEcVolumeRequest) (*volume_server_pb.ScrubEcVolumeResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.scrubs++ + if f.failScrub { + return nil, fmt.Errorf("scrub refused") + } + resp := &volume_server_pb.ScrubEcVolumeResponse{} + for _, vid := range req.VolumeIds { + for sid := range f.shards[vid] { + if f.corrupt[sid] { + resp.BrokenShardInfos = append(resp.BrokenShardInfos, &volume_server_pb.EcShardInfo{VolumeId: vid, ShardId: sid}) + } + } + } + return resp, nil +} + +func (f *ecRebuildFakeServer) VolumeEcShardsCopy(ctx context.Context, req *volume_server_pb.VolumeEcShardsCopyRequest) (*volume_server_pb.VolumeEcShardsCopyResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + for _, sid := range req.ShardIds { + f.copied[sid] = true + f.copiedIds = append(f.copiedIds, sid) + } + return &volume_server_pb.VolumeEcShardsCopyResponse{}, nil +} + +func (f *ecRebuildFakeServer) VolumeEcShardsRebuild(ctx context.Context, req *volume_server_pb.VolumeEcShardsRebuildRequest) (*volume_server_pb.VolumeEcShardsRebuildResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.rebuilds++ + resp := &volume_server_pb.VolumeEcShardsRebuildResponse{} + for sid := uint32(0); sid < erasure_coding.TotalShardsCount; sid++ { + mounted := f.shards[req.VolumeId][sid] + if (mounted || f.copied[sid]) && !f.corrupt[sid] { + continue + } + delete(f.corrupt, sid) + resp.RebuiltShardIds = append(resp.RebuiltShardIds, sid) + } + return resp, nil +} + +func (f *ecRebuildFakeServer) VolumeEcShardsUnmount(ctx context.Context, req *volume_server_pb.VolumeEcShardsUnmountRequest) (*volume_server_pb.VolumeEcShardsUnmountResponse, error) { + if f.failUnmount { + return nil, fmt.Errorf("unmount refused") + } + return &volume_server_pb.VolumeEcShardsUnmountResponse{}, nil +} + +func (f *ecRebuildFakeServer) VolumeEcShardsDelete(ctx context.Context, req *volume_server_pb.VolumeEcShardsDeleteRequest) (*volume_server_pb.VolumeEcShardsDeleteResponse, error) { + f.mu.Lock() + defer f.mu.Unlock() + for _, sid := range req.ShardIds { + delete(f.shards[req.VolumeId], sid) + delete(f.copied, sid) + f.deleted = append(f.deleted, sid) + } + return &volume_server_pb.VolumeEcShardsDeleteResponse{}, nil +} + +func startEcRebuildFakeServer(t *testing.T, vid uint32, present []uint32, corrupt ...uint32) *ecRebuildFakeServer { + t.Helper() + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + f := &ecRebuildFakeServer{ + ecBalanceFakeServer: &ecBalanceFakeServer{ + shards: map[uint32]map[uint32]bool{vid: {}}, + grpcAddr: listener.Addr().String(), + }, + corrupt: map[uint32]bool{}, + copied: map[uint32]bool{}, + } + for _, sid := range present { + f.shards[vid][sid] = true + } + for _, sid := range corrupt { + f.corrupt[sid] = true + } + server := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(server, f) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + return f +} + +func shardRange(from, to uint32) (ids []uint32) { + for sid := from; sid < to; sid++ { + ids = append(ids, sid) + } + return ids +} + +func runChecksumRebuild(t *testing.T, apply, verifyChecksums bool, servers ...*ecRebuildFakeServer) string { + t.Helper() + out, err := runChecksumRebuildLocked(t, apply, verifyChecksums, nil, servers...) + if err != nil { + t.Fatalf("rebuild: %v\n%s", err, out) + } + return out +} + +func runChecksumRebuildLocked(t *testing.T, apply, verifyChecksums bool, isLocked func() bool, servers ...*ecRebuildFakeServer) (string, error) { + t.Helper() + var balanceServers []*ecBalanceFakeServer + for _, f := range servers { + balanceServers = append(balanceServers, f.ecBalanceFakeServer) + } + env := &Env{ + GrpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + FetchTopology: fetchEcBalanceTopology(balanceServers), + IsLocked: isLocked, + } + ecNodes, _, err := CollectEcNodes(env, types.HardDriveType) + if err != nil { + t.Fatalf("collect ec nodes: %v", err) + } + var out bytes.Buffer + err = RebuildEcVolumes(env, ecNodes, &out, []string{"c1"}, nil, types.HardDriveType, 1, apply, verifyChecksums) + return out.String(), err +} + +func assertShards(t *testing.T, name string, f *ecRebuildFakeServer, vid uint32, want []uint32) { + t.Helper() + f.mu.Lock() + defer f.mu.Unlock() + var got []uint32 + for sid := range f.shards[vid] { + got = append(got, sid) + } + slices.Sort(got) + if !slices.Equal(got, want) { + t.Errorf("%s holds shards %v, want %v", name, got, want) + } +} + +// A complete volume whose shard fails the checksum on a node other than the +// rebuilder: the shard is regenerated on the rebuilder, never copied from the +// failing node, and the failing copy is removed afterwards. +func TestRebuildChecksumReplacesCorruptShardOfCompleteVolume(t *testing.T) { + const vid = uint32(1) + holder := startEcRebuildFakeServer(t, vid, shardRange(0, 9), 3) + rebuilder := startEcRebuildFakeServer(t, vid, shardRange(9, 14)) + + out := runChecksumRebuild(t, true, true, holder, rebuilder) + + if rebuilder.rebuilds != 1 || holder.rebuilds != 0 { + t.Fatalf("rebuilds: rebuilder %d, holder %d; want 1 and 0\n%s", rebuilder.rebuilds, holder.rebuilds, out) + } + if slices.Contains(rebuilder.copiedIds, 3) { + t.Errorf("the corrupt shard was copied to the rebuilder: %v", rebuilder.copiedIds) + } + assertShards(t, "rebuilder", rebuilder, vid, append([]uint32{3}, shardRange(9, 14)...)) + assertShards(t, "holder", holder, vid, []uint32{0, 1, 2, 4, 5, 6, 7, 8}) + if !strings.Contains(out, "removed corrupt shards 1.[3]") { + t.Errorf("expected the corrupt copy removal in the output:\n%s", out) + } +} + +// A corrupt shard on the rebuilder itself is regenerated in place and kept. +func TestRebuildChecksumRegeneratesCorruptShardOnRebuilder(t *testing.T) { + const vid = uint32(1) + holder := startEcRebuildFakeServer(t, vid, shardRange(0, 9)) + rebuilder := startEcRebuildFakeServer(t, vid, shardRange(9, 14), 11) + + out := runChecksumRebuild(t, true, true, holder, rebuilder) + + if rebuilder.rebuilds != 1 { + t.Fatalf("rebuilder ran %d rebuilds, want 1\n%s", rebuilder.rebuilds, out) + } + if slices.Contains(rebuilder.deleted, 11) || holder.deleted != nil { + t.Errorf("deleted: rebuilder %v, holder %v; the regenerated shard must stay", rebuilder.deleted, holder.deleted) + } + assertShards(t, "rebuilder", rebuilder, vid, shardRange(9, 14)) + assertShards(t, "holder", holder, vid, shardRange(0, 9)) +} + +// A corrupt copy of a shard another holder serves intact needs no rebuild: +// only the corrupt copy is removed. +func TestRebuildChecksumRemovesCorruptDuplicate(t *testing.T) { + const vid = uint32(1) + holder := startEcRebuildFakeServer(t, vid, shardRange(0, 9), 3) + other := startEcRebuildFakeServer(t, vid, append([]uint32{3}, shardRange(9, 14)...)) + + out := runChecksumRebuild(t, true, true, holder, other) + + if holder.rebuilds+other.rebuilds != 0 { + t.Fatalf("no rebuild expected, got %d\n%s", holder.rebuilds+other.rebuilds, out) + } + assertShards(t, "holder", holder, vid, []uint32{0, 1, 2, 4, 5, 6, 7, 8}) + assertShards(t, "other", other, vid, append([]uint32{3}, shardRange(9, 14)...)) +} + +// Dry-run scrubs and reports, but copies, rebuilds and removes nothing. +func TestRebuildChecksumDryRun(t *testing.T) { + const vid = uint32(1) + holder := startEcRebuildFakeServer(t, vid, shardRange(0, 9), 3) + rebuilder := startEcRebuildFakeServer(t, vid, shardRange(9, 14)) + + out := runChecksumRebuild(t, false, true, holder, rebuilder) + + if holder.scrubs != 1 || rebuilder.scrubs != 1 { + t.Errorf("scrubs: holder %d, rebuilder %d; want 1 each", holder.scrubs, rebuilder.scrubs) + } + if rebuilder.rebuilds != 0 || rebuilder.copiedIds != nil || holder.deleted != nil { + t.Errorf("dry run changed state: rebuilds %d, copied %v, deleted %v", rebuilder.rebuilds, rebuilder.copiedIds, holder.deleted) + } + for _, want := range []string{ + "ec volume 1 shard 3 on " + holder.nodeId() + " fails its checksum", + "missing shard 1.3", + "would remove corrupt shards 1.[3] from " + holder.nodeId(), + } { + if !strings.Contains(out, want) { + t.Errorf("expected %q in the output:\n%s", want, out) + } + } +} + +// Without -checksum a complete volume is neither scrubbed nor rebuilt. +func TestRebuildWithoutChecksumSkipsCompleteVolume(t *testing.T) { + const vid = uint32(1) + holder := startEcRebuildFakeServer(t, vid, shardRange(0, 9), 3) + rebuilder := startEcRebuildFakeServer(t, vid, shardRange(9, 14)) + + runChecksumRebuild(t, true, false, holder, rebuilder) + + if holder.scrubs+rebuilder.scrubs != 0 || holder.rebuilds+rebuilder.rebuilds != 0 { + t.Errorf("scrubs %d, rebuilds %d; want none", holder.scrubs+rebuilder.scrubs, holder.rebuilds+rebuilder.rebuilds) + } + assertShards(t, "holder", holder, vid, shardRange(0, 9)) +} + +// The removal of a corrupt duplicate is the whole repair of that volume, so a +// failed removal fails the run, and nothing is removed once the lock is lost. +func TestRebuildChecksumReportsDuplicateRemovalFailures(t *testing.T) { + const vid = uint32(1) + t.Run("unmount fails", func(t *testing.T) { + holder := startEcRebuildFakeServer(t, vid, shardRange(0, 9), 3) + other := startEcRebuildFakeServer(t, vid, append([]uint32{3}, shardRange(9, 14)...)) + holder.failUnmount = true + + out, err := runChecksumRebuildLocked(t, true, true, nil, holder, other) + if err == nil || !strings.Contains(err.Error(), "unmount refused") { + t.Fatalf("expected the failed removal to fail the run, got %v\n%s", err, out) + } + assertShards(t, "holder", holder, vid, shardRange(0, 9)) + }) + t.Run("lock lost", func(t *testing.T) { + holder := startEcRebuildFakeServer(t, vid, shardRange(0, 9), 3) + other := startEcRebuildFakeServer(t, vid, append([]uint32{3}, shardRange(9, 14)...)) + + out, err := runChecksumRebuildLocked(t, true, true, func() bool { return false }, holder, other) + if err == nil || !strings.Contains(err.Error(), "lock is lost") { + t.Fatalf("expected a lost lock error, got %v\n%s", err, out) + } + if holder.deleted != nil { + t.Errorf("removed %v without the lock", holder.deleted) + } + }) +} + +// A node whose scrub fails fails the run, and the corruption found on the +// other nodes is still repaired. +func TestRebuildChecksumScrubFailureFailsRun(t *testing.T) { + const vid = uint32(1) + holder := startEcRebuildFakeServer(t, vid, shardRange(0, 9), 3) + rebuilder := startEcRebuildFakeServer(t, vid, shardRange(9, 14)) + rebuilder.failScrub = true + + out, err := runChecksumRebuildLocked(t, true, true, nil, holder, rebuilder) + if err == nil || !strings.Contains(err.Error(), "scrub refused") { + t.Fatalf("expected the failed scrub to fail the run, got %v\n%s", err, out) + } + assertShards(t, "rebuilder", rebuilder, vid, append([]uint32{3}, shardRange(9, 14)...)) + assertShards(t, "holder", holder, vid, []uint32{0, 1, 2, 4, 5, 6, 7, 8}) +} diff --git a/weed/shell/command_ec_rebuild.go b/weed/shell/command_ec_rebuild.go index 51f810a56a..d21946e849 100644 --- a/weed/shell/command_ec_rebuild.go +++ b/weed/shell/command_ec_rebuild.go @@ -27,7 +27,7 @@ func (c *commandEcRebuild) Name() string { func (c *commandEcRebuild) Help() string { return `find and rebuild missing ec shards among volume servers - ec.rebuild [-c EACH_COLLECTION|] [-apply] [-maxParallelization N] [-diskType=] + ec.rebuild [-c EACH_COLLECTION|] [-apply] [-maxParallelization N] [-diskType=] [-checksum] Before rebuilding, asks volume servers to recover any shards left unmounted by a missing .ecx index (the index resides only on a peer server). Such shards are @@ -41,6 +41,11 @@ func (c *commandEcRebuild) Help() string { Increase for faster rebuilds with more system resources. Decrease if experiencing resource contention or instability. -diskType: disk type for EC shards (hdd, ssd, or empty for default hdd) + -checksum: first verify every shard against its checksum sidecar (as + ec.scrub -mode=checksum does), and rebuild the shards that fail + it, also in volumes that have every shard. A failing copy is + removed once a good copy of its shard is mounted elsewhere. + This reads every shard in full. Algorithm: @@ -79,6 +84,7 @@ func (c *commandEcRebuild) Do(args []string, commandEnv *CommandEnv, writer io.W maxParallelization := fixCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") applyChanges := fixCommand.Bool("apply", false, "apply the changes") diskTypeStr := fixCommand.String("diskType", "", "disk type for EC shards (hdd, ssd, or empty for default hdd)") + verifyChecksums := fixCommand.Bool("checksum", false, "verify every shard against its checksum sidecar first, and rebuild the shards that fail it, also in volumes that have every shard") // TODO: remove this alias applyChangesAlias := fixCommand.Bool("force", false, "apply the changes (alias for -apply)") if err = fixCommand.Parse(args); err != nil { @@ -119,5 +125,5 @@ func (c *commandEcRebuild) Do(args []string, commandEnv *CommandEnv, writer io.W } } - return ec.RebuildEcVolumes(commandEnv.ecEnv(), allEcNodes, writer, collections, volumeIds, diskType, *maxParallelization, *applyChanges) + return ec.RebuildEcVolumes(commandEnv.ecEnv(), allEcNodes, writer, collections, volumeIds, diskType, *maxParallelization, *applyChanges, *verifyChecksums) }