From d95b27a0cef8001ae0713c6281aa55b29e8b6c66 Mon Sep 17 00:00:00 2001 From: Milos Pesic Date: Wed, 15 Jul 2026 15:37:03 +0200 Subject: [PATCH 1/2] Cap RPC payload logging with size-limited proto summaries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit psrpc access logs marshal full request/response protos. Unbounded payloads (e.g. large ListEgressResponse pages) produce log lines that exceed collector line limits, getting split into unparseable chunks. Add logger.ProtoWithLimit: when a message's wire size exceeds the limit, log a summary instead — message type, wire size, top-level scalars (sensitivity redaction applied, long strings capped), counts for repeated/map fields. Use it in the psrpc logging interceptors with a 32KB cap. Co-Authored-By: Claude Fable 5 --- logger/proto.go | 64 ++++++++++++++++++++++++++++++++++ logger/proto_test.go | 82 ++++++++++++++++++++++++++++++++++++++++++++ rpc/logging.go | 23 ++++++++----- 3 files changed, 160 insertions(+), 9 deletions(-) diff --git a/logger/proto.go b/logger/proto.go index 65e2d797b..8dd02c067 100644 --- a/logger/proto.go +++ b/logger/proto.go @@ -53,6 +53,19 @@ 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{} @@ -60,6 +73,9 @@ 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 { @@ -67,6 +83,12 @@ func (p protoMarshaller) MarshalLogObject(e zapcore.ObjectEncoder) error { 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 diff --git a/logger/proto_test.go b/logger/proto_test.go index 5cf19bb6e..affb7b597 100644 --- a/logger/proto_test.go +++ b/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, int64(proto.Size(msg)), fields["protoBytes"]) + require.Equal(t, int64(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, int64(2), fields["attributesCount"]) + require.Equal(t, int64(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, "", 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)) diff --git a/rpc/logging.go b/rpc/logging.go index dca59815f..968cab3c6 100644 --- a/rpc/logging.go +++ b/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,7 +112,7 @@ 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) } @@ -115,9 +120,9 @@ func (s *streamLoggerInterceptor) Send(msg proto.Message, opts ...psrpc.StreamOp 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) From bf82616873a2ec5d0b921225069d579163209a6f Mon Sep 17 00:00:00 2001 From: Milos Pesic Date: Wed, 15 Jul 2026 15:44:34 +0200 Subject: [PATCH 2/2] Fix int type assertions in ProtoWithLimit tests zapcore.MapObjectEncoder stores AddInt values as int, not int64. Co-Authored-By: Claude Fable 5 --- logger/proto_test.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/logger/proto_test.go b/logger/proto_test.go index affb7b597..7fca72881 100644 --- a/logger/proto_test.go +++ b/logger/proto_test.go @@ -175,8 +175,8 @@ func TestProtoWithLimitSummary(t *testing.T) { fields := marshalFields(t, logger.ProtoWithLimit(msg, 8)) require.Equal(t, "livekit.ListEgressResponse", fields["truncatedProto"]) - require.Equal(t, int64(proto.Size(msg)), fields["protoBytes"]) - require.Equal(t, int64(3), fields["itemsCount"]) + 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. @@ -194,8 +194,8 @@ func TestProtoWithLimitSummaryFieldCounts(t *testing.T) { fields := marshalFields(t, logger.ProtoWithLimit(msg, 1)) require.Equal(t, "livekit.ParticipantInfo", fields["truncatedProto"]) - require.Equal(t, int64(2), fields["attributesCount"]) - require.Equal(t, int64(1), fields["tracksCount"]) + 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"])