grpctunnel: explain every dispatch failure (phase, stream state, agent report)

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

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},
 	}}})
 }
 

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
+}

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)
+}