[master] grow volumes if no writable volumes in current dataCenter (#5434)

* grow volumes if no writable volumes in current dataCenter
https://github.com/seaweedfs/seaweedfs/issues/3886

* fix tests with volume grow

* automatic volume grow one volume

* add ErrorChunkAssign metrics

weed/operation/assign_file_id.go

@@ -6,6 +6,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
 	"github.com/seaweedfs/seaweedfs/weed/security"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"google.golang.org/grpc"
 	"sync"
@@ -193,6 +194,7 @@ func Assign(masterFn GetMasterFn, grpcDialOption grpc.DialOption, primaryRequest
 		})
 
 		if lastError != nil {
+			stats.FilerHandlerCounter.WithLabelValues(stats.ErrorChunkAssign).Inc()
 			continue
 		}
 
@@ -262,6 +264,7 @@ func (so *StorageOption) ToAssignRequests(count int) (ar *VolumeAssignRequest, a
 		WritableVolumeCount: so.VolumeGrowthCount,
 	}
 	if so.DataCenter != "" || so.Rack != "" || so.DataNode != "" {
+		ar.WritableVolumeCount = uint32(count)
 		altRequest = &VolumeAssignRequest{
 			Count:               uint64(count),
 			Replication:         so.Replication,

weed/stats/metrics_names.go

@@ -40,6 +40,7 @@ const (
 	ErrorReadInternal        = "read.internal.error"
 	ErrorWriteEntry          = "write.entry.failed"
 	RepeatErrorUploadContent = "upload.content.repeat.failed"
+	ErrorChunkAssign         = "chunkAssign.failed"
 	ErrorReadCache           = "read.cache.failed"
 	ErrorReadStream          = "read.stream.failed"
 )

weed/topology/volume_growth_test.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 	"testing"
 
 	"github.com/seaweedfs/seaweedfs/weed/sequence"
@@ -419,11 +420,13 @@ func TestPickForWrite(t *testing.T) {
 		Rack:       "",
 		DataNode:   "",
 	}
+	v := util.GetViper()
+	v.Set("master.volume_growth.threshold", 0.9)
 	for _, rpStr := range []string{"001", "010", "100"} {
 		rp, _ := super_block.NewReplicaPlacementFromString(rpStr)
 		vl := topo.GetVolumeLayout("test", rp, needle.EMPTY_TTL, types.HardDriveType)
 		volumeGrowOption.ReplicaPlacement = rp
-		for _, dc := range []string{"", "dc1", "dc2", "dc3"} {
+		for _, dc := range []string{"", "dc1", "dc2", "dc3", "dc0"} {
 			volumeGrowOption.DataCenter = dc
 			for _, r := range []string{""} {
 				volumeGrowOption.Rack = r
@@ -432,8 +435,13 @@ func TestPickForWrite(t *testing.T) {
 						continue
 					}
 					volumeGrowOption.DataNode = dn
-					fileId, count, _, _, err := topo.PickForWrite(1, volumeGrowOption, vl)
+					fileId, count, _, shouldGrow, err := topo.PickForWrite(1, volumeGrowOption, vl)
-					if err != nil {
+					if dc == "dc0" {
+						if err == nil || count != 0 || !shouldGrow {
+							fmt.Println(dc, r, dn, "pick for write should be with error")
+							t.Fail()
+						}
+					} else if err != nil {
 						fmt.Println(dc, r, dn, "pick for write error :", err)
 						t.Fail()
 					} else if count == 0 {
@@ -442,6 +450,9 @@ func TestPickForWrite(t *testing.T) {
 					} else if len(fileId) == 0 {
 						fmt.Println(dc, r, dn, "pick for write file id is empty")
 						t.Fail()
+					} else if shouldGrow {
+						fmt.Println(dc, r, dn, "pick for write error : not should grow")
+						t.Fail()
 					}
 				}
 			}

weed/topology/volume_layout.go

@@ -336,7 +336,7 @@ func (vl *VolumeLayout) PickForWrite(count uint64, option *VolumeGrowOption) (vi
 			return
 		}
 	}
-	return vid, count, locationList, shouldGrow, fmt.Errorf("No writable volumes in DataCenter:%v Rack:%v DataNode:%v", option.DataCenter, option.Rack, option.DataNode)
+	return vid, count, locationList, true, fmt.Errorf("No writable volumes in DataCenter:%v Rack:%v DataNode:%v", option.DataCenter, option.Rack, option.DataNode)
 }
 
 func (vl *VolumeLayout) HasGrowRequest() bool {