diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go index ef103e4657..de2b7c3f77 100644 --- a/weed/topology/topology_vacuum.go +++ b/weed/topology/topology_vacuum.go @@ -479,8 +479,10 @@ func (t *Topology) deleteEmptyVolumes(grpcDialOption grpc.DialOption, vl *Volume nowUnixSeconds := time.Now().Unix() for vid, locationList := range todoVolumeMap { // A missing replica may still hold data; deleting the listed copies - // would leave it as the only one when it returns. - if len(locationList.list) < vl.rp.GetCopyCount() { + // would leave it as the only one when it returns. A vid still marked + // deleting lost copies to an earlier partial delete, so finish it. + resuming := vl.IsMarkedDeleting(vid) + if len(locationList.list) < vl.rp.GetCopyCount() && !resuming { continue } eligible := true @@ -492,12 +494,15 @@ func (t *Topology) deleteEmptyVolumes(grpcDialOption grpc.DialOption, vl *Volume } } if !eligible { + if resuming { + vl.UnmarkDeleting(vid) + } continue } // keep the volume out of assignments for the whole delete: a heartbeat // landing mid-delete must not re-add it while replicas are dropping vl.MarkDeleting(vid) - remaining := 0 + deleted := 0 kept := locationList.list[:0] for _, dn := range locationList.list { v, err := dn.GetVolumesById(vid) @@ -506,16 +511,19 @@ func (t *Topology) deleteEmptyVolumes(grpcDialOption grpc.DialOption, vl *Volume glog.V(0).Infof("deleting empty volume %d on %s", vid, dn.ServerAddress()) if err = t.deleteEmptyVolume(grpcDialOption, dn, vid, onlyGarbage); err == nil { t.UnRegisterVolumeLayout(v, dn) + deleted++ continue } } glog.Warningf("delete empty volume %d on %s: %v", vid, dn.ServerAddress(), err) kept = append(kept, dn) - remaining++ } - vl.UnmarkDeleting(vid) + // a partly deleted vid stays marked so the next sweep finishes it + if len(kept) == 0 || deleted == 0 && !resuming { + vl.UnmarkDeleting(vid) + } locationList.list = kept - if remaining == 0 { + if len(kept) == 0 { delete(todoVolumeMap, vid) } } diff --git a/weed/topology/topology_vacuum_test.go b/weed/topology/topology_vacuum_test.go index 897a2b0ae8..d64235a867 100644 --- a/weed/topology/topology_vacuum_test.go +++ b/weed/topology/topology_vacuum_test.go @@ -54,13 +54,18 @@ func TestIsEmptyVolumeDeleteCandidate(t *testing.T) { type fakeVolumeDeleteServer struct { volume_server_pb.UnimplementedVolumeServerServer - mu sync.Mutex - deletes []*volume_server_pb.VolumeDeleteRequest + mu sync.Mutex + deletes []*volume_server_pb.VolumeDeleteRequest + failures int } func (f *fakeVolumeDeleteServer) VolumeDelete(ctx context.Context, req *volume_server_pb.VolumeDeleteRequest) (*volume_server_pb.VolumeDeleteResponse, error) { f.mu.Lock() defer f.mu.Unlock() + if f.failures > 0 { + f.failures-- + return nil, errors.New("volume server busy") + } f.deletes = append(f.deletes, req) return &volume_server_pb.VolumeDeleteResponse{}, nil } @@ -252,6 +257,67 @@ func TestDeleteEmptyVolumesSkipsUnderReplicatedVid(t *testing.T) { } } +// A delete that removed one copy and failed on the other leaves the vid with +// fewer copies than its replica placement; the next sweep must still finish +// it instead of treating it as a volume with a replica offline. +func TestDeleteEmptyVolumesFinishesPartialDelete(t *testing.T) { + okFake := &fakeVolumeDeleteServer{} + okPort, dialOption := startFakeVolumeServer(t, okFake) + flakyFake := &fakeVolumeDeleteServer{failures: 1} + flakyPort, _ := startFakeVolumeServer(t, flakyFake) + + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + rack := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1") + okDn := rack.GetOrCreateDataNode("127.0.0.1", 8080, okPort, "127.0.0.1", "dn-ok", map[string]uint32{"": 10}) + flakyDn := rack.GetOrCreateDataNode("127.0.0.1", 8081, flakyPort, "127.0.0.1", "dn-flaky", map[string]uint32{"": 10}) + rp, err := super_block.NewReplicaPlacementFromString("001") + if err != nil { + t.Fatal(err) + } + v := storage.VolumeInfo{ + Id: needle.VolumeId(1), + Collection: "c", + ModifiedAtSecond: time.Now().Unix() - 7200, + Version: needle.GetCurrentVersion(), + ReplicaPlacement: rp, + Ttl: needle.EMPTY_TTL, + } + for _, dn := range []*DataNode{okDn, flakyDn} { + dn.UpdateVolumes([]storage.VolumeInfo{v}) + topo.RegisterVolumeLayout(v, dn) + } + vl := topo.GetVolumeLayout("c", rp, needle.EMPTY_TTL, types.ToDiskType("")) + + sweep := func() map[needle.VolumeId]*VolumeLocationList { + vl.accessLock.RLock() + todo := map[needle.VolumeId]*VolumeLocationList{} + for vid, ll := range vl.vid2location { + todo[vid] = ll.Copy() + } + vl.accessLock.RUnlock() + topo.deleteEmptyVolumes(dialOption, vl, todo, time.Hour) + return todo + } + + if todo := sweep(); todo[v.Id] == nil { + t.Fatal("vid left the sweep map while a copy delete failed") + } + if !vl.IsMarkedDeleting(v.Id) { + t.Fatal("partly deleted vid lost its deleting mark") + } + if todo := sweep(); todo[v.Id] != nil { + t.Fatal("next sweep left the partly deleted vid behind") + } + if vl.IsMarkedDeleting(v.Id) { + t.Error("fully deleted vid kept its deleting mark") + } + flakyFake.mu.Lock() + defer flakyFake.mu.Unlock() + if len(flakyFake.deletes) != 1 { + t.Fatalf("VolumeDelete on the flaky copy = %d, want 1 after the retry", len(flakyFake.deletes)) + } +} + // With every copy eligible a copy whose delete RPC fails keeps the vid in the // sweep map; the deleted copy is gone but the vid falls through to the normal // compaction path for its surviving replicas. diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index bd961565ae..49b6b4d9df 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -648,6 +648,13 @@ func (vl *VolumeLayout) MarkDeleting(vid needle.VolumeId) { vl.waitForPendingDrain(context.Background(), vid) } +func (vl *VolumeLayout) IsMarkedDeleting(vid needle.VolumeId) bool { + vl.accessLock.RLock() + defer vl.accessLock.RUnlock() + _, ok := vl.deletingVolumes[vid] + return ok +} + func (vl *VolumeLayout) UnmarkDeleting(vid needle.VolumeId) { vl.accessLock.Lock() delete(vl.deletingVolumes, vid)