diff --git a/agent/server/grpctunnel/call_handler.go b/agent/server/grpctunnel/call_handler.go index 5763fa6..cce3037 100644 --- a/agent/server/grpctunnel/call_handler.go +++ b/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) { diff --git a/agent/server/grpctunnel/tunnel_client.go b/agent/server/grpctunnel/tunnel_client.go index 8f04121..406b527 100644 --- a/agent/server/grpctunnel/tunnel_client.go +++ b/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() } diff --git a/agent/server/grpctunnel/tunnel_client_test.go b/agent/server/grpctunnel/tunnel_client_test.go index a88cbea..018d136 100644 --- a/agent/server/grpctunnel/tunnel_client_test.go +++ b/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 diff --git a/server/adapters/http.go b/server/adapters/http.go index 3fbf5fc..852e48c 100644 --- a/server/adapters/http.go +++ b/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), diff --git a/server/broker/broker_server_client.go b/server/broker/broker_server_client.go index a29d842..0682d9b 100644 --- a/server/broker/broker_server_client.go +++ b/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), diff --git a/server/tunnel/client_registry.go b/server/tunnel/client_registry.go index 3384e34..37050f4 100644 --- a/server/tunnel/client_registry.go +++ b/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), diff --git a/server/tunnel/service.go b/server/tunnel/service.go index 87b3574..ee37d22 100644 --- a/server/tunnel/service.go +++ b/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), )