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") + } +}