diff --git a/rpc/logging.go b/rpc/logging.go index 968cab3c6..5f41a6877 100644 --- a/rpc/logging.go +++ b/rpc/logging.go @@ -16,6 +16,7 @@ package rpc import ( "context" + "errors" "time" "github.com/puzpuzpuz/xsync/v4" @@ -25,6 +26,23 @@ import ( "github.com/livekit/psrpc" ) +// clientLocalError reports whether an RPC error surfaced on the client without +// the server handler having produced (and logged) it: transport, timeout, +// selection, and response-decoding failures. Errors returned by the server +// handler are logged server-side with full request/response detail, so +// clients only need to log them at debug. +func clientLocalError(err error) bool { + var pe psrpc.Error + if !errors.As(err, &pe) { + return true + } + switch pe.Code() { + case psrpc.Canceled, psrpc.DeadlineExceeded, psrpc.Unavailable, psrpc.MalformedResponse: + return true + } + return false +} + // maxPayloadLogSize is the wire size in bytes above which RPC payloads are // logged as a summary. Encoded JSON can be several times larger than the wire // size; this keeps log lines within collector line limits. @@ -67,8 +85,10 @@ func newClientRPCLoggerInterceptor(l logger.Logger) psrpc.ClientRPCInterceptor { return func(ctx context.Context, req proto.Message, opts ...psrpc.RequestOption) (res proto.Message, err error) { start := time.Now() defer func() { - if err != nil { + if err != nil && clientLocalError(err) { l.Warnw("client error", err, "topic", rpcInfo.Topic, "request", logger.ProtoWithLimit(req, maxPayloadLogSize), "response", logger.ProtoWithLimit(res, maxPayloadLogSize), "duration", time.Since(start)) + } else if err != nil { + l.Debugw("client error", "error", err, "topic", rpcInfo.Topic, "request", logger.ProtoWithLimit(req, maxPayloadLogSize), "duration", time.Since(start)) } else { l.Debugw("client response", "topic", rpcInfo.Topic, "request", logger.ProtoWithLimit(req, maxPayloadLogSize), "response", logger.ProtoWithLimit(res, maxPayloadLogSize), "duration", time.Since(start)) } @@ -150,6 +170,7 @@ type multiRPCLoggerInterceptor struct { start time.Time responseCount int errorCount int + lastErr error } func (r *multiRPCLoggerInterceptor) Send(ctx context.Context, req proto.Message, opts ...psrpc.RequestOption) error { @@ -160,8 +181,11 @@ func (r *multiRPCLoggerInterceptor) Send(ctx context.Context, req proto.Message, func (r *multiRPCLoggerInterceptor) Recv(msg proto.Message, err error) { if err != nil { - r.logger.Warnw("received error", err) + // individual errors are summarized in Close to avoid one warn line + // per responding node on fan-out RPCs + r.logger.Debugw("received error", "error", err) r.errorCount++ + r.lastErr = err } else { r.logger.Debugw("received response", "response", logger.ProtoWithLimit(msg, maxPayloadLogSize)) r.responseCount++ @@ -170,6 +194,10 @@ func (r *multiRPCLoggerInterceptor) Recv(msg proto.Message, err error) { } func (r *multiRPCLoggerInterceptor) Close() { - r.logger.Debugw("multirpc closed", "responseCount", r.responseCount, "errorCount", r.errorCount) + if r.errorCount > 0 { + r.logger.Warnw("multirpc closed with errors", r.lastErr, "responseCount", r.responseCount, "errorCount", r.errorCount, "duration", time.Since(r.start)) + } else { + r.logger.Debugw("multirpc closed", "responseCount", r.responseCount, "errorCount", r.errorCount) + } r.ClientMultiRPCHandler.Close() }