diff --git a/seaweed-volume/proto/volume_server.proto b/seaweed-volume/proto/volume_server.proto index 2caf8ebf1..6b64ef529 100644 --- a/seaweed-volume/proto/volume_server.proto +++ b/seaweed-volume/proto/volume_server.proto @@ -714,6 +714,7 @@ enum VolumeScrubMode { FULL = 2; LOCAL = 3; CHECKSUM = 4; // EC only: verify each local shard's raw bytes against the bitrot checksum sidecar + READS = 5; // like FULL, but EC intervals no shard can serve are reconstructed from parity } message ScrubVolumeRequest { diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 791eb2aba..c6d9bb7cd 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -38,6 +38,7 @@ fn scrub_mode_label(mode: i32) -> &'static str { 2 => "FULL", 3 => "LOCAL", 4 => "CHECKSUM", + 5 => "READS", _ => "UNKNOWN", } } @@ -4097,7 +4098,7 @@ impl VolumeServer for VolumeGrpcService { // Validate mode let mode = req.mode; match mode { - 1 | 2 | 3 => {} // INDEX=1, FULL=2, LOCAL=3 + 1 | 2 | 3 | 5 => {} // INDEX=1, FULL=2, LOCAL=3, READS=5 (FULL for regular volumes) _ => { return Err(Status::invalid_argument(format!( "unsupported volume scrub mode {}", @@ -4197,7 +4198,7 @@ impl VolumeServer for VolumeGrpcService { // Validate mode let mode = req.mode; match mode { - 1 | 2 | 3 | 4 => {} // INDEX=1, FULL=2, LOCAL=3, CHECKSUM=4 + 1 | 2 | 3 | 4 | 5 => {} // INDEX=1, FULL=2, LOCAL=3, CHECKSUM=4, READS=5 _ => { return Err(Status::invalid_argument(format!( "unsupported EC volume scrub mode {}", @@ -4206,6 +4207,14 @@ impl VolumeServer for VolumeGrpcService { } } + // Only the modes that walk needles can be strict about deleted ones. + let force_deleted_needles_check = req.force_deleted_needles_check; + if force_deleted_needles_check && mode != 2 && mode != 5 { + return Err(Status::invalid_argument( + "deleted needle checks are only supported for FULL and READS scrubs", + )); + } + // Collect the volume ids under a brief lock, then release it: FULL (mode 2) // reads remote shards and must not hold the !Send store guard across .await. let vids: Vec = { @@ -4247,8 +4256,8 @@ impl VolumeServer for VolumeGrpcService { } } } - 2 => { - // FULL: Go-parity per-needle local+remote walk, PLUS a TEMPORARY + 2 | 5 => { + // FULL/READS: Go-parity per-needle local+remote walk, PLUS a TEMPORARY // local Reed-Solomon parity check. The needle walk only reads // DATA-shard intervals of LIVE needles, so on its own it can't // catch silent bitrot in a PARITY shard or an unwalked cold @@ -4280,8 +4289,13 @@ impl VolumeServer for VolumeGrpcService { // (1) Per-needle local+remote walk (Go ScrubEcVolume parity). let (files, mut shard_infos, mut errs) = - crate::server::store_ec::scrub_ec_volume_distributed(&self.state, vid, false) - .await; + crate::server::store_ec::scrub_ec_volume_distributed( + &self.state, + vid, + force_deleted_needles_check, + mode == 5, + ) + .await; total_files += files as u64; // count comes from the needle walk only // (2) Local parity check, gated on all-shards-local. Blocking RS diff --git a/seaweed-volume/src/server/store_ec.rs b/seaweed-volume/src/server/store_ec.rs index 4b1e785bf..82277c500 100644 --- a/seaweed-volume/src/server/store_ec.rs +++ b/seaweed-volume/src/server/store_ec.rs @@ -246,7 +246,9 @@ pub async fn read_ec_shard_needle_distributed( /// without decoding (so genuine shard faults are reported rather than healed). /// Mirrors Go's `Store.ScrubEcVolume`. Returns (rows walked, broken shards, /// errors). `force_deleted_needles_check` disables the benign delete-state -/// size-mismatch suppression. +/// size-mismatch suppression. `recover_unreadable` (READS mode) rebuilds an +/// unreadable interval from the surviving shards: the same shards are reported +/// broken, but only needles parity can no longer recover become errors. /// /// Shard locations are refreshed once up front. Each needle is then processed via /// `scrub_snapshot_under_lock` + lock-drop + no-reconstruct `read_remote_ec_shard_interval`, @@ -255,6 +257,7 @@ pub async fn scrub_ec_volume_distributed( state: &Arc, vid: VolumeId, force_deleted_needles_check: bool, + recover_unreadable: bool, ) -> (i64, Vec, Vec) { // Phase A — under the Store read lock, run the index scrub and grab the // paths/scalars + shard-location staleness; release the lock before any await. @@ -395,8 +398,9 @@ pub async fn scrub_ec_volume_distributed( } }; - // Read each interval local-then-remote WITHOUT reconstructing: we verify - // the shards are valid, we do not heal them. Locations refreshed above. + // Read each interval local-then-remote. Neither read decodes: the point is to + // find shards that are themselves broken, not to heal around them. READS then + // rebuilds what it could not read. Locations refreshed above. let n_intervals = snapshot.intervals.len(); let mut data: Vec = Vec::with_capacity(snapshot.actual_size); for (i, res) in snapshot.intervals.iter().enumerate() { @@ -426,11 +430,9 @@ pub async fn scrub_ec_volume_distributed( // -> the delete-state suppression (mirrors Go's pre-zeroed buffer). Ok((_, true)) => data.resize(data.len() + *ssize, 0), Ok((buf, false)) => data.extend_from_slice(&buf), - Err(_) => { - errs.push(format!( - "failed to read EC shard {} for needle {} on volume {} (interval {}/{})", - shard_id, id.0, vid.0, i + 1, n_intervals - )); + Err(read_err) => { + // The shard is broken whether or not the needle survives it, + // so report it either way. broken_shards.insert( *shard_id, crate::pb::volume_server_pb::EcShardInfo { @@ -441,7 +443,41 @@ pub async fn scrub_ec_volume_distributed( ..Default::default() }, ); - break; + if !recover_unreadable { + errs.push(format!( + "failed to read EC shard {} for needle {} on volume {} (interval {}/{}): {}", + shard_id, id.0, vid.0, i + 1, n_intervals, read_err + )); + break; + } + match recover_one_remote_ec_shard_interval( + state, + vid, + id, + *shard_id, + *shard_offset, + *ssize, + &locations, + data_shards, + total_shards - data_shards, + snapshot.encode_ts_ns, + ) + .await + { + // Same as the direct read above: a holder reporting the + // needle deleted is authoritative and answers with no + // bytes, so zero-fill and let the delete-state + // suppression have it. + Ok((_, true)) => data.resize(data.len() + *ssize, 0), + Ok((buf, false)) => data.extend_from_slice(&buf), + Err(e) => { + errs.push(format!( + "failed to recover EC shard {} for needle {} on volume {} (interval {}/{}): {}", + shard_id, id.0, vid.0, i + 1, n_intervals, e + )); + break; + } + } } } } diff --git a/test/volume_server/grpc/scrub_integration_test.go b/test/volume_server/grpc/scrub_integration_test.go index 5b8da8af4..cd844f626 100644 --- a/test/volume_server/grpc/scrub_integration_test.go +++ b/test/volume_server/grpc/scrub_integration_test.go @@ -3,6 +3,7 @@ package volume_server_grpc_test import ( "context" "net/http" + "strings" "testing" "time" @@ -414,3 +415,87 @@ func TestScrubEcVolumeIndexCorruptEcx(t *testing.T) { t.Fatalf("expected broken volume after ECX corruption") } } + +// scrubEcUntilLocated runs an EC scrub, retrying while the server is still waiting on +// the master for the shard locations a distributed scrub needs. +func scrubEcUntilLocated(t *testing.T, ctx context.Context, grpcClient volume_server_pb.VolumeServerClient, volumeID uint32, mode volume_server_pb.VolumeScrubMode) *volume_server_pb.ScrubEcVolumeResponse { + t.Helper() + deadline := time.Now().Add(60 * time.Second) + for { + resp, err := grpcClient.ScrubEcVolume(ctx, &volume_server_pb.ScrubEcVolumeRequest{ + VolumeIds: []uint32{volumeID}, + Mode: mode, + }) + if err != nil { + t.Fatalf("ScrubEcVolume %s failed: %v", mode, err) + } + waiting := false + for _, d := range resp.GetDetails() { + if strings.Contains(d, "failed to locate shard via master grpc") { + waiting = true + break + } + } + if !waiting { + return resp + } + if time.Now().After(deadline) { + t.Fatalf("master never reported EC shard locations for volume %d: %v", volumeID, resp.GetDetails()) + } + time.Sleep(time.Second) + } +} + +// With a shard gone, FULL cannot read the intervals that lived on it, while READS +// rebuilds them from parity. Either way the missing shard has to be reported: a +// volume that scrubs clean is a volume nobody repairs. +func TestScrubEcVolumeReadsRecoversMissingShard(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + clusterHarness := framework.StartVolumeCluster(t, matrix.P1()) + conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress()) + defer conn.Close() + + const volumeID = uint32(216) + const missingShard = uint32(0) + httpClient := framework.NewHTTPClient() + ecSetup(t, grpcClient, httpClient, clusterHarness.VolumeAdminURL(), volumeID) + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + + if _, err := grpcClient.VolumeEcShardsUnmount(ctx, &volume_server_pb.VolumeEcShardsUnmountRequest{ + VolumeId: volumeID, + ShardIds: []uint32{missingShard}, + }); err != nil { + t.Fatalf("VolumeEcShardsUnmount shard %d failed: %v", missingShard, err) + } + if _, err := grpcClient.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{ + VolumeId: volumeID, + ShardIds: []uint32{missingShard}, + }); err != nil { + t.Fatalf("VolumeEcShardsDelete shard %d failed: %v", missingShard, err) + } + + fullResp := scrubEcUntilLocated(t, ctx, grpcClient, volumeID, volume_server_pb.VolumeScrubMode_FULL) + assertOnlyBrokenShard(t, "FULL", fullResp, missingShard) + if len(fullResp.GetDetails()) == 0 { + t.Fatalf("FULL should report the needles it could not read") + } + + readsResp := scrubEcUntilLocated(t, ctx, grpcClient, volumeID, volume_server_pb.VolumeScrubMode_READS) + assertOnlyBrokenShard(t, "READS", readsResp, missingShard) + if len(readsResp.GetDetails()) != 0 { + t.Fatalf("READS should rebuild every needle from parity, got: %v", readsResp.GetDetails()) + } +} + +func assertOnlyBrokenShard(t *testing.T, mode string, resp *volume_server_pb.ScrubEcVolumeResponse, shardID uint32) { + t.Helper() + infos := resp.GetBrokenShardInfos() + if len(infos) != 1 || infos[0].GetShardId() != shardID { + t.Fatalf("%s reported broken shards %v, want only shard %d (details: %v)", mode, infos, shardID, resp.GetDetails()) + } +} diff --git a/weed/pb/volume_server.proto b/weed/pb/volume_server.proto index 93215af0a..d44712c3f 100644 --- a/weed/pb/volume_server.proto +++ b/weed/pb/volume_server.proto @@ -718,6 +718,7 @@ enum VolumeScrubMode { FULL = 2; LOCAL = 3; CHECKSUM = 4; // EC only: verify each local shard's raw bytes against the bitrot checksum sidecar + READS = 5; // like FULL, but EC intervals no shard can serve are reconstructed from parity } message ScrubVolumeRequest { diff --git a/weed/pb/volume_server_pb/volume_server.pb.go b/weed/pb/volume_server_pb/volume_server.pb.go index 8159e8eaa..afe59bc08 100644 --- a/weed/pb/volume_server_pb/volume_server.pb.go +++ b/weed/pb/volume_server_pb/volume_server.pb.go @@ -76,6 +76,7 @@ const ( VolumeScrubMode_FULL VolumeScrubMode = 2 VolumeScrubMode_LOCAL VolumeScrubMode = 3 VolumeScrubMode_CHECKSUM VolumeScrubMode = 4 // EC only: verify each local shard's raw bytes against the bitrot checksum sidecar + VolumeScrubMode_READS VolumeScrubMode = 5 // like FULL, but EC intervals no shard can serve are reconstructed from parity ) // Enum value maps for VolumeScrubMode. @@ -86,6 +87,7 @@ var ( 2: "FULL", 3: "LOCAL", 4: "CHECKSUM", + 5: "READS", } VolumeScrubMode_value = map[string]int32{ "UNKNOWN": 0, @@ -93,6 +95,7 @@ var ( "FULL": 2, "LOCAL": 3, "CHECKSUM": 4, + "READS": 5, } ) @@ -7748,13 +7751,14 @@ const file_volume_server_proto_rawDesc = "" + "stopTimeNs*;\n" + "\x11ChecksumAlgorithm\x12\x11\n" + "\rCHECKSUM_NONE\x10\x00\x12\x13\n" + - "\x0fCHECKSUM_CRC32C\x10\x01*L\n" + + "\x0fCHECKSUM_CRC32C\x10\x01*W\n" + "\x0fVolumeScrubMode\x12\v\n" + "\aUNKNOWN\x10\x00\x12\t\n" + "\x05INDEX\x10\x01\x12\b\n" + "\x04FULL\x10\x02\x12\t\n" + "\x05LOCAL\x10\x03\x12\f\n" + - "\bCHECKSUM\x10\x042\xfa)\n" + + "\bCHECKSUM\x10\x04\x12\t\n" + + "\x05READS\x10\x052\xfa)\n" + "\fVolumeServer\x12\\\n" + "\vBatchDelete\x12$.volume_server_pb.BatchDeleteRequest\x1a%.volume_server_pb.BatchDeleteResponse\"\x00\x12n\n" + "\x11VacuumVolumeCheck\x12*.volume_server_pb.VacuumVolumeCheckRequest\x1a+.volume_server_pb.VacuumVolumeCheckResponse\"\x00\x12v\n" + diff --git a/weed/server/volume_grpc_scrub.go b/weed/server/volume_grpc_scrub.go index c2e7a3bcf..351159aa7 100644 --- a/weed/server/volume_grpc_scrub.go +++ b/weed/server/volume_grpc_scrub.go @@ -43,8 +43,9 @@ func (vs *VolumeServer) ScrubVolume(ctx context.Context, req *volume_server_pb.S switch m := req.GetMode(); m { case volume_server_pb.VolumeScrubMode_INDEX: files, serrs = v.ScrubIndex() - case volume_server_pb.VolumeScrubMode_LOCAL: - // LOCAL is equivalent to FULL for regular volumes + case volume_server_pb.VolumeScrubMode_LOCAL, volume_server_pb.VolumeScrubMode_READS: + // both are equivalent to FULL for regular volumes: there are no shards to + // stay local to, and nothing to reconstruct from fallthrough case volume_server_pb.VolumeScrubMode_FULL: files, serrs = v.Scrub() @@ -96,8 +97,9 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb if err := vs.checkGrpcAdminAuth(ctx); err != nil { return nil, err } - if req.GetForceDeletedNeedlesCheck() && req.GetMode() != volume_server_pb.VolumeScrubMode_FULL { - return nil, fmt.Errorf("deleted needle checks are only supported for FULL scrubs") + if m := req.GetMode(); req.GetForceDeletedNeedlesCheck() && + m != volume_server_pb.VolumeScrubMode_FULL && m != volume_server_pb.VolumeScrubMode_READS { + return nil, fmt.Errorf("deleted needle checks are only supported for FULL and READS scrubs") } vids := []needle.VolumeId{} @@ -130,8 +132,8 @@ func (vs *VolumeServer) ScrubEcVolume(ctx context.Context, req *volume_server_pb files, serrs = v.ScrubIndex() case volume_server_pb.VolumeScrubMode_LOCAL: files, shardInfos, serrs = v.ScrubLocal() - case volume_server_pb.VolumeScrubMode_FULL: - files, shardInfos, serrs = vs.store.ScrubEcVolume(v.VolumeId, req.GetForceDeletedNeedlesCheck()) + case volume_server_pb.VolumeScrubMode_FULL, volume_server_pb.VolumeScrubMode_READS: + files, shardInfos, serrs = vs.store.ScrubEcVolume(v.VolumeId, m, req.GetForceDeletedNeedlesCheck()) case volume_server_pb.VolumeScrubMode_CHECKSUM: // Verify each local shard's raw bytes against the bitrot sidecar, // exercising cold parity shards. Read-only. ChecksumScrub's first diff --git a/weed/shell/command_ec_scrub.go b/weed/shell/command_ec_scrub.go index 349bacc84..13414fba3 100644 --- a/weed/shell/command_ec_scrub.go +++ b/weed/shell/command_ec_scrub.go @@ -41,10 +41,10 @@ func (c *commandEcVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer volScrubCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) nodesStr := volScrubCommand.String("node", "", "comma-separated list of volume server : (optional)") volumeIDsStr := volScrubCommand.String("volumeId", "", "comma-separated EC volume IDs to process (optional)") - mode := volScrubCommand.String("mode", "local", "scrubbing mode (index/local/full/checksum)") + mode := volScrubCommand.String("mode", "local", "scrubbing mode (index/local/full/reads/checksum)") maxParallelization := volScrubCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") showDetails := volScrubCommand.Bool("details", false, "display scrub result details, if available") - forceDeletedNeedlesCheck := volScrubCommand.Bool("forceDeletedNeedlesCheck", false, "force strict verification of deleted needles (full mode only); may report false positives when EC indexes disagree") + forceDeletedNeedlesCheck := volScrubCommand.Bool("forceDeletedNeedlesCheck", false, "force strict verification of deleted needles (full and reads modes only); may report false positives when EC indexes disagree") if err = volScrubCommand.Parse(args); err != nil { return err @@ -97,12 +97,15 @@ func (c *commandEcVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer scrubMode = volume_server_pb.VolumeScrubMode_FULL case "CHECKSUM": scrubMode = volume_server_pb.VolumeScrubMode_CHECKSUM + case "READS": + scrubMode = volume_server_pb.VolumeScrubMode_READS default: return fmt.Errorf("unsupported scrubbing mode %q", *mode) } fmt.Fprintf(writer, "using %s mode\n", scrubMode.String()) - if *forceDeletedNeedlesCheck && scrubMode != volume_server_pb.VolumeScrubMode_FULL { - return fmt.Errorf("deleted needle checks are only supported for FULL scrubs") + if *forceDeletedNeedlesCheck && + scrubMode != volume_server_pb.VolumeScrubMode_FULL && scrubMode != volume_server_pb.VolumeScrubMode_READS { + return fmt.Errorf("deleted needle checks are only supported for FULL and READS scrubs") } return ec.ScrubEcVolumes(commandEnv.ecEnv(), writer, volumeServerAddrs, volumeIDs, scrubMode, *forceDeletedNeedlesCheck, *maxParallelization, *showDetails) diff --git a/weed/shell/command_volume_scrub.go b/weed/shell/command_volume_scrub.go index dfac14294..95afd3a32 100644 --- a/weed/shell/command_volume_scrub.go +++ b/weed/shell/command_volume_scrub.go @@ -50,7 +50,7 @@ func (c *commandVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer io volScrubCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError) nodesStr := volScrubCommand.String("node", "", "comma-separated list of volume server : (optional)") volumeIDsStr := volScrubCommand.String("volumeId", "", "comma-separated volume IDs to process (optional)") - mode := volScrubCommand.String("mode", "full", "scrubbing mode (index/local/full)") + mode := volScrubCommand.String("mode", "full", "scrubbing mode (index/local/full/reads)") markBrokenReadonly := volScrubCommand.Bool("markBrokenReadonly", false, "whether to flag volumes with scrub failures as read-only") maxParallelization := volScrubCommand.Int("maxParallelization", DefaultMaxParallelization, "run up to X tasks in parallel, whenever possible") showDetails := volScrubCommand.Bool("details", false, "display scrub result details, if available") @@ -99,6 +99,9 @@ func (c *commandVolumeScrub) Do(args []string, commandEnv *CommandEnv, writer io c.mode = volume_server_pb.VolumeScrubMode_LOCAL case "FULL": c.mode = volume_server_pb.VolumeScrubMode_FULL + case "READS": + // only meaningful for EC volumes; accepted here so one mode name works on both commands + c.mode = volume_server_pb.VolumeScrubMode_READS default: return fmt.Errorf("unsupported scrubbing mode %q", *mode) } diff --git a/weed/storage/store_ec_interval_read_test.go b/weed/storage/store_ec_interval_read_test.go index dbab475ee..8dba1bb03 100644 --- a/weed/storage/store_ec_interval_read_test.go +++ b/weed/storage/store_ec_interval_read_test.go @@ -2,90 +2,17 @@ package storage import ( "bytes" - "math/rand" - "os" "testing" - "time" - "github.com/seaweedfs/seaweedfs/weed/pb" - "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/types" - "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" ) // A needle larger than one EC block is split over consecutive blocks, which // live on different shards. The intervals are read concurrently, so check they // still come back in order. func TestReadEcShardNeedleSpanningBlocks(t *testing.T) { - store := newTestStore(t, 1) - dir := store.Locations[0].Directory const vid = needle.VolumeId(7) - - v, err := NewVolume(dir, dir, "", vid, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) - if err != nil { - t.Fatalf("new volume: %v", err) - } - n := new(needle.Needle) - n.Id = types.Uint64ToNeedleId(42) - n.Data = make([]byte, 3*erasure_coding.ErasureCodingSmallBlockSize+1234) - rand.New(rand.NewSource(42)).Read(n.Data) - n.Checksum = needle.NewCRC(n.Data) - if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil { - t.Fatalf("write needle: %v", err) - } - baseFileName := v.DataFileName() - v.Close() - - datSize, err := os.Stat(baseFileName + ".dat") - if err != nil { - t.Fatalf("stat .dat: %v", err) - } - if _, err := erasure_coding.WriteEcFiles(baseFileName, erasure_coding.BackgroundECContext()); err != nil { - t.Fatalf("write ec files: %v", err) - } - if err := erasure_coding.WriteSortedFileFromIdx(baseFileName, ".ecx"); err != nil { - t.Fatalf("write .ecx: %v", err) - } - if err := os.WriteFile(baseFileName+".ecj", nil, 0o644); err != nil { - t.Fatalf("write .ecj: %v", err) - } - if err := volume_info.SaveVolumeInfo(baseFileName+".vif", &volume_server_pb.VolumeInfo{ - Version: uint32(needle.GetCurrentVersion()), - DatFileSize: datSize.Size(), - EcShardConfig: &volume_server_pb.EcShardConfig{ - DataShards: erasure_coding.DataShardsCount, - ParityShards: erasure_coding.ParityShardsCount, - }, - }); err != nil { - t.Fatalf("save .vif: %v", err) - } - for _, ext := range []string{".dat", ".idx"} { - if err := os.Remove(baseFileName + ext); err != nil { - t.Fatalf("remove %s: %v", ext, err) - } - } - - for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ { - if err := store.MountEcShards("", vid, erasure_coding.ShardId(shardId), ""); err != nil { - t.Fatalf("mount shard %d: %v", shardId, err) - } - } - ecVolume, found := store.Locations[0].FindEcVolume(vid) - if !found { - t.Fatal("ec volume not mounted") - } - - // Every shard is local, so seed the location cache to keep the read off the - // master this test does not have. - ecVolume.ShardLocationsLock.Lock() - for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ { - ecVolume.ShardLocations[erasure_coding.ShardId(shardId)] = []pb.ServerAddress{"localhost:8080"} - } - ecVolume.ShardLocationsRefreshTime = time.Now() - ecVolume.ShardLocationsLock.Unlock() + store, ecVolume, n := newLocalEcVolume(t, vid) _, _, intervals, err := ecVolume.LocateEcShardNeedle(n.Id, ecVolume.Version) if err != nil { diff --git a/weed/storage/store_ec_scrub.go b/weed/storage/store_ec_scrub.go index a1aa52922..08c002682 100644 --- a/weed/storage/store_ec_scrub.go +++ b/weed/storage/store_ec_scrub.go @@ -13,7 +13,11 @@ import ( // ScrubEcVolume checks the full integrity of a EC volume, across both local and remote shards. // Returns a count of processed file entries, slice of found broken shards, and slice of found errors. -func (s *Store) ScrubEcVolume(vid needle.VolumeId, forceDeletedNeedlesCheck bool) (int64, []*volume_server_pb.EcShardInfo, []error) { +// +// FULL reports an unreadable shard and gives up on the needle. READS additionally rebuilds the +// interval from the surviving shards, so it reports the same broken shards but only errors on +// needles that parity can no longer recover - which is what exercises the parity data. +func (s *Store) ScrubEcVolume(vid needle.VolumeId, mode volume_server_pb.VolumeScrubMode, forceDeletedNeedlesCheck bool) (int64, []*volume_server_pb.EcShardInfo, []error) { ecv, found := s.FindEcVolume(vid) if !found { return 0, nil, []error{fmt.Errorf("EC volume id %d not found", vid)} @@ -25,6 +29,8 @@ func (s *Store) ScrubEcVolume(vid needle.VolumeId, forceDeletedNeedlesCheck bool // full scan means verifying indexes as well _, errs := ecv.ScrubIndex() + recoverUnreadable := mode == volume_server_pb.VolumeScrubMode_READS + var count int64 // reads for EC chunks can hit the same shard multiple times, so dedupe upon read errors brokenShardsMap := map[erasure_coding.ShardId]*volume_server_pb.EcShardInfo{} @@ -56,27 +62,39 @@ func (s *Store) ScrubEcVolume(vid needle.VolumeId, forceDeletedNeedlesCheck bool continue } - // ...then remote. note we do not try to recover EC-encoded data upon read failures; - // we want check that shards are valid without decoding + // ...then remote. neither read decodes: the point is to find shards that are + // themselves broken, not to heal around them ecv.ShardLocationsLock.RLock() sourceDataNodes, ok := ecv.ShardLocations[shardId] ecv.ShardLocationsLock.RUnlock() + readErr := errors.New("no known shard locations") if ok { - if _, _, err := s.readRemoteEcShardInterval(sourceDataNodes, id, ecv.VolumeId, shardId, chunk, offset, ecv.EncodeTsNs); err == nil { + if _, _, readErr = s.readRemoteEcShardInterval(sourceDataNodes, id, ecv.VolumeId, shardId, chunk, offset, ecv.EncodeTsNs); readErr == nil { data = append(data, chunk...) continue } } - // chunk read for shard failed :( - errs = append(errs, fmt.Errorf("failed to read EC shard %d for needle %d on volume %d (interval %d/%d)", shardId, id, ecv.VolumeId, i+1, len(intervals))) + // the shard is broken whether or not the needle survives it, so report it either way brokenShardsMap[shardId] = &volume_server_pb.EcShardInfo{ ShardId: uint32(shardId), Size: int64(iv.Size), Collection: ecv.Collection, VolumeId: uint32(ecv.VolumeId), } - break + + if !recoverUnreadable { + errs = append(errs, fmt.Errorf("failed to read EC shard %d for needle %d on volume %d (interval %d/%d): %v", shardId, id, ecv.VolumeId, i+1, len(intervals), readErr)) + break + } + // A holder reporting the needle deleted is authoritative, and it answers + // with no bytes: the chunk stays zeroed and reaches ReadBytes as the + // delete-state mismatch the walk below already tolerates. + if _, isDeleted, err := s.recoverOneRemoteEcShardInterval(id, ecv, shardId, chunk, offset); err != nil && !isDeleted { + errs = append(errs, fmt.Errorf("failed to recover EC shard %d for needle %d on volume %d (interval %d/%d): %v", shardId, id, ecv.VolumeId, i+1, len(intervals), err)) + break + } + data = append(data, chunk...) } if got, want := int64(len(data)), needle.GetActualSize(size, ecv.Version); got != want { diff --git a/weed/storage/store_ec_scrub_reads_test.go b/weed/storage/store_ec_scrub_reads_test.go new file mode 100644 index 000000000..d8a1758b3 --- /dev/null +++ b/weed/storage/store_ec_scrub_reads_test.go @@ -0,0 +1,172 @@ +package storage + +import ( + "math/rand" + "os" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "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/types" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" +) + +// newLocalEcVolume builds an EC volume holding one needle spread over several shards, +// mounts every shard locally, and seeds the shard-location cache with empty location +// lists: the cache then counts as fresh, so nothing calls the master these tests do not +// have, and no read leaves the process. Returns the needle that was encoded. +func newLocalEcVolume(t *testing.T, vid needle.VolumeId) (*Store, *erasure_coding.EcVolume, *needle.Needle) { + t.Helper() + + store := newTestStore(t, 1) + dir := store.Locations[0].Directory + + v, err := NewVolume(dir, dir, "", vid, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + if err != nil { + t.Fatalf("new volume: %v", err) + } + n := new(needle.Needle) + n.Id = types.Uint64ToNeedleId(42) + n.Data = make([]byte, 3*erasure_coding.ErasureCodingSmallBlockSize+1234) + rand.New(rand.NewSource(42)).Read(n.Data) + n.Checksum = needle.NewCRC(n.Data) + if _, _, _, err := v.writeNeedle2(n, true, false, false); err != nil { + t.Fatalf("write needle: %v", err) + } + baseFileName := v.DataFileName() + v.Close() + + datSize, err := os.Stat(baseFileName + ".dat") + if err != nil { + t.Fatalf("stat .dat: %v", err) + } + if _, err := erasure_coding.WriteEcFiles(baseFileName, erasure_coding.BackgroundECContext()); err != nil { + t.Fatalf("write ec files: %v", err) + } + if err := erasure_coding.WriteSortedFileFromIdx(baseFileName, ".ecx"); err != nil { + t.Fatalf("write .ecx: %v", err) + } + if err := os.WriteFile(baseFileName+".ecj", nil, 0o644); err != nil { + t.Fatalf("write .ecj: %v", err) + } + if err := volume_info.SaveVolumeInfo(baseFileName+".vif", &volume_server_pb.VolumeInfo{ + Version: uint32(needle.GetCurrentVersion()), + DatFileSize: datSize.Size(), + EcShardConfig: &volume_server_pb.EcShardConfig{ + DataShards: erasure_coding.DataShardsCount, + ParityShards: erasure_coding.ParityShardsCount, + }, + }); err != nil { + t.Fatalf("save .vif: %v", err) + } + for _, ext := range []string{".dat", ".idx"} { + if err := os.Remove(baseFileName + ext); err != nil { + t.Fatalf("remove %s: %v", ext, err) + } + } + + for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ { + if err := store.MountEcShards("", vid, erasure_coding.ShardId(shardId), ""); err != nil { + t.Fatalf("mount shard %d: %v", shardId, err) + } + } + ecVolume, found := store.Locations[0].FindEcVolume(vid) + if !found { + t.Fatal("ec volume not mounted") + } + + ecVolume.ShardLocationsLock.Lock() + for shardId := 0; shardId < erasure_coding.TotalShardsCount; shardId++ { + ecVolume.ShardLocations[erasure_coding.ShardId(shardId)] = []pb.ServerAddress{} + } + ecVolume.ShardLocationsRefreshTime = time.Now() + ecVolume.ShardLocationsLock.Unlock() + + return store, ecVolume, n +} + +func TestScrubEcVolumeReadsMatchesFullWhenHealthy(t *testing.T) { + const vid = needle.VolumeId(7) + store, _, _ := newLocalEcVolume(t, vid) + + for _, mode := range []volume_server_pb.VolumeScrubMode{ + volume_server_pb.VolumeScrubMode_FULL, + volume_server_pb.VolumeScrubMode_READS, + } { + count, brokenShards, errs := store.ScrubEcVolume(vid, mode, false) + if count != 1 { + t.Errorf("%s: scrubbed %d needles, want 1", mode, count) + } + if len(brokenShards) != 0 { + t.Errorf("%s: reported broken shards %v on a healthy volume", mode, brokenShards) + } + if len(errs) != 0 { + t.Errorf("%s: reported errors on a healthy volume: %v", mode, errs) + } + } +} + +// A READS scrub rebuilds what it cannot read, but the shard it could not read is still +// broken and must still be reported - otherwise a volume missing shards scrubs clean and +// the operator never learns to repair it. +func TestScrubEcVolumeReadsReportsTheShardItRebuilt(t *testing.T) { + const vid = needle.VolumeId(7) + const missingShard = erasure_coding.ShardId(0) + + store, _, _ := newLocalEcVolume(t, vid) + if err := store.UnmountEcShards(vid, missingShard, 0); err != nil { + t.Fatalf("unmount shard %d: %v", missingShard, err) + } + + // FULL gives up on the needles that lived on the missing shard... + _, brokenShards, errs := store.ScrubEcVolume(vid, volume_server_pb.VolumeScrubMode_FULL, false) + assertOnlyBrokenShard(t, "FULL", brokenShards, missingShard) + if len(errs) == 0 { + t.Fatalf("FULL reported no error for a shard it could not read") + } + if got := errs[0].Error(); !strings.Contains(got, "failed to read EC shard 0") { + t.Fatalf("FULL error %q, want it to name the shard it could not read", got) + } + + // ...READS rebuilds them from the shards that are left, and still names the shard. + _, brokenShards, errs = store.ScrubEcVolume(vid, volume_server_pb.VolumeScrubMode_READS, false) + assertOnlyBrokenShard(t, "READS", brokenShards, missingShard) + if len(errs) != 0 { + t.Fatalf("READS should rebuild every needle from parity, got: %v", errs) + } +} + +func TestScrubEcVolumeReadsErrorsWhenParityCannotCover(t *testing.T) { + const vid = needle.VolumeId(7) + + store, _, _ := newLocalEcVolume(t, vid) + // one shard more than parity can cover, so nothing can be rebuilt + for shardId := 0; shardId <= erasure_coding.ParityShardsCount; shardId++ { + if err := store.UnmountEcShards(vid, erasure_coding.ShardId(shardId), 0); err != nil { + t.Fatalf("unmount shard %d: %v", shardId, err) + } + } + + _, brokenShards, errs := store.ScrubEcVolume(vid, volume_server_pb.VolumeScrubMode_READS, false) + if len(brokenShards) == 0 { + t.Fatalf("no broken shard reported for an unrecoverable volume") + } + if len(errs) == 0 { + t.Fatalf("no error reported for an unrecoverable volume") + } + if got := errs[0].Error(); !strings.Contains(got, "failed to recover EC shard") { + t.Fatalf("error %q, want it to say the rebuild failed", got) + } +} + +func assertOnlyBrokenShard(t *testing.T, mode string, brokenShards []*volume_server_pb.EcShardInfo, shardId erasure_coding.ShardId) { + t.Helper() + if len(brokenShards) != 1 || brokenShards[0].GetShardId() != uint32(shardId) { + t.Fatalf("%s reported broken shards %v, want only shard %d", mode, brokenShards, shardId) + } +}