grpctunnel: log routine reconnects quietly, real failures at Warn/Error

agent/server/grpctunnel/call_handler.go

@@ -258,14 +258,7 @@ func (tc *tunnelClient) runCall(ctx context.Context, sc *streamCtx, c *agentCall
 		bodyR.CloseWithError(err)
 		tc.requestsTotal.WithLabelValues(method, "502").Inc()
 		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{
+		tc.logger.Log(callFailureLevel(cause), "Call execution failed", append([]zap.Field{
 			zap.String("callId", callID),
 			zap.String("method", method),
 			zap.String("targetURL", breq.URL.String()),
@@ -284,14 +277,26 @@ func (tc *tunnelClient) runCall(ctx context.Context, sc *streamCtx, c *agentCall
 	// anyone asks for when a call is slow or comes back wrong. Deferred so
 	// it is still written when the response body or a send fails partway
 	// through, with the latency that had elapsed by then.
+	//
+	// A response that did not reach CallEnd -- the upstream body or a send
+	// to the server failed partway -- is still logged here, but at Warn and
+	// marked aborted: its status is what the upstream started with, not what
+	// the caller got.
+	completed := false
 	defer func() {
-		tc.logger.Info("Request completed",
+		level := zapcore.InfoLevel
+		fields := []zap.Field{
 			zap.String("callId", callID),
 			zap.String("method", method),
 			zap.String("targetURL", breq.URL.String()),
 			zap.Int("status", resp.StatusCode),
 			zap.Int64("durationMs", time.Since(startTime).Milliseconds()),
-		)
+		}
+		if !completed {
+			level = zapcore.WarnLevel
+			fields = append(fields, zap.Bool("aborted", true))
+		}
+		tc.logger.Log(level, "Request completed", fields...)
 	}()
 
 	// Response start.
@@ -340,7 +345,7 @@ func (tc *tunnelClient) runCall(ctx context.Context, sc *streamCtx, c *agentCall
 		}
 		if readErr != nil {
 			cause := tc.stopCause(ctx, c)
-			tc.logger.Warn("Response body read failed mid-stream", append([]zap.Field{
+			tc.logger.Log(callFailureLevel(cause), "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))
@@ -362,10 +367,26 @@ func (tc *tunnelClient) runCall(ctx context.Context, sc *streamCtx, c *agentCall
 	}}}); err != nil {
 		return
 	}
+	completed = true
 
 	tc.requestDuration.WithLabelValues(method).Observe(float64(time.Since(startTime).Milliseconds()))
 }
 
+// callFailureLevel is the level for a call that failed with cause. 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. A call cut off by the agent shutting
+// down is expected.
+func callFailureLevel(cause pb.CallCancel_Cause) zapcore.Level {
+	switch cause {
+	case pb.CallCancel_UPSTREAM_ERROR:
+		return zapcore.ErrorLevel
+	case pb.CallCancel_AGENT_SHUTDOWN:
+		return zapcore.InfoLevel
+	}
+	return zapcore.WarnLevel
+}
+
 // 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) {

agent/server/grpctunnel/tunnel_client.go

@@ -28,6 +28,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/fx"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
@@ -538,7 +539,9 @@ func (tc *tunnelClient) startAsync() {
 	}
 
 	if err := tc.initialRegister(); err != nil {
-		tc.logger.Warn("Initial registration aborted", zap.Error(err))
+		// initialRegister retries until it succeeds, so it only returns an
+		// error when the client is shutting down.
+		tc.logger.Info("Initial registration aborted", zap.Error(err))
 		return
 	}
 
@@ -617,7 +620,6 @@ func (tc *tunnelClient) resolveRegistration() (token, serverAddr string, fromEnv
 	if envServerURL != "" {
 		serverAddr = envServerURL
 	}
-	tc.logger.Info("Registered with Cortex API", zap.String("serverUri", serverAddr))
 	return regInfo.Token, serverAddr, false, nil
 }
 
@@ -662,6 +664,8 @@ func (tc *tunnelClient) initialRegister() error {
 		if fromEnv {
 			tc.logger.Info("Using direct connection config (skipping registration)",
 				zap.String("serverUrl", serverAddr))
+		} else {
+			tc.logger.Info("Registered with Cortex API", zap.String("serverUri", serverAddr))
 		}
 		tc.storeRegistration(token, serverAddr)
 		return nil
@@ -711,6 +715,10 @@ func (tc *tunnelClient) reregister(reason string, force bool) {
 		// Direct-config mode — nothing to refresh.
 		return
 	}
+	// Debug: a refresh that changes nothing is routine. A rotation is the
+	// event worth seeing, and it is logged at Info below.
+	tc.logger.Debug("Registered with Cortex API",
+		zap.String("serverUri", serverAddr), zap.String("reason", reason))
 
 	if tc.storeRegistration(token, serverAddr) {
 		tc.tokenRotationsTotal.Inc()
@@ -1403,8 +1411,18 @@ func (tc *tunnelClient) manageStream(id int) {
 		}
 
 		recovering = true
-		tc.logger.Warn("Tunnel slot stream ended; will retry",
-			zap.Int("slot", id), zap.Error(err), zap.Duration("nextBackoff", backoff))
+		// A stream that got through the handshake and later dropped is
+		// routine -- a server restart, a load balancer reset, the 24h
+		// connection limit -- and they come in batches of every stream on
+		// the connection, so they are Info. A stream that could not be
+		// opened at all is the failure to chase, and stays a warning.
+		level := zapcore.WarnLevel
+		if established {
+			level = zapcore.InfoLevel
+		}
+		tc.logger.Log(level, "Tunnel slot stream ended; will retry",
+			zap.Int("slot", id), zap.Bool("established", established),
+			zap.Error(err), zap.Duration("nextBackoff", backoff))
 	}
 	tc.workerExited()
 }

agent/server/grpctunnel/tunnel_client_test.go

@@ -1084,7 +1084,7 @@ func TestManageStream_BackoffResetsAfterEstablishedStream(t *testing.T) {
 	defer stop()
 
 	tc, _ := newTestClient(t, config.AgentConfig{}, &countingRegistration{})
-	core, logs := observer.New(zapcore.WarnLevel)
+	core, logs := observer.New(zapcore.InfoLevel)
 	tc.logger = zap.New(core)
 	tc.mu.Lock()
 	tc.currentToken, tc.currentServerAddr = "tok", addr
@@ -1095,11 +1095,27 @@ func TestManageStream_BackoffResetsAfterEstablishedStream(t *testing.T) {
 	waitFor(t, 5*time.Second, func() bool { return logs.FilterMessage(msg).Len() >= 4 })
 
 	var got []time.Duration
+	var levels []zapcore.Level
 	for _, e := range logs.FilterMessage(msg).All() {
 		got = append(got, e.ContextMap()["nextBackoff"].(time.Duration))
+		levels = append(levels, e.Level)
 	}
 	require.Equal(t, []time.Duration{time.Second, 2 * time.Second, 4 * time.Second, minBackoff}, got[:4],
 		"three failed opens must back off exponentially, and the drop of the stream that then opened must not")
+	require.Equal(t,
+		[]zapcore.Level{zapcore.WarnLevel, zapcore.WarnLevel, zapcore.WarnLevel, zapcore.InfoLevel}, levels[:4],
+		"a failed open is a warning; the drop of a stream that had been established is routine")
+}
+
+func TestCallFailureLevel(t *testing.T) {
+	for cause, want := range map[pb.CallCancel_Cause]zapcore.Level{
+		pb.CallCancel_UPSTREAM_ERROR:     zapcore.ErrorLevel,
+		pb.CallCancel_CANCELED_BY_SERVER: zapcore.WarnLevel,
+		pb.CallCancel_AGENT_TIMEOUT:      zapcore.WarnLevel,
+		pb.CallCancel_AGENT_SHUTDOWN:     zapcore.InfoLevel,
+	} {
+		require.Equal(t, want, callFailureLevel(cause), cause.String())
+	}
 }
 
 // All workers failing at once cross the threshold together; that is one

server/adapters/http.go

@@ -280,11 +280,12 @@ func (h *HttpAdapter) handleBrokerDispatch(w http.ResponseWriter, r *http.Reques
 	if copyErr != nil {
 		// Response already committed; nothing to send but log it. A failed
 		// write is the caller going away; anything else is the tunnel side.
-		side := "tunnel"
+		// A caller going away is routine, as in failureLevel.
+		side, level := "tunnel", zapcore.WarnLevel
 		if errors.Is(copyErr, errCallerWrite) {
-			side = "caller"
+			side, level = "caller", zapcore.InfoLevel
 		}
-		h.logger.Warn("Response body aborted mid-stream",
+		h.logger.Log(level, "Response body aborted mid-stream",
 			append(respFields,
 				zap.Int("status", statusCode),
 				zap.String("side", side),

server/broker/broker_server_client.go

@@ -282,7 +282,9 @@ func (c *Client) doRequest(ctx context.Context, method, path string, params url.
 		if !isRetryable(lastErr) {
 			return lastErr
 		}
-		c.logger.Warn("BROKER_SERVER request failed; will retry",
+		// Info: each caller logs a warning if every attempt fails, so a
+		// warning per attempt only repeats it.
+		c.logger.Info("BROKER_SERVER request failed; will retry",
 			zap.String("method", method),
 			zap.String("path", path),
 			zap.Int("attempt", attempt),

server/tunnel/client_registry.go

@@ -254,8 +254,7 @@ func (r *ClientRegistry) RegisterStream(token broker.Token, identity ClientIdent
 		stream.onRelease = r.makeOnRelease(key, stream)
 		existing.pushIdle(stream)
 		// Debug: the entry already existed, so the token's reachability here
-		// has not changed — only its stream count. Registered new client
-		// below is the transition, and stays at Info.
+		// has not changed — only its stream count.
 		r.logger.Debug("Added stream to existing client entry",
 			append(identity.LogFields(token),
 				zap.String("streamId", stream.StreamID),
@@ -281,7 +280,9 @@ func (r *ClientRegistry) RegisterStream(token broker.Token, identity ClientIdent
 	entry.pushIdle(stream)
 	r.entries[key] = entry
 
-	r.logger.Info("Registered new client",
+	// Debug: the service logs "Agent instance connected" at Info for the
+	// same event, with the stream counts and peer address.
+	r.logger.Debug("Registered new client",
 		append(identity.LogFields(token), zap.String("streamId", stream.StreamID))...,
 	)
 	return RegisterResult{
@@ -364,14 +365,15 @@ func (r *ClientRegistry) UnregisterStream(token broker.Token, streamID string) U
 	if len(entry.Streams) == 0 {
 		delete(r.entries, key)
 		res.TokenRemoved = true
-		r.logger.Info("Removed client entry (last stream closed)",
+		// Debug: the service logs "Agent instance disconnected" at Info for
+		// the same event, with the close reason.
+		r.logger.Debug("Removed client entry (last stream closed)",
 			append(stream.Identity.LogFields(token), zap.String("streamId", streamID))...,
 		)
 		return res
 	}
 
-	// Debug: streams remain, so the token is still reachable here. Losing the
-	// last one is the event that matters, and it is logged at Info above.
+	// Debug: streams remain, so the token is still reachable here.
 	r.logger.Debug("Removed stream from client entry",
 		append(stream.Identity.LogFields(token),
 			zap.String("streamId", streamID),

server/tunnel/service.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cortexapps/axon-server/metrics"
 	"github.com/google/uuid"
 	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/peer"
 	"google.golang.org/grpc/status"
@@ -437,11 +438,14 @@ func (s *Service) cleanupStream(token broker.Token, streamID string, stopwatch i
 			fields = append(fields, zap.String("grpcCode", code.String()))
 		}
 	}
+	// A transport error is Info, not Warn: a load balancer reset or an
+	// agent restart ends every stream on the connection at once, and the
+	// agent reconnects. A heartbeat timeout has its own warning.
 	switch {
 	case unreg.InstanceRemoved:
 		s.logger.Info("Agent instance disconnected", fields...)
 	case reason == closeTransportError:
-		s.logger.Warn("Tunnel stream closed", fields...)
+		s.logger.Info("Tunnel stream closed", fields...)
 	default:
 		s.logger.Debug("Tunnel stream closed", fields...)
 	}
@@ -484,15 +488,23 @@ func (s *Service) notifyClientConnected(token broker.Token, clientID, clientVers
 		if err == nil {
 			s.registry.SetBrokerServerRegistered(token)
 			// One per connection edge now rather than per stream, but still
-			// Debug: the operator-visible transition is "Registered new
-			// client" at Info. A failure is logged loudly below.
+			// Debug: the operator-visible transition is "Agent instance
+			// connected" at Info. A failure is logged loudly below.
 			s.logger.Debug("BROKER_SERVER client-connected succeeded",
 				zap.String("clientId", clientID),
 			)
 			return
 		}
 
-		s.logger.Warn("BROKER_SERVER client-connected exhausted retries, will try again",
+		// Warn while backing off; once the backoff is at its ceiling this
+		// has been failing for minutes and BROKER_SERVER still believes the
+		// token is offline, so it becomes an error.
+		level := zapcore.WarnLevel
+		if backoff >= s.notifyMaxBackoff {
+			level = zapcore.ErrorLevel
+		}
+		s.logger.Log(level, "BROKER_SERVER client-connected exhausted retries, will try again",
+			zap.String("tokenHash", token.Hashed()),
 			zap.Error(err),
 			zap.Duration("backoff", backoff),
 		)