From 42b1ffb653487f7d83d7eaf194ec2c822c50224b Mon Sep 17 00:00:00 2001 From: Shawn Burke Date: Tue, 10 Nov 2026 07:51:20 +1100 Subject: [PATCH 1/2] grpctunnel: explain every dispatch failure (phase, stream state, agent report) A failed dispatch now says what went wrong, not just who gave up. Prod's 2s /__axon/info probe failures and 60s upstream timeouts both logged as caller_canceled with nothing to tell them apart. - DispatchError gains Phase (acquire vs awaiting_response), the token's stream state at failure (streams, busy, waiters, per-instance split, oldest holding calls with path and age), and a FailureCause: streams_exhausted, upstream_slow, agent_unresponsive, agent_at_capacity, ... - CallCancel carries the agent's account: stage, cause, stage/total elapsed, upstream host. The agent fills it in on every early end, including its answer to a server-sent cancel. - On a server-side timeout the call is held, stream busy, for up to AGENT_REPORT_GRACE (1s) for that answer. This also stops the server dispatching onto a stream the agent is still running the old call on. - The response goes out at once with X-Axon-Dispatch-Failure/-Phase; the log line waits for the agent and records the final cause, counted in a new dispatch.failures{cause} metric. (The old tagged dispatch.errors helper collided with the untagged counter and would have panicked if called.) - Repro tests for both prod shapes in dispatch and adapters. Co-Authored-By: Claude Opus 5.5 --- README.relay.md | 20 ++ agent/server/grpctunnel/call_handler.go | 64 +++- agent/server/grpctunnel/call_progress.go | 68 ++++ agent/server/grpctunnel/call_report_test.go | 141 ++++++++ proto/tunnel/v2/tunnel.proto | 48 +++ server/.generated/proto/tunnelpb/tunnel.pb.go | 289 ++++++++++++--- .../adapters/dispatch_failure_cause_test.go | 214 +++++++++++ server/adapters/http.go | 178 +++++++++- server/config/config.go | 24 ++ server/dispatch/dispatcher.go | 313 ++++++++++++++++- server/dispatch/failure_cause.go | 110 ++++++ server/dispatch/failure_cause_test.go | 61 ++++ server/dispatch/failure_diagnostics_test.go | 332 ++++++++++++++++++ server/metrics/metrics.go | 57 +-- server/tunnel/client_registry.go | 58 +++ 15 files changed, 1862 insertions(+), 115 deletions(-) create mode 100644 agent/server/grpctunnel/call_progress.go create mode 100644 agent/server/grpctunnel/call_report_test.go create mode 100644 server/adapters/dispatch_failure_cause_test.go create mode 100644 server/dispatch/failure_cause.go create mode 100644 server/dispatch/failure_cause_test.go create mode 100644 server/dispatch/failure_diagnostics_test.go diff --git a/README.relay.md b/README.relay.md index 17b49f00..b3d60c23 100644 --- a/README.relay.md +++ b/README.relay.md @@ -183,6 +183,26 @@ the backpressure, and it needs no configuration: `dispatch.acquire_wait_ms` and `dispatch.all_busy` on the server report it directly, and are what distinguish a saturated tunnel from a slow upstream. +A dispatch that fails says why. The server's `Dispatch failed` log line, the +`dispatch.failures` counter, and the `X-Axon-Dispatch-Failure` response header +carry a cause: + +| Cause | Meaning | +|-------|---------| +| `streams_exhausted` | Every stream for the token on that server stayed busy until the deadline. The log names the calls holding them (`holders`) and the per-instance split (`tokenInstances`). | +| `upstream_slow` | The agent sent the request upstream and no headers came back in time. `upstreamHost` and `agentStageElapsed` say where and for how long. | +| `upstream_error` | The upstream failed outright. | +| `agent_at_capacity` | The call queued for the agent's own in-flight capacity and never got it. | +| `agent_unresponsive` | The server cancelled the call and the agent did not answer within `AGENT_REPORT_GRACE`. | +| `route_rejected`, `agent_shutdown`, `tunnel_error`, `no_tunnel` | As named. | + +`X-Axon-Dispatch-Phase` says whether the call ever got a stream (`acquire`) or +was with the agent (`awaiting_response`). When the server gives up on a call +the agent has, it holds the stream for up to `AGENT_REPORT_GRACE` (default +`1s`) for the agent's account of where the call was. The response goes out +first, so its header can only say `awaiting_agent`; the log line carries the +final cause. + The design, wire protocol, and rollout plan are documented in [docs/design/grpc-tunnel-v2.md](docs/design/grpc-tunnel-v2.md). diff --git a/agent/server/grpctunnel/call_handler.go b/agent/server/grpctunnel/call_handler.go index ff412f70..5763fa6f 100644 --- a/agent/server/grpctunnel/call_handler.go +++ b/agent/server/grpctunnel/call_handler.go @@ -7,10 +7,12 @@ import ( "net/http" "strings" "sync" + "sync/atomic" "time" pb "github.com/cortexapps/axon/.generated/proto/github.com/cortexapps/axon/tunnelpb" "go.uber.org/zap" + "go.uber.org/zap/zapcore" ) // agentCall tracks one in-flight call received over a tunnel stream. @@ -21,6 +23,12 @@ type agentCall struct { bodyW *io.PipeWriter // cancel aborts the call's context (backend request included). cancel context.CancelFunc + // serverCancelled is set when the server sent CallCancel, so the call's + // own CallCancel can say the server ended it rather than the upstream + // failing or the agent's deadline passing -- all three end in the same + // context error. + serverCancelled atomic.Bool + progress callProgress } // callTable tracks the in-flight calls on one stream. The server sends at @@ -126,6 +134,7 @@ func (tc *tunnelClient) handleCallFrame(sc *streamCtx, table *callTable, frame * // written its log lines. An entry in the table has to mean a // goroutine is still running, which is what the heartbeat check // relies on and what the test barrier waits for. + c.serverCancelled.Store(true) c.bodyW.CloseWithError(fmt.Errorf("cancelled by server: %s", body.Cancel.Reason)) c.cancel() } @@ -159,7 +168,7 @@ func (tc *tunnelClient) startCall(sc *streamCtx, table *callTable, callID string ctx, cancel := context.WithTimeout(parent, timeout) bodyR, bodyW := io.Pipe() - c := &agentCall{id: callID, bodyW: bodyW, cancel: cancel} + c := &agentCall{id: callID, bodyW: bodyW, cancel: cancel, progress: newCallProgress(time.Now())} if !table.add(c) { cancel() bodyW.CloseWithError(fmt.Errorf("duplicate call id")) @@ -203,21 +212,27 @@ func (tc *tunnelClient) startCall(sc *streamCtx, table *callTable, callID string tc.requestsRejected.Inc() tc.requestsTotal.WithLabelValues(start.PseudoHeaders[":method"], "503").Inc() bodyR.CloseWithError(ctx.Err()) - tc.sendCancel(sc.sendFn, callID, http.StatusServiceUnavailable, - "timed out waiting for in-flight capacity") + cause := tc.stopCause(ctx, c) + if cause == pb.CallCancel_AGENT_TIMEOUT { + cause = pb.CallCancel_CAPACITY + } + tc.sendCallCancel(sc.sendFn, callID, c.progress.report(http.StatusServiceUnavailable, + "timed out waiting for in-flight capacity", cause)) return } } - tc.runCall(ctx, sc, callID, start, bodyR) + tc.runCall(ctx, sc, c, start, bodyR) }() } // runCall routes and executes one call, streaming the response back. -func (tc *tunnelClient) runCall(ctx context.Context, sc *streamCtx, callID string, start *pb.CallStart, bodyR *io.PipeReader) { +func (tc *tunnelClient) runCall(ctx context.Context, sc *streamCtx, c *agentCall, start *pb.CallStart, bodyR *io.PipeReader) { + callID := c.id method := start.PseudoHeaders[":method"] startTime := time.Now() + c.progress.enter(pb.CallCancel_ROUTING) breq, err := tc.router.Route(start) if err != nil { bodyR.CloseWithError(err) @@ -229,25 +244,38 @@ func (tc *tunnelClient) runCall(ctx context.Context, sc *streamCtx, callID strin zap.String("path", start.PseudoHeaders[":path"]), zap.Error(err), ) - tc.sendCancel(sc.sendFn, callID, code, reason) + tc.sendCallCancel(sc.sendFn, callID, c.progress.report(code, reason, pb.CallCancel_ROUTE_REJECTED)) return } breq.Body = bodyR + if breq.URL != nil { + c.progress.upstreamHost = breq.URL.Host + } + c.progress.enter(pb.CallCancel_AWAITING_UPSTREAM_HEADERS) resp, err := tc.backend.Do(ctx, breq) if err != nil { bodyR.CloseWithError(err) tc.requestsTotal.WithLabelValues(method, "502").Inc() - tc.logger.Error("Call execution failed", + cause := tc.stopCause(ctx, c) + // Only an upstream that failed is an error of ours to chase; a call + // the server gave up on, or that ran out its own deadline, is the + // upstream being slow, and the server logs the failure as well. + level := zapcore.WarnLevel + if cause == pb.CallCancel_UPSTREAM_ERROR { + level = zapcore.ErrorLevel + } + tc.logger.Log(level, "Call execution failed", append([]zap.Field{ zap.String("callId", callID), zap.String("method", method), zap.String("targetURL", breq.URL.String()), zap.Error(err), - ) - tc.sendCancel(sc.sendFn, callID, http.StatusBadGateway, err.Error()) + }, c.progress.fields(cause)...)...) + tc.sendCallCancel(sc.sendFn, callID, c.progress.report(http.StatusBadGateway, err.Error(), cause)) return } defer resp.Body.Close() + c.progress.enter(pb.CallCancel_STREAMING_BODY) tc.requestsTotal.WithLabelValues(method, fmt.Sprintf("%d", resp.StatusCode)).Inc() @@ -311,9 +339,11 @@ func (tc *tunnelClient) runCall(ctx context.Context, sc *streamCtx, callID strin break } if readErr != nil { - tc.logger.Warn("Response body read failed mid-stream", - zap.String("callId", callID), zap.Error(readErr)) - tc.sendCancel(sc.sendFn, callID, 0, fmt.Sprintf("upstream read failed: %v", readErr)) + cause := tc.stopCause(ctx, c) + tc.logger.Warn("Response body read failed mid-stream", append([]zap.Field{ + zap.String("callId", callID), zap.Error(readErr), + }, c.progress.fields(cause)...)...) + tc.sendCallCancel(sc.sendFn, callID, c.progress.report(0, fmt.Sprintf("upstream read failed: %v", readErr), cause)) return } } @@ -336,11 +366,17 @@ func (tc *tunnelClient) runCall(ctx context.Context, sc *streamCtx, callID strin tc.requestDuration.WithLabelValues(method).Observe(float64(time.Since(startTime).Milliseconds())) } -// sendCancel sends a CallCancel with an HTTP status hint. +// sendCancel sends a CallCancel with an HTTP status hint, for a call that +// never got as far as having a stage. func (tc *tunnelClient) sendCancel(sendFn sendFunc, callID string, code int32, reason string) { + tc.sendCallCancel(sendFn, callID, &pb.CallCancel{Code: code, Reason: reason}) +} + +// sendCallCancel sends cancel for callID. +func (tc *tunnelClient) sendCallCancel(sendFn sendFunc, callID string, cancel *pb.CallCancel) { _ = sendFn(&pb.ClientFrame{Msg: &pb.ClientFrame_Call{Call: &pb.CallFrame{ CallId: callID, - Body: &pb.CallFrame_Cancel{Cancel: &pb.CallCancel{Code: code, Reason: reason}}, + Body: &pb.CallFrame_Cancel{Cancel: cancel}, }}}) } diff --git a/agent/server/grpctunnel/call_progress.go b/agent/server/grpctunnel/call_progress.go new file mode 100644 index 00000000..2d173865 --- /dev/null +++ b/agent/server/grpctunnel/call_progress.go @@ -0,0 +1,68 @@ +package grpctunnel + +import ( + "context" + "errors" + "time" + + pb "github.com/cortexapps/axon/.generated/proto/github.com/cortexapps/axon/tunnelpb" + "go.uber.org/zap" +) + +// callProgress records where a call is, so that whatever stops it can say +// where it was and for how long. Only the call's own goroutine moves it on, +// and only that goroutine reads it. +type callProgress struct { + receivedAt time.Time + stage pb.CallCancel_Stage + stageAt time.Time + upstreamHost string +} + +func newCallProgress(now time.Time) callProgress { + return callProgress{receivedAt: now, stage: pb.CallCancel_QUEUED_FOR_CAPACITY, stageAt: now} +} + +func (p *callProgress) enter(stage pb.CallCancel_Stage) { + p.stage = stage + p.stageAt = time.Now() +} + +// report builds the CallCancel that tells the server where the call stopped +// and why. +func (p *callProgress) report(code int32, reason string, cause pb.CallCancel_Cause) *pb.CallCancel { + now := time.Now() + return &pb.CallCancel{ + Code: code, + Reason: reason, + Stage: p.stage, + Cause: cause, + StageElapsedMs: now.Sub(p.stageAt).Milliseconds(), + TotalElapsedMs: now.Sub(p.receivedAt).Milliseconds(), + UpstreamHost: p.upstreamHost, + } +} + +// fields renders the same account for the agent's own log line. +func (p *callProgress) fields(cause pb.CallCancel_Cause) []zap.Field { + now := time.Now() + return []zap.Field{ + zap.Stringer("stage", p.stage), + zap.Stringer("cause", cause), + zap.Int64("stageElapsedMs", now.Sub(p.stageAt).Milliseconds()), + zap.Int64("totalElapsedMs", now.Sub(p.receivedAt).Milliseconds()), + } +} + +// stopCause says why a call's context ended, or why its upstream failed. +func (tc *tunnelClient) stopCause(ctx context.Context, c *agentCall) pb.CallCancel_Cause { + switch { + case c.serverCancelled.Load(): + return pb.CallCancel_CANCELED_BY_SERVER + case tc.parentCtx != nil && tc.parentCtx.Err() != nil: + return pb.CallCancel_AGENT_SHUTDOWN + case errors.Is(ctx.Err(), context.DeadlineExceeded): + return pb.CallCancel_AGENT_TIMEOUT + } + return pb.CallCancel_UPSTREAM_ERROR +} diff --git a/agent/server/grpctunnel/call_report_test.go b/agent/server/grpctunnel/call_report_test.go new file mode 100644 index 00000000..27dc6635 --- /dev/null +++ b/agent/server/grpctunnel/call_report_test.go @@ -0,0 +1,141 @@ +package grpctunnel + +import ( + "errors" + "testing" + "time" + + pb "github.com/cortexapps/axon/.generated/proto/github.com/cortexapps/axon/tunnelpb" + "github.com/cortexapps/axon/config" + "github.com/cortexapps/axon/server/snykbroker/acceptfile" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +// Every CallCancel the agent sends says where the call was and why it +// stopped, so the server can tell a slow upstream from a call stuck in the +// agent. These pin that account for each way a call ends early. + +// onlyCancel returns the call's CallCancel, failing if it sent anything else. +func onlyCancel(t *testing.T, fc *frameCollector, callID string) *pb.CallCancel { + t.Helper() + frames := fc.byCall(callID) + require.Len(t, frames, 1, "want exactly one frame for %s", callID) + cancel := frames[0].GetCancel() + require.NotNil(t, cancel, "want CallCancel for %s, got %v", callID, frames[0]) + return cancel +} + +// The prod 60s case from the agent's side: the request went upstream, and +// the server cancelled while it waited for headers. The agent's answer names +// the stage, the server as the cause, how long the upstream had it, and the +// host. +func TestCallReport_ServerCancelWhileAwaitingUpstream(t *testing.T) { + tc, _ := newTestClient(t, config.AgentConfig{MaxInflightRequests: 4}, &fakeRegistration{serverURI: "x", tokens: []string{"x"}}) + tc.backend = &stubBackend{statusCode: 200, delay: 5 * time.Second, respectCtx: true} + + fc := newFrameCollector() + sc := newTestStreamCtx(fc.sendFn) + table := newTestCallTable(t) + + tc.startCall(sc, table, "c1", reqStart("GET", "/api/v4/groups/1/merge_requests", 0)) + tc.handleCallFrame(sc, table, &pb.CallFrame{CallId: "c1", Body: &pb.CallFrame_End{End: &pb.CallEnd{}}}) + time.Sleep(60 * time.Millisecond) + tc.handleCallFrame(sc, table, &pb.CallFrame{CallId: "c1", Body: &pb.CallFrame_Cancel{Cancel: &pb.CallCancel{Reason: "caller timeout"}}}) + fc.waitDone(t, 1) + + cancel := onlyCancel(t, fc, "c1") + assert.Equal(t, pb.CallCancel_AWAITING_UPSTREAM_HEADERS, cancel.Stage) + assert.Equal(t, pb.CallCancel_CANCELED_BY_SERVER, cancel.Cause) + assert.Equal(t, "stub.internal", cancel.UpstreamHost) + assert.GreaterOrEqual(t, cancel.StageElapsedMs, int64(50)) + assert.GreaterOrEqual(t, cancel.TotalElapsedMs, cancel.StageElapsedMs) + assert.Equal(t, int32(502), cancel.Code, "the status hint is unchanged") +} + +// The agent's own deadline for the call ran out while the upstream had it. +func TestCallReport_AgentDeadlineWhileAwaitingUpstream(t *testing.T) { + tc, _ := newTestClient(t, config.AgentConfig{MaxInflightRequests: 4}, &fakeRegistration{serverURI: "x", tokens: []string{"x"}}) + tc.backend = &stubBackend{statusCode: 200, delay: 5 * time.Second, respectCtx: true} + + fc := newFrameCollector() + sc := newTestStreamCtx(fc.sendFn) + table := newTestCallTable(t) + + tc.startCall(sc, table, "c1", reqStart("GET", "/slow", 50)) + tc.handleCallFrame(sc, table, &pb.CallFrame{CallId: "c1", Body: &pb.CallFrame_End{End: &pb.CallEnd{}}}) + fc.waitDone(t, 1) + + cancel := onlyCancel(t, fc, "c1") + assert.Equal(t, pb.CallCancel_AWAITING_UPSTREAM_HEADERS, cancel.Stage) + assert.Equal(t, pb.CallCancel_AGENT_TIMEOUT, cancel.Cause) +} + +// The upstream failed outright. +func TestCallReport_UpstreamError(t *testing.T) { + tc, _ := newTestClient(t, config.AgentConfig{MaxInflightRequests: 4}, &fakeRegistration{serverURI: "x", tokens: []string{"x"}}) + tc.backend = &stubBackend{err: errors.New("connection refused")} + + fc := newFrameCollector() + sc := newTestStreamCtx(fc.sendFn) + table := newTestCallTable(t) + + tc.startCall(sc, table, "c1", reqStart("GET", "/x", 0)) + tc.handleCallFrame(sc, table, &pb.CallFrame{CallId: "c1", Body: &pb.CallFrame_End{End: &pb.CallEnd{}}}) + fc.waitDone(t, 1) + + cancel := onlyCancel(t, fc, "c1") + assert.Equal(t, pb.CallCancel_AWAITING_UPSTREAM_HEADERS, cancel.Stage) + assert.Equal(t, pb.CallCancel_UPSTREAM_ERROR, cancel.Cause) + assert.Contains(t, cancel.Reason, "connection refused") +} + +// A call that queued for the agent's in-flight capacity and never got it says +// so: nothing went upstream. +func TestCallReport_QueuedForCapacity(t *testing.T) { + cfg := config.AgentConfig{MaxInflightRequests: 1, MaxRequestTimeout: 5 * time.Second} + tc, _ := newTestClient(t, cfg, &fakeRegistration{serverURI: "x", tokens: []string{"x"}}) + tc.backend = &stubBackend{statusCode: 200, delay: 2 * time.Second, respectCtx: true} + + fc := newFrameCollector() + sc := newTestStreamCtx(fc.sendFn) + table := newTestCallTable(t) + + tc.startCall(sc, table, "r1", reqStart("GET", "/", 0)) + tc.handleCallFrame(sc, table, &pb.CallFrame{CallId: "r1", Body: &pb.CallFrame_End{End: &pb.CallEnd{}}}) + time.Sleep(20 * time.Millisecond) + tc.startCall(sc, table, "r2", reqStart("GET", "/", 100)) + tc.handleCallFrame(sc, table, &pb.CallFrame{CallId: "r2", Body: &pb.CallFrame_End{End: &pb.CallEnd{}}}) + fc.waitDone(t, 1) + + cancel := onlyCancel(t, fc, "r2") + assert.Equal(t, pb.CallCancel_QUEUED_FOR_CAPACITY, cancel.Stage) + assert.Equal(t, pb.CallCancel_CAPACITY, cancel.Cause) + assert.Empty(t, cancel.UpstreamHost, "never routed") + assert.Equal(t, int32(503), cancel.Code) +} + +// No route for the request. +func TestCallReport_RouteRejected(t *testing.T) { + tc, _ := newTestClient(t, config.AgentConfig{MaxInflightRequests: 4}, &fakeRegistration{serverURI: "x", tokens: []string{"x"}}) + af, err := acceptfile.NewAcceptFile([]byte(`{ + "private": [{"method": "GET", "path": "/allowed", "origin": "http://stub.internal"}] + }`), config.AgentConfig{HttpServerPort: 8080, PluginDirs: []string{}}, zap.NewNop()) + require.NoError(t, err) + tc.router, err = NewRouter(af.Wrapper().PrivateRules(), zap.NewNop()) + require.NoError(t, err) + tc.backend = &stubBackend{statusCode: 200} + + fc := newFrameCollector() + sc := newTestStreamCtx(fc.sendFn) + table := newTestCallTable(t) + + tc.startCall(sc, table, "c1", reqStart("GET", "/denied", 0)) + tc.handleCallFrame(sc, table, &pb.CallFrame{CallId: "c1", Body: &pb.CallFrame_End{End: &pb.CallEnd{}}}) + fc.waitDone(t, 1) + + cancel := onlyCancel(t, fc, "c1") + assert.Equal(t, pb.CallCancel_ROUTING, cancel.Stage) + assert.Equal(t, pb.CallCancel_ROUTE_REJECTED, cancel.Cause) +} diff --git a/proto/tunnel/v2/tunnel.proto b/proto/tunnel/v2/tunnel.proto index ddda80e9..bd944f18 100644 --- a/proto/tunnel/v2/tunnel.proto +++ b/proto/tunnel/v2/tunnel.proto @@ -196,7 +196,55 @@ message CallEnd { // response (e.g. 404 no matching accept-file rule, 502 upstream // unreachable, 503 agent at in-flight cap). 0 means unspecified (the // adapter picks). +// +// The agent also says where the call was and why it stopped (fields 3-7), so +// a failed call explains itself on the server instead of only in the agent's +// own logs. That includes the agent's answer to a server-sent Cancel: the +// server waits briefly for it, which is how a server-side timeout learns +// whether the request ever left the agent and how long the upstream had it. +// All optional; a receiver that sees STAGE_UNSPECIFIED knows nothing more +// than an older peer would have told it. message CallCancel { string reason = 1; int32 code = 2; + + // Where the call was when it stopped. Agent-sent only. + Stage stage = 3; + enum Stage { + STAGE_UNSPECIFIED = 0; + // Waiting for the agent's own in-flight capacity; nothing sent upstream. + QUEUED_FOR_CAPACITY = 1; + // Matching the request to an upstream route. + ROUTING = 2; + // Request sent upstream; no response headers yet. + AWAITING_UPSTREAM_HEADERS = 3; + // Response headers received; relaying the body. + STREAMING_BODY = 4; + } + + // Why it stopped. Agent-sent only. + Cause cause = 4; + enum Cause { + CAUSE_UNSPECIFIED = 0; + // The server sent Cancel (its caller timed out or went away). + CANCELED_BY_SERVER = 1; + // The agent's own deadline for the call expired. + AGENT_TIMEOUT = 2; + // The upstream failed: connection error, reset, unreadable body. + UPSTREAM_ERROR = 3; + // No route matched, or the route refused the request. + ROUTE_REJECTED = 4; + // The agent was at its in-flight cap until the deadline. + CAPACITY = 5; + // The agent is shutting down or restarting. + AGENT_SHUTDOWN = 6; + } + + // How long the call had been in its final stage, and in the agent overall + // (from receipt of CallStart, queueing included). + int64 stage_elapsed_ms = 5; + int64 total_elapsed_ms = 6; + // Host the request was routed to, when it got that far. Host only: never a + // path or query, which can carry credentials. + string upstream_host = 7; } diff --git a/server/.generated/proto/tunnelpb/tunnel.pb.go b/server/.generated/proto/tunnelpb/tunnel.pb.go index 4b250c6d..f1c11ae0 100644 --- a/server/.generated/proto/tunnelpb/tunnel.pb.go +++ b/server/.generated/proto/tunnelpb/tunnel.pb.go @@ -73,6 +73,132 @@ func (CallStart_Kind) EnumDescriptor() ([]byte, []int) { return file_tunnel_proto_rawDescGZIP(), []int{6, 0} } +type CallCancel_Stage int32 + +const ( + CallCancel_STAGE_UNSPECIFIED CallCancel_Stage = 0 + // Waiting for the agent's own in-flight capacity; nothing sent upstream. + CallCancel_QUEUED_FOR_CAPACITY CallCancel_Stage = 1 + // Matching the request to an upstream route. + CallCancel_ROUTING CallCancel_Stage = 2 + // Request sent upstream; no response headers yet. + CallCancel_AWAITING_UPSTREAM_HEADERS CallCancel_Stage = 3 + // Response headers received; relaying the body. + CallCancel_STREAMING_BODY CallCancel_Stage = 4 +) + +// Enum value maps for CallCancel_Stage. +var ( + CallCancel_Stage_name = map[int32]string{ + 0: "STAGE_UNSPECIFIED", + 1: "QUEUED_FOR_CAPACITY", + 2: "ROUTING", + 3: "AWAITING_UPSTREAM_HEADERS", + 4: "STREAMING_BODY", + } + CallCancel_Stage_value = map[string]int32{ + "STAGE_UNSPECIFIED": 0, + "QUEUED_FOR_CAPACITY": 1, + "ROUTING": 2, + "AWAITING_UPSTREAM_HEADERS": 3, + "STREAMING_BODY": 4, + } +) + +func (x CallCancel_Stage) Enum() *CallCancel_Stage { + p := new(CallCancel_Stage) + *p = x + return p +} + +func (x CallCancel_Stage) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (CallCancel_Stage) Descriptor() protoreflect.EnumDescriptor { + return file_tunnel_proto_enumTypes[1].Descriptor() +} + +func (CallCancel_Stage) Type() protoreflect.EnumType { + return &file_tunnel_proto_enumTypes[1] +} + +func (x CallCancel_Stage) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use CallCancel_Stage.Descriptor instead. +func (CallCancel_Stage) EnumDescriptor() ([]byte, []int) { + return file_tunnel_proto_rawDescGZIP(), []int{9, 0} +} + +type CallCancel_Cause int32 + +const ( + CallCancel_CAUSE_UNSPECIFIED CallCancel_Cause = 0 + // The server sent Cancel (its caller timed out or went away). + CallCancel_CANCELED_BY_SERVER CallCancel_Cause = 1 + // The agent's own deadline for the call expired. + CallCancel_AGENT_TIMEOUT CallCancel_Cause = 2 + // The upstream failed: connection error, reset, unreadable body. + CallCancel_UPSTREAM_ERROR CallCancel_Cause = 3 + // No route matched, or the route refused the request. + CallCancel_ROUTE_REJECTED CallCancel_Cause = 4 + // The agent was at its in-flight cap until the deadline. + CallCancel_CAPACITY CallCancel_Cause = 5 + // The agent is shutting down or restarting. + CallCancel_AGENT_SHUTDOWN CallCancel_Cause = 6 +) + +// Enum value maps for CallCancel_Cause. +var ( + CallCancel_Cause_name = map[int32]string{ + 0: "CAUSE_UNSPECIFIED", + 1: "CANCELED_BY_SERVER", + 2: "AGENT_TIMEOUT", + 3: "UPSTREAM_ERROR", + 4: "ROUTE_REJECTED", + 5: "CAPACITY", + 6: "AGENT_SHUTDOWN", + } + CallCancel_Cause_value = map[string]int32{ + "CAUSE_UNSPECIFIED": 0, + "CANCELED_BY_SERVER": 1, + "AGENT_TIMEOUT": 2, + "UPSTREAM_ERROR": 3, + "ROUTE_REJECTED": 4, + "CAPACITY": 5, + "AGENT_SHUTDOWN": 6, + } +) + +func (x CallCancel_Cause) Enum() *CallCancel_Cause { + p := new(CallCancel_Cause) + *p = x + return p +} + +func (x CallCancel_Cause) String() string { + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) +} + +func (CallCancel_Cause) Descriptor() protoreflect.EnumDescriptor { + return file_tunnel_proto_enumTypes[2].Descriptor() +} + +func (CallCancel_Cause) Type() protoreflect.EnumType { + return &file_tunnel_proto_enumTypes[2] +} + +func (x CallCancel_Cause) Number() protoreflect.EnumNumber { + return protoreflect.EnumNumber(x) +} + +// Deprecated: Use CallCancel_Cause.Descriptor instead. +func (CallCancel_Cause) EnumDescriptor() ([]byte, []int) { + return file_tunnel_proto_rawDescGZIP(), []int{9, 1} +} + // Client → Server envelope. type ClientFrame struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -872,10 +998,29 @@ func (x *CallEnd) GetTrailers() map[string]string { // response (e.g. 404 no matching accept-file rule, 502 upstream // unreachable, 503 agent at in-flight cap). 0 means unspecified (the // adapter picks). +// +// The agent also says where the call was and why it stopped (fields 3-7), so +// a failed call explains itself on the server instead of only in the agent's +// own logs. That includes the agent's answer to a server-sent Cancel: the +// server waits briefly for it, which is how a server-side timeout learns +// whether the request ever left the agent and how long the upstream had it. +// All optional; a receiver that sees STAGE_UNSPECIFIED knows nothing more +// than an older peer would have told it. type CallCancel struct { - state protoimpl.MessageState `protogen:"open.v1"` - Reason string `protobuf:"bytes,1,opt,name=reason,proto3" json:"reason,omitempty"` - Code int32 `protobuf:"varint,2,opt,name=code,proto3" json:"code,omitempty"` + state protoimpl.MessageState `protogen:"open.v1"` + Reason string `protobuf:"bytes,1,opt,name=reason,proto3" json:"reason,omitempty"` + Code int32 `protobuf:"varint,2,opt,name=code,proto3" json:"code,omitempty"` + // Where the call was when it stopped. Agent-sent only. + Stage CallCancel_Stage `protobuf:"varint,3,opt,name=stage,proto3,enum=cortex.axon.tunnel.v2.CallCancel_Stage" json:"stage,omitempty"` + // Why it stopped. Agent-sent only. + Cause CallCancel_Cause `protobuf:"varint,4,opt,name=cause,proto3,enum=cortex.axon.tunnel.v2.CallCancel_Cause" json:"cause,omitempty"` + // How long the call had been in its final stage, and in the agent overall + // (from receipt of CallStart, queueing included). + StageElapsedMs int64 `protobuf:"varint,5,opt,name=stage_elapsed_ms,json=stageElapsedMs,proto3" json:"stage_elapsed_ms,omitempty"` + TotalElapsedMs int64 `protobuf:"varint,6,opt,name=total_elapsed_ms,json=totalElapsedMs,proto3" json:"total_elapsed_ms,omitempty"` + // Host the request was routed to, when it got that far. Host only: never a + // path or query, which can carry credentials. + UpstreamHost string `protobuf:"bytes,7,opt,name=upstream_host,json=upstreamHost,proto3" json:"upstream_host,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -924,6 +1069,41 @@ func (x *CallCancel) GetCode() int32 { return 0 } +func (x *CallCancel) GetStage() CallCancel_Stage { + if x != nil { + return x.Stage + } + return CallCancel_STAGE_UNSPECIFIED +} + +func (x *CallCancel) GetCause() CallCancel_Cause { + if x != nil { + return x.Cause + } + return CallCancel_CAUSE_UNSPECIFIED +} + +func (x *CallCancel) GetStageElapsedMs() int64 { + if x != nil { + return x.StageElapsedMs + } + return 0 +} + +func (x *CallCancel) GetTotalElapsedMs() int64 { + if x != nil { + return x.TotalElapsedMs + } + return 0 +} + +func (x *CallCancel) GetUpstreamHost() string { + if x != nil { + return x.UpstreamHost + } + return "" +} + var File_tunnel_proto protoreflect.FileDescriptor const file_tunnel_proto_rawDesc = "" + @@ -993,11 +1173,30 @@ const file_tunnel_proto_rawDesc = "" + "\btrailers\x18\x01 \x03(\v2,.cortex.axon.tunnel.v2.CallEnd.TrailersEntryR\btrailers\x1a;\n" + "\rTrailersEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + - "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"8\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xbe\x04\n" + "\n" + "CallCancel\x12\x16\n" + "\x06reason\x18\x01 \x01(\tR\x06reason\x12\x12\n" + - "\x04code\x18\x02 \x01(\x05R\x04code2e\n" + + "\x04code\x18\x02 \x01(\x05R\x04code\x12=\n" + + "\x05stage\x18\x03 \x01(\x0e2'.cortex.axon.tunnel.v2.CallCancel.StageR\x05stage\x12=\n" + + "\x05cause\x18\x04 \x01(\x0e2'.cortex.axon.tunnel.v2.CallCancel.CauseR\x05cause\x12(\n" + + "\x10stage_elapsed_ms\x18\x05 \x01(\x03R\x0estageElapsedMs\x12(\n" + + "\x10total_elapsed_ms\x18\x06 \x01(\x03R\x0etotalElapsedMs\x12#\n" + + "\rupstream_host\x18\a \x01(\tR\fupstreamHost\"w\n" + + "\x05Stage\x12\x15\n" + + "\x11STAGE_UNSPECIFIED\x10\x00\x12\x17\n" + + "\x13QUEUED_FOR_CAPACITY\x10\x01\x12\v\n" + + "\aROUTING\x10\x02\x12\x1d\n" + + "\x19AWAITING_UPSTREAM_HEADERS\x10\x03\x12\x12\n" + + "\x0eSTREAMING_BODY\x10\x04\"\x93\x01\n" + + "\x05Cause\x12\x15\n" + + "\x11CAUSE_UNSPECIFIED\x10\x00\x12\x16\n" + + "\x12CANCELED_BY_SERVER\x10\x01\x12\x11\n" + + "\rAGENT_TIMEOUT\x10\x02\x12\x12\n" + + "\x0eUPSTREAM_ERROR\x10\x03\x12\x12\n" + + "\x0eROUTE_REJECTED\x10\x04\x12\f\n" + + "\bCAPACITY\x10\x05\x12\x12\n" + + "\x0eAGENT_SHUTDOWN\x10\x062e\n" + "\rTunnelService\x12T\n" + "\x06Tunnel\x12\".cortex.axon.tunnel.v2.ClientFrame\x1a\".cortex.axon.tunnel.v2.ServerFrame(\x010\x01B,Z*github.com/cortexapps/axon-server/tunnelpbb\x06proto3" @@ -1013,48 +1212,52 @@ func file_tunnel_proto_rawDescGZIP() []byte { return file_tunnel_proto_rawDescData } -var file_tunnel_proto_enumTypes = make([]protoimpl.EnumInfo, 1) +var file_tunnel_proto_enumTypes = make([]protoimpl.EnumInfo, 3) var file_tunnel_proto_msgTypes = make([]protoimpl.MessageInfo, 14) var file_tunnel_proto_goTypes = []any{ - (CallStart_Kind)(0), // 0: cortex.axon.tunnel.v2.CallStart.Kind - (*ClientFrame)(nil), // 1: cortex.axon.tunnel.v2.ClientFrame - (*ServerFrame)(nil), // 2: cortex.axon.tunnel.v2.ServerFrame - (*ClientHello)(nil), // 3: cortex.axon.tunnel.v2.ClientHello - (*ServerHello)(nil), // 4: cortex.axon.tunnel.v2.ServerHello - (*Heartbeat)(nil), // 5: cortex.axon.tunnel.v2.Heartbeat - (*CallFrame)(nil), // 6: cortex.axon.tunnel.v2.CallFrame - (*CallStart)(nil), // 7: cortex.axon.tunnel.v2.CallStart - (*CallData)(nil), // 8: cortex.axon.tunnel.v2.CallData - (*CallEnd)(nil), // 9: cortex.axon.tunnel.v2.CallEnd - (*CallCancel)(nil), // 10: cortex.axon.tunnel.v2.CallCancel - nil, // 11: cortex.axon.tunnel.v2.ClientHello.MetadataEntry - nil, // 12: cortex.axon.tunnel.v2.CallStart.PseudoHeadersEntry - nil, // 13: cortex.axon.tunnel.v2.CallStart.HeadersEntry - nil, // 14: cortex.axon.tunnel.v2.CallEnd.TrailersEntry + (CallStart_Kind)(0), // 0: cortex.axon.tunnel.v2.CallStart.Kind + (CallCancel_Stage)(0), // 1: cortex.axon.tunnel.v2.CallCancel.Stage + (CallCancel_Cause)(0), // 2: cortex.axon.tunnel.v2.CallCancel.Cause + (*ClientFrame)(nil), // 3: cortex.axon.tunnel.v2.ClientFrame + (*ServerFrame)(nil), // 4: cortex.axon.tunnel.v2.ServerFrame + (*ClientHello)(nil), // 5: cortex.axon.tunnel.v2.ClientHello + (*ServerHello)(nil), // 6: cortex.axon.tunnel.v2.ServerHello + (*Heartbeat)(nil), // 7: cortex.axon.tunnel.v2.Heartbeat + (*CallFrame)(nil), // 8: cortex.axon.tunnel.v2.CallFrame + (*CallStart)(nil), // 9: cortex.axon.tunnel.v2.CallStart + (*CallData)(nil), // 10: cortex.axon.tunnel.v2.CallData + (*CallEnd)(nil), // 11: cortex.axon.tunnel.v2.CallEnd + (*CallCancel)(nil), // 12: cortex.axon.tunnel.v2.CallCancel + nil, // 13: cortex.axon.tunnel.v2.ClientHello.MetadataEntry + nil, // 14: cortex.axon.tunnel.v2.CallStart.PseudoHeadersEntry + nil, // 15: cortex.axon.tunnel.v2.CallStart.HeadersEntry + nil, // 16: cortex.axon.tunnel.v2.CallEnd.TrailersEntry } var file_tunnel_proto_depIdxs = []int32{ - 3, // 0: cortex.axon.tunnel.v2.ClientFrame.hello:type_name -> cortex.axon.tunnel.v2.ClientHello - 5, // 1: cortex.axon.tunnel.v2.ClientFrame.heartbeat:type_name -> cortex.axon.tunnel.v2.Heartbeat - 6, // 2: cortex.axon.tunnel.v2.ClientFrame.call:type_name -> cortex.axon.tunnel.v2.CallFrame - 4, // 3: cortex.axon.tunnel.v2.ServerFrame.hello:type_name -> cortex.axon.tunnel.v2.ServerHello - 5, // 4: cortex.axon.tunnel.v2.ServerFrame.heartbeat:type_name -> cortex.axon.tunnel.v2.Heartbeat - 6, // 5: cortex.axon.tunnel.v2.ServerFrame.call:type_name -> cortex.axon.tunnel.v2.CallFrame - 11, // 6: cortex.axon.tunnel.v2.ClientHello.metadata:type_name -> cortex.axon.tunnel.v2.ClientHello.MetadataEntry - 7, // 7: cortex.axon.tunnel.v2.CallFrame.start:type_name -> cortex.axon.tunnel.v2.CallStart - 8, // 8: cortex.axon.tunnel.v2.CallFrame.data:type_name -> cortex.axon.tunnel.v2.CallData - 9, // 9: cortex.axon.tunnel.v2.CallFrame.end:type_name -> cortex.axon.tunnel.v2.CallEnd - 10, // 10: cortex.axon.tunnel.v2.CallFrame.cancel:type_name -> cortex.axon.tunnel.v2.CallCancel - 12, // 11: cortex.axon.tunnel.v2.CallStart.pseudo_headers:type_name -> cortex.axon.tunnel.v2.CallStart.PseudoHeadersEntry - 13, // 12: cortex.axon.tunnel.v2.CallStart.headers:type_name -> cortex.axon.tunnel.v2.CallStart.HeadersEntry + 5, // 0: cortex.axon.tunnel.v2.ClientFrame.hello:type_name -> cortex.axon.tunnel.v2.ClientHello + 7, // 1: cortex.axon.tunnel.v2.ClientFrame.heartbeat:type_name -> cortex.axon.tunnel.v2.Heartbeat + 8, // 2: cortex.axon.tunnel.v2.ClientFrame.call:type_name -> cortex.axon.tunnel.v2.CallFrame + 6, // 3: cortex.axon.tunnel.v2.ServerFrame.hello:type_name -> cortex.axon.tunnel.v2.ServerHello + 7, // 4: cortex.axon.tunnel.v2.ServerFrame.heartbeat:type_name -> cortex.axon.tunnel.v2.Heartbeat + 8, // 5: cortex.axon.tunnel.v2.ServerFrame.call:type_name -> cortex.axon.tunnel.v2.CallFrame + 13, // 6: cortex.axon.tunnel.v2.ClientHello.metadata:type_name -> cortex.axon.tunnel.v2.ClientHello.MetadataEntry + 9, // 7: cortex.axon.tunnel.v2.CallFrame.start:type_name -> cortex.axon.tunnel.v2.CallStart + 10, // 8: cortex.axon.tunnel.v2.CallFrame.data:type_name -> cortex.axon.tunnel.v2.CallData + 11, // 9: cortex.axon.tunnel.v2.CallFrame.end:type_name -> cortex.axon.tunnel.v2.CallEnd + 12, // 10: cortex.axon.tunnel.v2.CallFrame.cancel:type_name -> cortex.axon.tunnel.v2.CallCancel + 14, // 11: cortex.axon.tunnel.v2.CallStart.pseudo_headers:type_name -> cortex.axon.tunnel.v2.CallStart.PseudoHeadersEntry + 15, // 12: cortex.axon.tunnel.v2.CallStart.headers:type_name -> cortex.axon.tunnel.v2.CallStart.HeadersEntry 0, // 13: cortex.axon.tunnel.v2.CallStart.kind:type_name -> cortex.axon.tunnel.v2.CallStart.Kind - 14, // 14: cortex.axon.tunnel.v2.CallEnd.trailers:type_name -> cortex.axon.tunnel.v2.CallEnd.TrailersEntry - 1, // 15: cortex.axon.tunnel.v2.TunnelService.Tunnel:input_type -> cortex.axon.tunnel.v2.ClientFrame - 2, // 16: cortex.axon.tunnel.v2.TunnelService.Tunnel:output_type -> cortex.axon.tunnel.v2.ServerFrame - 16, // [16:17] is the sub-list for method output_type - 15, // [15:16] is the sub-list for method input_type - 15, // [15:15] is the sub-list for extension type_name - 15, // [15:15] is the sub-list for extension extendee - 0, // [0:15] is the sub-list for field type_name + 16, // 14: cortex.axon.tunnel.v2.CallEnd.trailers:type_name -> cortex.axon.tunnel.v2.CallEnd.TrailersEntry + 1, // 15: cortex.axon.tunnel.v2.CallCancel.stage:type_name -> cortex.axon.tunnel.v2.CallCancel.Stage + 2, // 16: cortex.axon.tunnel.v2.CallCancel.cause:type_name -> cortex.axon.tunnel.v2.CallCancel.Cause + 3, // 17: cortex.axon.tunnel.v2.TunnelService.Tunnel:input_type -> cortex.axon.tunnel.v2.ClientFrame + 4, // 18: cortex.axon.tunnel.v2.TunnelService.Tunnel:output_type -> cortex.axon.tunnel.v2.ServerFrame + 18, // [18:19] is the sub-list for method output_type + 17, // [17:18] is the sub-list for method input_type + 17, // [17:17] is the sub-list for extension type_name + 17, // [17:17] is the sub-list for extension extendee + 0, // [0:17] is the sub-list for field type_name } func init() { file_tunnel_proto_init() } @@ -1083,7 +1286,7 @@ func file_tunnel_proto_init() { File: protoimpl.DescBuilder{ GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_tunnel_proto_rawDesc), len(file_tunnel_proto_rawDesc)), - NumEnums: 1, + NumEnums: 3, NumMessages: 14, NumExtensions: 0, NumServices: 1, diff --git a/server/adapters/dispatch_failure_cause_test.go b/server/adapters/dispatch_failure_cause_test.go new file mode 100644 index 00000000..011472da --- /dev/null +++ b/server/adapters/dispatch_failure_cause_test.go @@ -0,0 +1,214 @@ +package adapters + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + pb "github.com/cortexapps/axon-server/.generated/proto/tunnelpb" + "github.com/cortexapps/axon-server/broker" + "github.com/cortexapps/axon-server/config" + "github.com/cortexapps/axon-server/dispatch" + "github.com/cortexapps/axon-server/metrics" + "github.com/cortexapps/axon-server/tunnel" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" + "go.uber.org/zap/zapcore" + "go.uber.org/zap/zaptest/observer" +) + +// failureHarness is an HTTP adapter over a real dispatcher and registry, one +// agent stream whose frames the test reads, and an observed logger. +type failureHarness struct { + h *HttpAdapter + d *dispatch.Dispatcher + logs *observer.ObservedLogs + frames chan *pb.CallFrame +} + +const failureToken = "tok" + +func newFailureHarness(t *testing.T, grace time.Duration) *failureHarness { + t.Helper() + core, logs := observer.New(zapcore.DebugLevel) + logger := zap.New(core) + registry := tunnel.NewClientRegistry(logger) + cfg := config.Config{DispatchTimeout: 10 * time.Second, AgentReportGrace: grace} + m := metrics.New("test") + d := dispatch.NewDispatcher(cfg, registry, m, logger) + fh := &failureHarness{ + h: NewHttpAdapter(cfg, registry, d, m, logger), + d: d, + logs: logs, + frames: make(chan *pb.CallFrame, 64), + } + mustRegister(t, registry, broker.NewToken(failureToken), tunnel.ClientIdentity{InstanceID: "bitbucket-relay-1", Integration: "bitbucket"}, + &tunnel.StreamHandle{StreamID: "s1", Send: func(f *pb.ServerFrame) error { + if c := f.GetCall(); c != nil { + fh.frames <- c + } + return nil + }}) + return fh +} + +// next returns the next frame of the given type the agent was sent. +func (fh *failureHarness) next(t *testing.T, want func(*pb.CallFrame) bool) *pb.CallFrame { + t.Helper() + deadline := time.After(5 * time.Second) + for { + select { + case f := <-fh.frames: + if want(f) { + return f + } + case <-deadline: + t.Fatal("timed out waiting for a frame") + } + } +} + +func isStart(f *pb.CallFrame) bool { return f.GetStart() != nil } +func isCancel(f *pb.CallFrame) bool { return f.GetCancel() != nil } + +// serve runs a GET through the adapter with ctx as the caller's context. +func (fh *failureHarness) serve(ctx context.Context, path string) *httptest.ResponseRecorder { + req := httptest.NewRequest(http.MethodGet, "/broker/"+failureToken+path, nil).WithContext(ctx) + w := httptest.NewRecorder() + fh.h.ServeHTTP(w, req) + return w +} + +// failureLog waits for the "Dispatch failed" line for path. +func (fh *failureHarness) failureLog(t *testing.T, path string) observer.LoggedEntry { + t.Helper() + var entry observer.LoggedEntry + require.Eventually(t, func() bool { + for _, e := range fh.logs.FilterMessage("Dispatch failed").All() { + if e.ContextMap()["path"] == path { + entry = e + return true + } + } + return false + }, 5*time.Second, 5*time.Millisecond, "no Dispatch failed line for %s", path) + return entry +} + +// The prod 2s probe failure, end to end: a slow call holds the token's only +// stream, the dispatcher's probe gives up waiting, and both the response and +// the log say the streams were exhausted and by what. +func TestDispatchFailure_ProbeStarvedNamesTheHolder(t *testing.T) { + fh := newFailureHarness(t, time.Second) + + holderDone := make(chan struct{}) + go func() { + defer close(holderDone) + fh.serve(context.Background(), "/2.0/repositories/acme") + }() + holder := fh.next(t, isStart) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + w := fh.serve(ctx, "/__axon/info") + + assert.Equal(t, http.StatusGatewayTimeout, w.Code) + assert.Equal(t, "streams_exhausted", w.Header().Get("X-Axon-Dispatch-Failure")) + assert.Equal(t, "acquire", w.Header().Get("X-Axon-Dispatch-Phase")) + + entry := fh.failureLog(t, "/__axon/info") + assert.Equal(t, zapcore.WarnLevel, entry.Level) + f := entry.ContextMap() + assert.Equal(t, "streams_exhausted", f["cause"]) + assert.Equal(t, "acquire", f["phase"]) + assert.EqualValues(t, 1, f["tokenStreams"]) + assert.EqualValues(t, 1, f["tokenBusy"]) + assert.EqualValues(t, 1, f["tokenWaiters"]) + assert.Equal(t, []any{"bitbucket-relay-1 busy=1/1"}, f["tokenInstances"]) + holders, _ := f["holders"].([]any) + require.Len(t, holders, 1) + assert.Contains(t, holders[0], "GET /2.0/repositories/acme") + assert.Contains(t, holders[0], holder.CallId) + assert.NotContains(t, f, "agentOutcome", "nothing reached the agent to ask") + + // Let the holder finish. + fh.d.HandleFrame("s1", &pb.CallFrame{CallId: holder.CallId, Body: &pb.CallFrame_Start{Start: &pb.CallStart{PseudoHeaders: map[string]string{":status": "200"}}}}) + fh.d.HandleFrame("s1", &pb.CallFrame{CallId: holder.CallId, Body: &pb.CallFrame_End{End: &pb.CallEnd{}}}) + <-holderDone +} + +// The prod 60s failure, end to end: the call reached the agent and the caller +// gave up while the agent waited on the upstream. The response goes out at +// once; the log line waits for the agent's account and names the upstream. +func TestDispatchFailure_SlowUpstreamIsNamedByTheAgent(t *testing.T) { + fh := newFailureHarness(t, 5*time.Second) + + // Play the agent: answer the server's cancel the way a current agent does. + go func() { + c := fh.next(t, isCancel) + fh.d.HandleFrame("s1", &pb.CallFrame{CallId: c.CallId, Body: &pb.CallFrame_Cancel{Cancel: &pb.CallCancel{ + Code: 502, Reason: "context canceled", + Stage: pb.CallCancel_AWAITING_UPSTREAM_HEADERS, Cause: pb.CallCancel_CANCELED_BY_SERVER, + StageElapsedMs: 48, TotalElapsedMs: 50, UpstreamHost: "gitlab.com", + }}}) + }() + + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(50*time.Millisecond, cancel) + w := fh.serve(ctx, "/api/v4/groups/12345/merge_requests") + + assert.Equal(t, 499, w.Code) + assert.Equal(t, "awaiting_agent", w.Header().Get("X-Axon-Dispatch-Failure"), + "the response cannot wait for the agent; the log can") + assert.Equal(t, "awaiting_response", w.Header().Get("X-Axon-Dispatch-Phase")) + + entry := fh.failureLog(t, "/api/v4/groups/12345/merge_requests") + assert.Equal(t, zapcore.WarnLevel, entry.Level, "the caller gave up because the upstream was slow") + f := entry.ContextMap() + assert.Equal(t, "upstream_slow", f["cause"]) + assert.Equal(t, "caller_canceled", f["kind"]) + assert.Equal(t, "reported", f["agentOutcome"]) + assert.Equal(t, "AWAITING_UPSTREAM_HEADERS", f["agentStage"]) + assert.Equal(t, "CANCELED_BY_SERVER", f["agentCause"]) + assert.Equal(t, 48*time.Millisecond, f["agentStageElapsed"]) + assert.Equal(t, "gitlab.com", f["upstreamHost"]) + assert.NotEmpty(t, f["callId"]) +} + +// An agent that never answers the cancel is itself the finding. +func TestDispatchFailure_SilentAgentIsUnresponsive(t *testing.T) { + fh := newFailureHarness(t, 50*time.Millisecond) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + w := fh.serve(ctx, "/slow") + assert.Equal(t, http.StatusGatewayTimeout, w.Code) + + f := fh.failureLog(t, "/slow").ContextMap() + assert.Equal(t, "agent_unresponsive", f["cause"]) + assert.Equal(t, "silent", f["agentOutcome"]) +} + +// An agent at its in-flight cap says so, and the response carries it at once: +// the agent's cancel is the dispatch error itself. +func TestDispatchFailure_AgentAtCapacity(t *testing.T) { + fh := newFailureHarness(t, time.Second) + go func() { + c := fh.next(t, isStart) + fh.d.HandleFrame("s1", &pb.CallFrame{CallId: c.CallId, Body: &pb.CallFrame_Cancel{Cancel: &pb.CallCancel{ + Code: 503, Reason: "timed out waiting for in-flight capacity", + Stage: pb.CallCancel_QUEUED_FOR_CAPACITY, Cause: pb.CallCancel_CAPACITY, + }}}) + }() + + w := fh.serve(context.Background(), "/busy") + assert.Equal(t, http.StatusServiceUnavailable, w.Code) + assert.Equal(t, "agent_at_capacity", w.Header().Get("X-Axon-Dispatch-Failure")) + + f := fh.failureLog(t, "/busy").ContextMap() + assert.Equal(t, "agent_at_capacity", f["cause"]) + assert.Equal(t, "QUEUED_FOR_CAPACITY", f["agentStage"]) +} diff --git a/server/adapters/http.go b/server/adapters/http.go index f8c15b82..3fbf5fc6 100644 --- a/server/adapters/http.go +++ b/server/adapters/http.go @@ -327,8 +327,14 @@ func agentFields(id tunnel.ClientIdentity) []zap.Field { } // writeDispatchError maps Dispatch errors to HTTP responses and logs the -// failure once, with its kind and everything that names the call: fields -// carries the request's own (token hash, method, path, elapsed). +// failure once, with its kind, its cause and everything that names the call: +// fields carries the request's own (token hash, method, path, elapsed). +// +// The response goes out first, carrying the cause as far as it is known then +// (x-axon-dispatch-failure). When the call had reached the agent and the +// server gave up on it, the log line waits for the agent's own account -- +// bounded by the dispatcher's grace period, off the request path -- so it can +// say whether the upstream was slow or the agent never answered. func (h *HttpAdapter) writeDispatchError(w http.ResponseWriter, err error, token broker.Token, fields ...zap.Field) { h.metrics.DispatchErrors.Inc(1) @@ -338,32 +344,38 @@ func (h *HttpAdapter) writeDispatchError(w http.ResponseWriter, err error, token var dispErr *dispatch.DispatchError if errors.As(err, &dispErr) { fields = append(fields, zap.String("kind", string(dispErr.Kind))) + if dispErr.Phase != "" { + fields = append(fields, zap.String("phase", string(dispErr.Phase))) + } if dispErr.CallID != "" { fields = append(fields, zap.String("callId", dispErr.CallID), zap.String("streamId", dispErr.StreamID), + zap.Duration("callElapsed", dispErr.Elapsed), ) } fields = append(fields, agentFields(dispErr.Agent)...) if dispErr.AcquireWait > 0 { fields = append(fields, zap.Duration("acquireWait", dispErr.AcquireWait)) } - } - // level is Warn unless the failure is routine; logf records the outcome. - logf := func(level zapcore.Level, status int) { - h.logger.Log(level, "Dispatch failed", - append(fields, zap.Int("status", status), zap.Error(err))...) + if !errors.Is(err, dispatch.ErrNoTunnel) { + fields = append(fields, streamStateFields(dispErr.Streams)...) + } } var cancelErr *dispatch.CancelError + errors.As(err, &cancelErr) + cause := failureCause(err, dispErr, cancelErr, dispatch.AgentNotAsked) + + var status int switch { - case errors.As(err, &cancelErr): - status := int(cancelErr.Code) + case cancelErr != nil: + status = int(cancelErr.Code) if status < 100 || status > 599 { status = http.StatusBadGateway } fields = append(fields, zap.Int32("agentCode", cancelErr.Code)) - logf(zapcore.WarnLevel, status) + setFailureHeaders(w, cause, dispErr) writeFailedDispatch(w, status, cancelErr.Reason) case errors.Is(err, dispatch.ErrNoTunnel): @@ -376,23 +388,159 @@ func (h *HttpAdapter) writeDispatchError(w http.ResponseWriter, err error, token // // Routine rather than exceptional: it is how a caller gets moved off // an instance that has lost its streams, so it is logged at Info - // rather than Warn. - logf(zapcore.InfoLevel, http.StatusNotFound) + // rather than Warn. No failure headers: the dispatcher matches this + // response byte for byte. + status = http.StatusNotFound writeBrokerJSON(w, http.StatusNotFound, brokerOKFalse) case errors.Is(err, context.DeadlineExceeded): - logf(zapcore.WarnLevel, http.StatusGatewayTimeout) + status = http.StatusGatewayTimeout + setFailureHeaders(w, cause, dispErr) http.Error(w, "gateway timeout", http.StatusGatewayTimeout) case errors.Is(err, context.Canceled): // Caller went away; nothing useful to write. - logf(zapcore.InfoLevel, 499) + status = 499 + setFailureHeaders(w, cause, dispErr) http.Error(w, "client closed request", 499) default: - logf(zapcore.WarnLevel, http.StatusBadGateway) + status = http.StatusBadGateway + setFailureHeaders(w, cause, dispErr) http.Error(w, "tunnel dispatch failed", http.StatusBadGateway) } + + logFailure := func(report *dispatch.CancelError, outcome dispatch.AgentOutcome) { + cause := failureCause(err, dispErr, report, outcome) + logFields := append(fields, zap.String("cause", string(cause)), zap.Int("status", status)) + if dispErr != nil && dispErr.CallID != "" { + logFields = append(logFields, agentReportFields(report, outcome)...) + } + logFields = append(logFields, zap.Error(err)) + h.logger.Log(failureLevel(err, cause), "Dispatch failed", logFields...) + h.recordFailure(token, dispErr, cause) + } + + if dispErr != nil && dispErr.CallID != "" && cancelErr == nil { + // The server gave up on a call the agent had; ask the agent why, + // without holding the caller's connection for the answer. + go func() { logFailure(dispErr.AwaitAgentReport()) }() + return + } + // Nothing to wait for: either no call reached the agent, or the agent's + // account is the error itself. + if cancelErr != nil { + logFailure(cancelErr, dispatch.AgentReported) + return + } + logFailure(nil, dispatch.AgentNotAsked) +} + +// failureCause classifies err; see dispatch.FailureCause. +func failureCause(err error, d *dispatch.DispatchError, report *dispatch.CancelError, outcome dispatch.AgentOutcome) dispatch.FailureCause { + if d == nil { + if errors.Is(err, dispatch.ErrNoTunnel) { + return dispatch.CauseNoTunnel + } + return dispatch.CauseUnknown + } + return d.Cause(report, outcome) +} + +// failureLevel is Warn unless the failure is routine. A caller walking away +// is routine only when nothing on our side made it wait. +func failureLevel(err error, cause dispatch.FailureCause) zapcore.Level { + switch { + case errors.Is(err, dispatch.ErrNoTunnel): + return zapcore.InfoLevel + case errors.Is(err, context.Canceled) && !cause.MadeCallerWait(): + return zapcore.InfoLevel + } + return zapcore.WarnLevel +} + +// setFailureHeaders tells the caller why its dispatch failed, as far as the +// server knows when it responds. +func setFailureHeaders(w http.ResponseWriter, cause dispatch.FailureCause, d *dispatch.DispatchError) { + w.Header().Set("X-Axon-Dispatch-Failure", string(cause)) + if d != nil && d.Phase != "" { + w.Header().Set("X-Axon-Dispatch-Phase", string(d.Phase)) + } +} + +// recordFailure counts the failure by cause, tagged with the agent's +// identity where one is known. +func (h *HttpAdapter) recordFailure(token broker.Token, d *dispatch.DispatchError, cause dispatch.FailureCause) { + id := tunnel.ClientIdentity{} + if d != nil && d.Agent != (tunnel.ClientIdentity{}) { + id = d.Agent + } else if h.registry != nil { + if found := h.registry.GetIdentity(token); found != nil { + id = *found + } + } + h.metrics.DispatchFailure(id.TenantID, id.Integration, id.Alias, string(cause)) +} + +// streamStateFields renders the token's streams at the failure: how many, +// how many busy, how many dispatches waiting, per agent instance, and the +// oldest calls holding them. +func streamStateFields(s dispatch.StreamState) []zap.Field { + fields := []zap.Field{ + zap.Int("tokenStreams", s.Streams), + zap.Int("tokenBusy", s.Busy), + zap.Int("tokenWaiters", s.Waiters), + } + if len(s.Instances) > 0 { + instances := make([]string, 0, len(s.Instances)) + for _, inst := range s.Instances { + instances = append(instances, fmt.Sprintf("%s busy=%d/%d", inst.InstanceID, inst.Busy, inst.Streams)) + } + fields = append(fields, zap.Strings("tokenInstances", instances)) + } + if len(s.Holders) > 0 { + holders := make([]string, 0, len(s.Holders)) + for _, c := range s.Holders { + line := fmt.Sprintf("%s %s %s on %s callId=%s", c.Method, c.Path, + c.Age.Round(time.Millisecond), c.InstanceID, c.CallID) + if c.Cancelled { + line += " (cancelled, awaiting agent)" + } + holders = append(holders, line) + } + fields = append(fields, + zap.Duration("oldestHolderAge", s.Holders[0].Age), + zap.Strings("holders", holders), + ) + } + return fields +} + +// agentReportFields renders the agent's account of a failed call. +func agentReportFields(r *dispatch.CancelError, outcome dispatch.AgentOutcome) []zap.Field { + fields := []zap.Field{zap.String("agentOutcome", string(outcome))} + if r == nil { + return fields + } + if r.Stage != pb.CallCancel_STAGE_UNSPECIFIED { + fields = append(fields, + zap.String("agentStage", r.Stage.String()), + zap.Duration("agentStageElapsed", r.StageElapsed), + ) + } + if r.Cause != pb.CallCancel_CAUSE_UNSPECIFIED { + fields = append(fields, zap.String("agentCause", r.Cause.String())) + } + if r.TotalElapsed > 0 { + fields = append(fields, zap.Duration("agentElapsed", r.TotalElapsed)) + } + if r.UpstreamHost != "" { + fields = append(fields, zap.String("upstreamHost", r.UpstreamHost)) + } + if r.Reason != "" { + fields = append(fields, zap.String("agentReason", r.Reason)) + } + return fields } // writeFailedDispatch writes a JSON error response for a dispatch the agent diff --git a/server/config/config.go b/server/config/config.go index 039b43a9..7c34ff88 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -16,6 +16,11 @@ const ( DefaultHeartbeatInterval = 30 * time.Second DefaultMaxFrameBytes = 1 << 20 // 1 MiB DefaultMaxStreamsPerToken = 64 + // 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 + // hostage to an agent that will never answer. + DefaultAgentReportGrace = time.Second // DefaultBuildVersion is what an unstamped binary reports. The release // image sets AXON_BUILD_VERSION at build time; anything without it — a // local `go run`, a hand-built image — is honestly labeled "dev" rather @@ -48,6 +53,15 @@ type Config struct { // DispatchTimeout is the maximum time to wait for a client response // to a dispatched HTTP request. DispatchTimeout time.Duration + // AgentReportGrace is how long the server keeps a call it cancelled (its + // caller timed out or left) registered, with its stream still busy, waiting + // for the agent's own CallCancel. That reply says where the call was -- + // queued in the agent, waiting on the upstream, streaming -- and is what + // tells "the upstream was slow" from "the agent never answered". Holding + // the stream also stops the server dispatching onto it while the agent is + // still running the old call. 0 releases the stream at once and logs + // without the agent's side. + AgentReportGrace time.Duration // ServerID identifies this server instance. Used in metrics and // returned to clients in ServerHello for dedup. ServerID string @@ -83,6 +97,7 @@ func (c Config) Fields() []zap.Field { zap.String("server_id", c.ServerID), zap.Duration("heartbeat_interval", c.HeartbeatInterval), zap.Duration("dispatch_timeout", c.DispatchTimeout), + zap.Duration("agent_report_grace", c.AgentReportGrace), zap.Duration("re_registration_interval", c.ReRegistrationInterval), zap.Int("max_frame_bytes", c.MaxFrameBytes), zap.Int("max_streams_per_token", c.MaxStreamsPerToken), @@ -101,6 +116,7 @@ func NewConfigFromEnv() Config { HttpPort: DefaultHttpPort, HeartbeatInterval: DefaultHeartbeatInterval, DispatchTimeout: 60 * time.Second, + AgentReportGrace: DefaultAgentReportGrace, ServerID: getServerID(), ReRegistrationInterval: 5 * time.Minute, MaxFrameBytes: DefaultMaxFrameBytes, @@ -166,6 +182,14 @@ func NewConfigFromEnv() Config { cfg.DispatchTimeout = d } + if v := os.Getenv("AGENT_REPORT_GRACE"); v != "" { + d, err := time.ParseDuration(v) + if err != nil { + panic(fmt.Errorf("invalid AGENT_REPORT_GRACE: %w", err)) + } + cfg.AgentReportGrace = d + } + if v := os.Getenv("RE_REGISTRATION_INTERVAL"); v != "" { d, err := time.ParseDuration(v) if err != nil { diff --git a/server/dispatch/dispatcher.go b/server/dispatch/dispatcher.go index 50a397ee..4c3ecd68 100644 --- a/server/dispatch/dispatcher.go +++ b/server/dispatch/dispatcher.go @@ -13,6 +13,8 @@ import ( "errors" "fmt" "io" + "slices" + "strings" "sync" "sync/atomic" "time" @@ -40,6 +42,9 @@ var ( // errCallAborted covers a call that ended before its response started // without any more specific error, so Dispatch never returns (nil, nil). errCallAborted = errors.New("call ended before response started") + // errCallCancelled closes the response pipe of a call the server is + // ending, for frames the agent sends before it sees the cancel. + errCallCancelled = errors.New("call cancelled") ) // FailureKind classifies why a dispatch failed, for logs and callers. @@ -66,12 +71,28 @@ const ( FailureAborted FailureKind = "aborted" ) +// FailurePhase is where in its life a dispatch failed. Kind says what ended +// it and who gave up; Phase says what it was waiting for at the time, which is +// the difference between "no stream was free" and "a stream was free, the +// call went out, and nothing came back" -- both of which can end in the same +// caller timeout. +type FailurePhase string + +const ( + // PhaseAcquire: waiting for an idle stream. Nothing reached the agent. + PhaseAcquire FailurePhase = "acquire" + // PhaseAwaitingResponse: the call was sent to the agent and its response + // had not started. + PhaseAwaitingResponse FailurePhase = "awaiting_response" +) + // DispatchError is returned by Dispatch for every failure. It wraps the // underlying error, so errors.Is(err, ErrNoTunnel), errors.As(err, // *CancelError) and errors.Is(err, context.DeadlineExceeded) still work, and // adds what a log line needs to name the call and the agent that had it. type DispatchError struct { - Kind FailureKind + Kind FailureKind + Phase FailurePhase // CallID and StreamID are empty when no stream was acquired. CallID string StreamID string @@ -80,7 +101,93 @@ type DispatchError struct { Agent tunnel.ClientIdentity // AcquireWait is how long Dispatch waited for an idle stream. AcquireWait time.Duration - Err error + // Elapsed is how long the call had been with the agent when it failed: + // from CallStart sent to the failure. Zero when no stream was acquired. + Elapsed time.Duration + // Streams is the token's state on this server when the dispatch failed: + // its streams, how many were busy, how many dispatches were waiting, and + // the oldest calls holding streams. It is what tells "every stream was + // held by slow calls" apart from "this call alone was slow". + Streams StreamState + Err error + + // call is the failed call, when one was sent, so AwaitAgentReport can + // collect what the agent says about it. + call *call +} + +// StreamState is a token's streams on this server at the moment a dispatch +// failed. +type StreamState struct { + tunnel.StreamSnapshot + // Waiters is how many dispatches for the token were waiting for a stream, + // this one included when it failed waiting. + Waiters int + // Holders are the oldest calls for the token still in flight, oldest + // first, at most maxHolders of them. + Holders []HeldCall +} + +// HeldCall is one call holding a stream, for explaining why none was free. +type HeldCall struct { + CallID string + InstanceID string + Method string + Path string + Age time.Duration + // Cancelled is true while the server has cancelled the call and is + // waiting for the agent's answer, with the stream still held. + Cancelled bool +} + +// maxHolders bounds how many holding calls a failure reports. +const maxHolders = 3 + +// AgentOutcome is what the agent did about a call the server had to give up +// on. +type AgentOutcome string + +const ( + // AgentReported: the agent sent CallCancel explaining where the call was. + AgentReported AgentOutcome = "reported" + // AgentCompletedLate: the agent finished the call normally, after the + // server had already given up on it. + AgentCompletedLate AgentOutcome = "completed_late" + // AgentSilent: the agent said nothing within the grace period. + AgentSilent AgentOutcome = "silent" + // AgentNotAsked: there was nothing to ask -- no call reached the agent, or + // the server is configured not to wait. + AgentNotAsked AgentOutcome = "not_asked" +) + +// AwaitAgentReport waits for the agent's account of a call the server gave +// up on, and returns it. It returns at once when no call was sent, and +// otherwise blocks until the call finishes, which the grace timer bounds; +// call it after the caller has its response. The CancelError is the agent's +// own, if it sent one -- for a call the agent cancelled itself, that is the +// error Dispatch already returned. +func (e *DispatchError) AwaitAgentReport() (*CancelError, AgentOutcome) { + c := e.call + if c == nil { + var cancelErr *CancelError + if errors.As(e.Err, &cancelErr) { + return cancelErr, AgentReported + } + return nil, AgentNotAsked + } + <-c.doneC + c.mu.Lock() + defer c.mu.Unlock() + switch { + case c.agentReport != nil: + return c.agentReport, AgentReported + case c.agentEnded: + return nil, AgentCompletedLate + case !c.held: + return nil, AgentNotAsked + default: + return nil, AgentSilent + } } func (e *DispatchError) Error() string { @@ -122,9 +229,30 @@ const acquireRetryInterval = 10 * time.Millisecond // CancelError is returned when the agent cancels a call before starting its // response. Code carries the agent's HTTP status hint (404 no matching rule, // 502 upstream unreachable, 503 at in-flight cap); 0 means unspecified. +// +// The remaining fields are the agent's account of where the call was and why +// it stopped. An older agent leaves them zero (STAGE_UNSPECIFIED, +// CAUSE_UNSPECIFIED). type CancelError struct { - Reason string - Code int32 + Reason string + Code int32 + Stage pb.CallCancel_Stage + Cause pb.CallCancel_Cause + StageElapsed time.Duration + TotalElapsed time.Duration + UpstreamHost string +} + +func cancelErrorFrom(c *pb.CallCancel) *CancelError { + return &CancelError{ + Reason: c.Reason, + Code: c.Code, + Stage: c.Stage, + Cause: c.Cause, + StageElapsed: time.Duration(c.StageElapsedMs) * time.Millisecond, + TotalElapsed: time.Duration(c.TotalElapsedMs) * time.Millisecond, + UpstreamHost: c.UpstreamHost, + } } func (e *CancelError) Error() string { @@ -182,6 +310,13 @@ type Dispatcher struct { mu sync.Mutex calls map[string]*call // call_id → call byStream map[string]map[string]*call // stream_id → call_id → call + // waiters counts dispatches per token (hashed) currently waiting for a + // stream, for StreamState. Guarded by mu. + waiters map[string]int + + // agentReportGrace is how long a call the server cancelled stays open + // for the agent's answer; see config.Config.AgentReportGrace. + agentReportGrace time.Duration // Cumulative backpressure accounting, alongside the tagged metrics, so // /healthz can report it without a metrics scrape. acquireWaits counts @@ -213,6 +348,9 @@ func NewDispatcher( logger: logger.Named("dispatch"), calls: make(map[string]*call), byStream: make(map[string]map[string]*call), + waiters: make(map[string]int), + + agentReportGrace: cfg.AgentReportGrace, } } @@ -222,6 +360,17 @@ type call struct { stream *tunnel.StreamHandle logger *zap.Logger + // What a failure of another call reports about this one while it holds + // its stream: whose it is, what it asked for, and since when. + tokenHash string + method string + path string // without the query string, which can carry credentials + sentAt time.Time + + // doneC is closed when the call finishes and its stream is released. + doneC chan struct{} + doneOnce sync.Once + // respStartC delivers the agent's response CallStart (cap 1). respStartC chan *pb.CallStart // bodyR/bodyW pipe response CallData payloads to the adapter. @@ -242,6 +391,14 @@ type call struct { recvDone bool // agent sent its terminal frame (End or Cancel) cancelSent bool // we sent CallCancel finished bool // cleanup ran + // held is true when the server cancelled the call and kept it open, + // stream busy, for the agent's answer; holdTimer ends the wait. + held bool + holdTimer *time.Timer + // agentReport is the agent's CallCancel, if it sent one; agentEnded is + // true if it sent CallEnd instead. + agentReport *CancelError + agentEnded bool // chansClosed guards trailersC/errC. Delivery and close both take mu // and consult it, so a terminal frame arriving as the call finishes // cannot send on a closed channel. @@ -297,22 +454,37 @@ func (c *call) closeChans() { // // Every error it returns is a *DispatchError. func (d *Dispatcher) Dispatch(ctx context.Context, token broker.Token, req *Request) (*Response, error) { - stream, waited, err := d.acquireStream(ctx, token) + stream, waited, state, err := d.acquireStream(ctx, token) if err != nil { kind := FailureNoTunnel if !errors.Is(err, ErrNoTunnel) { kind = kindForContext(err, FailureAcquireTimeout) } - return nil, &DispatchError{Kind: kind, AcquireWait: waited, Err: err} + return nil, &DispatchError{ + Kind: kind, + Phase: PhaseAcquire, + AcquireWait: waited, + Streams: state, + Err: err, + } } callID := uuid.New().String() pumpCtx, pumpCancel := context.WithCancel(context.Background()) bodyR, bodyW := io.Pipe() + path := req.PseudoHeaders[":path"] + if i := strings.IndexByte(path, '?'); i >= 0 { + path = path[:i] + } c := &call{ id: callID, stream: stream, logger: d.logger.With(zap.String("callId", callID), zap.String("streamId", stream.StreamID)), + tokenHash: token.Hashed(), + method: req.PseudoHeaders[":method"], + path: path, + sentAt: time.Now(), + doneC: make(chan struct{}), respStartC: make(chan *pb.CallStart, 1), bodyR: bodyR, bodyW: bodyW, @@ -327,11 +499,15 @@ func (d *Dispatcher) Dispatch(ctx context.Context, token broker.Token, req *Requ fail := func(kind FailureKind, err error) error { return &DispatchError{ Kind: kind, + Phase: PhaseAwaitingResponse, CallID: callID, StreamID: stream.StreamID, Agent: stream.Identity, AcquireWait: waited, + Elapsed: time.Since(c.sentAt), + Streams: d.streamState(token), Err: err, + call: c, } } @@ -377,8 +553,11 @@ func (d *Dispatcher) Dispatch(ctx context.Context, token broker.Token, req *Requ return nil, fail(kindForPreStart(err), err) case <-ctx.Done(): + // Snapshot before cancelling, so this call is still counted among + // the token's busy streams -- it is holding one. + dispErr := fail(kindForContext(ctx.Err(), FailureResponseTimeout), ctx.Err()) d.cancelCall(c, "caller timeout", true) - return nil, fail(kindForContext(ctx.Err(), FailureResponseTimeout), ctx.Err()) + return nil, dispErr } } @@ -388,15 +567,22 @@ func (d *Dispatcher) Dispatch(ctx context.Context, token broker.Token, req *Requ // mechanism: an agent at capacity stops offering idle streams, and that // shows up as latency to whoever made the request rather than as work // piling up invisibly inside the agent. The wait is measured so the -// difference is visible, and returned so a failure can report it. -func (d *Dispatcher) acquireStream(ctx context.Context, token broker.Token) (*tunnel.StreamHandle, time.Duration, error) { +// difference is visible, and returned so a failure can report it, along with +// the token's stream state at the moment it gave up -- taken while this +// dispatch still counts as a waiter, which is the point of reporting it. +func (d *Dispatcher) acquireStream(ctx context.Context, token broker.Token) (*tunnel.StreamHandle, time.Duration, StreamState, error) { start := time.Now() waited := false + defer func() { + if waited { + d.addWaiter(token, -1) + } + }() for { handle, allBusy := d.registry.AcquireIdleStream(token) if handle != nil { if !waited { - return handle, 0, nil + return handle, 0, StreamState{}, nil } wait := time.Since(start) waitMs := wait.Milliseconds() @@ -404,13 +590,14 @@ func (d *Dispatcher) acquireStream(ctx context.Context, token broker.Token) (*tu d.withIdentity(token, func(tenant, integration, alias string) { d.metrics.DispatchAcquireWait(tenant, integration, alias, float64(waitMs)) }) - return handle, wait, nil + return handle, wait, StreamState{}, nil } if !allBusy { - return nil, time.Since(start), ErrNoTunnel + return nil, time.Since(start), StreamState{}, ErrNoTunnel } if !waited { waited = true + d.addWaiter(token, 1) d.acquireWaits.Add(1) d.withIdentity(token, func(tenant, integration, alias string) { d.metrics.DispatchAllBusy(tenant, integration, alias) @@ -418,12 +605,61 @@ func (d *Dispatcher) acquireStream(ctx context.Context, token broker.Token) (*tu } select { case <-ctx.Done(): - return nil, time.Since(start), ctx.Err() + return nil, time.Since(start), d.streamState(token), ctx.Err() case <-time.After(acquireRetryInterval): } } } +func (d *Dispatcher) addWaiter(token broker.Token, delta int) { + key := token.Hashed() + d.mu.Lock() + defer d.mu.Unlock() + if n := d.waiters[key] + delta; n > 0 { + d.waiters[key] = n + } else { + delete(d.waiters, key) + } +} + +// streamState reports the token's streams on this server, the dispatches +// waiting for one, and the oldest calls holding them. Only failures call +// it, so it may scan every in-flight call. +func (d *Dispatcher) streamState(token broker.Token) StreamState { + state := StreamState{StreamSnapshot: d.registry.Snapshot(token)} + key := token.Hashed() + now := time.Now() + + d.mu.Lock() + state.Waiters = d.waiters[key] + var holders []*call + for _, c := range d.calls { + if c.tokenHash == key { + holders = append(holders, c) + } + } + d.mu.Unlock() + + slices.SortFunc(holders, func(a, b *call) int { return a.sentAt.Compare(b.sentAt) }) + if len(holders) > maxHolders { + holders = holders[:maxHolders] + } + for _, c := range holders { + c.mu.Lock() + cancelled := c.held + c.mu.Unlock() + state.Holders = append(state.Holders, HeldCall{ + CallID: c.id, + InstanceID: c.stream.Identity.InstanceID, + Method: c.method, + Path: c.path, + Age: now.Sub(c.sentAt), + Cancelled: cancelled, + }) + } + return state +} + // withIdentity looks up a token's client identity for metric tagging. It is // only called off the fast path (when a dispatch actually had to wait), so // the registry lookup never costs an uncontended dispatch anything. @@ -528,6 +764,7 @@ func (d *Dispatcher) HandleFrame(streamID string, frame *pb.CallFrame) { c.mu.Lock() preStart := !c.started c.recvDone = true + c.agentEnded = true c.mu.Unlock() if preStart { c.logger.Warn("CallEnd before response CallStart; treating as abort") @@ -541,14 +778,14 @@ func (d *Dispatcher) HandleFrame(streamID string, frame *pb.CallFrame) { d.maybeFinish(c) case *pb.CallFrame_Cancel: - reason := body.Cancel.Reason - code := body.Cancel.Code + err := cancelErrorFrom(body.Cancel) c.logger.Debug("Call cancelled by agent", - zap.String("reason", reason), zap.Int32("code", code)) + zap.String("reason", err.Reason), zap.Int32("code", err.Code), + zap.Stringer("stage", err.Stage), zap.Stringer("cause", err.Cause)) c.mu.Lock() c.recvDone = true + c.agentReport = err c.mu.Unlock() - err := &CancelError{Reason: reason, Code: code} d.deliverErr(c, err) c.bodyW.CloseWithError(err) // The agent is done; stop pumping request body and finish. @@ -596,8 +833,21 @@ func (d *Dispatcher) deliverErr(c *call, err error) { // cancelCall aborts a call: stops the body pump, optionally notifies the // agent with CallCancel, and finishes the call. +// +// When the server is the one ending the call (notifyAgent) and the agent has +// not finished it, the call is held rather than finished: it stays +// registered, its stream stays busy, and it finishes when the agent answers +// the cancel or agentReportGrace runs out. The agent is still running the +// call until it sees the cancel, so the stream is not free yet, and its answer +// says where the call was -- which is the difference between an upstream that +// was slow and an agent that was not answering. func (d *Dispatcher) cancelCall(c *call, reason string, notifyAgent bool) { c.pumpCancel() + // Nothing reads the response any more. Closing the pipe makes a late + // CallData from the agent fail its write at once instead of blocking the + // stream's receive loop on a reader that is gone; CloseWithError keeps the + // first error, so an abort's own error stands. + c.bodyW.CloseWithError(errCallCancelled) c.mu.Lock() // Cancel is a call-level abort, not a direction terminal: it is valid @@ -608,19 +858,40 @@ func (d *Dispatcher) cancelCall(c *call, reason string, notifyAgent bool) { c.cancelSent = true } c.sendDone = true - c.recvDone = true + hold := sendCancel && !c.recvDone && d.agentReportGrace > 0 + switch { + case hold: + c.held = true + c.holdTimer = time.AfterFunc(d.agentReportGrace, func() { d.endHold(c) }) + case !notifyAgent || !c.held: + // Either the agent's own side is over (its Cancel or End, or the + // stream died under it), or there is nothing to wait for. A repeat + // server-side cancel of a held call keeps waiting. + c.recvDone = true + } c.mu.Unlock() if sendCancel { frame := callFrame(c.id, &pb.CallFrame_Cancel{Cancel: &pb.CallCancel{Reason: reason}}) if err := c.stream.Send(frame); err != nil { c.logger.Debug("Failed to send CallCancel", zap.Error(err)) + // The agent cannot answer a cancel it never got. + d.endHold(c) } } d.maybeFinish(c) } +// endHold stops waiting for the agent's answer to a cancel and lets the call +// finish. +func (d *Dispatcher) endHold(c *call) { + c.mu.Lock() + c.recvDone = true + c.mu.Unlock() + d.maybeFinish(c) +} + // maybeFinish cleans up once both directions have terminated. func (d *Dispatcher) maybeFinish(c *call) { c.mu.Lock() @@ -638,7 +909,13 @@ func (d *Dispatcher) maybeFinish(c *call) { func (d *Dispatcher) finish(c *call, closeChans bool) { d.unregister(c) c.pumpCancel() + c.mu.Lock() + if c.holdTimer != nil { + c.holdTimer.Stop() + } + c.mu.Unlock() c.stream.Release() + c.doneOnce.Do(func() { close(c.doneC) }) d.metrics.DispatchInflight.Update(float64(d.InflightCount())) if closeChans { c.closeChans() diff --git a/server/dispatch/failure_cause.go b/server/dispatch/failure_cause.go new file mode 100644 index 00000000..5f651862 --- /dev/null +++ b/server/dispatch/failure_cause.go @@ -0,0 +1,110 @@ +package dispatch + +import ( + "errors" + + pb "github.com/cortexapps/axon-server/.generated/proto/tunnelpb" +) + +// FailureCause names what went wrong with a dispatch in terms of what to fix, +// as opposed to its FailureKind, which names who gave up. A caller that times +// out after 60s is the same Kind whether every stream was held by slow calls +// or its own call sat waiting on the upstream; the cause tells those apart. +// It is logged, counted, and returned to the caller in a response header. +type FailureCause string + +const ( + // CauseNoTunnel: this server holds no stream for the token. + CauseNoTunnel FailureCause = "no_tunnel" + // CauseStreamsExhausted: every stream for the token on this server stayed + // busy until the deadline. The calls holding them are the problem, not + // this one; the failure's StreamState names them. + CauseStreamsExhausted FailureCause = "streams_exhausted" + // CauseUpstreamSlow: the agent sent the request upstream and no response + // headers came back before someone gave up -- or they came back after. + CauseUpstreamSlow FailureCause = "upstream_slow" + // CauseUpstreamError: the upstream failed outright (connection error, + // reset, unreadable body). + CauseUpstreamError FailureCause = "upstream_error" + // CauseAgentAtCapacity: the call queued for the agent's own in-flight + // capacity and never got it. + CauseAgentAtCapacity FailureCause = "agent_at_capacity" + // CauseRouteRejected: no route matched the request, or the route refused it. + CauseRouteRejected FailureCause = "route_rejected" + // CauseAgentShutdown: the agent was shutting down or restarting. + CauseAgentShutdown FailureCause = "agent_shutdown" + // CauseAgentUnresponsive: the server cancelled the call and the agent did + // not answer within the grace period. + CauseAgentUnresponsive FailureCause = "agent_unresponsive" + // CauseAgentCancel: the agent cancelled without saying why (an agent that + // predates stage and cause reporting). + CauseAgentCancel FailureCause = "agent_cancel" + // CauseAwaitingAgent: the call was with the agent when the server gave up, + // and the agent's account is not in (yet, or at all when the grace period + // is off). + CauseAwaitingAgent FailureCause = "awaiting_agent" + // CauseTunnelError: the stream failed under the call (send failure, closed). + CauseTunnelError FailureCause = "tunnel_error" + // CauseCallerBodyError: reading the caller's request body failed. + CauseCallerBodyError FailureCause = "caller_body_error" + // CauseUnknown: not a dispatch failure the server can classify. + CauseUnknown FailureCause = "unknown" +) + +// MadeCallerWait reports whether the cause is something on our side keeping +// the caller waiting, as opposed to a caller simply leaving. A caller giving +// up on one of these is the symptom, not the routine event a caller +// disconnect otherwise is. +func (c FailureCause) MadeCallerWait() bool { + switch c { + case CauseStreamsExhausted, CauseAgentUnresponsive, CauseAgentAtCapacity, CauseUpstreamSlow: + return true + } + return false +} + +// FailureCause reads the agent's account of where a call was and why it +// stopped. An agent too old to report a stage or cause gives CauseAgentCancel. +func (e *CancelError) FailureCause() FailureCause { + switch { + case e.Cause == pb.CallCancel_CAPACITY || e.Stage == pb.CallCancel_QUEUED_FOR_CAPACITY: + return CauseAgentAtCapacity + case e.Cause == pb.CallCancel_ROUTE_REJECTED: + return CauseRouteRejected + case e.Cause == pb.CallCancel_AGENT_SHUTDOWN: + return CauseAgentShutdown + case e.Cause == pb.CallCancel_UPSTREAM_ERROR: + return CauseUpstreamError + case e.Stage == pb.CallCancel_AWAITING_UPSTREAM_HEADERS: + return CauseUpstreamSlow + } + return CauseAgentCancel +} + +// Cause classifies the failure. report and outcome are the agent's account, +// from AwaitAgentReport; before it is in, pass (nil, AgentNotAsked) and the +// cause is what the server alone can tell. +func (e *DispatchError) Cause(report *CancelError, outcome AgentOutcome) FailureCause { + switch { + case errors.Is(e.Err, ErrNoTunnel): + return CauseNoTunnel + case e.Phase == PhaseAcquire: + return CauseStreamsExhausted + } + switch e.Kind { + case FailureStreamClosed, FailureSendFailed: + return CauseTunnelError + case FailureRequestBody: + return CauseCallerBodyError + } + if report != nil { + return report.FailureCause() + } + switch outcome { + case AgentCompletedLate: + return CauseUpstreamSlow + case AgentSilent: + return CauseAgentUnresponsive + } + return CauseAwaitingAgent +} diff --git a/server/dispatch/failure_cause_test.go b/server/dispatch/failure_cause_test.go new file mode 100644 index 00000000..c454114f --- /dev/null +++ b/server/dispatch/failure_cause_test.go @@ -0,0 +1,61 @@ +package dispatch + +import ( + "context" + "fmt" + "testing" + + pb "github.com/cortexapps/axon-server/.generated/proto/tunnelpb" + "github.com/stretchr/testify/assert" +) + +func TestCancelError_FailureCause(t *testing.T) { + for _, tc := range []struct { + stage pb.CallCancel_Stage + cause pb.CallCancel_Cause + want FailureCause + }{ + {pb.CallCancel_AWAITING_UPSTREAM_HEADERS, pb.CallCancel_CANCELED_BY_SERVER, CauseUpstreamSlow}, + {pb.CallCancel_AWAITING_UPSTREAM_HEADERS, pb.CallCancel_AGENT_TIMEOUT, CauseUpstreamSlow}, + {pb.CallCancel_AWAITING_UPSTREAM_HEADERS, pb.CallCancel_UPSTREAM_ERROR, CauseUpstreamError}, + {pb.CallCancel_STREAMING_BODY, pb.CallCancel_UPSTREAM_ERROR, CauseUpstreamError}, + {pb.CallCancel_QUEUED_FOR_CAPACITY, pb.CallCancel_CANCELED_BY_SERVER, CauseAgentAtCapacity}, + {pb.CallCancel_QUEUED_FOR_CAPACITY, pb.CallCancel_CAPACITY, CauseAgentAtCapacity}, + {pb.CallCancel_ROUTING, pb.CallCancel_ROUTE_REJECTED, CauseRouteRejected}, + {pb.CallCancel_AWAITING_UPSTREAM_HEADERS, pb.CallCancel_AGENT_SHUTDOWN, CauseAgentShutdown}, + // An agent that predates stage and cause reporting. + {pb.CallCancel_STAGE_UNSPECIFIED, pb.CallCancel_CAUSE_UNSPECIFIED, CauseAgentCancel}, + } { + t.Run(fmt.Sprintf("%s/%s", tc.stage, tc.cause), func(t *testing.T) { + e := &CancelError{Stage: tc.stage, Cause: tc.cause} + assert.Equal(t, tc.want, e.FailureCause()) + }) + } +} + +func TestDispatchError_Cause(t *testing.T) { + slowUpstream := &CancelError{Stage: pb.CallCancel_AWAITING_UPSTREAM_HEADERS, Cause: pb.CallCancel_CANCELED_BY_SERVER} + sent := func(kind FailureKind) *DispatchError { + return &DispatchError{Kind: kind, Phase: PhaseAwaitingResponse, CallID: "c", Err: context.DeadlineExceeded} + } + for _, tc := range []struct { + name string + err *DispatchError + report *CancelError + outcome AgentOutcome + want FailureCause + }{ + {"no tunnel", &DispatchError{Kind: FailureNoTunnel, Phase: PhaseAcquire, Err: ErrNoTunnel}, nil, AgentNotAsked, CauseNoTunnel}, + {"never got a stream", &DispatchError{Kind: FailureCallerCanceled, Phase: PhaseAcquire, Err: context.Canceled}, nil, AgentNotAsked, CauseStreamsExhausted}, + {"sent, agent says upstream", sent(FailureResponseTimeout), slowUpstream, AgentReported, CauseUpstreamSlow}, + {"sent, agent finished late", sent(FailureCallerCanceled), nil, AgentCompletedLate, CauseUpstreamSlow}, + {"sent, agent silent", sent(FailureResponseTimeout), nil, AgentSilent, CauseAgentUnresponsive}, + {"sent, not asked yet", sent(FailureResponseTimeout), nil, AgentNotAsked, CauseAwaitingAgent}, + {"stream closed", sent(FailureStreamClosed), nil, AgentNotAsked, CauseTunnelError}, + {"request body", sent(FailureRequestBody), nil, AgentNotAsked, CauseCallerBodyError}, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, tc.err.Cause(tc.report, tc.outcome)) + }) + } +} diff --git a/server/dispatch/failure_diagnostics_test.go b/server/dispatch/failure_diagnostics_test.go new file mode 100644 index 00000000..98799778 --- /dev/null +++ b/server/dispatch/failure_diagnostics_test.go @@ -0,0 +1,332 @@ +package dispatch + +import ( + "context" + "errors" + "testing" + "time" + + pb "github.com/cortexapps/axon-server/.generated/proto/tunnelpb" + "github.com/cortexapps/axon-server/broker" + "github.com/cortexapps/axon-server/config" + "github.com/cortexapps/axon-server/metrics" + "github.com/cortexapps/axon-server/tunnel" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap/zaptest" +) + +// These tests reproduce the two failure shapes seen in prod (Nov 2026) and +// pin down what a failed dispatch now says about itself: +// +// - a 2s /__axon/info probe that found the token's only stream on this +// server busy with another call, and gave up waiting for it; +// - a 60s call that got a stream, reached the agent, and was still waiting +// on the upstream when the caller gave up. + +// newHoldingDispatcher is newTestDispatcher with a grace period for the +// agent's answer to a server-side cancel, which the prod config has and the +// older tests predate. +func newHoldingDispatcher(t *testing.T, grace time.Duration) (*Dispatcher, broker.Token, *fakeStream, *tunnel.StreamHandle) { + t.Helper() + logger := zaptest.NewLogger(t) + registry := tunnel.NewClientRegistry(logger) + cfg := config.Config{MaxFrameBytes: 1024, AgentReportGrace: grace} + d := NewDispatcher(cfg, registry, metrics.New("test-server"), logger) + + token := broker.NewToken("test-token") + fs := newFakeStream() + handle := &tunnel.StreamHandle{StreamID: "stream-1", Send: fs.send, Cancel: func() {}} + mustRegister(t, registry, token, tunnel.ClientIdentity{InstanceID: "agent-a"}, handle) + return d, token, fs, handle +} + +// holdStream dispatches a call that the test never answers, so it keeps the +// token's stream busy, and returns its CallStart. +func holdStream(t *testing.T, d *Dispatcher, token broker.Token, fs *fakeStream, path string) *pb.CallFrame { + t.Helper() + go d.Dispatch(context.Background(), token, &Request{ + PseudoHeaders: map[string]string{":method": "GET", ":path": path}, + }) + start := fs.next(t) + require.NotNil(t, start.GetStart()) + fs.next(t) // request-side End: no body + return start +} + +func asDispatchError(t *testing.T, err error) *DispatchError { + t.Helper() + var dispErr *DispatchError + require.True(t, errors.As(err, &dispErr), "want *DispatchError, got %T: %v", err, err) + return dispErr +} + +// The prod 2s probe failure: the token's one stream on this server is held by +// a slow call, the probe waits for it, and the caller (the dispatcher's 2s +// client) gives up. The error must say it never got a stream, and who had it. +func TestDispatch_ProbeStarvedBySlowCall(t *testing.T) { + for _, tc := range []struct { + name string + ctx func() (context.Context, context.CancelFunc) + wantKind FailureKind + }{ + {"deadline", func() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 50*time.Millisecond) + }, FailureAcquireTimeout}, + {"caller went away", func() (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(50*time.Millisecond, cancel) + return ctx, cancel + }, FailureCallerCanceled}, + } { + t.Run(tc.name, func(t *testing.T) { + d, token, fs, _ := newHoldingDispatcher(t, 0) + holder := holdStream(t, d, token, fs, "/2.0/repositories?page=2") + + ctx, cancel := tc.ctx() + defer cancel() + _, err := d.Dispatch(ctx, token, &Request{ + PseudoHeaders: map[string]string{":method": "GET", ":path": "/__axon/info"}, + }) + + dispErr := asDispatchError(t, err) + assert.Equal(t, tc.wantKind, dispErr.Kind) + assert.Equal(t, PhaseAcquire, dispErr.Phase, "the probe never got a stream") + assert.Empty(t, dispErr.CallID, "nothing reached the agent") + assert.GreaterOrEqual(t, dispErr.AcquireWait, 40*time.Millisecond) + + s := dispErr.Streams + assert.Equal(t, 1, s.Streams) + assert.Equal(t, 1, s.Busy) + assert.Equal(t, 0, s.Idle()) + assert.Equal(t, 1, s.Waiters, "the probe itself was waiting") + assert.Equal(t, []tunnel.InstanceStreams{{InstanceID: "agent-a", Streams: 1, Busy: 1}}, s.Instances) + require.Len(t, s.Holders, 1) + h := s.Holders[0] + assert.Equal(t, holder.CallId, h.CallID) + assert.Equal(t, "GET", h.Method) + assert.Equal(t, "/2.0/repositories", h.Path, "query strings can carry credentials") + assert.Equal(t, "agent-a", h.InstanceID) + assert.GreaterOrEqual(t, h.Age, 40*time.Millisecond) + + rep, outcome := dispErr.AwaitAgentReport() + assert.Nil(t, rep) + assert.Equal(t, AgentNotAsked, outcome) + }) + } +} + +// Waiters drop back to zero once nobody is waiting, so a later snapshot does +// not report dispatches that are long gone. +func TestDispatch_WaitersAreCountedOnlyWhileWaiting(t *testing.T) { + d, token, fs, _ := newHoldingDispatcher(t, 0) + holdStream(t, d, token, fs, "/slow") + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _, err := d.Dispatch(ctx, token, &Request{PseudoHeaders: map[string]string{":method": "GET", ":path": "/x"}}) + require.Error(t, err) + + assert.Equal(t, 0, d.streamState(token).Waiters) +} + +// The prod 60s failure: the call got a stream at once and went to the agent; +// nothing came back before the caller's deadline. The error must say the call +// was with the agent, for how long, and count its own stream as busy. +func TestDispatch_SlowUpstreamTimesOutAfterSend(t *testing.T) { + d, token, _, _ := newHoldingDispatcher(t, 0) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _, err := d.Dispatch(ctx, token, &Request{ + PseudoHeaders: map[string]string{":method": "GET", ":path": "/api/v4/groups/1/merge_requests?per_page=100"}, + }) + + dispErr := asDispatchError(t, err) + assert.Equal(t, FailureResponseTimeout, dispErr.Kind) + assert.Equal(t, PhaseAwaitingResponse, dispErr.Phase) + assert.NotEmpty(t, dispErr.CallID) + assert.Zero(t, dispErr.AcquireWait, "a stream was free") + assert.GreaterOrEqual(t, dispErr.Elapsed, 40*time.Millisecond) + assert.Equal(t, 1, dispErr.Streams.Busy, "the failed call was holding the stream") + require.Len(t, dispErr.Streams.Holders, 1) + assert.Equal(t, dispErr.CallID, dispErr.Streams.Holders[0].CallID) + assert.Equal(t, "/api/v4/groups/1/merge_requests", dispErr.Streams.Holders[0].Path) +} + +// After a server-side timeout the agent is still running the call until it +// sees the cancel. The stream stays busy until the agent answers, and its +// answer -- where the call was -- is what the failure reports. +func TestDispatch_TimeoutHoldsStreamUntilAgentReports(t *testing.T) { + d, token, fs, handle := newHoldingDispatcher(t, 5*time.Second) + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + _, err := d.Dispatch(ctx, token, &Request{PseudoHeaders: map[string]string{":method": "GET", ":path": "/slow"}}) + dispErr := asDispatchError(t, err) + + start := fs.next(t) + fs.next(t) // End + require.NotNil(t, fs.next(t).GetCancel(), "agent must be told") + + assert.True(t, handle.Busy(), "the agent still has the call; the stream is not free") + assert.Equal(t, 1, d.InflightCount()) + + // A late CallData must not block the stream's receive loop on a pipe + // nobody reads. + dataDone := make(chan struct{}) + go func() { + d.HandleFrame("stream-1", &pb.CallFrame{CallId: start.CallId, Body: &pb.CallFrame_Data{Data: &pb.CallData{Payload: []byte("late")}}}) + close(dataDone) + }() + select { + case <-dataDone: + case <-time.After(time.Second): + t.Fatal("late CallData blocked the receive loop") + } + + d.HandleFrame("stream-1", &pb.CallFrame{CallId: start.CallId, Body: &pb.CallFrame_Cancel{Cancel: &pb.CallCancel{ + Code: 502, + Reason: "context canceled", + Stage: pb.CallCancel_AWAITING_UPSTREAM_HEADERS, + Cause: pb.CallCancel_CANCELED_BY_SERVER, + StageElapsedMs: 29, + TotalElapsedMs: 30, + UpstreamHost: "gitlab.com", + }}}) + + rep, outcome := dispErr.AwaitAgentReport() + assert.Equal(t, AgentReported, outcome) + require.NotNil(t, rep) + assert.Equal(t, pb.CallCancel_AWAITING_UPSTREAM_HEADERS, rep.Stage) + assert.Equal(t, pb.CallCancel_CANCELED_BY_SERVER, rep.Cause) + assert.Equal(t, 29*time.Millisecond, rep.StageElapsed) + assert.Equal(t, "gitlab.com", rep.UpstreamHost) + + assert.False(t, handle.Busy(), "released once the agent answered") + assert.Equal(t, 0, d.InflightCount()) +} + +// A held call shows up in another dispatch's failure as cancelled, so a +// starved probe can tell a stream held by a live call from one held only +// while the server waits for the agent. +func TestDispatch_HeldCallIsReportedAsCancelled(t *testing.T) { + d, token, fs, _ := newHoldingDispatcher(t, 5*time.Second) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _, err := d.Dispatch(ctx, token, &Request{PseudoHeaders: map[string]string{":method": "GET", ":path": "/slow"}}) + require.Error(t, err) + + probeCtx, probeCancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer probeCancel() + _, err = d.Dispatch(probeCtx, token, &Request{PseudoHeaders: map[string]string{":method": "GET", ":path": "/__axon/info"}}) + dispErr := asDispatchError(t, err) + require.Len(t, dispErr.Streams.Holders, 1) + assert.True(t, dispErr.Streams.Holders[0].Cancelled) + + start := fs.next(t) + d.HandleFrame("stream-1", &pb.CallFrame{CallId: start.CallId, Body: &pb.CallFrame_Cancel{Cancel: &pb.CallCancel{}}}) +} + +// An agent that never answers the cancel gets the grace period and no more: +// then the stream is released and the failure says the agent was silent. +func TestDispatch_SilentAgentReleasesStreamAfterGrace(t *testing.T) { + d, token, _, handle := newHoldingDispatcher(t, 50*time.Millisecond) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _, err := d.Dispatch(ctx, token, &Request{PseudoHeaders: map[string]string{":method": "GET", ":path": "/slow"}}) + dispErr := asDispatchError(t, err) + + rep, outcome := dispErr.AwaitAgentReport() + assert.Nil(t, rep) + assert.Equal(t, AgentSilent, outcome) + assert.False(t, handle.Busy()) + assert.Equal(t, 0, d.InflightCount()) +} + +// An agent that finishes the call just after the server gave up: the upstream +// answered, too late. +func TestDispatch_AgentCompletingLateIsReportedAsSuch(t *testing.T) { + d, token, fs, handle := newHoldingDispatcher(t, 5*time.Second) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _, err := d.Dispatch(ctx, token, &Request{PseudoHeaders: map[string]string{":method": "GET", ":path": "/slow"}}) + dispErr := asDispatchError(t, err) + start := fs.next(t) + + d.HandleFrame("stream-1", &pb.CallFrame{CallId: start.CallId, Body: &pb.CallFrame_Start{Start: &pb.CallStart{ + PseudoHeaders: map[string]string{":status": "200"}, + }}}) + d.HandleFrame("stream-1", &pb.CallFrame{CallId: start.CallId, Body: &pb.CallFrame_End{End: &pb.CallEnd{}}}) + + _, outcome := dispErr.AwaitAgentReport() + assert.Equal(t, AgentCompletedLate, outcome) + assert.False(t, handle.Busy()) +} + +// A stream that dies while its call is held cannot carry the agent's answer; +// the hold ends at once rather than at the grace deadline. +func TestDispatch_StreamCloseEndsHold(t *testing.T) { + d, token, _, handle := newHoldingDispatcher(t, time.Hour) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _, err := d.Dispatch(ctx, token, &Request{PseudoHeaders: map[string]string{":method": "GET", ":path": "/slow"}}) + dispErr := asDispatchError(t, err) + + d.HandleStreamClose("stream-1") + + done := make(chan AgentOutcome, 1) + go func() { + _, outcome := dispErr.AwaitAgentReport() + done <- outcome + }() + select { + case outcome := <-done: + assert.Equal(t, AgentSilent, outcome) + case <-time.After(time.Second): + t.Fatal("hold outlived its stream") + } + assert.False(t, handle.Busy()) +} + +// Without a grace period the old behaviour stands: the stream is released the +// moment the server gives up. +func TestDispatch_NoGraceReleasesAtOnce(t *testing.T) { + d, token, _, handle := newHoldingDispatcher(t, 0) + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _, err := d.Dispatch(ctx, token, &Request{PseudoHeaders: map[string]string{":method": "GET", ":path": "/slow"}}) + dispErr := asDispatchError(t, err) + + assert.False(t, handle.Busy()) + _, outcome := dispErr.AwaitAgentReport() + assert.Equal(t, AgentNotAsked, outcome) +} + +// An agent that cancels on its own says why, and the error carries it. +func TestDispatch_AgentCancelCarriesItsReport(t *testing.T) { + d, token, fs, _ := newHoldingDispatcher(t, time.Second) + + errC := make(chan error, 1) + go func() { + _, err := d.Dispatch(context.Background(), token, &Request{PseudoHeaders: map[string]string{":method": "GET", ":path": "/x"}}) + errC <- err + }() + start := fs.next(t) + d.HandleFrame("stream-1", &pb.CallFrame{CallId: start.CallId, Body: &pb.CallFrame_Cancel{Cancel: &pb.CallCancel{ + Code: 503, Reason: "timed out waiting for in-flight capacity", + Stage: pb.CallCancel_QUEUED_FOR_CAPACITY, Cause: pb.CallCancel_CAPACITY, + }}}) + + dispErr := asDispatchError(t, <-errC) + assert.Equal(t, FailureAgentCancel, dispErr.Kind) + rep, outcome := dispErr.AwaitAgentReport() + assert.Equal(t, AgentReported, outcome) + require.NotNil(t, rep) + assert.Equal(t, pb.CallCancel_QUEUED_FOR_CAPACITY, rep.Stage) +} diff --git a/server/metrics/metrics.go b/server/metrics/metrics.go index 63c0885e..33749f6a 100644 --- a/server/metrics/metrics.go +++ b/server/metrics/metrics.go @@ -17,7 +17,7 @@ const ( LabelAlias = "alias" LabelMethod = "method" LabelStatusCode = "status_code" - LabelErrorType = "error_type" + LabelCause = "cause" ) // Metrics holds all server-side metric instruments. @@ -26,16 +26,16 @@ type Metrics struct { Closer func() Registry *prom.Registry - ConnectionsActive tally.Gauge - ConnectionsTotal tally.Counter - HeartbeatSent tally.Counter - HeartbeatReceived tally.Counter - HeartbeatMissed tally.Counter - DispatchInflight tally.Gauge - DispatchErrors tally.Counter - DispatchBytesSent tally.Counter - DispatchBytesRecv tally.Counter - AuthFailures tally.Counter + ConnectionsActive tally.Gauge + ConnectionsTotal tally.Counter + HeartbeatSent tally.Counter + HeartbeatReceived tally.Counter + HeartbeatMissed tally.Counter + DispatchInflight tally.Gauge + DispatchErrors tally.Counter + DispatchBytesSent tally.Counter + DispatchBytesRecv tally.Counter + AuthFailures tally.Counter } // New creates a new Metrics instance backed by a Prometheus reporter. @@ -59,16 +59,16 @@ func New(serverID string) *Metrics { Closer: func() { closer.Close() }, Registry: registry, - ConnectionsActive: scope.Gauge("connections.active"), - ConnectionsTotal: scope.Counter("connections.total"), - HeartbeatSent: scope.Counter("heartbeat.sent"), - HeartbeatReceived: scope.Counter("heartbeat.received"), - HeartbeatMissed: scope.Counter("heartbeat.missed"), - DispatchInflight: scope.Gauge("dispatch.inflight"), - DispatchErrors: scope.Counter("dispatch.errors"), - DispatchBytesSent: scope.Counter("dispatch.bytes_sent"), - DispatchBytesRecv: scope.Counter("dispatch.bytes_received"), - AuthFailures: scope.Counter("auth.failures"), + ConnectionsActive: scope.Gauge("connections.active"), + ConnectionsTotal: scope.Counter("connections.total"), + HeartbeatSent: scope.Counter("heartbeat.sent"), + HeartbeatReceived: scope.Counter("heartbeat.received"), + HeartbeatMissed: scope.Counter("heartbeat.missed"), + DispatchInflight: scope.Gauge("dispatch.inflight"), + DispatchErrors: scope.Counter("dispatch.errors"), + DispatchBytesSent: scope.Counter("dispatch.bytes_sent"), + DispatchBytesRecv: scope.Counter("dispatch.bytes_received"), + AuthFailures: scope.Counter("auth.failures"), } return m @@ -94,14 +94,21 @@ func (m *Metrics) DispatchDuration(tenantID, integration, alias string, d float6 }).Histogram("dispatch.duration_ms", tally.DefaultBuckets).RecordValue(d) } -// DispatchError records a dispatch error. -func (m *Metrics) DispatchError(tenantID, integration, alias, errorType string) { +// DispatchFailure counts a failed dispatch by cause: what went wrong in +// terms of what to fix (streams_exhausted, upstream_slow, +// agent_unresponsive, ...), tagged with the agent's identity. +// +// Its own name, not dispatch.errors: the untagged DispatchErrors counter +// owns that one, and Prometheus refuses a second registration of a name with +// different labels -- by panicking, which is why the tagged variant that +// used to sit here could never be called. +func (m *Metrics) DispatchFailure(tenantID, integration, alias, cause string) { m.Scope.Tagged(map[string]string{ LabelTenantID: tenantID, LabelIntegration: integration, LabelAlias: alias, - LabelErrorType: errorType, - }).Counter("dispatch.errors").Inc(1) + LabelCause: cause, + }).Counter("dispatch.failures").Inc(1) } // DispatchAcquireWait records how long a call waited for an idle stream on diff --git a/server/tunnel/client_registry.go b/server/tunnel/client_registry.go index 75dc0935..3384e346 100644 --- a/server/tunnel/client_registry.go +++ b/server/tunnel/client_registry.go @@ -3,6 +3,8 @@ package tunnel import ( "errors" "fmt" + "slices" + "strings" "sync" "sync/atomic" @@ -429,6 +431,62 @@ func (r *ClientRegistry) AcquireIdleStream(token broker.Token) (handle *StreamHa return nil, true } +// StreamSnapshot is a token's streams on this server at one moment: how many +// there are, how many were carrying a call, and how they split across the +// agent instances that opened them. It exists to explain a failed dispatch, +// so it is taken only after one has failed. +type StreamSnapshot struct { + Streams int + Busy int + Instances []InstanceStreams +} + +// Idle is the streams that were free to take a call. +func (s StreamSnapshot) Idle() int { return s.Streams - s.Busy } + +// InstanceStreams is one agent instance's share of a token's streams. +type InstanceStreams struct { + InstanceID string + Streams int + Busy int +} + +// Snapshot reports the token's streams on this server. Busy comes from each +// stream's own flag, not the idle stack, which is only a hint. Instances are +// sorted by ID so the same state always logs the same way. A token with no +// streams here returns the zero snapshot. +func (r *ClientRegistry) Snapshot(token broker.Token) StreamSnapshot { + r.mu.RLock() + defer r.mu.RUnlock() + + entry, ok := r.entries[token.Hashed()] + if !ok { + return StreamSnapshot{} + } + snap := StreamSnapshot{Streams: len(entry.Streams)} + byInstance := make(map[string]*InstanceStreams, len(entry.instances)) + for _, s := range entry.Streams { + inst := byInstance[s.Identity.InstanceID] + if inst == nil { + inst = &InstanceStreams{InstanceID: s.Identity.InstanceID} + byInstance[s.Identity.InstanceID] = inst + } + inst.Streams++ + if s.Busy() { + inst.Busy++ + snap.Busy++ + } + } + snap.Instances = make([]InstanceStreams, 0, len(byInstance)) + for _, inst := range byInstance { + snap.Instances = append(snap.Instances, *inst) + } + slices.SortFunc(snap.Instances, func(a, b InstanceStreams) int { + return strings.Compare(a.InstanceID, b.InstanceID) + }) + return snap +} + // SetBrokerServerRegistered marks a token as successfully registered with BROKER_SERVER. func (r *ClientRegistry) SetBrokerServerRegistered(token broker.Token) { r.mu.RLock() From 8ac580cd931a0b11c31d401698fe8cf417d77bff Mon Sep 17 00:00:00 2001 From: Shawn Burke Date: Tue, 10 Nov 2026 11:56:32 +1100 Subject: [PATCH 2/2] dispatch: never release a held call from a cancel that does not notify the agent Review finding: when the caller's request body fails mid-read, the body pump cancels the call toward the agent and holds it, and Dispatch then cancels it again, without notifying, on seeing the error on errC. cancelCall took "don't notify the agent" to mean "the agent is done" and set recvDone, releasing the stream while the agent was still running the call, so the next dispatch could land on a busy stream. A held call is now released only by the agent's own terminal frame, the grace timer, a cancel that could not be sent, or the stream closing -- HandleStreamClose ends the hold explicitly. With no grace period nothing is ever held, so that path is unchanged. Co-Authored-By: Claude Opus 5.5 --- server/dispatch/dispatcher.go | 15 +++++++--- server/dispatch/failure_diagnostics_test.go | 33 +++++++++++++++++++++ 2 files changed, 44 insertions(+), 4 deletions(-) diff --git a/server/dispatch/dispatcher.go b/server/dispatch/dispatcher.go index 4c3ecd68..1fa9393e 100644 --- a/server/dispatch/dispatcher.go +++ b/server/dispatch/dispatcher.go @@ -805,6 +805,8 @@ func (d *Dispatcher) HandleStreamClose(streamID string) { for _, c := range calls { d.abortCall(c, ErrStreamClosed, false) + // The agent cannot answer on a stream that is gone. + d.endHold(c) } } @@ -863,12 +865,17 @@ func (d *Dispatcher) cancelCall(c *call, reason string, notifyAgent bool) { case hold: c.held = true c.holdTimer = time.AfterFunc(d.agentReportGrace, func() { d.endHold(c) }) - case !notifyAgent || !c.held: - // Either the agent's own side is over (its Cancel or End, or the - // stream died under it), or there is nothing to wait for. A repeat - // server-side cancel of a held call keeps waiting. + case !c.held: + // Nothing to wait for. c.recvDone = true } + // A held call is never released here, whoever cancels it next: only the + // agent's own terminal frame (HandleFrame), the grace timer, a cancel + // that could not be sent, or the stream closing (HandleStreamClose) ends + // the hold. Not notifying the agent is not the same as the agent being + // done -- Dispatch cancels without notifying after a request-body failure + // has already cancelled and held the call, and releasing it there handed + // a stream the agent was still using to the next dispatch. c.mu.Unlock() if sendCancel { diff --git a/server/dispatch/failure_diagnostics_test.go b/server/dispatch/failure_diagnostics_test.go index 98799778..3c0f9e37 100644 --- a/server/dispatch/failure_diagnostics_test.go +++ b/server/dispatch/failure_diagnostics_test.go @@ -330,3 +330,36 @@ func TestDispatch_AgentCancelCarriesItsReport(t *testing.T) { require.NotNil(t, rep) assert.Equal(t, pb.CallCancel_QUEUED_FOR_CAPACITY, rep.Stage) } + +// Review finding on #186: the caller's request body failing mid-read cancels +// the call toward the agent and holds it -- and then Dispatch, seeing the +// error on errC, cancelled it again without notifying the agent. That second +// cancel took "don't notify" to mean "the agent is done" and released the +// stream while the agent was still running the call, so the next dispatch +// could land on it. A held call is released only by the agent, the grace +// period, or the stream dying. +func TestDispatch_RequestBodyFailureHoldsStreamUntilAgentReports(t *testing.T) { + d, token, fs, handle := newHoldingDispatcher(t, 5*time.Second) + + readErr := errors.New("caller hung up mid-upload") + _, err := d.Dispatch(context.Background(), token, &Request{ + PseudoHeaders: map[string]string{":method": "POST", ":path": "/upload"}, + Body: failingReader{err: readErr}, + }) + dispErr := asDispatchError(t, err) + assert.Equal(t, FailureRequestBody, dispErr.Kind) + + start := fs.next(t) + require.NotNil(t, start.GetStart()) + require.NotNil(t, fs.next(t).GetCancel(), "the agent must be told to stop") + + assert.True(t, handle.Busy(), "the agent is still running the call; its stream is not free") + assert.Equal(t, 1, d.InflightCount()) + + d.HandleFrame("stream-1", &pb.CallFrame{CallId: start.CallId, Body: &pb.CallFrame_Cancel{Cancel: &pb.CallCancel{ + Stage: pb.CallCancel_AWAITING_UPSTREAM_HEADERS, Cause: pb.CallCancel_CANCELED_BY_SERVER, + }}}) + _, outcome := dispErr.AwaitAgentReport() + assert.Equal(t, AgentReported, outcome) + assert.False(t, handle.Busy(), "released once the agent answered") +}