diff --git a/weed/server/master_grpc_server_assign.go b/weed/server/master_grpc_server_assign.go index 3a07d35ee8..1fbe8fd843 100644 --- a/weed/server/master_grpc_server_assign.go +++ b/weed/server/master_grpc_server_assign.go @@ -118,6 +118,7 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest startTime = time.Now() initiatedGrow bool repickedAfterGrow bool + recheckedBlocker bool unservedLayoutLogged bool ) @@ -139,8 +140,33 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest stats.MasterPickForWriteErrorCounter.Inc() lastErr = err if (req.DataCenter != "" || req.Rack != "") && strings.Contains(err.Error(), topology.NoWritableVolumes) { - glog.V(0).Infof("assign %v %v: %v", req, option.String(), err) - return nil, err + // Fail fast only when no growth can produce a writable volume in + // the requested location; a location with a pending or just- + // concluded grow gets the same wait/shed path as an + // unconstrained assign, so its client retry can re-trigger + // growth. + growBlocker := err + growthInFlight := !ms.option.VolumeGrowthDisabled && (initiatedGrow || vl.HasGrowRequest()) + if growthInFlight { + growBlocker = ms.Topo.GrowthBlocker(option) + } + if growBlocker != nil { + // The blocker's slot counts can predate an in-flight grow's + // delivery by a heartbeat: a volume minted on the last slot + // still registers writable a moment later. Give a pending + // grow one repick before surfacing the placement error. + if growthInFlight && !recheckedBlocker { + recheckedBlocker = true + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-time.After(200 * time.Millisecond): + } + continue + } + glog.V(0).Infof("assign %v %v: %v", req, option.String(), err) + return nil, growBlocker + } } if shouldGrow { if ms.Topo.AvailableSpaceFor(option) <= 0 { diff --git a/weed/server/master_grpc_server_assign_test.go b/weed/server/master_grpc_server_assign_test.go index a6f99fc625..a712147659 100644 --- a/weed/server/master_grpc_server_assign_test.go +++ b/weed/server/master_grpc_server_assign_test.go @@ -151,6 +151,147 @@ func TestAssignInitiatorShedsWhenGrowthConcludesUnfulfilled(t *testing.T) { assert.Len(t, ms.volumeGrowthRequestChan, 0) } +// A location-constrained assign whose location still has room waits for the +// growth it triggered, like an unconstrained one, instead of failing the first +// pick outright. +func TestAssignConstrainedInitiatorWaitsForGrowth(t *testing.T) { + ms := newLeaderMaster() + dn := ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "dn1", map[string]uint32{"": 100}) + + req := &master_pb.AssignRequest{Count: 1, Replication: "000", DataCenter: "dc1"} + rp, err := super_block.NewReplicaPlacementFromString(req.Replication) + require.NoError(t, err) + + go func() { + req := <-ms.volumeGrowthRequestChan + v := storage.VolumeInfo{ + Id: needle.VolumeId(1), + Version: needle.GetCurrentVersion(), + ReplicaPlacement: rp, + Ttl: needle.EMPTY_TTL, + } + dn.UpdateVolumes([]storage.VolumeInfo{v}) + ms.Topo.RegisterVolumeLayout(v, dn) + ms.Topo.GetVolumeLayout(req.Option.Collection, rp, needle.EMPTY_TTL, types.ToDiskType("")).DoneGrowRequest() + }() + + start := time.Now() + resp, err := ms.Assign(context.Background(), req) + elapsed := time.Since(start) + + require.NoError(t, err) + require.NotNil(t, resp) + assert.NotEmpty(t, resp.Fid) + assert.Less(t, elapsed, 5*time.Second) +} + +// A constrained assign joining a growth already in flight sheds retryably so +// the client retry picks up the fresh volume, instead of failing outright. +func TestAssignConstrainedShedsWhileGrowthPending(t *testing.T) { + ms := newLeaderMaster() + ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "dn1", map[string]uint32{"": 100}) + + req := &master_pb.AssignRequest{Count: 1, Replication: "000", DataCenter: "dc1"} + markGrowthInFlight(t, ms.Topo, req) + + start := time.Now() + resp, err := ms.Assign(context.Background(), req) + elapsed := time.Since(start) + + require.Error(t, err) + require.Nil(t, resp) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.ResourceExhausted, st.Code()) + assert.Less(t, elapsed, 2*time.Second) + assert.Len(t, ms.volumeGrowthRequestChan, 0) +} + +// A constrained initiator whose growth concludes without yielding a writable +// volume sheds retryably so its client's retry can re-trigger growth, instead +// of returning the plain pick error that would stop the retries. +func TestAssignConstrainedShedsWhenGrowthConcludesUnfulfilled(t *testing.T) { + ms := newLeaderMaster() + ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "dn1", map[string]uint32{"": 100}) + + req := &master_pb.AssignRequest{Count: 1, Replication: "000", DataCenter: "dc1"} + rp, err := super_block.NewReplicaPlacementFromString(req.Replication) + require.NoError(t, err) + vl := ms.Topo.GetVolumeLayout("", rp, needle.EMPTY_TTL, types.ToDiskType("")) + + go func() { + <-ms.volumeGrowthRequestChan + vl.DoneGrowRequest() + }() + + start := time.Now() + resp, err := ms.Assign(context.Background(), req) + elapsed := time.Since(start) + + require.Error(t, err) + require.Nil(t, resp) + st, ok := status.FromError(err) + require.True(t, ok) + assert.Equal(t, codes.ResourceExhausted, st.Code()) + assert.Less(t, elapsed, 2*time.Second) + assert.Len(t, ms.volumeGrowthRequestChan, 0) +} + +// A constrained assign fails fast only when no growth can satisfy the +// constraint: the named location is full (even though another data center has +// room), or volume growth is disabled outright. +func TestAssignConstrainedFailsFastWithoutRoom(t *testing.T) { + tests := []struct { + name string + setup func(t *testing.T, ms *MasterServer, req *master_pb.AssignRequest) + }{ + {"location full", func(t *testing.T, ms *MasterServer, req *master_pb.AssignRequest) { + ms.Topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, 18080, "127.0.0.1", "dn1", map[string]uint32{"": 100}) + dn2 := ms.Topo.GetOrCreateDataCenter("dc2").GetOrCreateRack("rack2"). + GetOrCreateDataNode("127.0.0.2", 8080, 18080, "127.0.0.2", "dn2", map[string]uint32{"": 1}) + rp, err := super_block.NewReplicaPlacementFromString(req.Replication) + require.NoError(t, err) + v := storage.VolumeInfo{ + Id: needle.VolumeId(1), + Version: needle.GetCurrentVersion(), + ReplicaPlacement: rp, + Ttl: needle.EMPTY_TTL, + } + dn2.UpdateVolumes([]storage.VolumeInfo{v}) + ms.Topo.RegisterVolumeLayout(v, dn2) + }}, + {"growth disabled", func(t *testing.T, ms *MasterServer, req *master_pb.AssignRequest) { + ms.Topo.GetOrCreateDataCenter("dc2").GetOrCreateRack("rack2"). + GetOrCreateDataNode("127.0.0.2", 8080, 18080, "127.0.0.2", "dn2", map[string]uint32{"": 100}) + ms.option.VolumeGrowthDisabled = true + }}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ms := newLeaderMaster() + req := &master_pb.AssignRequest{Count: 1, Replication: "000", Collection: "fresh", DataCenter: "dc2"} + tc.setup(t, ms, req) + + start := time.Now() + resp, err := ms.Assign(context.Background(), req) + elapsed := time.Since(start) + + require.Error(t, err) + require.Nil(t, resp) + if st, ok := status.FromError(err); ok { + assert.NotEqual(t, codes.Unavailable, st.Code()) + assert.NotEqual(t, codes.ResourceExhausted, st.Code()) + } + assert.Contains(t, err.Error(), topology.NoWritableVolumes) + assert.Less(t, elapsed, 2*time.Second) + }) + } +} + // A cancelled request stops waiting instead of sleeping out the 10s budget. func TestAssignAbortsOnCancel(t *testing.T) { ms := newLeaderMaster() diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index de1e54333c..b38cc52450 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -147,8 +147,16 @@ func (ms *MasterServer) ProcessGrowRequest() { return !found }) + // A targeted request is judged by the location it names: the + // layout-wide check would discard it whenever another location's + // writables are healthy, leaving the assign that asked for it + // without a volume no matter how often it retries. + needsGrowth := vl.ShouldGrowVolumes() + if option.DataCenter != "" || option.Rack != "" || option.DataNode != "" { + needsGrowth = vl.ShouldGrowVolumesWithinLocation(option) + } // not atomic but it's okay - if found || (!req.Force && !vl.ShouldGrowVolumes()) { + if found || (!req.Force && !needsGrowth) { glog.V(4).Infoln("discard volume grow request") time.Sleep(time.Millisecond * 211) vl.DoneGrowRequest() diff --git a/weed/topology/maintenance_mode_test.go b/weed/topology/maintenance_mode_test.go index 72fe3d66b8..8977efa12e 100644 --- a/weed/topology/maintenance_mode_test.go +++ b/weed/topology/maintenance_mode_test.go @@ -329,3 +329,49 @@ func TestMaintenanceModeAfterVolumeLeftLayout(t *testing.T) { } } } + +// GrowthBlocker reads room as well as shape: a location whose racks are all +// full cannot host the replica placement, and the rack or data node the +// option names must have room for the first replica itself. The assign +// layer, not this probe, makes the exception for a pending grow spending +// those slots on the request's own volume. +func TestGrowthBlockerNeedsRoom(t *testing.T) { + topo, dc1 := newMaintenanceTestTopology() + rack1 := NewRack("rack1") + dc1.LinkChildNode(rack1) + full1 := addMaintenanceTestNode(rack1, "s1", 5) + full1.getOrCreateDisk("").UpAdjustDiskUsageDelta("", &DiskUsageCounts{volumeCount: 8}) + rack2 := NewRack("rack2") + dc1.LinkChildNode(rack2) + full2 := addMaintenanceTestNode(rack2, "s2", 3) + full2.getOrCreateDisk("").UpAdjustDiskUsageDelta("", &DiskUsageCounts{volumeCount: 3}) + + rp010, _ := super_block.NewReplicaPlacementFromString("010") + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp010, DataCenter: "dc1"}); err == nil { + t.Fatal("two full racks can never place 010") + } + + // One freed slot per rack puts the placement back in reach. + full1.getOrCreateDisk("").UpAdjustDiskUsageDelta("", &DiskUsageCounts{volumeCount: -4}) + full2.getOrCreateDisk("").UpAdjustDiskUsageDelta("", &DiskUsageCounts{volumeCount: -3}) + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp010, DataCenter: "dc1"}); err != nil { + t.Fatalf("one free slot per rack fits 010: %v", err) + } + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp010, Rack: "rack1"}); err != nil { + t.Fatalf("rack1 hosts the first 010 replica, rack2 the second: %v", err) + } + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp010, Rack: "rack3"}); err == nil { + t.Fatal("no rack named rack3: 010 can never place its first replica") + } + + topo2, dc2 := newMaintenanceTestTopology() + rack := NewRack("rack1") + dc2.LinkChildNode(rack) + addMaintenanceTestNode(rack, "s1", 10) + if err := topo2.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp010, DataCenter: "dc1"}); err == nil { + t.Fatal("a single rack can never place 010") + } + if err := topo2.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp010, DataCenter: "dc9"}); err == nil { + t.Fatal("a missing data center can never place 010") + } +} diff --git a/weed/topology/topology.go b/weed/topology/topology.go index af73eb0cef..177c66a63d 100644 --- a/weed/topology/topology.go +++ b/weed/topology/topology.go @@ -1195,6 +1195,83 @@ func (t *Topology) GetDataCenter(dcName string) (dc *DataCenter) { return dc } +// GrowthBlocker reports why a volume grow under option could never place the +// replica set it needs, mirroring findEmptySlotsForOneVolume's feasibility: +// the option's location binds only the first replica's data center, rack, and +// data node — later replicas land anywhere with room — so counting every +// candidate against the named location condemns growth that can succeed. Nil +// means a grow can still place. Free slots are leaf-summed and ignore +// reservations, so an in-flight grow's own holds still read as room and its +// delivery lands as a writable the assign re-picks. +func (t *Topology) GrowthBlocker(option *VolumeGrowOption) error { + rp := option.ReplicaPlacement + needDCs := rp.DiffDataCenterCount + 1 + needRacks := rp.DiffRackCount + 1 + needNodes := rp.SameRackCount + 1 + t.RLock() + defer t.RUnlock() + roomDCs := 0 + mainFits := false + var errs []string + for _, c := range t.children { + var dcFree int64 + okRacks := 0 + mainRackFits := false + for _, r := range c.Children() { + var rackFree int64 + nodesWithRoom := 0 + namedNodeFits := option.DataNode == "" + for _, dn := range r.Children() { + free := dn.AvailableSpaceFor(option) + if free <= 0 { + continue + } + nodesWithRoom++ + rackFree += free + if string(dn.Id()) == option.DataNode { + namedNodeFits = true + } + } + dcFree += rackFree + if nodesWithRoom >= needNodes { + okRacks++ + } + if nodesWithRoom >= needNodes && namedNodeFits && + (option.Rack == "" || string(r.Id()) == option.Rack) { + mainRackFits = true + } + } + if dcFree > 0 { + roomDCs++ + } + if option.DataCenter != "" && string(c.Id()) != option.DataCenter { + continue + } + switch { + case len(c.Children()) < needRacks: + errs = append(errs, fmt.Sprintf("%s: %d racks, needs %d", c.Id(), len(c.Children()), needRacks)) + case okRacks < needRacks: + errs = append(errs, fmt.Sprintf("%s: %d racks with %d data nodes with room, needs %d", c.Id(), okRacks, needNodes, needRacks)) + case (option.Rack != "" || option.DataNode != "") && !mainRackFits: + errs = append(errs, fmt.Sprintf("%s: no rack fits the named placement", c.Id())) + case dcFree < int64(needRacks-1+needNodes): + errs = append(errs, fmt.Sprintf("%s: %d free volume slots, needs %d", c.Id(), dcFree, needRacks-1+needNodes)) + default: + mainFits = true + } + } + if mainFits && roomDCs >= needDCs { + return nil + } + if mainFits { + return fmt.Errorf("%s can grow in %s: %d data centers with room, needs %d", NoWritableVolumes, option, roomDCs, needDCs) + } + if len(errs) == 0 { + return fmt.Errorf("%s: no data center matches %s", NoWritableVolumes, option) + } + return fmt.Errorf("%s can grow in %s: %s", NoWritableVolumes, option, strings.Join(errs, "; ")) +} + func (t *Topology) GetOrCreateDataCenter(dcName string) *DataCenter { t.Lock() defer t.Unlock() diff --git a/weed/topology/volume_growth_test.go b/weed/topology/volume_growth_test.go index da9cf5affd..b70566989b 100644 --- a/weed/topology/volume_growth_test.go +++ b/weed/topology/volume_growth_test.go @@ -155,6 +155,39 @@ func TestFindEmptySlotsForOneVolume(t *testing.T) { } } +func TestGrowthBlocker(t *testing.T) { + topo := setup(topologyLayout) + + rp010, _ := super_block.NewReplicaPlacementFromString("010") + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp010, DataCenter: "dc1"}); err != nil { + t.Fatalf("dc1 has the two racks 010 needs, got blocker: %v", err) + } + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp010, DataCenter: "dc3"}); err == nil { + t.Fatal("dc3 has a single rack, 010 should report a blocker") + } + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp010, DataCenter: "dc2"}); err == nil { + t.Fatal("dc2 has no racks, 010 should report a blocker") + } + + rp000, _ := super_block.NewReplicaPlacementFromString("000") + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp000, DataCenter: "dc3"}); err != nil { + t.Fatalf("dc3 has a node for 000, got blocker: %v", err) + } + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp000, DataCenter: "dc9"}); err == nil { + t.Fatal("dc9 does not exist, 000 should report a blocker") + } + + // The named location binds only the first replica: dc1 hosts it for 100 + // and dc3 has room for the second, so the placement is reachable. + rp100, _ := super_block.NewReplicaPlacementFromString("100") + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp100, DataCenter: "dc1"}); err != nil { + t.Fatalf("dc3 has room for the second 100 replica, got blocker: %v", err) + } + if err := topo.GrowthBlocker(&VolumeGrowOption{ReplicaPlacement: rp100, DataCenter: "dc2"}); err == nil { + t.Fatal("dc2 has no room for the first 100 replica") + } +} + var topologyLayout2 = ` { "dc1":{ diff --git a/weed/topology/volume_layout.go b/weed/topology/volume_layout.go index c05598620e..bd961565ae 100644 --- a/weed/topology/volume_layout.go +++ b/weed/topology/volume_layout.go @@ -973,6 +973,36 @@ func (vl *VolumeLayout) ShouldGrowVolumesByDcAndRack(writables *[]needle.VolumeI return true } +// ShouldGrowVolumesWithinLocation reports whether the location the option +// names still lacks a writable it can use: true when every writable volume +// holding a replica inside the option's DataCenter/Rack/DataNode is crowded, +// or none exists at all. The layout-wide ShouldGrowVolumes answers the same +// question cluster-wide, so a targeted grow judged by it would be discarded +// whenever another location's writables are healthy — leaving the requesting +// location without a volume no matter how often the assign retries. +func (vl *VolumeLayout) ShouldGrowVolumesWithinLocation(option *VolumeGrowOption) bool { + writables := vl.CloneWritableVolumes() + for _, v := range writables { + for _, dn := range vl.Lookup(v) { + if !dn.matchesLocation(option) { + continue + } + if _, err := dn.GetVolumesById(v); err == nil { + vl.accessLock.RLock() + var size uint64 + if st := vl.sizeTracking[v]; st != nil { + size = st.effectiveSize + } + vl.accessLock.RUnlock() + if float64(size) <= float64(vl.volumeSizeLimit)*VolumeGrowStrategy.Threshold { + return false + } + } + } + } + return true +} + // listVolumeDataCenters returns the data centers hosting at least one volume // of this layout, stopping once limit distinct DCs are seen (0 = no limit). // The walk holds accessLock and a layout can span millions of volumes, so diff --git a/weed/topology/volume_layout_pick_test.go b/weed/topology/volume_layout_pick_test.go index aa8cc99c1f..6d44d557be 100644 --- a/weed/topology/volume_layout_pick_test.go +++ b/weed/topology/volume_layout_pick_test.go @@ -784,3 +784,50 @@ func TestShouldGrowVolumesByDcAndRack_WithPendingSize(t *testing.T) { t.Error("should grow after pending pushes volume past crowded threshold") } } + +// A targeted grow asks whether the location it names still lacks a usable +// writable, so healthy writables in other data centers must not mask the need. +func TestShouldGrowVolumesWithinLocation(t *testing.T) { + layout := ` +{ + "dc1":{ + "rack1":{ + "server1":{ + "ip":"10.0.0.1", + "volumes":[ + {"id":1, "size":2000, "replication":"000"}, + {"id":2, "size":9500, "replication":"000"} + ], + "limit":10 + } + } + }, + "dc2":{ + "rack2":{ + "server2":{ + "ip":"10.0.0.2", + "volumes":[], + "limit":10 + } + } + } +} +` + _, vl := setupPickTest(t, layout, 10000) + + if vl.ShouldGrowVolumesWithinLocation(&VolumeGrowOption{DataCenter: "dc1"}) { + t.Error("dc1 holds a non-crowded writable: a dc1 grow is not needed") + } + if !vl.ShouldGrowVolumesWithinLocation(&VolumeGrowOption{DataCenter: "dc1", Rack: "rack2"}) { + t.Error("no writable has a replica in dc1/rack2: a rack grow is needed") + } + if !vl.ShouldGrowVolumesWithinLocation(&VolumeGrowOption{DataCenter: "dc2"}) { + t.Error("dc2 has no writable at all: a dc2 grow is needed") + } + if vl.ShouldGrowVolumesWithinLocation(&VolumeGrowOption{DataCenter: "dc1", DataNode: "server1"}) { + t.Error("server1 holds a non-crowded writable: a datanode grow is not needed") + } + if vl.ShouldGrowVolumesWithinLocation(&VolumeGrowOption{}) { + t.Error("an unconstrained grow sees the non-crowded writable and is not needed") + } +}