Log large packets receive/send. (#4521)

* Log large packets receive/send.

Seeing cases of servers reporting need for segmentation/re-assembly of
packets. So, logging packet receive/send for RTP/RTCP to check if
anything is seeing more than 1400 byte packets.

* log downtrack RTCP too

pkg/rtc/mediatrack.go

@@ -301,16 +301,23 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe
 
 	var lastRR uint32
 	rtcpReader.OnPacket(func(bytes []byte) {
+		isLargePacket := len(bytes) > 1400
 		pkts, err := rtcp.Unmarshal(bytes)
 		if err != nil {
-			t.params.Logger.Errorw("could not unmarshal RTCP", err)
+			t.params.Logger.Errorw("could not unmarshal RTCP", err, "size", len(bytes))
 			return
 		}
 
 		for _, pkt := range pkts {
 			switch pkt := pkt.(type) {
 			case *rtcp.SourceDescription:
+				if isLargePacket {
+					t.params.Logger.Infow("large RTCP packet received with SDES", "size", len(bytes))
+				}
 			case *rtcp.SenderReport:
+				if isLargePacket {
+					t.params.Logger.Infow("large RTCP packet received with sender report", "size", len(bytes), "SSRC", pkt.SSRC)
+				}
 				if pkt.SSRC == uint32(track.SSRC()) {
 					buff.SetSenderReportData(&livekit.RTCPSenderReportState{
 						RtpTimestamp: pkt.RTPTime,
@@ -321,6 +328,9 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track sfu.TrackRe
 					})
 				}
 			case *rtcp.ExtendedReport:
+				if isLargePacket {
+					t.params.Logger.Infow("large RTCP packet received with extendedreport", "size", len(bytes))
+				}
 			rttFromXR:
 				for _, report := range pkt.Reports {
 					if rr, ok := report.(*rtcp.DLRRReportBlock); ok {

pkg/rtc/transport.go

@@ -1432,6 +1432,11 @@ func (t *PCTransport) GetICEConnectionType() types.ICEConnectionType {
 }
 
 func (t *PCTransport) WriteRTCP(pkts []rtcp.Packet) error {
+	// TODO-CLEANUP-PACKET-SIZE: remove after checking for large packets
+	raw, _ := rtcp.Marshal(pkts)
+	if len(raw) > 1400 {
+		t.params.Logger.Infow("large RTCP packet send", "size", len(raw))
+	}
 	return t.pc.WriteRTCP(pkts)
 }
 

pkg/sfu/buffer/buffer.go

@@ -138,9 +138,11 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili
 }
 
 // Write adds an RTP Packet, ordering is not guaranteed, newer packets may arrive later
-//
-//go:noinline
 func (b *Buffer) Write(pkt []byte) (n int, err error) {
+	if len(pkt) > 1400 {
+		b.logger.Infow("large RTP packet received", "size", len(pkt))
+	}
+
 	var rtpPacket rtp.Packet
 	err = rtpPacket.Unmarshal(pkt)
 	if err != nil {

pkg/sfu/downtrack.go

@@ -633,6 +633,9 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
 		d.writeStream = t.WriteStream()
 		if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrc).(*buffer.RTCPReader); rr != nil {
 			rr.OnPacket(func(pkt []byte) {
+				if len(pkt) > 1400 {
+					d.params.Logger.Infow("large RTCP packet received primary", "size", len(pkt))
+				}
 				d.handleRTCP(pkt)
 			})
 			d.rtcpReader = rr
@@ -640,6 +643,9 @@ func (d *DownTrack) Bind(t webrtc.TrackLocalContext) (webrtc.RTPCodecParameters,
 		if d.ssrcRTX != 0 {
 			if rr := d.params.BufferFactory.GetOrNew(packetio.RTCPBufferPacket, d.ssrcRTX).(*buffer.RTCPReader); rr != nil {
 				rr.OnPacket(func(pkt []byte) {
+					if len(pkt) > 1400 {
+						d.params.Logger.Infow("large RTCP packet received rtx", "size", len(pkt))
+					}
 					d.handleRTCPRTX(pkt)
 				})
 				d.rtcpReaderRTX = rr

pkg/sfu/pacer/base.go

@@ -76,6 +76,10 @@ func (b *Base) SendPacket(p *Packet) (int, error) {
 		return 0, err
 	}
 
+	if (p.HeaderSize + len(p.Payload)) > 1400 {
+		b.logger.Infow("large RTP packet send", "size", p.HeaderSize+len(p.Payload))
+	}
+
 	var written int
 	written, err = p.WriteStream.WriteRTP(p.Header, p.Payload)
 	if err != nil {