[pull] main from livekit:main

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