diff --git a/weed/storage/store.go b/weed/storage/store.go index 21ac971ed7..273490c05c 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -291,7 +291,7 @@ func (s *Store) DeleteCollection(collection string) (e error) { // volume grown and destroyed between two of them was never reported at // all, so nothing else would ever tell the master its slot came free. for _, v := range deleted { - readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons() + readOnly, readOnlyCanDelete := v.reportedReadOnly() s.DeletedVolumesChan <- &master_pb.VolumeShortInformationMessage{ Id: uint32(v.Id), Collection: v.Collection, @@ -428,7 +428,7 @@ func (s *Store) addVolume(vid needle.VolumeId, collection string, needleMapKind volume.diskId = diskId // Set the disk ID location.SetVolume(vid, volume) glog.V(0).Infof("add volume %d on disk ID %d", vid, diskId) - readOnly, _, readOnlyCanDelete, _ := volume.ReadOnlyReasons() + readOnly, readOnlyCanDelete := volume.reportedReadOnly() s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{ Id: uint32(vid), Collection: collection, @@ -523,7 +523,7 @@ func collectStatsForOneLocation(location *DiskLocation) (stats []*VolumeInfo) { func collectStatForOneVolume(vid needle.VolumeId, v *Volume) (s *VolumeInfo) { - readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons() + readOnly, readOnlyCanDelete := v.reportedReadOnly() s = &VolumeInfo{ Id: vid, Collection: v.Collection, @@ -1151,7 +1151,7 @@ func (s *Store) mountVolume(i needle.VolumeId, collection *string, ownLoad bool, return err } } - readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons() + readOnly, readOnlyCanDelete := v.reportedReadOnly() s.NewVolumesChan <- &master_pb.VolumeShortInformationMessage{ Id: uint32(v.Id), Collection: v.Collection, @@ -1183,7 +1183,7 @@ func (s *Store) UnmountVolume(i needle.VolumeId) error { if !found { continue } - readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons() + readOnly, readOnlyCanDelete := v.reportedReadOnly() message := master_pb.VolumeShortInformationMessage{ Id: uint32(v.Id), Collection: v.Collection, @@ -1289,7 +1289,7 @@ func (s *Store) DeleteVolume(i needle.VolumeId, onlyEmpty bool, onlyGarbage bool if !found { continue } - readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons() + readOnly, readOnlyCanDelete := v.reportedReadOnly() message := master_pb.VolumeShortInformationMessage{ Id: uint32(v.Id), Collection: v.Collection, @@ -1369,7 +1369,7 @@ func (s *Store) deleteVolumeGuarded(i needle.VolumeId, onlyEmpty bool, onlyGarba var deletedMessages []*master_pb.VolumeShortInformationMessage for _, location := range lockedLocations { v := location.volumes[i] - readOnly, _, readOnlyCanDelete, _ := v.ReadOnlyReasons() + readOnly, readOnlyCanDelete := v.reportedReadOnly() message := &master_pb.VolumeShortInformationMessage{ Id: uint32(v.Id), Collection: v.Collection, diff --git a/weed/storage/store_mark_readonly_can_delete_test.go b/weed/storage/store_mark_readonly_can_delete_test.go index 00376df8cf..63839a563d 100644 --- a/weed/storage/store_mark_readonly_can_delete_test.go +++ b/weed/storage/store_mark_readonly_can_delete_test.go @@ -152,3 +152,27 @@ func TestMountVolumeAnnouncesReadOnlyState(t *testing.T) { }) } } + +func TestLowDiskVolumeReportsCanDelete(t *testing.T) { + store := newSingleDirStore(t, t.TempDir()) + defer store.Close() + const vid = needle.VolumeId(14) + require.NoError(t, store.AddVolume(vid, "", NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0)) + v := store.GetVolume(vid) + n := newRandomNeedle(1) + _, err := store.WriteVolumeNeedle(vid, n, true, false) + require.NoError(t, err) + + store.Locations[0].isDiskSpaceLow.Store(true) + _, message := v.ToVolumeInformationMessage(nil) + require.True(t, message.ReadOnly) + require.True(t, message.ReadOnlyCanDelete) + require.Equal(t, message.ReadOnlyCanDelete, store.CollectHeartbeat().Volumes[0].ReadOnlyCanDelete) + _, err = store.DeleteVolumeNeedle(vid, n) + require.NoError(t, err) + + require.NoError(t, store.MarkVolumeReadonly(vid, false, false)) + _, message = v.ToVolumeInformationMessage(nil) + require.True(t, message.ReadOnly) + require.False(t, message.ReadOnlyCanDelete) +} diff --git a/weed/storage/volume.go b/weed/storage/volume.go index fd615bb452..ac4d2e0f88 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -549,7 +549,7 @@ func (v *Volume) ToVolumeInformationMessage(into *master_pb.VolumeInformationMes volumeInfo.FileCount = fileCount volumeInfo.DeleteCount = deletedCount volumeInfo.DeletedByteCount = deletedSize - volumeInfo.ReadOnly, _, volumeInfo.ReadOnlyCanDelete, _ = v.ReadOnlyReasons() + volumeInfo.ReadOnly, volumeInfo.ReadOnlyCanDelete = v.reportedReadOnly() volumeInfo.ReplicaPlacement = uint32(v.ReplicaPlacement.Byte()) volumeInfo.Version = uint32(v.Version()) volumeInfo.Ttl = v.Ttl.ToUint32() @@ -592,6 +592,13 @@ func (v *Volume) ReadOnlyReasons() (readOnly, noWriteOrDelete, noWriteCanDelete, return noWriteOrDelete || noWriteCanDelete || diskSpaceLow, noWriteOrDelete, noWriteCanDelete, diskSpaceLow } +// reportedReadOnly is what the master is told: low disk space still lets +// deletes through, only noWriteOrDelete refuses them. +func (v *Volume) reportedReadOnly() (readOnly, readOnlyCanDelete bool) { + readOnly, noWriteOrDelete, _, _ := v.ReadOnlyReasons() + return readOnly, readOnly && !noWriteOrDelete +} + var errVolumeUnavailable = errors.New("volume unavailable") func (v *Volume) UnavailableError() error {