From 492e700d1fa5b806730f1be2e70fab7f2185565f Mon Sep 17 00:00:00 2001 From: Bruce Zou Date: Tue, 13 Oct 2026 05:30:03 +0800 Subject: [PATCH] Volume grow avg fill threshold (#11725) * master: add conditional grow by average fill ratio to VolumeGrow gRPC Callers that pre-grow volumes on fill level currently have to list all volumes of a collection and sum sizes themselves, which is expensive on clusters with millions of volumes. The master already tracks each writable volume's effective size in memory, so it can answer the question directly. Add grow_avg_fill_threshold to VolumeGrowRequest: when set (e.g. 0.6), growth only happens once the layout's writable volumes have reached that average fill ratio. A layout with no writable volumes skips the check, since that is the case growth exists to fix. VolumeGrowResponse now reports the observed avg_fill and writable_count whether or not growth was triggered. * master: unit tests for WritableAvgFill and FindVolumeLayout Covers the average-fill computation (mixed fills, no writables, a writable volume without a size-tracking entry yet) and that FindVolumeLayout resolves existing layouts without creating missing collections. * address review: report avg_fill and writable_count for every VolumeGrow The stats were only read when grow_avg_fill_threshold was set, so a request without a threshold got AvgFill=0 and WritableCount=0 even when writable volumes existed. Read the layout stats unconditionally and use the threshold only to decide whether growth happens. * 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> --------- Co-authored-by: Chris Lu Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/pb/master.proto | 10 ++ weed/pb/master_pb/master.pb.go | 49 ++++- weed/server/master_grpc_server_volume.go | 26 ++- weed/topology/data_node.go | 13 ++ weed/topology/topology.go | 16 ++ weed/topology/volume_layout.go | 47 ++++- weed/topology/volume_layout_avg_fill_test.go | 179 +++++++++++++++++++ 7 files changed, 325 insertions(+), 15 deletions(-) create mode 100644 weed/topology/volume_layout_avg_fill_test.go diff --git a/weed/pb/master.proto b/weed/pb/master.proto index 153ed085b..aafa55887 100644 --- a/weed/pb/master.proto +++ b/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; } diff --git a/weed/pb/master_pb/master.pb.go b/weed/pb/master_pb/master.pb.go index 69833d906..49278c403 100644 --- a/weed/pb/master_pb/master.pb.go +++ b/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" + diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index b009c2ef1..716d06779 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -509,6 +509,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 { @@ -517,5 +538,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 } diff --git a/weed/topology/data_node.go b/weed/topology/data_node.go index 00325efb7..ca110ab4b 100644 --- a/weed/topology/data_node.go +++ b/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 { diff --git a/weed/topology/topology.go b/weed/topology/topology.go index f5717dd44..a892a0bad 100644 --- a/weed/topology/topology.go +++ b/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. diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index 31847adb5..1213b30fc 100644 --- a/weed/topology/volume_layout.go +++ b/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 diff --git a/weed/topology/volume_layout_avg_fill_test.go b/weed/topology/volume_layout_avg_fill_test.go new file mode 100644 index 000000000..23a85512d --- /dev/null +++ b/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") + } +}