diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index 52274d5b0..d4dc682e9 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -3203,11 +3203,17 @@ fn get_append_at_ns(last: u64) -> u64 { } /// Remove all files associated with a volume. +/// .dat/.idx removals log at info level so destructive calls are traceable. pub(crate) fn remove_volume_files(base: &str) { for ext in &[ ".dat", ".idx", ".vif", ".sdx", ".cpd", ".cpx", ".note", ".rdb", ] { - let _ = fs::remove_file(format!("{}{}", base, ext)); + let path = format!("{}{}", base, ext); + let size = fs::metadata(&path).map(|m| m.len()).unwrap_or(0); + let existed = fs::remove_file(&path).is_ok(); + if existed && (*ext == ".dat" || *ext == ".idx") { + tracing::info!("removed volume file {} (size={})", path, size); + } } // leveldb uses a directory let _ = fs::remove_dir_all(format!("{}.ldb", base)); diff --git a/test/plugin_workers/fake_volume_server.go b/test/plugin_workers/fake_volume_server.go index 6978b7c76..e38aa77a1 100644 --- a/test/plugin_workers/fake_volume_server.go +++ b/test/plugin_workers/fake_volume_server.go @@ -7,6 +7,7 @@ import ( "net" "os" "path/filepath" + "strings" "sync" "testing" "time" @@ -291,6 +292,60 @@ func (v *VolumeServer) VolumeEcShardsMount(ctx context.Context, req *volume_serv return &volume_server_pb.VolumeEcShardsMountResponse{}, nil } +func (v *VolumeServer) VolumeEcShardsInfo(ctx context.Context, req *volume_server_pb.VolumeEcShardsInfoRequest) (*volume_server_pb.VolumeEcShardsInfoResponse, error) { + if req == nil { + return nil, fmt.Errorf("VolumeEcShardsInfo request is nil") + } + v.mu.Lock() + defer v.mu.Unlock() + + // Report whichever shards exist on disk: seeded or mounted. Collection + // comes from the matching mount request when one exists. + collectionByShard := make(map[uint32]string) + for _, mr := range v.mountRequests { + if mr == nil || mr.VolumeId != req.VolumeId { + continue + } + for _, shardId := range mr.ShardIds { + if _, ok := collectionByShard[shardId]; !ok { + collectionByShard[shardId] = mr.Collection + } + } + } + + resp := &volume_server_pb.VolumeEcShardsInfoResponse{} + prefix := fmt.Sprintf("%d.ec", req.VolumeId) + entries, _ := os.ReadDir(v.baseDir) + for _, entry := range entries { + if entry.IsDir() { + continue + } + name := entry.Name() + if !strings.HasPrefix(name, prefix) { + continue + } + suffix := strings.TrimPrefix(name, prefix) + if len(suffix) < 2 { + continue + } + var shardId uint32 + if _, err := fmt.Sscanf(suffix[:2], "%d", &shardId); err != nil { + continue + } + var size int64 + if info, err := entry.Info(); err == nil { + size = info.Size() + } + resp.EcShardInfos = append(resp.EcShardInfos, &volume_server_pb.EcShardInfo{ + ShardId: shardId, + Size: size, + Collection: collectionByShard[shardId], + VolumeId: req.VolumeId, + }) + } + return resp, nil +} + func (v *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb.VolumeDeleteRequest) (*volume_server_pb.VolumeDeleteResponse, error) { v.mu.Lock() v.deleteRequests = append(v.deleteRequests, req) diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index 151fbe8a9..a9b309b67 100644 --- a/weed/server/volume_grpc_admin.go +++ b/weed/server/volume_grpc_admin.go @@ -176,7 +176,8 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb. if err != nil { glog.Errorf("volume delete %v: %v", req, err) } else { - glog.V(2).Infof("volume delete %v", req) + // V(0) so destructive RPCs are always traceable. + glog.Infof("volume delete %v", req) } return resp, err diff --git a/weed/shell/command_ec_decode.go b/weed/shell/command_ec_decode.go index 5c906c297..9da96a69a 100644 --- a/weed/shell/command_ec_decode.go +++ b/weed/shell/command_ec_decode.go @@ -7,6 +7,7 @@ import ( "io" "strings" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/storage/types" @@ -175,6 +176,13 @@ func doEcDecode(commandEnv *CommandEnv, topoInfo *master_pb.TopologyInfo, collec return fmt.Errorf("mount decoded volume %d on %s: %v", vid, targetNodeLocation, err) } + // Confirm the regenerated .dat is present and non-empty before destroying + // the shards. Without this gate, a silent failure in generate/mount could + // leave the cluster with neither shards nor volume. + if err := verifyDecodedVolumeBeforeDelete(commandEnv.option.GrpcDialOption, targetNodeLocation, vid); err != nil { + return fmt.Errorf("verify decoded volume %d on %s before deleting shards: %w", vid, targetNodeLocation, err) + } + // delete the previous ec shards err = unmountAndDeleteEcShardsWithPrefix("deleteDecodedEcShards", commandEnv.option.GrpcDialOption, collection, nodeToEcShardsInfo, vid) if err != nil { @@ -225,6 +233,30 @@ func unmountAndDeleteEcShardsWithPrefix(prefix string, grpcDialOption grpc.DialO return ewg.Wait() } +func verifyDecodedVolumeBeforeDelete(grpcDialOption grpc.DialOption, target pb.ServerAddress, vid needle.VolumeId) error { + var resp *volume_server_pb.ReadVolumeFileStatusResponse + if err := operation.WithVolumeServerClient(false, target, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + r, e := client.ReadVolumeFileStatus(context.Background(), &volume_server_pb.ReadVolumeFileStatusRequest{ + VolumeId: uint32(vid), + }) + if e != nil { + return e + } + resp = r + return nil + }); err != nil { + return fmt.Errorf("read volume file status: %w", err) + } + if resp.DatFileSize == 0 { + return fmt.Errorf("decoded .dat is 0 bytes") + } + if resp.IdxFileSize == 0 { + return fmt.Errorf("decoded .idx is 0 bytes") + } + glog.V(0).Infof("ec decode verification ok for volume %d on %s: dat=%d idx=%d", vid, target, resp.DatFileSize, resp.IdxFileSize) + return nil +} + func mountDecodedVolume(grpcDialOption grpc.DialOption, targetNodeLocation pb.ServerAddress, vid needle.VolumeId) error { return operation.WithVolumeServerClient(false, targetNodeLocation, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { _, mountErr := volumeServerClient.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{ diff --git a/weed/shell/command_ec_encode.go b/weed/shell/command_ec_encode.go index df8a6a26d..3824a38ef 100644 --- a/weed/shell/command_ec_encode.go +++ b/weed/shell/command_ec_encode.go @@ -199,6 +199,10 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr if err := EcBalance(commandEnv, balanceCollections, "", rp, diskType, *maxParallelization, *applyBalancing); 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 { @@ -335,6 +339,39 @@ func doEcEncode(commandEnv *CommandEnv, writer io.Writer, volumeIdToCollection m return nil } +func verifyEcShardsBeforeDelete(commandEnv *CommandEnv, volumeIds []needle.VolumeId, diskType types.DiskType) error { + topoInfo, _, err := collectTopologyInfo(commandEnv, 0) + if err != nil { + return fmt.Errorf("fetch topology for shard verification: %w", err) + } + + for _, vid := range volumeIds { + nodeShards := collectEcNodeShardsInfo(topoInfo, vid, diskType) + + var union erasure_coding.ShardBits + for _, info := range nodeShards { + union = erasure_coding.ShardBits(uint32(union) | info.Bitmap()) + } + + totalShards := erasure_coding.TotalShardsCount + if err := erasure_coding.RequireFullShardSet(uint32(vid), union, totalShards); err != nil { + summary := make([]string, 0, len(nodeShards)) + for node, info := range nodeShards { + summary = append(summary, fmt.Sprintf("%s=%s", node, info.String())) + } + sort.Strings(summary) + glog.Errorf("EC shard verification failed for volume %d on diskType %q: %v; observed: %v", + vid, diskType.ReadableString(), err, summary) + return fmt.Errorf("volume %d: %w (observed: %v)", vid, err, summary) + } + + glog.V(0).Infof("EC shard verification ok for volume %d on diskType %q: %d/%d shards present across %d nodes", + vid, diskType.ReadableString(), union.Count(), totalShards, len(nodeShards)) + } + + return nil +} + // doDeleteVolumesWithLocations deletes volumes using pre-collected location information // This avoids race conditions where master metadata is updated after EC encoding func doDeleteVolumesWithLocations(commandEnv *CommandEnv, volumeIds []needle.VolumeId, volumeLocationsMap map[needle.VolumeId][]wdclient.Location, maxParallelization int) error { diff --git a/weed/storage/erasure_coding/verification.go b/weed/storage/erasure_coding/verification.go new file mode 100644 index 000000000..0be6fa7c3 --- /dev/null +++ b/weed/storage/erasure_coding/verification.go @@ -0,0 +1,125 @@ +package erasure_coding + +import ( + "context" + "fmt" + "sort" + + "github.com/seaweedfs/seaweedfs/weed/operation" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "google.golang.org/grpc" +) + +type ServerShardInventory struct { + Bits ShardBits + QueryError error +} + +// Query errors are recorded per-server and treated as zero shards rather +// than aborting the scan, so the caller still sees partial coverage from +// healthy peers when one server is down. The caller gates destructive +// actions on RequireFullShardSet against the returned union. +func VerifyShardsAcrossServers(ctx context.Context, volumeID uint32, + servers []string, dialOption grpc.DialOption) ( + union ShardBits, perServer map[string]ServerShardInventory) { + + perServer = make(map[string]ServerShardInventory, len(servers)) + + for _, server := range servers { + if server == "" { + continue + } + if _, seen := perServer[server]; seen { + continue + } + + var inv ServerShardInventory + + callErr := operation.WithVolumeServerClient(false, pb.ServerAddress(server), dialOption, + func(client volume_server_pb.VolumeServerClient) error { + resp, e := client.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{ + VolumeId: volumeID, + }) + if e != nil { + return e + } + for _, s := range resp.EcShardInfos { + if s.VolumeId != volumeID || s.ShardId >= MaxShardCount { + continue + } + inv.Bits = inv.Bits.Set(ShardId(s.ShardId)) + } + return nil + }) + if callErr != nil { + inv.QueryError = callErr + } + + perServer[server] = inv + union = ShardBits(uint32(union) | uint32(inv.Bits)) + } + + return union, perServer +} + +// totalShards is the configured DataShards+ParityShards for this volume. +// Passed as a parameter (not derived from TotalShardsCount) so enterprise +// builds with custom EC ratios share this helper verbatim. +func RequireFullShardSet(volumeID uint32, shardsPresent ShardBits, totalShards int) error { + if totalShards <= 0 || totalShards > MaxShardCount { + return fmt.Errorf("invalid totalShards %d for volume %d (must be in [1, %d])", + totalShards, volumeID, MaxShardCount) + } + var missing []int + for id := 0; id < totalShards; id++ { + if !shardsPresent.Has(ShardId(id)) { + missing = append(missing, id) + } + } + if len(missing) == 0 { + return nil + } + sort.Ints(missing) + return fmt.Errorf("EC shard set incomplete for volume %d: %d/%d shards present, missing shard ids %v", + volumeID, shardsPresent.Count(), totalShards, missing) +} + +func SummarizeShardInventory(perServer map[string]ServerShardInventory) string { + servers := make([]string, 0, len(perServer)) + for s := range perServer { + servers = append(servers, s) + } + sort.Strings(servers) + + var b []byte + for i, s := range servers { + if i > 0 { + b = append(b, ' ') + } + inv := perServer[s] + b = append(b, s...) + b = append(b, '=') + b = append(b, '[') + ids := make([]int, 0) + for id := 0; id < MaxShardCount; id++ { + if inv.Bits.Has(ShardId(id)) { + ids = append(ids, id) + } + } + for j, id := range ids { + if j > 0 { + b = append(b, ' ') + } + b = append(b, []byte(fmt.Sprintf("%d", id))...) + } + if inv.QueryError != nil { + if len(ids) > 0 { + b = append(b, ' ') + } + b = append(b, []byte("ERR:"+inv.QueryError.Error())...) + } + b = append(b, ']') + } + return string(b) +} diff --git a/weed/storage/erasure_coding/verification_test.go b/weed/storage/erasure_coding/verification_test.go new file mode 100644 index 000000000..8a268069a --- /dev/null +++ b/weed/storage/erasure_coding/verification_test.go @@ -0,0 +1,109 @@ +package erasure_coding + +import ( + "strings" + "testing" +) + +func TestRequireFullShardSet_AllPresent(t *testing.T) { + var bits ShardBits + for id := 0; id < TotalShardsCount; id++ { + bits = bits.Set(ShardId(id)) + } + if err := RequireFullShardSet(42, bits, TotalShardsCount); err != nil { + t.Fatalf("unexpected error for full set: %v", err) + } +} + +func TestRequireFullShardSet_ReportsMissingIds(t *testing.T) { + var bits ShardBits + for id := 0; id < TotalShardsCount; id++ { + if id == 3 || id == 7 { + continue + } + bits = bits.Set(ShardId(id)) + } + err := RequireFullShardSet(42, bits, TotalShardsCount) + if err == nil { + t.Fatal("expected error for incomplete set, got nil") + } + msg := err.Error() + if !strings.Contains(msg, "volume 42") { + t.Errorf("error should name the volume id: %s", msg) + } + if !strings.Contains(msg, "[3 7]") { + t.Errorf("error should list missing ids 3 and 7: %s", msg) + } + if !strings.Contains(msg, "12/14") { + t.Errorf("error should report 12/14 shards present: %s", msg) + } +} + +func TestRequireFullShardSet_EmptyBitmap(t *testing.T) { + err := RequireFullShardSet(1, 0, TotalShardsCount) + if err == nil { + t.Fatal("expected error for empty bitmap") + } + if !strings.Contains(err.Error(), "0/14") { + t.Errorf("error should report 0/14 shards: %s", err.Error()) + } +} + +func TestRequireFullShardSet_CustomRatio(t *testing.T) { + // 6+3 ratio: total=9, all present + var bits ShardBits + for id := 0; id < 9; id++ { + bits = bits.Set(ShardId(id)) + } + if err := RequireFullShardSet(7, bits, 9); err != nil { + t.Fatalf("unexpected error for full 6+3 set: %v", err) + } + + // 6+3, missing shard 5 + bits = bits.Clear(5) + err := RequireFullShardSet(7, bits, 9) + if err == nil { + t.Fatal("expected error when shard 5 is missing in 6+3 ratio") + } + if !strings.Contains(err.Error(), "8/9") { + t.Errorf("error should report 8/9: %s", err.Error()) + } + if !strings.Contains(err.Error(), "[5]") { + t.Errorf("error should list missing id 5: %s", err.Error()) + } +} + +func TestRequireFullShardSet_RejectsInvalidTotal(t *testing.T) { + if err := RequireFullShardSet(1, 0, 0); err == nil { + t.Error("expected error for totalShards=0") + } + if err := RequireFullShardSet(1, 0, MaxShardCount+1); err == nil { + t.Errorf("expected error for totalShards > MaxShardCount") + } +} + +func TestSummarizeShardInventory_Deterministic(t *testing.T) { + perServer := map[string]ServerShardInventory{ + "10.0.0.2:8080": {Bits: ShardBits(0).Set(4).Set(5).Set(6)}, + "10.0.0.1:8080": {Bits: ShardBits(0).Set(0).Set(1).Set(2).Set(3)}, + } + got := SummarizeShardInventory(perServer) + want := "10.0.0.1:8080=[0 1 2 3] 10.0.0.2:8080=[4 5 6]" + if got != want { + t.Errorf("summary mismatch\n got: %q\n want: %q", got, want) + } +} + +func TestSummarizeShardInventory_IncludesError(t *testing.T) { + perServer := map[string]ServerShardInventory{ + "10.0.0.1:8080": {Bits: ShardBits(0).Set(0).Set(1), QueryError: errStr("dial timeout")}, + } + got := SummarizeShardInventory(perServer) + if !strings.Contains(got, "ERR:dial timeout") { + t.Errorf("expected error tag in summary, got %q", got) + } +} + +type errStr string + +func (e errStr) Error() string { return string(e) } diff --git a/weed/storage/volume_write.go b/weed/storage/volume_write.go index 8d5916786..686a19d7d 100644 --- a/weed/storage/volume_write.go +++ b/weed/storage/volume_write.go @@ -101,11 +101,17 @@ func (v *Volume) Destroy(onlyEmpty bool, keepRemoteData bool) (err error) { } func removeVolumeFiles(filename string) { - // basic + // .dat/.idx removals log at V(0) so destructive calls are traceable. deleteAndLog := func(ext string) { fullFilename := filename + "." + ext - if err := os.RemoveAll(fullFilename); err != nil { + st, statErr := os.Stat(fullFilename) + err := os.RemoveAll(fullFilename) + if err != nil { glog.V(0).Infof("failed to remove volume file %s: %s", fullFilename, err) + return + } + if statErr == nil && (ext == "dat" || ext == "idx") { + glog.Infof("removed volume file %s (size=%d)", fullFilename, st.Size()) } } deleteAndLog("dat") diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index da537b2a5..85335f39b 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -190,7 +190,15 @@ func (t *ErasureCodingTask) Execute(ctx context.Context, params *worker_pb.TaskP return fmt.Errorf("failed to mount EC shards: %v", err) } - // Step 6: Delete original volume + // Without this gate, a partial distribute/mount lets the next step + // zero the only intact .dat while the cluster is missing shards. + t.ReportProgressWithStage(85.0, "Verifying EC shards across destinations") + t.GetLogger().Info("Verifying EC shards across destinations") + if err := t.verifyEcShardsBeforeDelete(ctx); err != nil { + return fmt.Errorf("EC shard verification failed; refusing to delete source volume %d: %w", t.volumeID, err) + } + + // Step 7: Delete original volume t.ReportProgressWithStage(90.0, "Deleting original volume") t.GetLogger().Info("Deleting original volume") if err := t.deleteOriginalVolume(ctx); err != nil { @@ -545,6 +553,37 @@ func (t *ErasureCodingTask) mountEcShards() error { return erasure_coding.MountEcShards(t.volumeID, t.collection, t.shardAssignment, t.sourceDiskType, t.grpcDialOption, t.GetLogger()) } +func (t *ErasureCodingTask) verifyEcShardsBeforeDelete(ctx context.Context) error { + servers := make([]string, 0, len(t.shardAssignment)) + for node := range t.shardAssignment { + servers = append(servers, node) + } + if len(servers) == 0 { + return fmt.Errorf("no destinations to verify; shardAssignment is empty") + } + + totalShards := int(t.dataShards + t.parityShards) + union, perServer := erasure_coding.VerifyShardsAcrossServers(ctx, t.volumeID, servers, t.grpcDialOption) + + summary := erasure_coding.SummarizeShardInventory(perServer) + t.GetLogger().WithFields(map[string]interface{}{ + "volume_id": t.volumeID, + "shards_seen": union.Count(), + "shards_needed": totalShards, + "per_server": summary, + }).Info("EC shard inventory before source deletion") + + if err := erasure_coding.RequireFullShardSet(t.volumeID, union, totalShards); err != nil { + t.GetLogger().WithFields(map[string]interface{}{ + "volume_id": t.volumeID, + "per_server": summary, + "error": err.Error(), + }).Error("EC shard verification failed — source volume will be kept") + return err + } + return nil +} + // deleteOriginalVolume deletes the original volume and all its replicas from all servers func (t *ErasureCodingTask) deleteOriginalVolume(ctx context.Context) error { // Get replicas from task parameters (set during detection)