From 41260a55f68eaa1950caddbab168091388855551 Mon Sep 17 00:00:00 2001 From: ingardm Date: Mon, 12 Oct 2026 12:10:41 +0200 Subject: [PATCH] master: record writableVolumeCount on HTTP /dir/assign like gRPC Assign (#11716) * master: extract the periodic per-layout grow decision Move the must-grow and crowded-step decision of the periodic growth scan in ProcessGrowRequest into periodicGrowCount, a pure function of the layout's lastGrowCount, writable and crowded counts, so it can be tested without running the scan loop or growing real volumes. No behavior change. Co-Authored-By: Claude Opus 5.5 * master: record writableVolumeCount on HTTP /dir/assign The periodic growth scan (#5781) keeps a layout supplied ahead of demand only when it knows how many writable volumes clients want: lastGrowCount, which gRPC Assign records with SetLastGrowCount. HTTP /dir/assign parsed writableVolumeCount but only used it to size the emergency grow when no writable volume was left, so the scan always saw lastGrowCount 0 for HTTP-only writers. Since #5821 removed the per-assign threshold grow, such a layout only grew by the rack-aware step of 2, and only once every writable volume was crowded. When a batch of volumes filled in lockstep, those 2 took every write and could fill before the next scan, leaving no writable volumes until an assign-triggered emergency grow landed, with 503s for every other assign meanwhile. Record writableVolumeCount on HTTP assigns as gRPC Assign does. Clients that do not send it are unaffected. Co-Authored-By: Claude Opus 5.5 * master: reject unknown rack/dataNode before recording grow targets An assign naming a rack or data node outside the topology can never be served, but lastGrowCount was recorded first. The periodic growth scan drops the requested placement, so it kept topping the layout up with volumes the requester could not use. Both assign paths now validate the placement like they already validate the data center. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * master: scope placement existence checks to the requested path A rack in another data center or a data node under another rack passed validation yet could never serve the request, and the grow count was still recorded. The rack and data-node lookups now take the requested parents, matching the filters the placement engine applies. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: Claude Opus 5.5 Co-authored-by: Chris Lu Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/server/master_grpc_server_assign.go | 6 + weed/server/master_grpc_server_volume.go | 27 ++- weed/server/master_server_handlers.go | 16 ++ .../master_server_handlers_assign_test.go | 209 ++++++++++++++++++ weed/topology/topology.go | 48 ++++ 5 files changed, 301 insertions(+), 5 deletions(-) create mode 100644 weed/server/master_server_handlers_assign_test.go diff --git a/weed/server/master_grpc_server_assign.go b/weed/server/master_grpc_server_assign.go index f981d3424..703a6ab3a 100644 --- a/weed/server/master_grpc_server_assign.go +++ b/weed/server/master_grpc_server_assign.go @@ -92,6 +92,12 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest if !ms.Topo.DataCenterExists(option.DataCenter) { return nil, fmt.Errorf("data center %v not found in topology", option.DataCenter) } + if !ms.Topo.RackExists(option.DataCenter, option.Rack) { + return nil, fmt.Errorf("rack %v not found in topology", option.Rack) + } + if !ms.Topo.DataNodeExists(option.DataCenter, option.Rack, option.DataNode) { + return nil, fmt.Errorf("data node %v not found in topology", option.DataNode) + } vl := ms.Topo.GetVolumeLayout(option.Collection, option.ReplicaPlacement, option.Ttl, option.DiskType) if req.DiskType == "" { diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index deb4a4f0e..b009c2ef1 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -45,6 +45,23 @@ func (ms *MasterServer) DoAutomaticVolumeGrow(req *topology.VolumeGrowRequest) { ms.broadcastVolumeLocationsToClients(newVidLocations) } +// periodicGrowCount decides how many volumes the periodic growth scan adds to +// a layout ahead of its rack-aware pass, from the writableVolumeCount clients +// last asked for (lastGrowCount) and the layout's writable and crowded counts. +// mustGrow reports a refill back up to lastGrowCount writable volumes, which +// the caller only grows when there is room for all of it. A layout that never +// received a writableVolumeCount (lastGrowCount 0) is left to the rack-aware +// pass. +func periodicGrowCount(lastGrowCount uint32, writable, crowded int) (count uint32, mustGrow bool) { + if n := int(lastGrowCount) - writable; n > 0 { + return uint32(n), true + } + if lastGrowCount > 0 && writable < int(lastGrowCount*2) && float64(crowded+volumeGrowStepCount) > float64(writable)*topology.VolumeGrowStrategy.Threshold { + return volumeGrowStepCount, false + } + return 0, false +} + func (ms *MasterServer) ProcessGrowRequest() { go func() { ctx := context.Background() @@ -67,25 +84,25 @@ func (ms *MasterServer) ProcessGrowRequest() { continue } writable, crowded := vl.GetWritableVolumeCount() - mustGrow := int(lastGrowCount) - writable vgr := vlc.ToVolumeGrowRequest() underReplicated := vl.CountUnderReplicatedVolumes() stats.MasterVolumeLayoutWritable.WithLabelValues(vlc.Collection, vgr.DiskType, vgr.Replication, vgr.Ttl).Set(float64(writable)) stats.MasterVolumeLayoutCrowded.WithLabelValues(vlc.Collection, vgr.DiskType, vgr.Replication, vgr.Ttl).Set(float64(crowded)) stats.MasterUnderReplicatedVolumes.WithLabelValues(vlc.Collection, vgr.DiskType, vgr.Replication, vgr.Ttl).Set(float64(underReplicated)) + growCount, mustGrow := periodicGrowCount(lastGrowCount, writable, crowded) switch { - case mustGrow > 0: + case mustGrow: if rp, rpErr := super_block.NewReplicaPlacementFromString(vgr.Replication); rpErr != nil { glog.V(0).Infof("failed to parse replica placement %s: %v", vgr.Replication, rpErr) } else { - vgr.WritableVolumeCount = uint32(mustGrow) + vgr.WritableVolumeCount = growCount if ms.Topo.AvailableSpaceFor(&topology.VolumeGrowOption{DiskType: types.ToDiskType(vgr.DiskType)}) >= int64(vgr.WritableVolumeCount*uint32(rp.GetCopyCount())) { _, err = ms.VolumeGrow(ctx, vgr) } } - case lastGrowCount > 0 && writable < int(lastGrowCount*2) && float64(crowded+volumeGrowStepCount) > float64(writable)*topology.VolumeGrowStrategy.Threshold: - vgr.WritableVolumeCount = volumeGrowStepCount + case growCount > 0: + vgr.WritableVolumeCount = growCount _, err = ms.VolumeGrow(ctx, vgr) } if err != nil { diff --git a/weed/server/master_server_handlers.go b/weed/server/master_server_handlers.go index dc40b62cc..4bb583dc7 100644 --- a/weed/server/master_server_handlers.go +++ b/weed/server/master_server_handlers.go @@ -190,6 +190,22 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) }) return } + if !ms.Topo.RackExists(option.DataCenter, option.Rack) { + writeJsonQuiet(w, r, http.StatusBadRequest, operation.AssignResult{ + Error: fmt.Sprintf("rack %v not found in topology", option.Rack), + }) + return + } + if !ms.Topo.DataNodeExists(option.DataCenter, option.Rack, option.DataNode) { + writeJsonQuiet(w, r, http.StatusBadRequest, operation.AssignResult{ + Error: fmt.Sprintf("data node %v not found in topology", option.DataNode), + }) + return + } + + // As in gRPC Assign: the periodic growth scan keeps this layout topped up + // to the writableVolumeCount clients ask for. 0 (not sent) is ignored. + vl.SetLastGrowCount(uint32(writableVolumeCount)) for time.Since(startTime) < maxTimeout { fid, count, dnList, shouldGrow, err := ms.Topo.PickForWrite(requestedCount, option, vl, expectedDataSize) diff --git a/weed/server/master_server_handlers_assign_test.go b/weed/server/master_server_handlers_assign_test.go new file mode 100644 index 000000000..93b87e337 --- /dev/null +++ b/weed/server/master_server_handlers_assign_test.go @@ -0,0 +1,209 @@ +package weed_server + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/topology" +) + +const assignTestVolumeSizeLimit = 1 << 30 + +// assignTestCluster is a leader master over one data center with one volume +// server per rack, like a production cluster spreading "010" replicas. +type assignTestCluster struct { + ms *MasterServer + nodes []*topology.DataNode + rp *super_block.ReplicaPlacement + nextVid uint32 +} + +func newAssignTestCluster(t *testing.T, racks int, replication string) *assignTestCluster { + t.Helper() + rp, err := super_block.NewReplicaPlacementFromString(replication) + require.NoError(t, err) + c := &assignTestCluster{ms: newLeaderMaster(), rp: rp} + // Room for the per-assign size estimate, which would fill newLeaderMaster's + // tiny volumes on the first assign. + c.ms.Topo = topology.NewTopology("test", sequence.NewMemorySequencer(), assignTestVolumeSizeLimit, 5, false) + c.ms.Topo.RaftServer = leaderRaftServer{} + dc := c.ms.Topo.GetOrCreateDataCenter("dc1") + for i := 0; i < racks; i++ { + c.nodes = append(c.nodes, dc.GetOrCreateRack(fmt.Sprintf("rack%d", i)). + GetOrCreateDataNode("127.0.0.1", 8080+i, 18080+i, "127.0.0.1", fmt.Sprintf("dn%d", i), map[string]uint32{"": 1000})) + } + return c +} + +// grow registers n new empty volumes, each with its copies on consecutive +// racks, the way a finished volume grow lands in the topology. +func (c *assignTestCluster) grow(collection string, n uint32) { + for i := uint32(0); i < n; i++ { + c.nextVid++ + v := storage.VolumeInfo{ + Id: needle.VolumeId(c.nextVid), + Collection: collection, + ReplicaPlacement: c.rp, + Ttl: needle.EMPTY_TTL, + Version: needle.GetCurrentVersion(), + } + for r := 0; r < c.rp.GetCopyCount(); r++ { + dn := c.nodes[(int(c.nextVid)+r)%len(c.nodes)] + dn.AddOrUpdateVolume(v) + c.ms.Topo.RegisterVolumeLayout(v, dn) + } + } +} + +func (c *assignTestCluster) layout(collection string) *topology.VolumeLayout { + return c.ms.Topo.GetVolumeLayout(collection, c.rp, needle.EMPTY_TTL, "") +} + +func (c *assignTestCluster) httpAssign(t *testing.T, query string) { + t.Helper() + w := httptest.NewRecorder() + c.ms.dirAssignHandler(w, httptest.NewRequest(http.MethodGet, "/dir/assign?"+query, nil)) + require.Equal(t, http.StatusOK, w.Code, w.Body.String()) +} + +// A client's writableVolumeCount is what the periodic growth scan keeps a +// layout topped up to. gRPC Assign has recorded it since the scan was added; +// HTTP /dir/assign parsed it but dropped it, so HTTP-only writers never got +// proactive growth. Both entry points must leave the layout in the same state. +func TestDirAssignRecordsWritableVolumeCountLikeGrpcAssign(t *testing.T) { + tests := []struct { + name string + counts []uint32 // writableVolumeCount per assign, 0 = not sent + want uint32 + }{ + {name: "sent", counts: []uint32{30}, want: 30}, + {name: "not sent", counts: []uint32{0}, want: 0}, + {name: "later assign without it keeps the last value", counts: []uint32{30, 0}, want: 30}, + {name: "latest value wins", counts: []uint32{30, 12}, want: 12}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + httpCluster := newAssignTestCluster(t, 1, "000") + httpCluster.grow("c", 1) + for _, n := range tt.counts { + query := "collection=c&replication=000" + if n > 0 { + query += fmt.Sprintf("&writableVolumeCount=%d", n) + } + httpCluster.httpAssign(t, query) + } + + grpcCluster := newAssignTestCluster(t, 1, "000") + grpcCluster.grow("c", 1) + for _, n := range tt.counts { + _, err := grpcCluster.ms.Assign(context.Background(), &master_pb.AssignRequest{ + Count: 1, Collection: "c", Replication: "000", WritableVolumeCount: n, + }) + require.NoError(t, err) + } + + assert.Equal(t, tt.want, grpcCluster.layout("c").GetLastGrowCount(), "gRPC Assign") + assert.Equal(t, tt.want, httpCluster.layout("c").GetLastGrowCount(), "HTTP /dir/assign") + }) + } +} + +// An assign naming a placement the topology cannot serve must fail before it +// records a growth target: the periodic scan drops the requested placement, so +// the volumes it would keep growing could never serve the requester. +func TestDirAssignRejectsUnknownPlacementBeforeGrowCount(t *testing.T) { + c := newAssignTestCluster(t, 1, "000") + c.grow("c", 1) + vl := c.layout("c") + + // A rack and data node that exist, but in a different data center: the + // requester named dc1, so these placements can never be served either. + c.ms.Topo.GetOrCreateDataCenter("dc2").GetOrCreateRack("rackX"). + GetOrCreateDataNode("127.0.0.2", 8090, 18090, "127.0.0.2", "dnX", map[string]uint32{"": 1000}) + + for _, query := range []string{ + "collection=c&replication=000&writableVolumeCount=30&rack=missing", + "collection=c&replication=000&writableVolumeCount=30&dataNode=missing", + "collection=c&replication=000&writableVolumeCount=30&dataCenter=dc1&rack=rackX", + "collection=c&replication=000&writableVolumeCount=30&dataCenter=dc1&dataNode=dnX", + "collection=c&replication=000&writableVolumeCount=30&dataCenter=dc1&rack=rack0&dataNode=dnX", + } { + w := httptest.NewRecorder() + c.ms.dirAssignHandler(w, httptest.NewRequest(http.MethodGet, "/dir/assign?"+query, nil)) + assert.Equal(t, http.StatusBadRequest, w.Code, w.Body.String()) + } + assert.Zero(t, vl.GetLastGrowCount()) + + for _, req := range []*master_pb.AssignRequest{ + {Count: 1, Collection: "c", Replication: "000", WritableVolumeCount: 30, Rack: "missing"}, + {Count: 1, Collection: "c", Replication: "000", WritableVolumeCount: 30, DataCenter: "dc1", Rack: "rackX"}, + {Count: 1, Collection: "c", Replication: "000", WritableVolumeCount: 30, DataCenter: "dc1", Rack: "rack0", DataNode: "dnX"}, + } { + _, err := c.ms.Assign(context.Background(), req) + assert.Error(t, err) + } + assert.Zero(t, vl.GetLastGrowCount()) +} + +// Replays the outage of an HTTP-only writer with writableVolumeCount=30 on a +// "010" collection across 5 racks. Writes are spread by remaining capacity, so +// the 30 volumes fill in lockstep. Before the fix the periodic scan only ever +// saw lastGrowCount 0: it added the rack-aware step of 2 when everything was +// crowded and nothing more while those 2 stayed uncrowded. When the old batch +// went full, the 2 volumes took every write, filled before the next scan, and +// assigns failed until an emergency grow landed. +func TestPeriodicGrowthKeepsHTTPAssignLayoutWritable(t *testing.T) { + const n = 30 + c := newAssignTestCluster(t, 5, "010") + c.grow("photo", n) + vl := c.layout("photo") + c.httpAssign(t, fmt.Sprintf("collection=photo&replication=010&writableVolumeCount=%d", n)) + + const limit = int64(assignTestVolumeSizeLimit) + batch := vl.CloneWritableVolumes() + writeTo := func(vids []needle.VolumeId, size int64) { + for _, vid := range vids { + vl.RecordAssign(vid, size) + } + } + // scan runs the per-layout steps of ProcessGrowRequest, with a grow that + // lands at once. + scan := func() { + writable, crowded := vl.GetWritableVolumeCount() + count, _ := periodicGrowCount(vl.GetLastGrowCount(), writable, crowded) + c.grow("photo", count) + for _, plan := range vl.PlanRackAwareGrowth(c.ms.Topo.ListDCAndRacks(), vl.GetLastGrowCount(), volumeGrowStepCount) { + c.grow("photo", plan.WritableVolumeCount) + } + writable, _ = vl.GetWritableVolumeCount() + require.LessOrEqual(t, writable, 2*n, "growth must stay bounded by twice the requested writable count") + } + + // The batch crosses the crowded threshold together, and stays crowded + // over two scans. + writeTo(batch, limit*95/100) + _, crowded := vl.GetWritableVolumeCount() + require.Equal(t, n, crowded) + scan() + scan() + + // The batch goes full; whatever the scans added is all that is left. + writeTo(batch, limit) + scan() + + writable, crowded := vl.GetWritableVolumeCount() + assert.Zero(t, crowded) + assert.GreaterOrEqual(t, writable, n, + "after the batch went full the scan must restore the %d writable volumes the client asked for, not leave %d to take every write", n, writable) +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index 847379249..f5717dd44 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -585,6 +585,54 @@ func (t *Topology) DataCenterExists(dcName string) bool { return dcName == "" || t.GetDataCenter(dcName) != nil } +// RackExists reports whether a rack with the id exists under dcName; an +// empty dcName searches every data center, an empty rackName always matches. +func (t *Topology) RackExists(dcName, rackName string) bool { + if rackName == "" { + return true + } + t.RLock() + defer t.RUnlock() + for _, c := range t.children { + if dcName != "" && string(c.Id()) != dcName { + continue + } + for _, r := range c.Children() { + if string(r.Id()) == rackName { + return true + } + } + } + return false +} + +// DataNodeExists reports whether a data node with the id exists under +// dcName/rackName; empty parents search every data center and rack, an empty +// nodeId always matches. +func (t *Topology) DataNodeExists(dcName, rackName, nodeId string) bool { + if nodeId == "" { + return true + } + t.RLock() + defer t.RUnlock() + for _, c := range t.children { + if dcName != "" && string(c.Id()) != dcName { + continue + } + for _, r := range c.Children() { + if rackName != "" && string(r.Id()) != rackName { + continue + } + for _, dn := range r.Children() { + if string(dn.Id()) == nodeId { + return true + } + } + } + } + return false +} + func (t *Topology) GetDataCenter(dcName string) (dc *DataCenter) { t.RLock() defer t.RUnlock()