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/server/master_grpc_server_volume.go

@@ -499,7 +499,7 @@ func (ms *MasterServer) VolumeGrow(ctx context.Context, req *master_pb.VolumeGro
 	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()
+		avgFill, writableCount = vl.WritableAvgFill(&volumeGrowOption)
 	}
 
 	// Conditional grow: with a threshold set, grow only when the layout's

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/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
@@ -766,22 +760,37 @@ func (vl *VolumeLayout) ShouldGrowVolumes() bool {
 // 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. This lets a caller decide whether to grow proactively on fill level,
-// without listing all volumes to compute the ratio itself.
-func (vl *VolumeLayout) WritableAvgFill() (avgFill float64, writableCount int) {
+// 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()
 
-	writableCount = len(vl.writables)
-	if writableCount == 0 || vl.volumeSizeLimit == 0 {
-		return 0, writableCount
-	}
 	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
 }

weed/topology/volume_layout_avg_fill_test.go

@@ -30,7 +30,7 @@ func TestWritableAvgFill(t *testing.T) {
 `
 	_, vl := setupPickTest(t, layout, 10000)
 
-	avgFill, writableCount := vl.WritableAvgFill()
+	avgFill, writableCount := vl.WritableAvgFill(nil)
 	if writableCount != 3 {
 		t.Fatalf("expected 3 writable volumes, got %d", writableCount)
 	}
@@ -60,7 +60,7 @@ func TestWritableAvgFillNoWritables(t *testing.T) {
 	vl.removeFromWritable(1)
 	vl.accessLock.Unlock()
 
-	avgFill, writableCount := vl.WritableAvgFill()
+	avgFill, writableCount := vl.WritableAvgFill(nil)
 	if writableCount != 0 || avgFill != 0 {
 		t.Errorf("expected (0, 0) with no writables, got (%v, %d)", avgFill, writableCount)
 	}
@@ -90,7 +90,7 @@ func TestWritableAvgFillUntrackedVolumeCountsAsEmpty(t *testing.T) {
 	delete(vl.sizeTracking, 2)
 	vl.accessLock.Unlock()
 
-	avgFill, writableCount := vl.WritableAvgFill()
+	avgFill, writableCount := vl.WritableAvgFill(nil)
 	if writableCount != 2 {
 		t.Fatalf("expected 2 writable volumes, got %d", writableCount)
 	}
@@ -99,6 +99,47 @@ func TestWritableAvgFillUntrackedVolumeCountsAsEmpty(t *testing.T) {
 	}
 }
 
+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 := `
 {