Cap RPC payload logging with size-limited proto summaries

logger/proto.go

@@ -53,20 +53,42 @@ func UnredactedProto(val proto.Message) zapcore.ObjectMarshaler {
 	return protoMarshaller{m: val.ProtoReflect(), maxLevel: logger.Sensitivity_SENSITIVITY_PII}
 }
 
+// ProtoWithLimit behaves like Proto, but when the message's wire size exceeds
+// maxBytes it logs a compact summary instead of the full contents: message
+// type, wire size, and top-level scalar fields (redacted per sensitivity,
+// long strings truncated), with repeated/map fields reduced to counts and
+// nested messages dropped. Use for messages of unbounded size, such as RPC
+// payloads.
+func ProtoWithLimit(val proto.Message, maxBytes int) zapcore.ObjectMarshaler {
+	if val == nil {
+		return nil
+	}
+	return protoMarshaller{m: val.ProtoReflect(), maxLevel: logger.Sensitivity_SENSITIVITY_UNSPECIFIED, maxSize: maxBytes}
+}
+
 var _ zapcore.ObjectMarshaler = protoMarshaller{}
 var _ zapcore.ObjectMarshaler = protoMapMarshaller{}
 var _ zapcore.ArrayMarshaler = protoListMarshaller{}
 
 type protoMarshaller struct {
 	m        protoreflect.Message
 	maxLevel logger.Sensitivity
+	// maxSize is the wire size in bytes above which the message is logged as
+	// a summary instead of its full contents. 0 means no limit.
+	maxSize int
 }
 
 func (p protoMarshaller) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	if !p.m.IsValid() {
 		return nil
 	}
 
+	if p.maxSize > 0 {
+		if size := proto.Size(p.m.Interface()); size > p.maxSize {
+			return p.marshalSummary(e, size)
+		}
+	}
+
 	fields := p.m.Descriptor().Fields()
 	for i := 0; i < fields.Len(); i++ {
 		f := fields.Get(i)
@@ -100,6 +122,48 @@ func (p protoMarshaller) MarshalLogObject(e zapcore.ObjectEncoder) error {
 	return nil
 }
 
+// marshalSummary keeps top-level scalar fields, which do not meaningfully
+// contribute to message size, and reduces lists, maps, and long strings to
+// counts and sizes.
+func (p protoMarshaller) marshalSummary(e zapcore.ObjectEncoder, size int) error {
+	e.AddString("truncatedProto", string(p.m.Descriptor().FullName()))
+	e.AddInt("protoBytes", size)
+
+	fields := p.m.Descriptor().Fields()
+	for i := 0; i < fields.Len(); i++ {
+		f := fields.Get(i)
+		v := p.m.Get(f)
+		if protoFieldIsZero(f, v) {
+			continue
+		}
+		k := f.JSONName()
+		if proto.HasExtension(f.Options(), logger.E_Name) {
+			k = proto.GetExtension(f.Options(), logger.E_Name).(string)
+		}
+		switch {
+		case f.IsMap():
+			e.AddInt(k+"Count", v.Map().Len())
+		case f.IsList():
+			e.AddInt(k+"Count", v.List().Len())
+		case f.Kind() == protoreflect.MessageKind, f.Kind() == protoreflect.GroupKind:
+		case fieldSensitivity(f) > p.maxLevel:
+			e.AddString(k, marshalRedacted(f, v))
+		case f.Kind() == protoreflect.StringKind:
+			e.AddString(k, marshalTruncatedString(v.String()))
+		default:
+			marshalProtoField(k, f, v, e, p.maxLevel)
+		}
+	}
+	return nil
+}
+
+func marshalTruncatedString(s string) string {
+	if len(s) <= 256 {
+		return s
+	}
+	return fmt.Sprintf("%s... (%d bytes)", s[:256], len(s))
+}
+
 type protoMapMarshaller struct {
 	f        protoreflect.FieldDescriptor
 	m        protoreflect.Map

logger/proto_test.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap/zapcore"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/livekit/protocol/livekit"
 	"github.com/livekit/protocol/logger"
@@ -149,6 +150,87 @@ func TestProtoNestedSecretAlwaysRedacted(t *testing.T) {
 	}
 }
 
+func TestProtoWithLimitUnderLimit(t *testing.T) {
+	msg := &livekit.ListEgressResponse{
+		Items: []*livekit.EgressInfo{
+			{EgressId: "EG_1", RoomName: "room-1"},
+			{EgressId: "EG_2", RoomName: "room-2"},
+		},
+	}
+
+	fields := marshalFields(t, logger.ProtoWithLimit(msg, 1<<20))
+	require.NotContains(t, fields, "truncatedProto")
+	require.Len(t, fields["items"].([]any), 2)
+}
+
+func TestProtoWithLimitSummary(t *testing.T) {
+	msg := &livekit.ListEgressResponse{
+		Items: []*livekit.EgressInfo{
+			{EgressId: "EG_1", RoomName: "room-1"},
+			{EgressId: "EG_2", RoomName: "room-2"},
+			{EgressId: "EG_3", RoomName: "room-3"},
+		},
+		NextPageToken: &livekit.TokenPagination{Token: "next"},
+	}
+
+	fields := marshalFields(t, logger.ProtoWithLimit(msg, 8))
+	require.Equal(t, "livekit.ListEgressResponse", fields["truncatedProto"])
+	require.Equal(t, proto.Size(msg), fields["protoBytes"])
+	require.Equal(t, 3, fields["itemsCount"])
+
+	// Lists become counts and nested messages are dropped; nothing else at the
+	// root of this message, so the summary is exactly these three keys.
+	require.NotContains(t, fields, "items")
+	require.NotContains(t, fields, "nextPageToken")
+	require.Len(t, fields, 3)
+}
+
+func TestProtoWithLimitSummaryFieldCounts(t *testing.T) {
+	msg := &livekit.ParticipantInfo{
+		Identity:   "user-123",
+		Attributes: map[string]string{"a": "1", "b": "2"},
+		Tracks:     []*livekit.TrackInfo{{Sid: "TR1"}},
+	}
+
+	fields := marshalFields(t, logger.ProtoWithLimit(msg, 1))
+	require.Equal(t, "livekit.ParticipantInfo", fields["truncatedProto"])
+	require.Equal(t, 2, fields["attributesCount"])
+	require.Equal(t, 1, fields["tracksCount"])
+
+	// Scalars are preserved; collection contents are not.
+	require.Equal(t, "user-123", fields["identity"])
+	require.NotContains(t, fields, "attributes")
+	require.NotContains(t, fields, "tracks")
+}
+
+func TestProtoWithLimitSummaryScalars(t *testing.T) {
+	msg := &livekit.S3Upload{
+		AccessKey: "AKIAEXAMPLE",
+		Region:    "us-east-1",
+		Bucket:    strings.Repeat("b", 300),
+	}
+
+	fields := marshalFields(t, logger.ProtoWithLimit(msg, 1))
+	require.Equal(t, "livekit.S3Upload", fields["truncatedProto"])
+	require.Equal(t, "us-east-1", fields["region"])
+
+	// Sensitivity redaction applies in the summary.
+	require.Equal(t, "<redacted>", fields["accessKey"])
+
+	// Long strings are capped with a size marker.
+	bucket := fields["bucket"].(string)
+	require.True(t, strings.HasPrefix(bucket, strings.Repeat("b", 256)))
+	require.Contains(t, bucket, "(300 bytes)")
+}
+
+func TestProtoWithLimitNilSafe(t *testing.T) {
+	require.Nil(t, logger.ProtoWithLimit(nil, 1))
+
+	require.NotPanics(t, func() {
+		marshalFields(t, logger.ProtoWithLimit((*livekit.S3Upload)(nil), 1))
+	})
+}
+
 func TestProtoNilSafe(t *testing.T) {
 	require.Nil(t, logger.Proto(nil))
 	require.Nil(t, logger.UnredactedProto(nil))

rpc/logging.go

@@ -25,6 +25,11 @@ import (
 	"github.com/livekit/psrpc"
 )
 
+// 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.
+const maxPayloadLogSize = 32 << 10
+
 type loggerCache struct {
 	m *xsync.Map[string, logger.Logger]
 }
@@ -63,9 +68,9 @@ func newClientRPCLoggerInterceptor(l logger.Logger) psrpc.ClientRPCInterceptor {
 			start := time.Now()
 			defer func() {
 				if err != nil {
-					l.Warnw("client error", err, "topic", rpcInfo.Topic, "request", logger.Proto(req), "response", logger.Proto(res), "duration", time.Since(start))
+					l.Warnw("client error", err, "topic", rpcInfo.Topic, "request", logger.ProtoWithLimit(req, maxPayloadLogSize), "response", logger.ProtoWithLimit(res, maxPayloadLogSize), "duration", time.Since(start))
 				} else {
-					l.Debugw("client response", "topic", rpcInfo.Topic, "request", logger.Proto(req), "response", logger.Proto(res), "duration", time.Since(start))
+					l.Debugw("client response", "topic", rpcInfo.Topic, "request", logger.ProtoWithLimit(req, maxPayloadLogSize), "response", logger.ProtoWithLimit(res, maxPayloadLogSize), "duration", time.Since(start))
 				}
 			}()
 			return next(ctx, req, opts...)
@@ -80,9 +85,9 @@ func newServerRPCLoggerInterceptor(l logger.Logger) psrpc.ServerRPCInterceptor {
 		start := time.Now()
 		defer func() {
 			if err != nil {
-				l.Warnw("server error", err, "topic", rpcInfo.Topic, "request", logger.Proto(req), "response", logger.Proto(res), "duration", time.Since(start))
+				l.Warnw("server error", err, "topic", rpcInfo.Topic, "request", logger.ProtoWithLimit(req, maxPayloadLogSize), "response", logger.ProtoWithLimit(res, maxPayloadLogSize), "duration", time.Since(start))
 			} else {
-				l.Debugw("server response", "topic", rpcInfo.Topic, "request", logger.Proto(req), "response", logger.Proto(res), "duration", time.Since(start))
+				l.Debugw("server response", "topic", rpcInfo.Topic, "request", logger.ProtoWithLimit(req, maxPayloadLogSize), "response", logger.ProtoWithLimit(res, maxPayloadLogSize), "duration", time.Since(start))
 			}
 		}()
 		return handler(ctx, req)
@@ -107,17 +112,17 @@ type streamLoggerInterceptor struct {
 }
 
 func (s *streamLoggerInterceptor) Recv(msg proto.Message) (err error) {
-	s.logger.Debugw("received message", "message", logger.Proto(msg))
+	s.logger.Debugw("received message", "message", logger.ProtoWithLimit(msg, maxPayloadLogSize))
 	return s.StreamHandler.Recv(msg)
 }
 
 func (s *streamLoggerInterceptor) Send(msg proto.Message, opts ...psrpc.StreamOption) (err error) {
 	start := time.Now()
 	defer func() {
 		if err != nil {
-			s.logger.Warnw("failed to send message", err, "message", logger.Proto(msg), "duration", time.Since(start))
+			s.logger.Warnw("failed to send message", err, "message", logger.ProtoWithLimit(msg, maxPayloadLogSize), "duration", time.Since(start))
 		} else {
-			s.logger.Debugw("sent message", "message", logger.Proto(msg), "duration", time.Since(start))
+			s.logger.Debugw("sent message", "message", logger.ProtoWithLimit(msg, maxPayloadLogSize), "duration", time.Since(start))
 		}
 	}()
 	return s.StreamHandler.Send(msg, opts...)
@@ -149,7 +154,7 @@ type multiRPCLoggerInterceptor struct {
 
 func (r *multiRPCLoggerInterceptor) Send(ctx context.Context, req proto.Message, opts ...psrpc.RequestOption) error {
 	r.start = time.Now()
-	r.logger.Debugw("multirpc opened", "request", logger.Proto(req))
+	r.logger.Debugw("multirpc opened", "request", logger.ProtoWithLimit(req, maxPayloadLogSize))
 	return r.ClientMultiRPCHandler.Send(ctx, req, opts...)
 }
 
@@ -158,7 +163,7 @@ func (r *multiRPCLoggerInterceptor) Recv(msg proto.Message, err error) {
 		r.logger.Warnw("received error", err)
 		r.errorCount++
 	} else {
-		r.logger.Debugw("received response", "response", logger.Proto(msg))
+		r.logger.Debugw("received response", "response", logger.ProtoWithLimit(msg, maxPayloadLogSize))
 		r.responseCount++
 	}
 	r.ClientMultiRPCHandler.Recv(msg, err)