From 11439452b52780924679cefaf86bfc736a1f8eac Mon Sep 17 00:00:00 2001 From: Shawn Burke Date: Tue, 10 Nov 2026 11:03:57 +1100 Subject: [PATCH] =?UTF-8?q?grpctunnel:=20size=20streams=20per=20tunnel=20s?= =?UTF-8?q?erver=20=E2=80=94=20grow=20greedily=20under=20load,=20prune=20f?= =?UTF-8?q?irmly=20at=20rest?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A tunnel server (one relay-grpc-tunnel instance) can only hand a call to a stream this agent holds on that same server, and the dispatcher sends a token's calls to one tunnel server at a time. The agent sized its streams as a total across servers and left placement to the load balancer, which favours the nearest zone, so the server receiving calls could hold a single stream: one call in flight made it "all busy" and callers there queued, with idle streams on other servers and nothing telling the agent to grow. Streams carry I/O, not work: an open is a round trip on a pooled connection (the dispatcher is notified only for a token's first stream) and an idle stream costs its shared tunnel server a heartbeat and a few goroutines. So the agent now opens greedily and prunes firmly instead of rationing. Coverage (coverage.go) runs each 1s sizing tick. The agent's streams are grouped by the tunnel server they landed on (ServerHello.ServerId) into one serverCoverage each, which reports its state; coverage plans the tick from those states and balanceCoverage carries the plan out: covered streams >= busy + 2 nothing growing short, calls in flight open max(8, calls in flight, shortfall) that tick -- doubling under pressure; unbudgeted, paused while idle streams elsewhere reach 4x demand, capped filling short, idle open toward the reserve, budgeted overdue short, idle, 30s fill, then evacuate: recycle its idle streams so it drops the token -- never a server with calls in flight, only while another is covered, one per 30s evacuated last 10 minutes recycle idle streams; no grow or fill While anything is short the target is held. retireExcess prunes idle streams once 2 minutes old and unused for 10, never taking a server below its reserve except to obey a cap. The resting reserve stays at 2 per tunnel server: a larger one was harder to place under zone-skewed placement and costs every shared tunnel server more heartbeats. Each server's budget counts calls in flight, not streams used recently: a tunnel server hands each call to its newest idle stream, so counting streams made a quiet server look short and got it evacuated. Caps: agent maxStreams and MaxInflightRequests 256 -> 2048; tunnel server MaxStreamsPerToken default 64 -> 1024, since one tunnel server carries a token's whole concurrency. Visibility: grpc_tunnel_server_streams{server_id,state}, grpc_tunnel_coverage_actions_total{action=grow|fill|evacuate}, and a servers array in /systemcheck with each server's streams, busy, budget and coverage state. Metric construction moves to initMetrics, shared with the test client. Tests (coverage_test.go) drive the real sizeTick against simulated workers and tunnel servers under uniform, zone-skewed and unfillable placement; the incident cases fail with coverage disabled. Growth is bracketed without depending on placement: one tick asks for a chunk; with every stream landing on the busy server a 1000-call burst is in flight in 8 ticks; with none landing the agent stays bounded and keeps the busy server; after a burst it prunes back to rest. Older sizing tests are restated in terms of the reserve and cap constants. Co-Authored-By: Claude Opus 5.5 --- README.relay.md | 22 + agent/config/config.go | 2 +- agent/config/config_test.go | 2 +- agent/server/grpctunnel/coverage.go | 440 +++++++++++++ agent/server/grpctunnel/coverage_test.go | 598 ++++++++++++++++++ agent/server/grpctunnel/sizing_test.go | 55 +- agent/server/grpctunnel/tunnel_client.go | 230 +++++-- agent/server/grpctunnel/tunnel_client_test.go | 25 +- server/config/config.go | 14 +- 9 files changed, 1288 insertions(+), 100 deletions(-) create mode 100644 agent/server/grpctunnel/coverage.go create mode 100644 agent/server/grpctunnel/coverage_test.go diff --git a/README.relay.md b/README.relay.md index b3d60c2..229f9d7 100644 --- a/README.relay.md +++ b/README.relay.md @@ -177,6 +177,28 @@ across every server instance, so losing one costs a fraction of capacity for one redial and never all of it. Streams are opened as calls arrive and given back as they finish, so concurrency is whatever the agent can actually serve. +A tunnel server (one `relay-grpc-tunnel` instance) dispatches only onto the +streams the agent holds on that server, and the dispatcher sends a token's +calls to one tunnel server at a time. So the agent sizes per tunnel server: + +- **Grow.** A tunnel server with calls in flight and fewer idle streams than + the reserve gets at least 8 more streams opened that tick, or as many as it + has calls in flight, so capacity doubles each second under pressure. Streams + carry I/O, not work, so opening is cheap; the limits are the stream caps. +- **Prune.** Streams idle for 10 minutes, and at least 2 minutes old, are given + back down to a small resting reserve (2 per tunnel server). Every idle + stream costs its shared tunnel server a heartbeat and a few goroutines, so + what is not in use does not stay open. +- **Cover.** The load balancer places each new stream and does not spread them + evenly. A tunnel server with nothing in flight that is short of the reserve + gets streams opened toward it, within a budget; one still short after 30s is + left, so it drops the token. A tunnel server with calls in flight is never + left. + +Per-server counts are in `grpc_tunnel_server_streams{server_id,state}` and +`/systemcheck`, and what the agent did is in +`grpc_tunnel_coverage_actions_total{action="grow"|"fill"|"evacuate"}`. + When the agent falls behind, it runs out of idle streams, the server's dispatch blocks briefly, and the delay reaches the caller as latency. That is the backpressure, and it needs no configuration: `dispatch.acquire_wait_ms` diff --git a/agent/config/config.go b/agent/config/config.go index ac19e9f..7e0e71b 100644 --- a/agent/config/config.go +++ b/agent/config/config.go @@ -439,7 +439,7 @@ func NewAgentEnvConfig() AgentConfig { cfg.GrpcTunnelServer = os.Getenv("GRPC_TUNNEL_SERVER") - cfg.MaxInflightRequests = 256 + cfg.MaxInflightRequests = 2048 if maxInflight := os.Getenv("MAX_INFLIGHT_REQUESTS"); maxInflight != "" { v, err := strconv.Atoi(maxInflight) if err != nil { diff --git a/agent/config/config_test.go b/agent/config/config_test.go index bd009d4..10d2f01 100644 --- a/agent/config/config_test.go +++ b/agent/config/config_test.go @@ -19,7 +19,7 @@ func TestEnvConfig_TunnelDefaults(t *testing.T) { assert.Equal(t, 4, cfg.TunnelConns, "connections are the only tunnel dial; default is a blast-radius choice") assert.Equal(t, 128, cfg.UpstreamMaxConnsPerHost) - assert.Equal(t, 256, cfg.MaxInflightRequests) + assert.Equal(t, 2048, cfg.MaxInflightRequests) assert.False(t, cfg.GrpcInsecure, "TLS stays on unless explicitly disabled") } diff --git a/agent/server/grpctunnel/coverage.go b/agent/server/grpctunnel/coverage.go new file mode 100644 index 0000000..3459ed1 --- /dev/null +++ b/agent/server/grpctunnel/coverage.go @@ -0,0 +1,440 @@ +package grpctunnel + +import ( + "sort" + "sync" + "time" +) + +// Per-tunnel-server coverage. +// +// Why: a tunnel server (one relay-grpc-tunnel instance) can only hand a call +// to a stream this agent holds on that same server. If the server the +// dispatcher is using has no idle stream, callers there wait -- even when the +// agent has idle streams on other servers. +// +// What it tracks. Every sizing tick (1s), the agent's streams are grouped by +// the tunnel server each one landed on (ServerHello.ServerId). Per server: +// +// streams open streams on that server +// busy of those, how many are carrying a call right now +// budget busy + idleStreamsPerServer (2): what the server should have +// short budget - streams, when positive +// +// Each server (a serverCoverage) answers what state it is in, and each state +// has one response: +// +// State When Action +// --------- ---------------------------------- --------------------------- +// covered streams >= budget nothing to do +// growing short, with calls in flight GROW +// filling short, nothing in flight FILL +// overdue short, nothing in flight, for 30s FILL, then EVACUATE +// evacuated given up in the last 10 minutes recycle its idle streams +// +// What each action does: +// +// - GROW opens max(growthChunk (8), the server's calls in flight, its +// shortfall) streams that tick, so capacity roughly doubles each second +// while callers are waiting. No budget. It pauses once streams sitting +// idle on covered servers reach growthSurplusFactor (4) times the calls +// in flight (or 4 x 8, if more), and is capped by maxStreams and the +// servers' per-token cap. +// +// - FILL opens enough streams to cover the shortfall, from a shared budget +// of coverageActionsPerServer (4) opens per server the agent is on, per +// 10 minutes. +// +// - EVACUATE recycles the server's idle streams so they reopen somewhere +// else, and the server drops the token. +// +// When: a server has had nothing in flight and been short of its +// reserve for 30s (overdue) -- FILL has had its chance and the load +// balancer is not placing streams there. Typically it got one stream by +// chance, often an anchor, and rarely gets another because the balancer +// favours a different zone. +// +// Why: a server that holds even one of the agent's streams stays in the +// dispatcher's list for the token, so the dispatcher's liveness probe +// and any calls routed there land on that one stream. While it is busy +// they wait; a 2s probe that times out records the agent as +// unresponsive on that server. The agent cannot make that server +// healthy -- it cannot choose where streams land -- but it can leave: +// once the server holds none of its streams it drops the token, the +// dispatcher stops sending there, and the recycled streams reopen +// wherever the balancer places them, usually on servers already +// covered. A server the agent is not on cannot make a caller wait. +// +// Limits: never a server with calls in flight (that is where the +// dispatcher is sending, so leaving it would strand those callers -- +// GROW handles it instead); only while another server is covered, so +// the agent is never left with nowhere good to be; at most one server +// per 30s; from its own shared budget of 4 recycles per server per 10 +// minutes; and not refilled for 10 minutes, so it is not filled and +// given up in a loop. +// +// While anything is short, the target is held at the current stream count +// plus what was just opened, so nothing is pruned until the shortfall +// clears. +// +// The agent cannot choose where a new stream lands -- the load balancer does, +// and it favours the nearest zone -- so growth and fill open streams and some +// land on servers that did not need them. Pruning gives those back: when the +// agent holds more than its target, retireExcess retires streams that are +// idle, older than pruneMinAge (2m) and unused for streamIdleWindow (10m), +// keeping on each server one stream per call it carried in that window plus +// the reserve. +const ( + // coverageGrace is how long a server may stay short while fill tries to + // reach it before it is evacuated. + coverageGrace = 30 * time.Second + // coverageActionsPerServer is the budget per streamIdleWindow, per server + // the agent holds streams on, for fill opens and, separately, for + // evacuation recycles. Growth for calls in flight is not budgeted. + coverageActionsPerServer = 4 + // growthChunk is the fewest streams growth opens for a server queueing + // callers in one tick. + growthChunk = 8 + // growthSurplusFactor bounds the streams growth leaves idle on servers + // that did not need them: it stops opening while those exceed this many + // times the calls in flight (or growthChunk, if more). It is what keeps a + // balancer that never places on the busy server from growing the agent to + // its cap. + growthSurplusFactor = 4 +) + +// coverage is what the tunnel client's sizing tick remembers between ticks: one +// serverCoverage per tunnel server, and the budgets. Timestamps are UnixNano. +type coverage struct { + mu sync.Mutex + servers map[string]*serverCoverage + // lastEvacuation is when a server was last given up. At most one goes per + // coverageGrace, so servers running short together are not given up in a + // cascade. + lastEvacuation int64 + // windowStart, filled and recycled meter the budgets. + windowStart int64 + filled int + recycled int + // warned is set once the fill budget runs out with a server still short, + // so that is logged once per window. + warned bool +} + +func (c *coverage) reset() { + c.mu.Lock() + defer c.mu.Unlock() + c.servers = map[string]*serverCoverage{} + c.lastEvacuation = 0 + c.windowStart = 0 + c.filled = 0 + c.recycled = 0 + c.warned = false +} + +// serverStreamCounts is one tunnel server's streams as coverage sees them. +type serverStreamCounts struct { + have int // open and not being retired + busy int // in flight now + // recyclable are idle streams quiet for a full tick: the ones an + // evacuation may take without cancelling a call the server is about to + // dispatch onto them. + recyclable []*tunnelStream +} + +// budget is how many streams the server needs now: one for each call in +// flight on it, plus the idle reserve. +// +// Calls, not the streams that carried them recently. The server hands each +// call to the newest idle stream it holds, so a server seeing one call every +// few minutes touches a different stream each time; counting those made a +// quiet server look short, spent the fill budget on it, and then gave it up. +// retireExcess keeps one stream per recent call plus the reserve, which is +// never less than this, so what fill opens toward retirement leaves alone. +func (s *serverStreamCounts) budget() int { return s.busy + idleStreamsPerServer } + +func (s *serverStreamCounts) short() int { + if d := s.budget() - s.have; d > 0 { + return d + } + return 0 +} + +// countServerStreams groups streams by the tunnel server each landed on. +// Streams not yet placed, and ones being retired, are left out. The caller +// holds whatever guards streams. +func countServerStreams(streams map[*tunnelStream]struct{}, now time.Time) map[string]*serverStreamCounts { + quietBefore := now.Add(-sizeInterval).UnixNano() + servers := make(map[string]*serverStreamCounts) + for st := range streams { + if st.serverID == "" || st.retiring.Load() { + continue + } + s := servers[st.serverID] + if s == nil { + s = &serverStreamCounts{} + servers[st.serverID] = s + } + s.have++ + inflight := st.inflight.Load() != 0 + if inflight { + s.busy++ + } + if !inflight && st.quietSince() < quietBefore { + s.recyclable = append(s.recyclable, st) + } + } + return servers +} + +// serverCoverage is coverage's view of one tunnel server: its streams this +// tick, and what coverage remembers about it across ticks. +type serverCoverage struct { + serverStreamCounts + // shortSince is when the server fell short of its budget; zero while it + // is not short. + shortSince int64 + // evacuatedAt is when the server was given up; zero unless that was + // within the last streamIdleWindow. + evacuatedAt int64 +} + +// coverageState is where a tunnel server stands, and so what coverage does +// about it. +type coverageState int + +const ( + // stateCovered: the server holds its budget. Nothing to do. + stateCovered coverageState = iota + // stateGrowing: short, with calls in flight -- callers may be waiting. + // Grow, unbudgeted. + stateGrowing + // stateFilling: short, nothing in flight, for less than coverageGrace. + // Fill, within the budget. + stateFilling + // stateOverdue: short, nothing in flight, for coverageGrace or longer. + // Still filled, and evacuated as soon as evacuation is allowed. + stateOverdue + // stateEvacuated: given up within the last streamIdleWindow. Its idle + // streams are recycled; it is neither grown nor filled. + stateEvacuated +) + +func (st coverageState) String() string { + switch st { + case stateCovered: + return "covered" + case stateGrowing: + return "growing" + case stateFilling: + return "filling" + case stateOverdue: + return "overdue" + case stateEvacuated: + return "evacuated" + } + return "unknown" +} + +// state is where the server stands at now. +func (s *serverCoverage) state(now int64) coverageState { + switch { + case s.evacuatedAt != 0: + return stateEvacuated + case s.short() == 0: + return stateCovered + case s.busy > 0: + return stateGrowing + case now-s.shortSince >= int64(coverageGrace): + return stateOverdue + default: + return stateFilling + } +} + +// growth is how many streams a growing server asks for in one tick: at least +// growthChunk, and at least as many as it has calls in flight, so capacity +// doubles per tick under pressure. +func (s *serverCoverage) growth() int { return max(s.short(), growthChunk, s.busy) } + +// tick runs one sizing tick of coverage: it folds counts -- the agent's +// streams per tunnel server, from countServerStreams -- into what it tracks, +// and plans what to do. workers is the stream workers running, pending those +// still opening a stream. The caller carries the plan out. +func (c *coverage) tick(counts map[string]*serverStreamCounts, now int64, workers, pending int) coveragePlan { + c.mu.Lock() + defer c.mu.Unlock() + c.observe(counts, now) + return c.plan(now, workers, pending) +} + +// observe folds one tick's stream counts into the servers coverage tracks: +// new servers are added, ones the agent has left are dropped (unless still +// marked evacuated), evacuations expire, and each server's shortSince starts +// or clears. Callers hold c.mu. +func (c *coverage) observe(counts map[string]*serverStreamCounts, now int64) { + if c.servers == nil { + c.servers = map[string]*serverCoverage{} + } + if now-c.windowStart >= int64(streamIdleWindow) { + c.windowStart, c.filled, c.recycled, c.warned = now, 0, 0, false + } + for id, s := range c.servers { + if s.evacuatedAt != 0 && now-s.evacuatedAt >= int64(streamIdleWindow) { + s.evacuatedAt = 0 + } + if _, ok := counts[id]; !ok && s.evacuatedAt == 0 { + delete(c.servers, id) + continue + } + s.serverStreamCounts = serverStreamCounts{} + } + for id, n := range counts { + s := c.servers[id] + if s == nil { + s = &serverCoverage{} + c.servers[id] = s + } + s.serverStreamCounts = *n + } + for _, s := range c.servers { + switch { + case s.evacuatedAt != 0, s.short() == 0: + s.shortSince = 0 + case s.shortSince == 0: + s.shortSince = now + } + } +} + +// coveragePlan is what one tick of coverage decided. +type coveragePlan struct { + // floor is the stream count resize must not go below; zero when nothing + // is short. + floor int + // grown and filled are the streams opened for growth and for the + // reserve this tick. + grown, filled int + // evacuate are the servers given up this tick. + evacuate []string + // recycle are the idle streams on evacuated servers to reopen elsewhere. + recycle []*tunnelStream + // budget is this window's fill and recycle budget, each. + budget int + // exhausted is set the first tick in a window the fill budget runs out + // with servers still short; short names them. + exhausted bool + short []string +} + +// plan asks each server its state and decides the tick: what to open, what +// to evacuate and recycle, and the floor to hold the target at. workers is +// the stream workers running, pending those still opening a stream. Callers +// hold c.mu. +func (c *coverage) plan(now int64, workers, pending int) coveragePlan { + ids := make([]string, 0, len(c.servers)) + present := 0 + for id, s := range c.servers { + ids = append(ids, id) + if s.have > 0 { + present++ + } + } + sort.Strings(ids) + p := coveragePlan{budget: coverageActionsPerServer * max(present, 1)} + + growWant, reserveDeficit, covered, surplus, busyTotal := 0, 0, 0, 0, 0 + var overdue []string + for _, id := range ids { + s := c.servers[id] + busyTotal += s.busy + switch s.state(now) { + case stateCovered: + covered++ + surplus += s.have - s.budget() + case stateGrowing: + growWant += s.growth() + case stateOverdue: + overdue = append(overdue, id) + fallthrough + case stateFilling: + reserveDeficit += s.short() + p.short = append(p.short, id) + } + } + + // Evacuate an overdue server -- only while another is covered, and one + // per coverageGrace. + for _, id := range overdue { + if covered == 0 || now-c.lastEvacuation < int64(coverageGrace) { + continue + } + s := c.servers[id] + c.lastEvacuation = now + s.evacuatedAt, s.shortSince = now, 0 + reserveDeficit -= s.short() + p.evacuate = append(p.evacuate, id) + } + + // Recycle the idle streams of evacuated servers, as the budget allows; a + // busy one waits for its call to finish. + for _, id := range ids { + s := c.servers[id] + if s.evacuatedAt == 0 { + continue + } + for _, st := range s.recyclable { + if c.recycled >= p.budget { + break + } + c.recycled++ + p.recycle = append(p.recycle, st) + } + } + + // Open for what is short, less what is already on its way: growth first, + // unbudgeted but paused while streams that landed elsewhere pile up idle; + // then the reserve, within its budget. Hold the target while they land, + // and let it go once nothing is short. + if surplus >= growthSurplusFactor*max(growthChunk, busyTotal) { + growWant = 0 + } + if growWant > 0 || reserveDeficit > 0 { + inFlight := pending + p.grown = max(growWant-inFlight, 0) + inFlight = max(inFlight-growWant, 0) + p.filled = min(max(reserveDeficit-inFlight, 0), max(p.budget-c.filled, 0)) + c.filled += p.filled + p.floor = workers + p.grown + p.filled + } + if reserveDeficit > 0 && c.filled >= p.budget && !c.warned { + c.warned = true + p.exhausted = true + } + return p +} + +// serverSummary is one tunnel server's streams and coverage state for +// /systemcheck, as of the last sizing tick. +type serverSummary struct { + ServerID string `json:"server_id"` + Streams int `json:"streams"` + Busy int `json:"busy"` + Budget int `json:"budget"` + State string `json:"state"` +} + +// summaries reports each tunnel server's streams and state as of the last +// tick, for /systemcheck. +func (c *coverage) summaries(now int64) []serverSummary { + c.mu.Lock() + defer c.mu.Unlock() + out := make([]serverSummary, 0, len(c.servers)) + for id, s := range c.servers { + out = append(out, serverSummary{ + ServerID: id, Streams: s.have, Busy: s.busy, Budget: s.budget(), State: s.state(now).String(), + }) + } + sort.Slice(out, func(i, j int) bool { return out[i].ServerID < out[j].ServerID }) + return out +} diff --git a/agent/server/grpctunnel/coverage_test.go b/agent/server/grpctunnel/coverage_test.go new file mode 100644 index 0000000..ef637a9 --- /dev/null +++ b/agent/server/grpctunnel/coverage_test.go @@ -0,0 +1,598 @@ +package grpctunnel + +import ( + "fmt" + "math/rand/v2" + "slices" + "sort" + "testing" + "time" + + pb "github.com/cortexapps/axon/.generated/proto/github.com/cortexapps/axon/tunnelpb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These replay the production failure: an agent on four tunnel servers ended up +// with a single stream on one of them, so one in-flight call made that server +// "all busy" and a 2s health probe there timed out -- with the agent holding +// streams to spare elsewhere. Placement is the load balancer's, and its +// rate-based mode prefers the nearest zone over spreading, so the tests run +// against uneven placement, not round-robin. + +// coverageSim drives the real sizing tick against simulated stream workers +// and tunnel servers. Workers behave as manageStream does: a stream cancelled on +// purpose is reopened unless tryRetireWorker says the worker should go, and +// every open is placed by place, standing in for the balancer. Time moves by +// backdating, which is all the sizing rule reads. +type coverageSim struct { + t *testing.T + tc *tunnelClient + place func() string + opened int + starved int + // deferOpens makes streams a call's admission asks for arrive on the + // next tick rather than at once, as a real open's round trip does. + deferOpens bool + + cancelled []*tunnelStream +} + +func newCoverageSim(t *testing.T, conns int, place func() string) *coverageSim { + t.Helper() + tc := newSizingClient(t) + tc.pool = newConnPool(conns) + tc.streams = map[*tunnelStream]struct{}{} + tc.cov.reset() + return &coverageSim{t: t, tc: tc, place: place} +} + +// openLocked opens a stream for worker id where the balancer puts it. Callers +// hold tc.mu. +func (s *coverageSim) openLocked(id int) { + var ts *tunnelStream + ts = newTunnelStream(id, &pb.ServerHello{ServerId: s.place()}, nil, + func() { s.cancelled = append(s.cancelled, ts) }, id <= s.tc.anchorSlots()) + s.tc.streams[ts] = struct{}{} + s.opened++ +} + +// openToTarget spawns workers up to the target, each opening a stream. +func (s *coverageSim) openToTarget() { + tc := s.tc + tc.mu.Lock() + defer tc.mu.Unlock() + for tc.runningWorkers < int(tc.targetSlots.Load()) { + tc.runningWorkers++ + tc.slotSeq++ + s.openLocked(tc.slotSeq) + } +} + +// reap plays the workers whose streams were cancelled: each either retires or +// reopens, exactly as manageStream decides. +func (s *coverageSim) reap() { + for len(s.cancelled) > 0 { + ts := s.cancelled[0] + s.cancelled = s.cancelled[1:] + s.tc.mu.Lock() + delete(s.tc.streams, ts) + s.tc.mu.Unlock() + if s.tc.tryRetireWorker(ts.id) { + continue + } + s.tc.mu.Lock() + s.openLocked(ts.id) + s.tc.mu.Unlock() + } +} + +// tick is one second of the agent: a sizing pass, the workers' response, and +// the clock moving on. +func (s *coverageSim) tick() { + s.tc.sizeTick() + s.reap() + s.openToTarget() + s.elapse(sizeInterval) +} + +func (s *coverageSim) run(d time.Duration) { + for i := 0; i < int(d/sizeInterval); i++ { + s.tick() + } +} + +// elapse moves the clock by backdating every timestamp sizing and coverage +// read. +func (s *coverageSim) elapse(d time.Duration) { + tc := s.tc + tc.mu.Lock() + for st := range tc.streams { + if last := st.lastCallAt.Load(); last != 0 { + st.lastCallAt.Store(last - int64(d)) + } + st.openedAt.Add(-int64(d)) + } + tc.mu.Unlock() + tc.peak.mu.Lock() + for i := range tc.peak.samples { + tc.peak.samples[i].at -= int64(d) + } + tc.peak.mu.Unlock() + c := &tc.cov + c.mu.Lock() + for _, sc := range c.servers { + if sc.shortSince != 0 { + sc.shortSince -= int64(d) + } + if sc.evacuatedAt != 0 { + sc.evacuatedAt -= int64(d) + } + } + c.windowStart -= int64(d) + c.lastEvacuation -= int64(d) + c.mu.Unlock() +} + +// evacuated is the servers coverage has given up and not yet forgiven. +func (s *coverageSim) evacuated() []string { + c := &s.tc.cov + c.mu.Lock() + defer c.mu.Unlock() + var out []string + for id, sc := range c.servers { + if sc.evacuatedAt != 0 { + out = append(out, id) + } + } + sort.Strings(out) + return out +} + +// begin dispatches a call to server the way a tunnel server does: onto the most +// recently active idle stream the server holds for this agent. It returns nil, +// and counts the call starved, when the server holds streams but none is idle -- +// the caller waits, which is the failure. +func (s *coverageSim) begin(server string) *tunnelStream { + tc := s.tc + tc.mu.Lock() + var pick *tunnelStream + held := false + for st := range tc.streams { + if st.serverID != server || st.retiring.Load() { + continue + } + held = true + if st.inflight.Load() == 0 && (pick == nil || st.quietSince() > pick.quietSince()) { + pick = st + } + } + tc.mu.Unlock() + if pick == nil { + if held { + s.starved++ + } + return nil + } + tc.callStarted(pick) + if !s.deferOpens { + s.openToTarget() + } + return pick +} + +func (s *coverageSim) end(st *tunnelStream) { + s.tc.callFinished(st) +} + +// servers is every server holding at least one of the agent's live streams, with +// how many and how many idle. +func (s *coverageSim) servers() map[string][2]int { + s.tc.mu.Lock() + defer s.tc.mu.Unlock() + out := map[string][2]int{} + for st := range s.tc.streams { + if st.retiring.Load() { + continue + } + c := out[st.serverID] + c[0]++ + if st.inflight.Load() == 0 { + c[1]++ + } + out[st.serverID] = c + } + return out +} + +// Placement models. + +var fourServers = []string{"srv-a", "srv-b", "srv-c", "srv-d"} + +// uniform places each stream on a server at random. +func uniform(seed uint64, servers ...string) func() string { + r := rand.New(rand.NewPCG(seed, 1)) + return func() string { return servers[r.IntN(len(servers))] } +} + +// zoneSkewed sends most streams to the first server, the nearest zone under the +// balancer's rate-based mode, and spills the rest across the others. +func zoneSkewed(seed uint64, servers ...string) func() string { + r := rand.New(rand.NewPCG(seed, 2)) + return func() string { + if r.Float64() < 0.7 { + return servers[0] + } + return servers[1+r.IntN(len(servers)-1)] + } +} + +// scripted places the first opens as listed, then always on fallback: a server +// that got a stream once and is never chosen again. +func scripted(fallback string, first ...string) func() string { + i := 0 + return func() string { + if i < len(first) { + i++ + return first[i-1] + } + return fallback + } +} + +type placementCase struct { + name string + place func() string +} + +func placements() []placementCase { + var cases []placementCase + for seed := uint64(1); seed <= 25; seed++ { + cases = append(cases, + placementCase{fmt.Sprintf("uniform/%d", seed), uniform(seed, fourServers...)}, + placementCase{fmt.Sprintf("skewed/%d", seed), zoneSkewed(seed, fourServers...)}, + ) + } + return cases +} + +// The incident's precondition, removed: once settled, every server the agent is +// on holds its idle reserve, so no server is one call away from all busy. +func TestCoverage_EveryServerHoldsItsReserve(t *testing.T) { + for _, pc := range placements() { + t.Run(pc.name, func(t *testing.T) { + s := newCoverageSim(t, 4, pc.place) + s.run(2 * time.Minute) + + servers := s.servers() + require.NotEmpty(t, servers) + for server, c := range servers { + assert.GreaterOrEqual(t, c[1], idleStreamsPerServer, + "%s holds %d streams, %d idle: one call from all busy", server, c[0], c[1]) + } + }) + } +} + +// The incident itself: a call lands on a server, and the health probe that +// follows it to the same server must find a stream free. +func TestCoverage_ProbeBehindACallFindsAStream(t *testing.T) { + for _, pc := range placements() { + t.Run(pc.name, func(t *testing.T) { + s := newCoverageSim(t, 4, pc.place) + s.run(2 * time.Minute) + + for server := range s.servers() { + call := s.begin(server) + require.NotNil(t, call, "%s starved its first call", server) + probe := s.begin(server) + assert.NotNil(t, probe, "probe on %s waited behind one call", server) + if probe != nil { + s.end(probe) + } + s.end(call) + } + assert.Zero(t, s.starved) + }) + } +} + +// Slow calls piling onto one server take its reserve; the agent must top that server +// back up, or give it up, while the calls are still running -- not leave it at +// zero idle for as long as they last. +func TestCoverage_SlowCallsOnOneServerAreCoveredWhileTheyRun(t *testing.T) { + for _, pc := range placements() { + t.Run(pc.name, func(t *testing.T) { + s := newCoverageSim(t, 4, pc.place) + s.run(2 * time.Minute) + + var server string + for p := range s.servers() { + if server == "" || p < server { + server = p + } + } + var calls []*tunnelStream + for i := 0; i < 2; i++ { + c := s.begin(server) + require.NotNil(t, c) + calls = append(calls, c) + } + s.run(time.Minute) // the calls are still in flight + + c := s.servers()[server] + evacuated := slices.Contains(s.evacuated(), server) + assert.True(t, c[1] >= idleStreamsPerServer || evacuated, + "%s has %d of %d streams idle with slow calls running, and was not given up", server, c[1], c[0]) + for _, c := range calls { + s.end(c) + } + }) + } +} + +// A server the balancer will not place on again cannot be filled. The agent +// gives it up -- its streams move to servers it can cover -- and the attempts +// stay bounded: no open-and-retire loop against a balancer that keeps +// missing. +func TestCoverage_UnfillableServerIsGivenUpWithBoundedChurn(t *testing.T) { + // The first stream lands on srv-b; every one after it on srv-a. + s := newCoverageSim(t, 4, scripted("srv-a", "srv-b")) + s.run(2 * time.Minute) + + servers := s.servers() + assert.NotContains(t, servers, "srv-b", "the server that could not be covered still holds the token") + assert.GreaterOrEqual(t, servers["srv-a"][1], idleStreamsPerServer) + + // An hour on, nothing should still be churning. + settled := s.opened + s.run(time.Hour) + assert.LessOrEqual(t, s.opened-settled, 2, + "a settled agent kept opening streams: %d in an hour", s.opened-settled) +} + +// Evacuating is how the agent leaves a server it cannot cover; it must never be +// how it leaves every server. With a stream cap too low for any server to reach its +// budget, nothing is evacuated. +func TestCoverage_NeverGivesUpEveryServer(t *testing.T) { + s := newCoverageSim(t, 2, uniform(7, "srv-a", "srv-b")) + s.tc.serverMaxStreams.Store(1) // cap: one stream per server + s.run(5 * time.Minute) + + assert.Empty(t, s.evacuated()) + assert.NotEmpty(t, s.servers()) +} + +// Servers running short together are given up one at a time, so a burst of slow +// calls everywhere does not strip the agent down to a single server at once. +func TestCoverage_GivesUpOneServerAtATime(t *testing.T) { + // Every stream after the first three lands on srv-a: srv-b, srv-c and + // srv-d each get one and can never be filled. + s := newCoverageSim(t, 4, scripted("srv-a", "srv-b", "srv-c", "srv-d")) + s.run(coverageGrace + 5*time.Second) + + assert.Len(t, s.evacuated(), 1, "one short server given up per grace period") + + s.run(3 * coverageGrace) + assert.Equal(t, []string{"srv-a"}, keys(s.servers()), "in time, every server that cannot be covered is left") +} + +func keys(m map[string][2]int) []string { + out := make([]string, 0, len(m)) + for k := range m { + out = append(out, k) + } + sort.Strings(out) + return out +} + +// Coverage must not reintroduce churn: once every server is covered, an idle +// agent opens nothing. +func TestCoverage_QuietOnceCovered(t *testing.T) { + for _, pc := range placements() { + t.Run(pc.name, func(t *testing.T) { + s := newCoverageSim(t, 4, pc.place) + s.run(2 * time.Minute) + settled := s.opened + s.run(30 * time.Minute) + assert.Zero(t, s.opened-settled, "an idle, covered agent kept opening streams") + }) + } +} + +// Periodic probe traffic across every server -- the production steady state -- +// must neither starve nor churn. +func TestCoverage_PeriodicProbesNeitherStarveNorChurn(t *testing.T) { + for _, pc := range placements() { + t.Run(pc.name, func(t *testing.T) { + s := newCoverageSim(t, 4, pc.place) + s.run(2 * time.Minute) + settled := s.opened + for i := 0; i < 12; i++ { + for server := range s.servers() { + if c := s.begin(server); c != nil { + s.end(c) + } + } + s.run(285 * time.Second) + } + assert.Zero(t, s.starved) + assert.LessOrEqual(t, s.opened-settled, 8, "steady periodic traffic kept opening streams") + }) + } +} + +// retireExcess spends a surplus only from servers above their budget. A server at +// exactly its reserve keeps it, even with the fleet over target. +func TestRetireExcess_NeverTakesAServerBelowItsReserve(t *testing.T) { + tc := newSizingClient(t) + placement := append(repeatServer("srv-a", idleStreamsPerServer+1), repeatServer("srv-b", idleStreamsPerServer)...) + _, cancelled := seedRetirable(tc, placement) + + tc.targetSlots.Store(int32(idleStreamsPerServer + 1)) + tc.retireExcess() + + assert.Equal(t, map[string]int{"srv-a": 1}, cancelled(), + "only srv-a's stream above its reserve is surplus") +} + +// ----------------------------------------------------------------------------- +// Growth under load +// ----------------------------------------------------------------------------- + +// The dispatcher sends a token's calls to one tunnel server at a time, so +// that server's queue is the agent's whole demand. These pin how fast the +// agent grows into it, without depending on where the balancer places +// streams: the first asserts on what the agent asks for, and the rest bracket +// placement between every new stream landing on the busy server and none of +// them doing so. + +// queue models callers waiting on one tunnel server: each tick, the server +// hands waiting calls to whatever idle streams it holds, and they stay in +// flight -- slow upstream calls. +type queue struct { + s *coverageSim + server string + waiting int + inFlight []*tunnelStream +} + +func (q *queue) dispatch() { + for q.waiting > 0 { + st := q.s.begin(q.server) + if st == nil { + return + } + q.waiting-- + q.inFlight = append(q.inFlight, st) + } +} + +func (q *queue) finishAll() { + for _, st := range q.inFlight { + q.s.end(st) + } + q.inFlight = nil +} + +// A server with every stream busy gets a chunk of streams asked for on the +// next tick, not one: growth is greedy, and the floor says so before a single +// stream has landed. +func TestGrowth_AsksForAChunkWhenAServerIsQueueing(t *testing.T) { + s := newCoverageSim(t, 4, scripted("srv-a")) + s.run(time.Minute) + s.deferOpens = true + q := &queue{s: s, server: "srv-a", waiting: 1000} + q.dispatch() + busy := len(q.inFlight) + require.Positive(t, busy) + + workers := s.tc.runningWorkers + s.tc.sizeTick() + asked := int(s.tc.targetSlots.Load()) - workers + t.Logf("%d in flight, asked for %d more in one tick", busy, asked) + assert.GreaterOrEqual(t, asked, max(growthChunk, busy), + "one tick with %d calls in flight and callers waiting asked for %d streams", busy, asked) + q.finishAll() +} + +// Best case: every stream the agent opens lands on the busy server. A burst +// of a thousand waiting calls is fully in flight within a handful of ticks, +// because capacity doubles each tick. +func TestGrowth_DrainsABurstInAFewTicksWhenStreamsLand(t *testing.T) { + s := newCoverageSim(t, 4, scripted("srv-a")) + s.run(time.Minute) + s.deferOpens = true + q := &queue{s: s, server: "srv-a", waiting: 1000} + + ticks := 0 + for q.dispatch(); q.waiting > 0 && ticks < 60; q.dispatch() { + s.tick() + ticks++ + } + t.Logf("1000-call burst in flight after %d ticks", ticks) + assert.Zero(t, q.waiting, "callers still waiting after %d ticks", ticks) + assert.LessOrEqual(t, ticks, 10, "a 1000-call burst took %d ticks to reach streams", ticks) + q.finishAll() +} + +// Worst case: the balancer never places on the busy server. The agent cannot +// relieve that server's queue by opening streams, so it must not grow without +// bound trying, and it must never give up the server the dispatcher is +// sending to. +func TestGrowth_IsBoundedAndKeepsTheBusyServerWhenStreamsNeverLand(t *testing.T) { + // srv-b gets the first stream and nothing after it. + s := newCoverageSim(t, 4, scripted("srv-a", "srv-b")) + s.deferOpens = true + s.tick() // the first streams open; srv-b holds one + q := &queue{s: s, server: "srv-b", waiting: 100} + q.dispatch() + require.NotEmpty(t, q.inFlight, "srv-b should hold a stream to queue behind") + + s.run(10 * time.Minute) + q.dispatch() + + servers := s.servers() + assert.Contains(t, servers, "srv-b", "the busy server was given up") + total := 0 + for _, c := range servers { + total += c[0] + } + bound := growthSurplusFactor*max(growthChunk, len(q.inFlight)) + 4*idleStreamsPerServer + s.tc.anchorSlots() + growthChunk + t.Logf("never-landing balancer: %d streams in total after 10m, bound %d", total, bound) + assert.LessOrEqual(t, total, bound, "growth ran away to %d streams against a balancer that never helps", total) + q.finishAll() +} + +// Greedy on open, firm on prune: once a burst is over and its streams have +// aged out, the agent is back to the resting reserve -- what keeps shared +// tunnel servers from carrying every agent's last burst forever. +func TestGrowth_PrunesBackToRestAfterABurst(t *testing.T) { + s := newCoverageSim(t, 4, scripted("srv-a")) + s.run(time.Minute) + rest := s.servers()["srv-a"][0] + + s.deferOpens = true + q := &queue{s: s, server: "srv-a", waiting: 500} + for q.dispatch(); q.waiting > 0; q.dispatch() { + s.tick() + } + peak := s.servers()["srv-a"][0] + require.Greater(t, peak, 500) + t.Logf("burst peak %d streams, resting %d", peak, rest) + q.finishAll() + + s.run(streamIdleWindow + pruneMinAge + time.Minute) + assert.Equal(t, rest, s.servers()["srv-a"][0], "the burst's streams were not given back") +} + +// Each server answers what state it is in from its own counts and history; +// these pin the rule the table in coverage.go describes. +func TestServerCoverage_State(t *testing.T) { + now := time.Now().UnixNano() + grace := int64(coverageGrace) + for _, tc := range []struct { + name string + s serverCoverage + want coverageState + }{ + {"at budget", serverCoverage{serverStreamCounts: serverStreamCounts{have: idleStreamsPerServer}}, stateCovered}, + {"above budget, busy", serverCoverage{serverStreamCounts: serverStreamCounts{have: 10, busy: 3}}, stateCovered}, + {"short with calls in flight", serverCoverage{serverStreamCounts: serverStreamCounts{have: 2, busy: 2}, shortSince: now}, stateGrowing}, + {"short with calls in flight, long ago", serverCoverage{serverStreamCounts: serverStreamCounts{have: 1, busy: 1}, shortSince: now - 10*grace}, stateGrowing}, + {"short and idle, just now", serverCoverage{serverStreamCounts: serverStreamCounts{have: 1}, shortSince: now}, stateFilling}, + {"short and idle, past grace", serverCoverage{serverStreamCounts: serverStreamCounts{have: 1}, shortSince: now - grace}, stateOverdue}, + {"evacuated", serverCoverage{serverStreamCounts: serverStreamCounts{have: 1}, evacuatedAt: now}, stateEvacuated}, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, tc.s.state(now)) + }) + } +} + +// growth asks for at least a chunk, at least the calls in flight, and at least +// the shortfall. +func TestServerCoverage_Growth(t *testing.T) { + assert.Equal(t, growthChunk, (&serverCoverage{serverStreamCounts: serverStreamCounts{have: 1, busy: 1}}).growth()) + assert.Equal(t, 40, (&serverCoverage{serverStreamCounts: serverStreamCounts{have: 40, busy: 40}}).growth()) +} diff --git a/agent/server/grpctunnel/sizing_test.go b/agent/server/grpctunnel/sizing_test.go index e0fc732..12ca373 100644 --- a/agent/server/grpctunnel/sizing_test.go +++ b/agent/server/grpctunnel/sizing_test.go @@ -102,24 +102,25 @@ func TestSizing_OpensAStreamPerCallOnTopOfTheReserve(t *testing.T) { tc.resize() - // 5 in flight + 2 reserve. The reserve stays whole under load; that is + // 5 in flight + the reserve. The reserve stays whole under load; that is // what keeps the next call from waiting on a stream open. - assert.Equal(t, int32(7), tc.targetSlots.Load()) + assert.Equal(t, int32(5+idleStreamsPerServer), tc.targetSlots.Load()) } func TestSizing_GivesStreamsBackOnceTheBurstHasAged(t *testing.T) { tc := newSizingClient(t) - seedStreams(tc, 20, "srv-a") + burst := 18 + idleStreamsPerServer + seedStreams(tc, burst, "srv-a") require.Equal(t, 18, markUsed(tc, 18)) tc.resize() - require.Equal(t, int32(20), tc.targetSlots.Load()) + require.Equal(t, int32(burst), tc.targetSlots.Load()) // The burst passes. The streams it needed are kept, because reopening one // costs a round trip and a dispatcher notification and the next request is // probably seconds away. tc.resize() - assert.Equal(t, int32(20), tc.targetSlots.Load(), + assert.Equal(t, int32(burst), tc.targetSlots.Load(), "the streams that served the burst are still inside the window") // Once they fall out of it, the target drops straight to the reserve. @@ -520,6 +521,15 @@ func seedRetirable(tc *tunnelClient, placement []string) (map[string][]*tunnelSt } } +// repeatServer is n placements on the same server. +func repeatServer(server string, n int) []string { + out := make([]string, n) + for i := range out { + out[i] = server + } + return out +} + // useStream says st has just carried a call, alone: the stream is warm and the // peak saw one call in flight. func useStream(tc *tunnelClient, st *tunnelStream) { @@ -535,10 +545,10 @@ func useStream(tc *tunnelClient, st *tunnelStream) { // leaving the second exactly as fat as it was. func TestRetire_PrunesFromTheServerHoldingTheSurplus(t *testing.T) { tc := newSizingClient(t) - placement := make([]string, 0, 9) - for i := 0; i < 8; i++ { - placement = append(placement, "srv-a") - } + // srv-a holds its budget (one recent call plus the reserve) and seven + // more; srv-b holds one stream. + over := idleStreamsPerServer + 3 + placement := repeatServer("srv-a", 1+idleStreamsPerServer+over) placement = append(placement, "srv-b") byServer, cancelled := seedRetirable(tc, placement) @@ -549,9 +559,9 @@ func TestRetire_PrunesFromTheServerHoldingTheSurplus(t *testing.T) { tc.retireExcess() got := cancelled() - // Four, not five: five is what srv-a holds above its own budget, but the - // fleet is only four above target, and the fleet count is what says how - // many streams to give back. Shedding the fifth would leave the agent + // Four, not seven: seven is what srv-a holds above its own budget, but + // the fleet is only four above target, and the fleet count is what says + // how many streams to give back. Shedding more would leave the agent // below target and a replacement would open on the next tick — the churn // that per-server quotas used to cause. assert.Equal(t, 4, got["srv-a"], @@ -564,7 +574,7 @@ func TestRetire_PrunesFromTheServerHoldingTheSurplus(t *testing.T) { // ability to reach this agent, is never what pays for someone else's idleness. func TestRetire_NeverTakesAServerBelowItsOwnReserve(t *testing.T) { tc := newSizingClient(t) - placement := []string{"srv-a", "srv-a", "srv-a", "srv-a", "srv-a", "srv-a", "srv-b", "srv-b"} + placement := append(repeatServer("srv-a", idleStreamsPerServer+4), repeatServer("srv-b", idleStreamsPerServer)...) _, cancelled := seedRetirable(tc, placement) tc.resize() @@ -710,9 +720,9 @@ func TestStreamCap_UnknownServerCountAssumesOneServer(t *testing.T) { func TestStreamCap_NeverExceedsOurOwnPolicy(t *testing.T) { tc := newSizingClient(t) seedStreams(tc, 8, "s1", "s2", "s3", "s4", "s5", "s6", "s7", "s8") - tc.serverMaxStreams.Store(64) + tc.serverMaxStreams.Store(int32(maxStreams / 4)) - assert.Equal(t, maxStreams, tc.streamCap(), "8 x 64 exceeds our own ceiling") + assert.Equal(t, maxStreams, tc.streamCap(), "8 servers x their cap exceeds our own ceiling") } // ----------------------------------------------------------------------------- @@ -817,12 +827,15 @@ func TestRetireExcess_SparesStreamsThatJustOpened(t *testing.T) { // slot spent most of its life waiting rather than serving. func TestRetireExcess_DoesNotChurnWhenTheFleetIsAtTarget(t *testing.T) { tc := newSizingClient(t) - _, cancelled := seedRetirable(tc, []string{ - "srv-a", "srv-a", "srv-a", "srv-b", "srv-b", "srv-c", "srv-c", "srv-d", - }) + // Uneven, but exactly the reserve across four servers in total. + placement := repeatServer("srv-a", idleStreamsPerServer+1) + placement = append(placement, repeatServer("srv-b", idleStreamsPerServer)...) + placement = append(placement, repeatServer("srv-c", idleStreamsPerServer)...) + placement = append(placement, repeatServer("srv-d", idleStreamsPerServer-1)...) + _, cancelled := seedRetirable(tc, placement) tc.resize() - require.Equal(t, int32(8), tc.targetSlots.Load(), "the reserve across four servers") + require.Equal(t, int32(4*idleStreamsPerServer), tc.targetSlots.Load(), "the reserve across four servers") tc.retireExcess() total := 0 @@ -838,9 +851,9 @@ func TestRetireExcess_DoesNotChurnWhenTheFleetIsAtTarget(t *testing.T) { // backoff, so a slot retired for being surplus came back slower every time. func TestRetire_MarksTheStreamSoTheWorkerKnowsItWasDeliberate(t *testing.T) { tc := newSizingClient(t) - byServer, _ := seedRetirable(tc, []string{"srv-a", "srv-a", "srv-a", "srv-a"}) + byServer, _ := seedRetirable(tc, repeatServer("srv-a", idleStreamsPerServer+2)) - tc.targetSlots.Store(2) + tc.targetSlots.Store(int32(idleStreamsPerServer)) tc.retireExcess() retiring := 0 diff --git a/agent/server/grpctunnel/tunnel_client.go b/agent/server/grpctunnel/tunnel_client.go index 6e054e0..8f04121 100644 --- a/agent/server/grpctunnel/tunnel_client.go +++ b/agent/server/grpctunnel/tunnel_client.go @@ -4,6 +4,7 @@ import ( "context" "crypto/tls" "crypto/x509" + "encoding/json" "errors" "fmt" "math/rand/v2" @@ -55,22 +56,38 @@ const ( failuresBeforeReregister = 5 // idleStreamsPerServer is how many streams are kept idle and ready on - // each server instance the agent is connected to. It sets how large a - // burst is absorbed without waiting for a stream to open — a latency - // property, not a capacity one, which is why it is a constant and not a - // knob. + // each tunnel server the agent is connected to, at rest. It sets how large + // a burst is absorbed before growth (coverage.go) has opened more — a + // latency property, not a capacity one, which is why it is a constant and + // not a knob. Every idle stream costs its tunnel server a heartbeat each + // way per interval and a few goroutines, multiplied across every agent + // sharing that server, so the resting reserve is kept small -- enough + // that a call in flight never makes the next one wait -- and bursts are + // met by opening greedily instead. A larger one also proved harder to + // place: the balancer favours the nearest zone, so a server that gets one + // stream in ten rarely collects several, and the agent ended up giving + // such servers up rather than covering them. idleStreamsPerServer = 2 // maxStreams is the concurrency backstop. Reaching it is not a failure: // the agent stops offering idle streams, the server's dispatch blocks, // and the wait reaches the caller as latency. Without a ceiling the // agent would accept work it cannot do and the queue would move into its - // heap, where nothing can see it. - maxStreams = 256 + // heap, where nothing can see it. Streams carry I/O, not work, so the + // ceiling is set by what one agent should relay at once -- thousands of + // calls a second -- not by the agent's CPU. + maxStreams = 2048 // sizeInterval is how often the stream count is re-evaluated. sizeInterval = time.Second + // pruneMinAge is how long a stream lives before it may be pruned as + // surplus. Growth opens greedily and the balancer places where it likes, + // so under sustained load some of what opens lands on servers that do not + // need it; pruning those the next tick, while the load that asked for them + // is still running, would close and reopen streams every tick. + pruneMinAge = 2 * time.Minute + // streamIdleWindow is how long the agent keeps the streams a burst // needed after the burst has passed. Sizing holds the peak concurrency // seen inside it, and a stream that has carried a call inside it is not @@ -164,6 +181,11 @@ type tunnelClient struct { // is what resize sizes on. peak busyPeak + // cov is per-tunnel-server coverage state, and coverageFloor the stream + // count it holds the target at while a server is short; see coverage.go. + cov coverage + coverageFloor atomic.Int32 + // streamLifetime is how long this instance lets a stream live before // recycling it; zero disables recycling. Picked on every Start. streamLifetime time.Duration @@ -200,6 +222,8 @@ type tunnelClient struct { heartbeatTimeoutsTotal *prometheus.CounterVec tokenRotationsTotal prometheus.Counter connectErrorsTotal *prometheus.CounterVec + serverStreamsGauge *prometheus.GaugeVec + coverageActions *prometheus.CounterVec } type tunnelStream struct { @@ -321,6 +345,29 @@ func NewTunnelClient(p TunnelClientParams) snykbroker.RelayInstanceManager { tc.inflightSem = make(chan struct{}, p.Config.MaxInflightRequests) } + collectors := tc.initMetrics() + + p.HttpServer.RegisterHandler(tc) + + if p.Registry != nil { + p.Registry.MustRegister(collectors...) + } + + if p.Lifecycle != nil { + p.Lifecycle.Append(fx.Hook{ + OnStart: func(ctx context.Context) error { return tc.Start() }, + OnStop: func(ctx context.Context) error { return tc.Close() }, + }) + } + + return tc +} + +// RegisterRoutes implements cortexHttp.Handler. Exposes /restart and +// /systemcheck under the broker path prefix to preserve the snyk-broker URL +// shape. +// initMetrics creates the client's metrics and returns them for registering. +func (tc *tunnelClient) initMetrics() []prometheus.Collector { tc.connectionsActive = prometheus.NewGaugeVec( prometheus.GaugeOpts{Name: "grpc_tunnel_connections_active", Help: "Active gRPC tunnel streams"}, []string{"server_id"}, @@ -359,35 +406,30 @@ func NewTunnelClient(p TunnelClientParams) snykbroker.RelayInstanceManager { []string{"phase"}, ) - p.HttpServer.RegisterHandler(tc) - - if p.Registry != nil { - p.Registry.MustRegister( - tc.connectionsActive, - tc.requestsTotal, - tc.requestsInflight, - tc.requestsRejected, - tc.reconnectsTotal, - tc.requestDuration, - tc.heartbeatTimeoutsTotal, - tc.tokenRotationsTotal, - tc.connectErrorsTotal, - ) - } + tc.serverStreamsGauge = prometheus.NewGaugeVec( + prometheus.GaugeOpts{Name: "grpc_tunnel_server_streams", Help: "Tunnel streams on each tunnel server, idle and busy"}, + []string{"server_id", "state"}, + ) + tc.coverageActions = prometheus.NewCounterVec( + prometheus.CounterOpts{Name: "grpc_tunnel_coverage_actions_total", Help: "Streams opened to cover a short tunnel server (fill), or recycled off one given up on (evacuate)"}, + []string{"action"}, + ) - if p.Lifecycle != nil { - p.Lifecycle.Append(fx.Hook{ - OnStart: func(ctx context.Context) error { return tc.Start() }, - OnStop: func(ctx context.Context) error { return tc.Close() }, - }) + return []prometheus.Collector{ + tc.connectionsActive, + tc.requestsTotal, + tc.requestsInflight, + tc.requestsRejected, + tc.reconnectsTotal, + tc.requestDuration, + tc.heartbeatTimeoutsTotal, + tc.tokenRotationsTotal, + tc.connectErrorsTotal, + tc.serverStreamsGauge, + tc.coverageActions, } - - return tc } -// RegisterRoutes implements cortexHttp.Handler. Exposes /restart and -// /systemcheck under the broker path prefix to preserve the snyk-broker URL -// shape. func (tc *tunnelClient) RegisterRoutes(mux *mux.Router) error { sub := mux.PathPrefix(fmt.Sprintf("%s/broker", cortexHttp.AxonPathRoot)).Subrouter() sub.HandleFunc("/restart", tc.handleRestart) @@ -421,8 +463,14 @@ func (tc *tunnelClient) handleSystemCheck(w http.ResponseWriter, req *http.Reque active := len(tc.streams) tc.mu.Unlock() w.Header().Set("Content-Type", "application/json") - fmt.Fprintf(w, `{"status":"ok","relay_mode":"grpc-tunnel","streams":%d,"busy":%d,"target":%d}`, - active, tc.busySlots.Load(), tc.targetSlots.Load()) + _ = json.NewEncoder(w).Encode(struct { + Status string `json:"status"` + RelayMode string `json:"relay_mode"` + Streams int `json:"streams"` + Busy int32 `json:"busy"` + Target int32 `json:"target"` + Servers []serverSummary `json:"servers"` + }{"ok", "grpc-tunnel", active, tc.busySlots.Load(), tc.targetSlots.Load(), tc.cov.summaries(time.Now().UnixNano())}) } func (tc *tunnelClient) Start() error { @@ -469,6 +517,8 @@ func (tc *tunnelClient) resetForStart() { tc.streamLifetime = pickStreamLifetime() tc.observedServers.Store(0) tc.targetSlots.Store(int32(tc.idleReserve())) + tc.cov.reset() + tc.coverageFloor.Store(0) ctx, cancel := context.WithCancel(context.Background()) tc.parentCtx = ctx tc.cancelAll = cancel @@ -815,6 +865,13 @@ func (tc *tunnelClient) resize() { } desired := busy + tc.idleReserve() + // Hold whatever coverage needs while a tunnel server is short of its + // budget: the streams it opened to reach that server, including the ones + // that landed elsewhere, until the server is covered or given up on. + if floor := int(tc.coverageFloor.Load()); desired < floor { + desired = floor + } + // Never ask for fewer streams than there are connections to anchor. // Without this the target can sit below the anchor count, and every // tick retireExcess would try to cancel streams that refuse to retire. @@ -957,14 +1014,88 @@ func (tc *tunnelClient) sizeLoop() { case <-tc.parentCtx.Done(): return case <-ticker.C: - tc.refreshObservedServers() - tc.resize() - tc.retireExcess() - tc.recycleAged() + tc.sizeTick() + } + } +} + +// sizeTick is one pass of the sizing loop. +func (tc *tunnelClient) sizeTick() { + tc.refreshObservedServers() + tc.balanceCoverage() + tc.resize() + tc.retireExcess() + tc.recycleAged() + tc.recordServerStreams() +} + +// balanceCoverage runs once per sizing tick, before resize: it hands the +// agent's streams per tunnel server to coverage (coverage.go) and carries out +// the plan it returns -- the floor resize honours, metrics, logs, and +// recycling evacuated servers' idle streams. +func (tc *tunnelClient) balanceCoverage() { + now := time.Now() + + tc.mu.Lock() + counts := countServerStreams(tc.streams, now) + workers := tc.runningWorkers + pending := workers - len(tc.streams) // workers still opening a stream + tc.mu.Unlock() + + p := tc.cov.tick(counts, now.UnixNano(), workers, pending) + + tc.coverageFloor.Store(int32(p.floor)) + if p.grown > 0 { + tc.coverageActions.WithLabelValues("grow").Add(float64(p.grown)) + } + if p.filled > 0 { + tc.coverageActions.WithLabelValues("fill").Add(float64(p.filled)) + } + for _, id := range p.evacuate { + s := counts[id] + tc.logger.Info("Leaving tunnel server that cannot be covered", + zap.String("serverId", id), + zap.Int("streams", s.have), + zap.Int("busy", s.busy), + zap.Int("budget", s.budget()), + zap.Duration("shortFor", coverageGrace)) + } + if p.exhausted { + tc.logger.Warn("Tunnel server coverage budget spent with servers still short", + zap.Strings("shortServers", p.short), + zap.Int("budget", p.budget), + zap.Duration("window", streamIdleWindow)) + } + + if len(p.recycle) > 0 { + // Marked under the lock, like every retirement, so the worker sees + // the reason before the cancel. + tc.mu.Lock() + for _, st := range p.recycle { + st.recycling.Store(true) + st.retiring.Store(true) + } + tc.mu.Unlock() + tc.coverageActions.WithLabelValues("evacuate").Add(float64(len(p.recycle))) + for _, st := range p.recycle { + st.cancel() } } } +// recordServerStreams publishes each tunnel server's streams, idle and busy, +// replacing the previous tick's so a server the agent has left drops out. +func (tc *tunnelClient) recordServerStreams() { + tc.mu.Lock() + servers := countServerStreams(tc.streams, time.Now()) + tc.mu.Unlock() + tc.serverStreamsGauge.Reset() + for srv, s := range servers { + tc.serverStreamsGauge.WithLabelValues(srv, "busy").Set(float64(s.busy)) + tc.serverStreamsGauge.WithLabelValues(srv, "idle").Set(float64(s.have - s.busy)) + } +} + // recycleDeadline is when a stream opened at openedAt has lived its lifetime: // this instance's lifetime, jittered ±10% for the stream. func (tc *tunnelClient) recycleDeadline(openedAt int64) int64 { @@ -1049,6 +1180,7 @@ func (tc *tunnelClient) retireExcess() { // narrow the window in which the server dispatches onto a stream we are // cancelling; the dispatcher fails such a call back to its caller. cutoff := now.Add(-sizeInterval).UnixNano() + youngAfter := now.Add(-pruneMinAge).UnixNano() tc.mu.Lock() @@ -1079,7 +1211,7 @@ func (tc *tunnelClient) retireExcess() { switch { case st.inflight.Load() != 0 || st.lastCallAt.Load() >= staleBefore: g.used++ - case st.anchor || st.quietSince() >= cutoff: + case st.anchor || st.quietSince() >= cutoff || st.openedAt.Load() >= youngAfter: g.pinned++ default: g.prunable = append(g.prunable, st) @@ -1089,9 +1221,10 @@ func (tc *tunnelClient) retireExcess() { // The order the surplus is spent in. Streams with no server yet go first: // they protect no coverage, because no dispatcher knows about them. Then // whatever sits above a server's own budget, which is the surplus wherever - // the fleet count came from. Only then the stalest anywhere, for the case - // the first two do not cover the count — a cap dropping under us, or every - // server sitting exactly at budget while the fleet is still over target. + // the fleet count came from. A server's budget is otherwise a veto: taking + // it below its reserve is how a server ends up one in-flight call from making + // every caller there wait. The one exception is a stream cap dropping under + // us, which must be obeyed, so only then is the stalest anywhere fair game. var spend, overBudget, rest []*tunnelStream for serverID, g := range groups { sortStalestFirst(g.prunable) @@ -1118,9 +1251,11 @@ func (tc *tunnelClient) retireExcess() { rest = append(rest, g.prunable[over:]...) } sortStalestFirst(overBudget) - sortStalestFirst(rest) spend = append(spend, overBudget...) - spend = append(spend, rest...) + if tc.runningWorkers > tc.streamCap() { + sortStalestFirst(rest) + spend = append(spend, rest...) + } if surplus > len(spend) { surplus = len(spend) @@ -1143,7 +1278,7 @@ func (tc *tunnelClient) retireExcess() { // serverStreams buckets one server's streams by what may be done with them. type serverStreams struct { used int // carried a call inside the window - pinned int // anchors, and streams quiet for under a tick + pinned int // anchors, streams quiet for under a tick, and streams younger than pruneMinAge prunable []*tunnelStream // the rest, stalest first once sorted } @@ -1401,10 +1536,9 @@ func (tc *tunnelClient) openSlot(id int, recovering bool) (*streamCtx, error) { tc.serverMaxStreams.Store(serverHello.MaxStreams) } - // The per-server cap counts connections, not streams: on a shared - // connection only the first stream takes a slot, since every stream on it - // talks to the same backend. That keeps MaxStreamsPerServer meaning "how - // many connections may land on one server instance" in every mode. + // Behind the load balancer each stream is placed on its own, so streams on + // one connection can land on different server instances; ServerId is the + // only record of where this one went. ts := newTunnelStream(id, serverHello, conn, cancel, id <= tc.anchorSlots()) ts.recycleAt = tc.recycleDeadline(ts.openedAt.Load()) diff --git a/agent/server/grpctunnel/tunnel_client_test.go b/agent/server/grpctunnel/tunnel_client_test.go index 7529ade..a88cbea 100644 --- a/agent/server/grpctunnel/tunnel_client_test.go +++ b/agent/server/grpctunnel/tunnel_client_test.go @@ -240,30 +240,7 @@ func newTestClient(t *testing.T, cfg config.AgentConfig, reg snykbroker.Registra tc.inflightSem = make(chan struct{}, cfg.MaxInflightRequests) } - tc.connectionsActive = prometheus.NewGaugeVec( - prometheus.GaugeOpts{Name: "grpc_tunnel_connections_active"}, []string{"server_id"}) - tc.requestsTotal = prometheus.NewCounterVec( - prometheus.CounterOpts{Name: "grpc_tunnel_requests_total"}, []string{"method", "status"}) - tc.requestsInflight = prometheus.NewGauge( - prometheus.GaugeOpts{Name: "grpc_tunnel_requests_inflight"}) - tc.requestsRejected = prometheus.NewCounter( - prometheus.CounterOpts{Name: "grpc_tunnel_request_rejected_total"}) - tc.reconnectsTotal = prometheus.NewCounterVec( - prometheus.CounterOpts{Name: "grpc_tunnel_reconnects_total"}, []string{"server_id"}) - tc.requestDuration = prometheus.NewHistogramVec( - prometheus.HistogramOpts{Name: "grpc_tunnel_request_duration_ms"}, []string{"method"}) - tc.heartbeatTimeoutsTotal = prometheus.NewCounterVec( - prometheus.CounterOpts{Name: "grpc_tunnel_heartbeat_timeouts_total"}, []string{"server_id"}) - tc.tokenRotationsTotal = prometheus.NewCounter( - prometheus.CounterOpts{Name: "grpc_tunnel_token_rotations_total"}) - tc.connectErrorsTotal = prometheus.NewCounterVec( - prometheus.CounterOpts{Name: "grpc_tunnel_connect_errors_total"}, []string{"phase"}) - - registry.MustRegister( - tc.connectionsActive, tc.requestsTotal, tc.requestsInflight, - tc.requestsRejected, tc.reconnectsTotal, tc.requestDuration, - tc.heartbeatTimeoutsTotal, tc.tokenRotationsTotal, tc.connectErrorsTotal, - ) + registry.MustRegister(tc.initMetrics()...) return tc, registry } diff --git a/server/config/config.go b/server/config/config.go index 7c34ff8..69f7429 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -11,11 +11,15 @@ import ( ) const ( - DefaultGrpcPort = 50052 - DefaultHttpPort = 8080 - DefaultHeartbeatInterval = 30 * time.Second - DefaultMaxFrameBytes = 1 << 20 // 1 MiB - DefaultMaxStreamsPerToken = 64 + DefaultGrpcPort = 50052 + DefaultHttpPort = 8080 + DefaultHeartbeatInterval = 30 * time.Second + DefaultMaxFrameBytes = 1 << 20 // 1 MiB + // DefaultMaxStreamsPerToken is high because the dispatcher sends a + // token's calls to one tunnel server at a time, so this server's share is + // all of the agent's concurrency. It guards against a runaway agent, not + // ordinary load. + DefaultMaxStreamsPerToken = 1024 // DefaultAgentReportGrace is how long a call the server cancelled stays // open for the agent's answer. The agent replies within a round trip of // seeing the cancel; a second covers a slow path without holding a stream