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()