fix no more writables volumes while disk free space is sufficient (#4491)

Co-authored-by: wang wusong <wangwusong@virtaitech.com>

2 files changed, 17 insertions, 0 deletions

diff --git a/weed/topology/node.go b/weed/topology/node.go
index ae58891e4..89a6305e8 100644
--- a/weed/topology/node.go
+++ b/weed/topology/node.go
@@ -243,6 +243,7 @@ func (n *NodeImpl) CollectDeadNodeAndFullVolumes(freshThreshHold int64, volumeSi
 	if n.IsRack() {
 		for _, c := range n.Children() {
 			dn := c.(*DataNode) //can not cast n to DataNode
+			dn.RLock()
 			for _, v := range dn.GetVolumes() {
 				if v.Size >= volumeSizeLimit {
 					//fmt.Println("volume",v.Id,"size",v.Size,">",volumeSizeLimit)
@@ -259,6 +260,7 @@ func (n *NodeImpl) CollectDeadNodeAndFullVolumes(freshThreshHold int64, volumeSi
 					}
 				}
 			}
+			dn.RUnlock()
 		}
 	} else {
 		for _, c := range n.Children() {
diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go
index d302afe8c..1581d52fb 100644
--- a/weed/topology/topology_vacuum.go
+++ b/weed/topology/topology_vacuum.go
@@ -172,6 +172,21 @@ func (t *Topology) batchVacuumVolumeCommit(grpcDialOption grpc.DialOption, vl *V
 	}
 
 	if isCommitSuccess {
+		//reset all vacuumed volumes size
+		for _, dn := range vacuumLocationList.list {
+			vInfo, err := dn.GetVolumesById(vid)
+			if err != nil {
+				glog.V(0).Infof("get volume info for volume: %d failed %v", vid, err)
+				return false
+			}
+
+			dn.Lock()
+			disk := dn.getOrCreateDisk(vInfo.DiskType)
+			vInfo.Size = 0
+			disk.doAddOrUpdateVolume(vInfo)
+			dn.Unlock()
+		}
+
 		for _, dn := range vacuumLocationList.list {
 			vl.SetVolumeAvailable(dn, vid, isReadOnly)
 		}