Detect underflow when calculating unused space (#5758)

* Detect underflow when calculating unused space

* Detect underflow when calculating unused space

2 files changed, 84 insertions, 2 deletions

diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go
index 9f61ad872..dd78735d2 100644
--- a/weed/storage/disk_location.go
+++ b/weed/storage/disk_location.go
@@ -60,6 +60,7 @@ func GenerateDirUuid(dir string) (dirUuidString string, err error) {
 }
 
 func NewDiskLocation(dir string, maxVolumeCount int32, minFreeSpace util.MinFreeSpace, idxDir string, diskType types.DiskType) *DiskLocation {
+	glog.V(4).Infof("Added new Disk %s: maxVolumes=%d", dir, maxVolumeCount)
 	dir = util.ResolvePath(dir)
 	if idxDir == "" {
 		idxDir = dir
@@ -417,7 +418,6 @@ func (l *DiskLocation) LocateVolume(vid needle.VolumeId) (os.DirEntry, bool) {
 }
 
 func (l *DiskLocation) UnUsedSpace(volumeSizeLimit uint64) (unUsedSpace uint64) {
-
 	l.volumesLock.RLock()
 	defer l.volumesLock.RUnlock()
 
@@ -426,7 +426,11 @@ func (l *DiskLocation) UnUsedSpace(volumeSizeLimit uint64) (unUsedSpace uint64)
 			continue
 		}
 		datSize, idxSize, _ := vol.FileStat()
-		unUsedSpace += volumeSizeLimit - (datSize + idxSize)
+		unUsedSpaceVolume := int64(volumeSizeLimit) - int64(datSize+idxSize)
+		glog.V(4).Infof("Volume stats for %d: volumeSizeLimit=%d, datSize=%d idxSize=%d unused=%d", vol.Id, volumeSizeLimit, datSize, idxSize, unUsedSpaceVolume)
+		if unUsedSpaceVolume >= 0 {
+			unUsedSpace += uint64(unUsedSpaceVolume)
+		}
 	}
 
 	return
diff --git a/weed/storage/disk_location_test.go b/weed/storage/disk_location_test.go
new file mode 100644
index 000000000..d105a477f
--- /dev/null
+++ b/weed/storage/disk_location_test.go
@@ -0,0 +1,78 @@
+package storage
+
+import (
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+)
+
+type (
+	mockBackendStorageFile struct {
+		backend.DiskFile
+
+		datSize int64
+	}
+)
+
+func (df *mockBackendStorageFile) GetStat() (datSize int64, modTime time.Time, err error) {
+	return df.datSize, time.Now(), nil
+}
+
+type (
+	mockNeedleMapper struct {
+		NeedleMap
+
+		idxSize uint64
+	}
+)
+
+func (nm *mockNeedleMapper) IndexFileSize() (idxSize uint64) {
+	return nm.idxSize
+}
+
+func TestUnUsedSpace(t *testing.T) {
+	minFreeSpace := util.MinFreeSpace{Type: util.AsPercent, Percent: 1, Raw: "1"}
+
+	diskLocation := DiskLocation{
+		Directory:              "/test/",
+		DirectoryUuid:          "1234",
+		IdxDirectory:           "/test/",
+		DiskType:               "hdd",
+		MaxVolumeCount:         0,
+		OriginalMaxVolumeCount: 0,
+		MinFreeSpace:           minFreeSpace,
+	}
+	diskLocation.volumes = make(map[needle.VolumeId]*Volume)
+
+	volumes := [3]*Volume{
+		{dir: diskLocation.Directory, dirIdx: diskLocation.IdxDirectory, Collection: "", Id: 0, DataBackend: &mockBackendStorageFile{datSize: 990}, nm: &mockNeedleMapper{idxSize: 10}},
+		{dir: diskLocation.Directory, dirIdx: diskLocation.IdxDirectory, Collection: "", Id: 1, DataBackend: &mockBackendStorageFile{datSize: 990}, nm: &mockNeedleMapper{idxSize: 10}},
+		{dir: diskLocation.Directory, dirIdx: diskLocation.IdxDirectory, Collection: "", Id: 2, DataBackend: &mockBackendStorageFile{datSize: 990}, nm: &mockNeedleMapper{idxSize: 10}},
+	}
+
+	for i, vol := range volumes {
+		diskLocation.SetVolume(needle.VolumeId(i), vol)
+	}
+
+	// Testing when there's still space
+	unUsedSpace := diskLocation.UnUsedSpace(1200)
+	if unUsedSpace != 600 {
+		t.Errorf("unUsedSpace incorrect: %d != %d", unUsedSpace, 1500)
+	}
+
+	// Testing when there's exactly 0 space
+	unUsedSpace = diskLocation.UnUsedSpace(1000)
+	if unUsedSpace != 0 {
+		t.Errorf("unUsedSpace incorrect: %d != %d", unUsedSpace, 0)
+	}
+
+	// Testing when there's negative free space
+	unUsedSpace = diskLocation.UnUsedSpace(900)
+	if unUsedSpace != 0 {
+		t.Errorf("unUsedSpace incorrect: %d != %d", unUsedSpace, 0)
+	}
+
+}