[master] Do Automatic Volume Grow in background (#5781)

* Do Automatic Volume Grow in backgound

* pass lastGrowCount to master

* fix build

* fix count to uint64

weed/server/master_grpc_server_assign.go

@@ -74,6 +74,7 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest
 	}
 
 	vl := ms.Topo.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType)
+	vl.SetLastGrowCount(req.WritableVolumeCount)
 
 	var (
 		lastErr    error
@@ -91,7 +92,7 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest
 			vl.AddGrowRequest()
 			ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{
 				Option: option,
-				Count:  int(req.WritableVolumeCount),
+				Count:  req.WritableVolumeCount,
 			}
 		}
 		if err != nil {

weed/server/master_grpc_server_volume.go

@@ -3,6 +3,8 @@ package weed_server
 import (
 	"context"
 	"fmt"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+	"math/rand"
 	"reflect"
 	"strings"
 	"sync"
@@ -18,7 +20,38 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 )
 
+func (ms *MasterServer) DoAutomaticVolumeGrow(req *topology.VolumeGrowRequest) {
+	glog.V(1).Infoln("starting automatic volume grow")
+	start := time.Now()
+	newVidLocations, err := ms.vg.AutomaticGrowByType(req.Option, ms.grpcDialOption, ms.Topo, req.Count)
+	glog.V(1).Infoln("finished automatic volume grow, cost ", time.Now().Sub(start))
+	if err != nil {
+		glog.V(1).Infof("automatic volume grow failed: %+v", err)
+		return
+	}
+	for _, newVidLocation := range newVidLocations {
+		ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: newVidLocation})
+	}
+}
+
 func (ms *MasterServer) ProcessGrowRequest() {
+	go func() {
+		for {
+			time.Sleep(14*time.Minute + time.Duration(120*rand.Float32())*time.Second)
+			if !ms.Topo.IsLeader() {
+				continue
+			}
+			for _, vl := range ms.Topo.ListVolumeLyauts() {
+				if !vl.HasGrowRequest() && vl.ShouldGrowVolumes(&topology.VolumeGrowOption{}) {
+					vl.AddGrowRequest()
+					ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{
+						Option: vl.ToGrowOption(),
+						Count:  vl.GetLastGrowCount(),
+					}
+				}
+			}
+		}
+	}()
 	go func() {
 		filter := sync.Map{}
 		for {
@@ -50,23 +83,11 @@ func (ms *MasterServer) ProcessGrowRequest() {
 			if !found && vl.ShouldGrowVolumes(option) {
 				filter.Store(req, nil)
 				// we have lock called inside vg
-				go func() {
-					glog.V(1).Infoln("starting automatic volume grow")
-					start := time.Now()
-					newVidLocations, err := ms.vg.AutomaticGrowByType(req.Option, ms.grpcDialOption, ms.Topo, req.Count)
-					glog.V(1).Infoln("finished automatic volume grow, cost ", time.Now().Sub(start))
-					if err == nil {
-						for _, newVidLocation := range newVidLocations {
-							ms.broadcastToClients(&master_pb.KeepConnectedResponse{VolumeLocation: newVidLocation})
-						}
-					} else {
-						glog.V(1).Infof("automatic volume grow failed: %+v", err)
-					}
+				go func(req *topology.VolumeGrowRequest, vl *topology.VolumeLayout) {
+					ms.DoAutomaticVolumeGrow(req)
 					vl.DoneGrowRequest()
-
 					filter.Delete(req)
-				}()
-
+				}(req, vl)
 			} else {
 				glog.V(4).Infoln("discard volume grow request")
 				time.Sleep(time.Millisecond * 211)

weed/server/master_server.go

@@ -92,15 +92,15 @@ func NewMasterServer(r *mux.Router, option *MasterOption, peers map[string]pb.Se
 	v.SetDefault("master.replication.treat_replication_as_minimums", false)
 	replicationAsMin := v.GetBool("master.replication.treat_replication_as_minimums")
 
-	v.SetDefault("master.volume_growth.copy_1", 7)
-	v.SetDefault("master.volume_growth.copy_2", 6)
-	v.SetDefault("master.volume_growth.copy_3", 3)
-	v.SetDefault("master.volume_growth.copy_other", 1)
-	v.SetDefault("master.volume_growth.threshold", 0.9)
-	topology.VolumeGrowStrategy.Copy1Count = v.GetInt("master.volume_growth.copy_1")
-	topology.VolumeGrowStrategy.Copy2Count = v.GetInt("master.volume_growth.copy_2")
-	topology.VolumeGrowStrategy.Copy3Count = v.GetInt("master.volume_growth.copy_3")
-	topology.VolumeGrowStrategy.CopyOtherCount = v.GetInt("master.volume_growth.copy_other")
+	v.SetDefault("master.volume_growth.copy_1", topology.VolumeGrowStrategy.Copy1Count)
+	v.SetDefault("master.volume_growth.copy_2", topology.VolumeGrowStrategy.Copy2Count)
+	v.SetDefault("master.volume_growth.copy_3", topology.VolumeGrowStrategy.Copy3Count)
+	v.SetDefault("master.volume_growth.copy_other", topology.VolumeGrowStrategy.CopyOtherCount)
+	v.SetDefault("master.volume_growth.threshold", topology.VolumeGrowStrategy.Threshold)
+	topology.VolumeGrowStrategy.Copy1Count = v.GetUint32("master.volume_growth.copy_1")
+	topology.VolumeGrowStrategy.Copy2Count = v.GetUint32("master.volume_growth.copy_2")
+	topology.VolumeGrowStrategy.Copy3Count = v.GetUint32("master.volume_growth.copy_3")
+	topology.VolumeGrowStrategy.CopyOtherCount = v.GetUint32("master.volume_growth.copy_other")
 	topology.VolumeGrowStrategy.Threshold = v.GetFloat64("master.volume_growth.threshold")
 
 	var preallocateSize int64

weed/server/master_server_handlers.go

@@ -107,7 +107,7 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request)
 		requestedCount = 1
 	}
 
-	writableVolumeCount, e := strconv.Atoi(r.FormValue("writableVolumeCount"))
+	writableVolumeCount, e := strconv.ParseUint(r.FormValue("writableVolumeCount"), 10, 32)
 	if e != nil {
 		writableVolumeCount = 0
 	}
@@ -145,7 +145,7 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request)
 			vl.AddGrowRequest()
 			ms.volumeGrowthRequestChan <- &topology.VolumeGrowRequest{
 				Option: option,
-				Count:  writableVolumeCount,
+				Count:  uint32(writableVolumeCount),
 			}
 		}
 		if err != nil {

weed/server/master_server_handlers_admin.go

@@ -70,7 +70,7 @@ func (ms *MasterServer) volumeVacuumHandler(w http.ResponseWriter, r *http.Reque
 }
 
 func (ms *MasterServer) volumeGrowHandler(w http.ResponseWriter, r *http.Request) {
-	count := 0
+	count := uint64(0)
 	option, err := ms.getVolumeGrowOption(r)
 	if err != nil {
 		writeJsonError(w, r, http.StatusNotAcceptable, err)
@@ -78,15 +78,16 @@ func (ms *MasterServer) volumeGrowHandler(w http.ResponseWriter, r *http.Request
 	}
 	glog.V(0).Infof("volumeGrowHandler received %v from %v", option.String(), r.RemoteAddr)
 
-	if count, err = strconv.Atoi(r.FormValue("count")); err == nil {
-		if ms.Topo.AvailableSpaceFor(option) < int64(count*option.ReplicaPlacement.GetCopyCount()) {
-			err = fmt.Errorf("only %d volumes left, not enough for %d", ms.Topo.AvailableSpaceFor(option), count*option.ReplicaPlacement.GetCopyCount())
+	if count, err = strconv.ParseUint(r.FormValue("count"), 10, 32); err == nil {
+		replicaCount := int64(count * uint64(option.ReplicaPlacement.GetCopyCount()))
+		if ms.Topo.AvailableSpaceFor(option) < replicaCount {
+			err = fmt.Errorf("only %d volumes left, not enough for %d", ms.Topo.AvailableSpaceFor(option), replicaCount)
 		} else if !ms.Topo.DataCenterExists(option.DataCenter) {
 			err = fmt.Errorf("data center %v not found in topology", option.DataCenter)
 		} else {
 			var newVidLocations []*master_pb.VolumeLocation
-			newVidLocations, err = ms.vg.GrowByCountAndType(ms.grpcDialOption, count, option, ms.Topo)
-			count = len(newVidLocations)
+			newVidLocations, err = ms.vg.GrowByCountAndType(ms.grpcDialOption, uint32(count), option, ms.Topo)
+			count = uint64(len(newVidLocations))
 		}
 	} else {
 		err = fmt.Errorf("can not parse parameter count %s", r.FormValue("count"))

weed/topology/topology_info.go

@@ -42,6 +42,15 @@ func (t *Topology) ToInfo() (info TopologyInfo) {
 	return
 }
 
+func (t *Topology) ListVolumeLyauts() (volumeLayouts []*VolumeLayout) {
+	for _, col := range t.collectionMap.Items() {
+		for _, volumeLayout := range col.(*Collection).storageType2VolumeLayout.Items() {
+			volumeLayouts = append(volumeLayouts, volumeLayout.(*VolumeLayout))
+		}
+	}
+	return volumeLayouts
+}
+
 func (t *Topology) ToVolumeMap() interface{} {
 	m := make(map[string]interface{})
 	m["Max"] = t.diskUsages.GetMaxVolumeCount()

weed/topology/volume_growth.go

@@ -27,14 +27,14 @@ This package is created to resolve these replica placement issues:
 
 type VolumeGrowRequest struct {
 	Option *VolumeGrowOption
-	Count  int
+	Count  uint32
 }
 
 type volumeGrowthStrategy struct {
-	Copy1Count     int
-	Copy2Count     int
-	Copy3Count     int
-	CopyOtherCount int
+	Copy1Count     uint32
+	Copy2Count     uint32
+	Copy3Count     uint32
+	CopyOtherCount uint32
 	Threshold      float64
 }
 
@@ -75,7 +75,7 @@ func NewDefaultVolumeGrowth() *VolumeGrowth {
 
 // one replication type may need rp.GetCopyCount() actual volumes
 // given copyCount, how many logical volumes to create
-func (vg *VolumeGrowth) findVolumeCount(copyCount int) (count int) {
+func (vg *VolumeGrowth) findVolumeCount(copyCount int) (count uint32) {
 	switch copyCount {
 	case 1:
 		count = VolumeGrowStrategy.Copy1Count
@@ -89,7 +89,7 @@ func (vg *VolumeGrowth) findVolumeCount(copyCount int) (count int) {
 	return
 }
 
-func (vg *VolumeGrowth) AutomaticGrowByType(option *VolumeGrowOption, grpcDialOption grpc.DialOption, topo *Topology, targetCount int) (result []*master_pb.VolumeLocation, err error) {
+func (vg *VolumeGrowth) AutomaticGrowByType(option *VolumeGrowOption, grpcDialOption grpc.DialOption, topo *Topology, targetCount uint32) (result []*master_pb.VolumeLocation, err error) {
 	if targetCount == 0 {
 		targetCount = vg.findVolumeCount(option.ReplicaPlacement.GetCopyCount())
 	}
@@ -99,11 +99,11 @@ func (vg *VolumeGrowth) AutomaticGrowByType(option *VolumeGrowOption, grpcDialOp
 	}
 	return result, err
 }
-func (vg *VolumeGrowth) GrowByCountAndType(grpcDialOption grpc.DialOption, targetCount int, option *VolumeGrowOption, topo *Topology) (result []*master_pb.VolumeLocation, err error) {
+func (vg *VolumeGrowth) GrowByCountAndType(grpcDialOption grpc.DialOption, targetCount uint32, option *VolumeGrowOption, topo *Topology) (result []*master_pb.VolumeLocation, err error) {
 	vg.accessLock.Lock()
 	defer vg.accessLock.Unlock()
 
-	for i := 0; i < targetCount; i++ {
+	for i := uint32(0); i < targetCount; i++ {
 		if res, e := vg.findAndGrow(grpcDialOption, topo, option); e == nil {
 			result = append(result, res...)
 		} else {

weed/topology/volume_layout.go

@@ -107,6 +107,7 @@ func (v *volumesBinaryState) copyState(list *VolumeLocationList) copyState {
 // mapping from volume to its locations, inverted from server to volume
 type VolumeLayout struct {
 	growRequest      atomic.Bool
+	lastGrowCount    atomic.Uint32
   rp               *super_block.ReplicaPlacement
 	ttl              *needle.TTL
 	diskType         types.DiskType
@@ -354,6 +355,16 @@ func (vl *VolumeLayout) DoneGrowRequest() {
 	vl.growRequest.Store(false)
 }
 
+func (vl *VolumeLayout) SetLastGrowCount(count uint32) {
+	if vl.lastGrowCount.Load() != count {
+		vl.lastGrowCount.Store(count)
+	}
+}
+
+func (vl *VolumeLayout) GetLastGrowCount() uint32 {
+	return vl.lastGrowCount.Load()
+}
+
 func (vl *VolumeLayout) ShouldGrowVolumes(option *VolumeGrowOption) bool {
 	total, active, crowded := vl.GetActiveVolumeCount(option)
 	stats.MasterVolumeLayout.WithLabelValues(option.Collection, option.DataCenter, "total").Set(float64(total))
@@ -539,6 +550,13 @@ func (vl *VolumeLayout) ToInfo() (info VolumeLayoutInfo) {
 	return
 }
 
+func (vl *VolumeLayout) ToGrowOption() (option *VolumeGrowOption) {
+	option.ReplicaPlacement = vl.rp
+	option.Ttl = vl.ttl
+	option.DiskType = vl.diskType
+	return
+}
+
 func (vl *VolumeLayout) Stats() *VolumeLayoutStats {
 	vl.accessLock.RLock()
 	defer vl.accessLock.RUnlock()