gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

volume: deletion checks all disk locations

Browse source 
fix https://github.com/chrislusf/seaweedfs/issues/1283
This commit is contained in:
Chris Lu 2020-04-21 14:49:58 -07:00
parent 57df14f76f
commit bafa95045b
2 changed files with 14 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
weed/storage/disk_location.go
View file

@ -167,7 +167,7 @@ func (l *DiskLocation) DeleteCollectionFromDiskLocation(collection string) (e er
return return
} }
func (l *DiskLocation) deleteVolumeById(vid needle.VolumeId) (e error) { func (l *DiskLocation) deleteVolumeById(vid needle.VolumeId) (found bool, e error) {
v, ok := l.volumes[vid] v, ok := l.volumes[vid]
if !ok { if !ok {
return return
@ -176,6 +176,7 @@ func (l *DiskLocation) deleteVolumeById(vid needle.VolumeId) (e error) {
if e != nil { if e != nil {
return return
} }
found = true
delete(l.volumes, vid) delete(l.volumes, vid)
return return
} }
@ -195,7 +196,8 @@ func (l *DiskLocation) DeleteVolume(vid needle.VolumeId) error {
if !ok { if !ok {
return fmt.Errorf("Volume not found, VolumeId: %d", vid) return fmt.Errorf("Volume not found, VolumeId: %d", vid)
} }
return l.deleteVolumeById(vid) _, err := l.deleteVolumeById(vid)
return err
} }
func (l *DiskLocation) UnloadVolume(vid needle.VolumeId) error { func (l *DiskLocation) UnloadVolume(vid needle.VolumeId) error {

14
weed/storage/store.go
View file

@ -221,8 +221,14 @@ func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
// delete expired volumes. // delete expired volumes.
location.volumesLock.Lock() location.volumesLock.Lock()
for _, vid := range deleteVids { for _, vid := range deleteVids {
location.deleteVolumeById(vid) found, err := location.deleteVolumeById(vid)
glog.V(0).Infoln("volume", vid, "is deleted.") if found {
if err == nil {
glog.V(0).Infof("volume %d is deleted", vid)
} else {
glog.V(0).Infof("delete volume %d: %v", vid, err)
}
}
} }
location.volumesLock.Unlock() location.volumesLock.Unlock()
} }
@ -355,7 +361,7 @@ func (s *Store) UnmountVolume(i needle.VolumeId) error {
func (s *Store) DeleteVolume(i needle.VolumeId) error { func (s *Store) DeleteVolume(i needle.VolumeId) error {
v := s.findVolume(i) v := s.findVolume(i)
if v == nil { if v == nil {
return nil return fmt.Errorf("delete volume %d not found on disk", i)
} }
message := master_pb.VolumeShortInformationMessage{ message := master_pb.VolumeShortInformationMessage{
Id: uint32(v.Id), Id: uint32(v.Id),
@ -365,7 +371,7 @@ func (s *Store) DeleteVolume(i needle.VolumeId) error {
Ttl: v.Ttl.ToUint32(), Ttl: v.Ttl.ToUint32(),
} }
for _, location := range s.Locations { for _, location := range s.Locations {
if error := location.deleteVolumeById(i); error == nil { if found, error := location.deleteVolumeById(i); found && error == nil {
glog.V(0).Infof("DeleteVolume %d", i) glog.V(0).Infof("DeleteVolume %d", i)
s.DeletedVolumesChan <- message s.DeletedVolumesChan <- message
return nil return nil