From f6438938910b9fd0950da50a86f5693a09ac6f80 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 26 Jun 2026 14:23:40 -0700 Subject: [PATCH] fix(master): shed assign load when volume growth is already in flight (#10121) Under a herd of concurrent assigns with no writable volume, Assign spun PickForWrite for the full 10s timeout, pinning a goroutine per request and starving the master of the cycles it needs to process growth and answer heartbeats. When growth is the relevant remedy and already in flight, stop spinning: if free space exists, shed with a fast retryable error so clients back off and retry once growth lands; if the cluster is out of space, fail fast with the real out-of-space error instead of masking it as retryable. The gRPC shed uses ResourceExhausted, not Unavailable: operation.Assign retries it, but the client connection layer doesn't treat it as a dead channel, so a per-request shed across a herd doesn't tear down the shared master connection and cancel every other in-flight assign. The HTTP dirAssignHandler sheds with 503 + Retry-After. --- weed/operation/assign_file_id.go | 4 +- weed/pb/grpc_client_server.go | 7 ++ weed/pb/grpc_client_server_test.go | 16 +++ weed/server/master_grpc_server_assign.go | 18 +++- weed/server/master_grpc_server_assign_test.go | 97 +++++++++++++++++++ weed/server/master_server_handlers.go | 11 +++ 6 files changed, 149 insertions(+), 4 deletions(-) create mode 100644 weed/server/master_grpc_server_assign_test.go diff --git a/weed/operation/assign_file_id.go b/weed/operation/assign_file_id.go index 70c54b9ed..8ae2049b3 100644 --- a/weed/operation/assign_file_id.go +++ b/weed/operation/assign_file_id.go @@ -74,7 +74,9 @@ func Assign(ctx context.Context, masterFn GetMasterFn, grpcDialOption grpc.DialO return false } switch st.Code() { - case codes.Unavailable: + case codes.Unavailable, codes.ResourceExhausted: + // ResourceExhausted: the master is shedding because volume growth + // is in flight; retry so we pick up the new volume once it lands. return true case codes.Canceled, codes.DeadlineExceeded: // A stale cached gRPC channel (e.g., master restart behind diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index d7960255a..f1c5bbcc9 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -391,6 +391,13 @@ func shouldInvalidateConnection(ctx context.Context, err error) bool { switch code { case codes.Unavailable, codes.Aborted, codes.Internal: return true + case codes.ResourceExhausted: + // Server-side backpressure on a healthy channel (e.g. the master's + // assign shed while volume growth is in flight), not a dead connection. + // Invalidating would Close() the shared conn and cancel every other + // in-flight RPC with "the client connection is closing"; keep it and let + // the caller retry. + return false case codes.Canceled, codes.DeadlineExceeded: // Ambiguous: this fires both when a stale cached channel rejects // RPCs (e.g. a peer restart behind a k8s Service VIP), where we must diff --git a/weed/pb/grpc_client_server_test.go b/weed/pb/grpc_client_server_test.go index e38514111..89b6d6ea5 100644 --- a/weed/pb/grpc_client_server_test.go +++ b/weed/pb/grpc_client_server_test.go @@ -66,6 +66,22 @@ func TestShouldInvalidateConnection_StaleChannelStillInvalidates(t *testing.T) { } } +// TestShouldInvalidateConnection_ResourceExhaustedIsPerRequest ensures the +// master's growth-in-progress assign shed (codes.ResourceExhausted) does NOT +// tear down the shared cached ClientConn. The shed fires per-request across a +// herd of concurrent assigns; invalidating on it would cancel every other +// in-flight assign with "the client connection is closing" — the cascade in +// seaweedfs#10118. It is retried by the caller without touching the channel. +func TestShouldInvalidateConnection_ResourceExhaustedIsPerRequest(t *testing.T) { + shed := status.Error(codes.ResourceExhausted, "no writable volumes for x, volume growth in progress") + if shouldInvalidateConnection(context.Background(), shed) { + t.Fatalf("ResourceExhausted shed must not invalidate the shared connection") + } + if shouldInvalidateConnection(nil, shed) { + t.Fatalf("ResourceExhausted shed must not invalidate the shared connection (nil ctx)") + } +} + // TestShouldInvalidateConnection_GenuineInternalStillInvalidates ensures the // marshal-error carve-out does not swallow real server-side Internal errors, // which previously caused — and should continue to cause — connection diff --git a/weed/server/master_grpc_server_assign.go b/weed/server/master_grpc_server_assign.go index 081ec298d..6626bf0bc 100644 --- a/weed/server/master_grpc_server_assign.go +++ b/weed/server/master_grpc_server_assign.go @@ -30,9 +30,10 @@ func (ms *MasterServer) StreamAssign(server master_pb.Seaweed_StreamAssignServer } resp, err := ms.Assign(context.Background(), req) if err != nil { - // Return transient errors (e.g. warmup) as in-band error responses - // instead of killing the stream, so pooled connections survive. - if st, ok := status.FromError(err); ok && st.Code() == codes.Unavailable { + // Return transient errors (warmup, growth-in-progress shed) as in-band + // error responses instead of killing the stream, so pooled connections + // survive. + if st, ok := status.FromError(err); ok && (st.Code() == codes.Unavailable || st.Code() == codes.ResourceExhausted) { glog.V(1).Infof("StreamAssign transient error: %v", err) resp = &master_pb.AssignResponse{Error: st.Message()} } else { @@ -130,6 +131,17 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest if (req.DataCenter != "" || req.Rack != "") && strings.Contains(err.Error(), topology.NoWritableVolumes) { break } + // Growth is the remedy and already in flight; don't pin a goroutine + // spinning out the timeout under an assign herd. + if shouldGrow && vl.HasGrowRequest() { + if ms.Topo.AvailableSpaceFor(option) <= 0 { + break // out of space: surface the real error, not a retryable shed + } + // ResourceExhausted, not Unavailable: clients retry it (assign_file_id.go) + // but the gRPC layer doesn't treat it as a dead channel, so the shed + // doesn't tear down the shared master connection mid-herd. + return nil, status.Errorf(codes.ResourceExhausted, "no writable volumes for %s, volume growth in progress", option.String()) + } time.Sleep(200 * time.Millisecond) continue } diff --git a/weed/server/master_grpc_server_assign_test.go b/weed/server/master_grpc_server_assign_test.go new file mode 100644 index 000000000..3d2b45732 --- /dev/null +++ b/weed/server/master_grpc_server_assign_test.go @@ -0,0 +1,97 @@ +package weed_server + +import ( + "context" + "testing" + "time" + + "github.com/seaweedfs/raft" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/sequence" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/topology" +) + +// leaderRaftServer answers only State(); the embedded nil interface panics if any +// other method is called. +type leaderRaftServer struct { + raft.Server +} + +func (leaderRaftServer) State() string { return raft.Leader } + +func newLeaderMaster() *MasterServer { + topo := topology.NewTopology("test", sequence.NewMemorySequencer(), 1<<20, 5, false) + topo.RaftServer = leaderRaftServer{} + return &MasterServer{ + Topo: topo, + option: &MasterOption{}, + volumeGrowthRequestChan: make(chan *topology.VolumeGrowRequest, 1<<6), + } +} + +// markGrowthInFlight flags growth on the same VolumeLayout the handler resolves, +// by mirroring its exact lookup. +func markGrowthInFlight(t *testing.T, topo *topology.Topology, req *master_pb.AssignRequest) { + t.Helper() + rp, err := super_block.NewReplicaPlacementFromString(req.Replication) + require.NoError(t, err) + ttl, err := needle.ReadTTL(req.Ttl) + require.NoError(t, err) + topo.GetVolumeLayout(req.Collection, rp, ttl, types.ToDiskType(req.DiskType)).AddGrowRequest() +} + +// With free space but no writable volume and growth already in flight, Assign +// sheds with a retryable ResourceExhausted immediately instead of spinning to +// timeout. ResourceExhausted (not Unavailable) so the shed isn't mistaken for a +// dead channel and doesn't tear down the shared master connection. +func TestAssignShedsLoadWhenGrowthInFlight(t *testing.T) { + ms := newLeaderMaster() + // Free volume slots but no writable volume yet, so growth is the remedy. + 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"} + 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()) + // Shed immediately rather than spinning out the 10s retry budget. + assert.Less(t, elapsed, 2*time.Second) + // Growth was already pending, so we must not enqueue another grow request. + assert.Len(t, ms.volumeGrowthRequestChan, 0) +} + +// Out of space, Assign fails fast with the real error rather than masking it as +// a retryable "growth in progress". +func TestAssignFailsFastWhenOutOfSpace(t *testing.T) { + ms := newLeaderMaster() // no data nodes -> no free space + + req := &master_pb.AssignRequest{Count: 1, Replication: "000"} + + 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.Contains(t, err.Error(), "no free volumes left") + assert.Less(t, elapsed, 2*time.Second) +} diff --git a/weed/server/master_server_handlers.go b/weed/server/master_server_handlers.go index 102eb6669..6d5eb6c28 100644 --- a/weed/server/master_server_handlers.go +++ b/weed/server/master_server_handlers.go @@ -187,6 +187,17 @@ func (ms *MasterServer) dirAssignHandler(w http.ResponseWriter, r *http.Request) if err != nil { stats.MasterPickForWriteErrorCounter.Inc() lastErr = err + // See Assign: shed instead of spinning when growth is already in flight. + if shouldGrow && vl.HasGrowRequest() { + if ms.Topo.AvailableSpaceFor(option) <= 0 { + break // out of space: surface the real error (406 below) + } + w.Header().Set("Retry-After", "1") + writeJsonQuiet(w, r, http.StatusServiceUnavailable, operation.AssignResult{ + Error: fmt.Sprintf("no writable volumes for %s, volume growth in progress", option.String()), + }) + return + } time.Sleep(200 * time.Millisecond) continue } else {