master: scope the grow fill check to the requested target

A VolumeGrow aimed at a data center, rack or node must not skip growth
because fill elsewhere keeps the layout average below the threshold.
Measure only writables serving the target; a target with none still
grows.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>

weed/pb/master.proto

@@ -305,6 +305,11 @@ message VolumeGrowRequest {
   string data_node = 7;
   uint32 memory_map_max_size_mb = 8;
   string disk_type = 9;
+  // If > 0, grow only when the average fill ratio of the layout's writable
+  // volumes has reached this threshold (e.g. 0.6). Lets callers pre-grow on
+  // fill level with one call, instead of listing volumes to compute the
+  // ratio themselves. Ignored when there are no writable volumes.
+  float grow_avg_fill_threshold = 10;
 }
 
 message AssignResponse {
@@ -611,4 +616,9 @@ message RaftLeadershipTransferResponse {
 }
 
 message VolumeGrowResponse {
+  // Average fill ratio of the layout's writable volumes at request time
+  // (0 when there are none). Reported whether or not growth was triggered.
+  float avg_fill = 1;
+  // Number of writable volumes at request time.
+  uint32 writable_count = 2;
 }

weed/pb/master_pb/master.pb.go

@@ -1739,8 +1739,13 @@ type VolumeGrowRequest struct {
 	DataNode            string                 `protobuf:"bytes,7,opt,name=data_node,json=dataNode,proto3" json:"data_node,omitempty"`
 	MemoryMapMaxSizeMb  uint32                 `protobuf:"varint,8,opt,name=memory_map_max_size_mb,json=memoryMapMaxSizeMb,proto3" json:"memory_map_max_size_mb,omitempty"`
 	DiskType            string                 `protobuf:"bytes,9,opt,name=disk_type,json=diskType,proto3" json:"disk_type,omitempty"`
-	unknownFields       protoimpl.UnknownFields
-	sizeCache           protoimpl.SizeCache
+	// If > 0, grow only when the average fill ratio of the layout's writable
+	// volumes has reached this threshold (e.g. 0.6). Lets callers pre-grow on
+	// fill level with one call, instead of listing volumes to compute the
+	// ratio themselves. Ignored when there are no writable volumes.
+	GrowAvgFillThreshold float32 `protobuf:"fixed32,10,opt,name=grow_avg_fill_threshold,json=growAvgFillThreshold,proto3" json:"grow_avg_fill_threshold,omitempty"`
+	unknownFields        protoimpl.UnknownFields
+	sizeCache            protoimpl.SizeCache
 }
 
 func (x *VolumeGrowRequest) Reset() {
@@ -1836,6 +1841,13 @@ func (x *VolumeGrowRequest) GetDiskType() string {
 	return ""
 }
 
+func (x *VolumeGrowRequest) GetGrowAvgFillThreshold() float32 {
+	if x != nil {
+		return x.GrowAvgFillThreshold
+	}
+	return 0
+}
+
 type AssignResponse struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Fid           string                 `protobuf:"bytes,1,opt,name=fid,proto3" json:"fid,omitempty"`
@@ -4659,7 +4671,12 @@ func (x *RaftLeadershipTransferResponse) GetNewLeader() string {
 }
 
 type VolumeGrowResponse struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
+	state protoimpl.MessageState `protogen:"open.v1"`
+	// Average fill ratio of the layout's writable volumes at request time
+	// (0 when there are none). Reported whether or not growth was triggered.
+	AvgFill float32 `protobuf:"fixed32,1,opt,name=avg_fill,json=avgFill,proto3" json:"avg_fill,omitempty"`
+	// Number of writable volumes at request time.
+	WritableCount uint32 `protobuf:"varint,2,opt,name=writable_count,json=writableCount,proto3" json:"writable_count,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -4694,6 +4711,20 @@ func (*VolumeGrowResponse) Descriptor() ([]byte, []int) {
 	return file_master_proto_rawDescGZIP(), []int{68}
 }
 
+func (x *VolumeGrowResponse) GetAvgFill() float32 {
+	if x != nil {
+		return x.AvgFill
+	}
+	return 0
+}
+
+func (x *VolumeGrowResponse) GetWritableCount() uint32 {
+	if x != nil {
+		return x.WritableCount
+	}
+	return 0
+}
+
 type SuperBlockExtra_ErasureCoding struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	Data          uint32                 `protobuf:"varint,1,opt,name=data,proto3" json:"data,omitempty"`
@@ -5229,7 +5260,7 @@ const file_master_proto_rawDesc = "" +
 	"\x15writable_volume_count\x18\t \x01(\rR\x13writableVolumeCount\x12\x1b\n" +
 	"\tdisk_type\x18\n" +
 	" \x01(\tR\bdiskType\x12,\n" +
-	"\x12expected_data_size\x18\v \x01(\x04R\x10expectedDataSize\"\xbe\x02\n" +
+	"\x12expected_data_size\x18\v \x01(\x04R\x10expectedDataSize\"\xf5\x02\n" +
 	"\x11VolumeGrowRequest\x122\n" +
 	"\x15writable_volume_count\x18\x01 \x01(\rR\x13writableVolumeCount\x12 \n" +
 	"\vreplication\x18\x02 \x01(\tR\vreplication\x12\x1e\n" +
@@ -5242,7 +5273,9 @@ const file_master_proto_rawDesc = "" +
 	"\x04rack\x18\x06 \x01(\tR\x04rack\x12\x1b\n" +
 	"\tdata_node\x18\a \x01(\tR\bdataNode\x122\n" +
 	"\x16memory_map_max_size_mb\x18\b \x01(\rR\x12memoryMapMaxSizeMb\x12\x1b\n" +
-	"\tdisk_type\x18\t \x01(\tR\bdiskType\"\xc4\x01\n" +
+	"\tdisk_type\x18\t \x01(\tR\bdiskType\x125\n" +
+	"\x17grow_avg_fill_threshold\x18\n" +
+	" \x01(\x02R\x14growAvgFillThreshold\"\xc4\x01\n" +
 	"\x0eAssignResponse\x12\x10\n" +
 	"\x03fid\x18\x01 \x01(\tR\x03fid\x12\x14\n" +
 	"\x05count\x18\x04 \x01(\x04R\x05count\x12\x14\n" +
@@ -5477,8 +5510,10 @@ const file_master_proto_rawDesc = "" +
 	"\x1eRaftLeadershipTransferResponse\x12'\n" +
 	"\x0fprevious_leader\x18\x01 \x01(\tR\x0epreviousLeader\x12\x1d\n" +
 	"\n" +
-	"new_leader\x18\x02 \x01(\tR\tnewLeader\"\x14\n" +
-	"\x12VolumeGrowResponse2\xf1\x12\n" +
+	"new_leader\x18\x02 \x01(\tR\tnewLeader\"V\n" +
+	"\x12VolumeGrowResponse\x12\x19\n" +
+	"\bavg_fill\x18\x01 \x01(\x02R\aavgFill\x12%\n" +
+	"\x0ewritable_count\x18\x02 \x01(\rR\rwritableCount2\xf1\x12\n" +
 	"\aSeaweed\x12I\n" +
 	"\rSendHeartbeat\x12\x14.master_pb.Heartbeat\x1a\x1c.master_pb.HeartbeatResponse\"\x00(\x010\x01\x12X\n" +
 	"\rKeepConnected\x12\x1f.master_pb.KeepConnectedRequest\x1a .master_pb.KeepConnectedResponse\"\x00(\x010\x01\x12Q\n" +

weed/server/master_grpc_server_volume.go

@@ -492,6 +492,27 @@ func (ms *MasterServer) VolumeGrow(ctx context.Context, req *master_pb.VolumeGro
 		Force:  true,
 		Reason: "grpc volume grow",
 	}
+
+	// Report the layout's fill observed at request time whether or not
+	// growth was triggered, so callers can tell an empty layout from a
+	// skipped one. A missing collection/layout reports zeros.
+	var avgFill float64
+	var writableCount int
+	if vl := ms.Topo.FindVolumeLayout(req.Collection, replicaPlacement, ttl, types.ToDiskType(req.DiskType)); vl != nil {
+		avgFill, writableCount = vl.WritableAvgFill(&volumeGrowOption)
+	}
+
+	// Conditional grow: with a threshold set, grow only when the layout's
+	// writable volumes have reached this average fill. A layout with no
+	// writable volumes (none yet, or all just filled) skips the check — that
+	// is the case growth exists to fix.
+	if req.GrowAvgFillThreshold > 0 && writableCount > 0 && avgFill < float64(req.GrowAvgFillThreshold) {
+		return &master_pb.VolumeGrowResponse{
+			AvgFill:       float32(avgFill),
+			WritableCount: uint32(writableCount),
+		}, nil
+	}
+
 	replicaCount := int64(req.WritableVolumeCount * uint32(replicaPlacement.GetCopyCount()))
 
 	if ms.Topo.AvailableSpaceFor(&volumeGrowOption) < replicaCount {
@@ -500,5 +521,8 @@ func (ms *MasterServer) VolumeGrow(ctx context.Context, req *master_pb.VolumeGro
 
 	ms.DoAutomaticVolumeGrow(&volumeGrowRequest)
 
-	return &master_pb.VolumeGrowResponse{}, nil
+	return &master_pb.VolumeGrowResponse{
+		AvgFill:       float32(avgFill),
+		WritableCount: uint32(writableCount),
+	}, nil
 }

weed/topology/data_node.go

@@ -343,6 +343,19 @@ func (dn *DataNode) GetRack() *Rack {
 	return dn.Parent().(*NodeImpl).value.(*Rack)
 }
 
+func (dn *DataNode) matchesLocation(option *VolumeGrowOption) bool {
+	if option.DataCenter != "" && dn.GetDataCenter().Id() != NodeId(option.DataCenter) {
+		return false
+	}
+	if option.Rack != "" && dn.GetRack().Id() != NodeId(option.Rack) {
+		return false
+	}
+	if option.DataNode != "" && dn.Id() != NodeId(option.DataNode) {
+		return false
+	}
+	return true
+}
+
 func (dn *DataNode) GetTopology() *Topology {
 	p := dn.Parent()
 	for p.Parent() != nil {

weed/topology/topology.go

@@ -448,6 +448,22 @@ func (t *Topology) GetVolumeLayout(collectionName string, rp *super_block.Replic
 	}).(*Collection).GetOrCreateVolumeLayout(rp, ttl, diskType)
 }
 
+// FindVolumeLayout is the read-only counterpart of GetVolumeLayout: it returns
+// nil when the collection or its layout does not exist, instead of creating
+// one. Lets callers check a layout's state without polluting the collection
+// map with an empty collection.
+func (t *Topology) FindVolumeLayout(collectionName string, rp *super_block.ReplicaPlacement, ttl *needle.TTL, diskType types.DiskType) *VolumeLayout {
+	c, ok := t.collectionMap.Find(collectionName)
+	if !ok {
+		return nil
+	}
+	vl, ok := c.(*Collection).GetVolumeLayout(rp, ttl, diskType)
+	if !ok {
+		return nil
+	}
+	return vl
+}
+
 // DecayQuietVolumeSizes decays pending assign estimates across every layout.
 // A volume that changed reports within a pulse, so two quiet pulses mean the
 // size on record is the size there is.

weed/topology/volume_layout.go

@@ -650,13 +650,7 @@ func (vl *VolumeLayout) PickForWrite(count uint64, option *VolumeGrowOption) (vi
 		writableVolumeId := vl.writables[(start+i)%lenWriters]
 		volumeLocationList := vl.vid2location[writableVolumeId]
 		for _, dn := range volumeLocationList.list {
-			if option.DataCenter != "" && dn.GetDataCenter().Id() != NodeId(option.DataCenter) {
-				continue
-			}
-			if option.Rack != "" && dn.GetRack().Id() != NodeId(option.Rack) {
-				continue
-			}
-			if option.DataNode != "" && dn.Id() != NodeId(option.DataNode) {
+			if !dn.matchesLocation(option) {
 				continue
 			}
 			sample[found] = writableVolumeId
@@ -762,6 +756,45 @@ func (vl *VolumeLayout) ShouldGrowVolumes() bool {
 	return writable <= crowded
 }
 
+// WritableAvgFill returns the average fill ratio of the writable volumes and
+// how many there are. The fill of a single volume is its tracked effective
+// size over the layout's volume size limit; volumes without a size-tracking
+// entry yet (freshly created, no heartbeat size report so far) count as
+// empty. A non-empty location filter in option narrows the measure to
+// writables serving that target, so a targeted grow is not blocked or
+// triggered by fill elsewhere in the layout.
+func (vl *VolumeLayout) WritableAvgFill(option *VolumeGrowOption) (avgFill float64, writableCount int) {
+	vl.accessLock.RLock()
+	defer vl.accessLock.RUnlock()
+
+	var total uint64
+	for _, vid := range vl.writables {
+		if option != nil && (option.DataCenter != "" || option.Rack != "" || option.DataNode != "") {
+			matched := false
+			if loc := vl.vid2location[vid]; loc != nil {
+				for _, dn := range loc.list {
+					if dn.matchesLocation(option) {
+						matched = true
+						break
+					}
+				}
+			}
+			if !matched {
+				continue
+			}
+		}
+		writableCount++
+		if st := vl.sizeTracking[vid]; st != nil {
+			total += st.effectiveSize
+		}
+	}
+	if writableCount == 0 || vl.volumeSizeLimit == 0 {
+		return 0, writableCount
+	}
+	avgFill = float64(total) / (float64(writableCount) * float64(vl.volumeSizeLimit))
+	return avgFill, writableCount
+}
+
 func (vl *VolumeLayout) ShouldGrowVolumesByDcAndRack(writables *[]needle.VolumeId, dcId NodeId, rackId NodeId) bool {
 	// When replication spans multiple racks (DiffRackCount > 0), a writable
 	// volume's replicas only cover some racks in a DC. It is wrong to

weed/topology/volume_layout_avg_fill_test.go

@@ -0,0 +1,179 @@
+package topology
+
+import (
+	"math"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
+	"github.com/seaweedfs/seaweedfs/weed/storage/types"
+)
+
+func TestWritableAvgFill(t *testing.T) {
+	// 3 writable volumes at 20%, 50%, 80% fill (sizes 2000, 5000, 8000 of
+	// limit 10000): average fill is 50%.
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":2000, "replication":"000"},
+          {"id":2, "size":5000, "replication":"000"},
+          {"id":3, "size":8000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	avgFill, writableCount := vl.WritableAvgFill(nil)
+	if writableCount != 3 {
+		t.Fatalf("expected 3 writable volumes, got %d", writableCount)
+	}
+	if math.Abs(avgFill-0.5) > 1e-9 {
+		t.Errorf("expected avg fill 0.5, got %v", avgFill)
+	}
+}
+
+func TestWritableAvgFillNoWritables(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":2000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	vl.accessLock.Lock()
+	vl.removeFromWritable(1)
+	vl.accessLock.Unlock()
+
+	avgFill, writableCount := vl.WritableAvgFill(nil)
+	if writableCount != 0 || avgFill != 0 {
+		t.Errorf("expected (0, 0) with no writables, got (%v, %d)", avgFill, writableCount)
+	}
+}
+
+// A writable volume without a size-tracking entry yet — freshly created, no
+// heartbeat size report so far — counts as empty, not as an error.
+func TestWritableAvgFillUntrackedVolumeCountsAsEmpty(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":5000, "replication":"000"},
+          {"id":2, "size":5000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	vl.accessLock.Lock()
+	delete(vl.sizeTracking, 2)
+	vl.accessLock.Unlock()
+
+	avgFill, writableCount := vl.WritableAvgFill(nil)
+	if writableCount != 2 {
+		t.Fatalf("expected 2 writable volumes, got %d", writableCount)
+	}
+	if math.Abs(avgFill-0.25) > 1e-9 {
+		t.Errorf("expected avg fill 0.25 (one of two volumes empty), got %v", avgFill)
+	}
+}
+
+func TestWritableAvgFillScopedToTarget(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":9000, "replication":"000"},
+          {"id":2, "size":9000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  },
+  "dc2":{
+    "rack2":{
+      "server2":{
+        "volumes":[
+          {"id":3, "size":1000, "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	_, vl := setupPickTest(t, layout, 10000)
+
+	avgFill, writableCount := vl.WritableAvgFill(&VolumeGrowOption{DataCenter: "dc2"})
+	if writableCount != 1 || math.Abs(avgFill-0.1) > 1e-9 {
+		t.Fatalf("expected dc2 fill (0.1, 1), got (%v, %d)", avgFill, writableCount)
+	}
+
+	// A target with no writable volumes reports none, so growth is not
+	// blocked by fill elsewhere.
+	avgFill, writableCount = vl.WritableAvgFill(&VolumeGrowOption{DataCenter: "dc2", DataNode: "missing"})
+	if writableCount != 0 || avgFill != 0 {
+		t.Fatalf("expected (0, 0) for a target with no writables, got (%v, %d)", avgFill, writableCount)
+	}
+}
+
+func TestFindVolumeLayout(t *testing.T) {
+	layout := `
+{
+  "dc1":{
+    "rack1":{
+      "server1":{
+        "volumes":[
+          {"id":1, "size":2000, "collection":"colA", "replication":"000"}
+        ],
+        "limit":10
+      }
+    }
+  }
+}
+`
+	topo := setupWithLimit(t, layout, 10000)
+	rp, _ := super_block.NewReplicaPlacementFromString("000")
+
+	// Existing layout is found.
+	if vl := topo.FindVolumeLayout("colA", rp, needle.EMPTY_TTL, types.HardDriveType); vl == nil {
+		t.Error("expected layout for colA to be found")
+	}
+
+	// A missing collection resolves to nil and must not be created.
+	if vl := topo.FindVolumeLayout("noSuchCollection", rp, needle.EMPTY_TTL, types.HardDriveType); vl != nil {
+		t.Error("expected nil layout for unknown collection")
+	}
+	if _, ok := topo.collectionMap.Find("noSuchCollection"); ok {
+		t.Error("FindVolumeLayout must not create the collection")
+	}
+
+	// A missing layout within an existing collection also resolves to nil.
+	rpOther, _ := super_block.NewReplicaPlacementFromString("001")
+	if vl := topo.FindVolumeLayout("colA", rpOther, needle.EMPTY_TTL, types.HardDriveType); vl != nil {
+		t.Error("expected nil layout for unused replication")
+	}
+}