diff --git a/weed/server/volume_grpc_batch_delete.go b/weed/server/volume_grpc_batch_delete.go index 06a2eeb3ea..1a3dc90655 100644 --- a/weed/server/volume_grpc_batch_delete.go +++ b/weed/server/volume_grpc_batch_delete.go @@ -10,6 +10,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/topology" ) func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.BatchDeleteRequest) (*volume_server_pb.BatchDeleteResponse, error) { @@ -24,6 +25,7 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B } now := uint64(time.Now().Unix()) + replicaErrs := make(map[needle.VolumeId]error) for _, fid := range req.FileIds { vid, id_cookie, err := operation.ParseFileId(fid) @@ -56,6 +58,14 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B } } + if err := vs.checkDeleteReplicas(volumeId, replicaErrs); err != nil { + resp.Results = append(resp.Results, &volume_server_pb.DeleteResult{ + FileId: fid, + Status: http.StatusServiceUnavailable, + Error: err.Error(), + }) + continue + } ecVolume, isEcVolume := vs.store.FindEcVolume(volumeId) if !req.SkipCookieCheck { cookie := n.Cookie @@ -145,3 +155,18 @@ func (vs *VolumeServer) BatchDelete(ctx context.Context, req *volume_server_pb.B return resp, nil } + +// checkDeleteReplicas refuses a delete the master cannot route to every +// replica, as the HTTP delete path does, so an offline replica does not keep +// the needle once it returns. +func (vs *VolumeServer) checkDeleteReplicas(volumeId needle.VolumeId, checked map[needle.VolumeId]error) error { + if err, found := checked[volumeId]; found { + return err + } + var err error + if v := vs.store.GetVolume(volumeId); v != nil && v.ReplicaPlacement.GetCopyCount() > 1 { + _, err = topology.GetRemoteReplications(vs.store, vs.grpcDialOption, volumeId, vs.GetMaster) + } + checked[volumeId] = err + return err +} diff --git a/weed/server/volume_grpc_batch_delete_replica_test.go b/weed/server/volume_grpc_batch_delete_replica_test.go new file mode 100644 index 0000000000..fcdf2f29db --- /dev/null +++ b/weed/server/volume_grpc_batch_delete_replica_test.go @@ -0,0 +1,109 @@ +package weed_server + +import ( + "context" + "net" + "net/http" + "strings" + "sync" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/peer" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" +) + +type replicaLookupMaster struct { + master_pb.UnimplementedSeaweedServer + mu sync.Mutex + locations []*master_pb.Location +} + +func (m *replicaLookupMaster) LookupVolume(_ context.Context, req *master_pb.LookupVolumeRequest) (*master_pb.LookupVolumeResponse, error) { + m.mu.Lock() + defer m.mu.Unlock() + resp := &master_pb.LookupVolumeResponse{} + for _, vid := range req.VolumeOrFileIds { + resp.VolumeIdLocations = append(resp.VolumeIdLocations, &master_pb.LookupVolumeResponse_VolumeIdLocation{ + VolumeOrFileId: vid, + Locations: m.locations, + }) + } + return resp, nil +} + +func (m *replicaLookupMaster) setLocations(locations ...*master_pb.Location) { + m.mu.Lock() + defer m.mu.Unlock() + m.locations = locations +} + +// A replica the master no longer lists would miss the delete and keep the +// needle when it returns, so the delete waits until every replica is back. +func TestBatchDelete_WaitsForEveryReplica(t *testing.T) { + const vid = needle.VolumeId(9) + const cookie = uint32(0x12345678) + + store := newVolumeCopyTestStore(t, t.TempDir()) + if err := store.AddVolume(vid, "", storage.NeedleMapInMemory, "001", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add volume: %v", err) + } + data := []byte("hello") + if _, err := store.WriteVolumeNeedle(vid, &needle.Needle{Id: 1, Cookie: types.Cookie(cookie), Data: data, Checksum: needle.NewCRC(data)}, true, false); err != nil { + t.Fatalf("write needle: %v", err) + } + + self := &master_pb.Location{Url: "127.0.0.1:0"} + peerReplica := &master_pb.Location{Url: "127.0.0.2:8080"} + master := &replicaLookupMaster{} + master.setLocations(self) + masterGrpcPort := serveGrpc(t, func(s *grpc.Server) { + master_pb.RegisterSeaweedServer(s, master) + }) + + vs := &VolumeServer{ + store: store, + guard: security.NewGuard(nil, "", 0, "", 0), + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + } + vs.setCurrentMaster(pb.NewServerAddress("127.0.0.1", masterGrpcPort-10000, masterGrpcPort)) + + ctx := peer.NewContext(context.Background(), &peer.Peer{ + Addr: &net.TCPAddr{IP: net.ParseIP("127.0.0.1"), Port: 12345}, + }) + req := &volume_server_pb.BatchDeleteRequest{ + FileIds: []string{needle.NewFileId(vid, 1, cookie).String()}, + SkipCookieCheck: true, + } + + resp, err := vs.BatchDelete(ctx, req) + if err != nil { + t.Fatalf("BatchDelete: %v", err) + } + if len(resp.Results) != 1 { + t.Fatalf("got %d results, want 1", len(resp.Results)) + } + if got := resp.Results[0]; got.Status != http.StatusServiceUnavailable || !strings.Contains(got.Error, "replication copy count") { + t.Fatalf("with a replica offline: status %d error %q, want %d and a replica count error", got.Status, got.Error, http.StatusServiceUnavailable) + } + if _, err := store.ReadVolumeNeedle(vid, &needle.Needle{Id: 1}, nil, nil); err != nil { + t.Fatalf("needle deleted while a replica was offline: %v", err) + } + + master.setLocations(self, peerReplica) + resp, err = vs.BatchDelete(ctx, req) + if err != nil { + t.Fatalf("BatchDelete: %v", err) + } + if got := resp.Results[0]; got.Status != http.StatusAccepted || got.Error != "" { + t.Fatalf("with every replica back: status %d error %q, want %d", got.Status, got.Error, http.StatusAccepted) + } +}