diff --git a/pkg/sfu/rtpstats/rtpstats_base.go b/pkg/sfu/rtpstats/rtpstats_base.go index c53197327..c06043559 100644 --- a/pkg/sfu/rtpstats/rtpstats_base.go +++ b/pkg/sfu/rtpstats/rtpstats_base.go @@ -428,32 +428,13 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime( firstTime := now - samplesDuration.Nanoseconds() adjustment = r.firstTime - firstTime - getFields := func() []any { - return []any{ - "startTime", time.Unix(0, r.startTime), - "nowTime", time.Unix(0, now), - "before", time.Unix(0, r.firstTime), - "after", time.Unix(0, firstTime), - "adjustment", time.Duration(adjustment), - "firstTimeAdjustment", r.firstTimeAdjustment, - "extNowTS", extNowTS, - "extStartTS", extStartTS, - "srData", WrappedRTCPSenderReportStateLogger{srData}, - "tsOffset", tsOffset, - "timeSinceReceive", timeSinceReceive, - "timeSinceFirst", timeSinceFirst, - "samplesDiff", samplesDiff, - "samplesDuration", samplesDuration, - } - } - if firstTime < r.firstTime { if adjustment > cFirstPacketTimeAdjustThreshold { err = errors.New("adjusting first packet time, too big, ignoring") - loggingFields = getFields() + loggingFields = r.firstPacketTimeAdjustLogFields(now, firstTime, adjustment, extNowTS, extStartTS, srData, tsOffset, timeSinceReceive, timeSinceFirst, samplesDuration, samplesDiff) } else { r.firstTimeAdjustment += time.Duration(adjustment) - r.logger.Debugw("adjusting first packet time", getFields()...) + r.logger.Debugw("adjusting first packet time", r.firstPacketTimeAdjustLogFields(now, firstTime, adjustment, extNowTS, extStartTS, srData, tsOffset, timeSinceReceive, timeSinceFirst, samplesDuration, samplesDiff)...) r.firstTime = firstTime } } @@ -461,6 +442,25 @@ func (r *rtpStatsBase) maybeAdjustFirstPacketTime( return } +func (r *rtpStatsBase) firstPacketTimeAdjustLogFields(now, firstTime int64, adjustment int64, extNowTS, extStartTS uint64, srData *livekit.RTCPSenderReportState, tsOffset uint64, timeSinceReceive, timeSinceFirst, samplesDuration time.Duration, samplesDiff int64) []any { + return []any{ + "startTime", time.Unix(0, r.startTime), + "nowTime", time.Unix(0, now), + "before", time.Unix(0, r.firstTime), + "after", time.Unix(0, firstTime), + "adjustment", time.Duration(adjustment), + "firstTimeAdjustment", r.firstTimeAdjustment, + "extNowTS", extNowTS, + "extStartTS", extStartTS, + "srData", WrappedRTCPSenderReportStateLogger{srData}, + "tsOffset", tsOffset, + "timeSinceReceive", timeSinceReceive, + "timeSinceFirst", timeSinceFirst, + "samplesDiff", samplesDiff, + "samplesDuration", samplesDuration, + } +} + func (r *rtpStatsBase) getPacketsSeenMinusPadding(extStartSN, extHighestSN uint64) uint64 { packetsSeen := r.getPacketsSeen(extStartSN, extHighestSN) if r.packetsPadding > packetsSeen { @@ -742,12 +742,14 @@ func (r *rtpStatsBase) getDrift(extStartTS, extHighestTS uint64) ( if r.srFirst != nil && r.srNewest != nil && r.srFirst.RtpTimestamp != r.srNewest.RtpTimestamp { rtpClockTicks := r.srNewest.RtpTimestampExt - r.srFirst.RtpTimestampExt - elapsed := mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time().Sub(mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time()) + srFirstTime := mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time() + srNewestTime := mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time() + elapsed := srNewestTime.Sub(srFirstTime) if elapsed.Seconds() > 0.0 { driftSamples := int64(rtpClockTicks - r.rtpConverter.ToRTPExt(elapsed)) ntpReportDrift = &livekit.RTPDrift{ - StartTime: timestamppb.New(mediatransportutil.NtpTime(r.srFirst.NtpTimestamp).Time()), - EndTime: timestamppb.New(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time()), + StartTime: timestamppb.New(srFirstTime), + EndTime: timestamppb.New(srNewestTime), Duration: elapsed.Seconds(), StartTimestamp: r.srFirst.RtpTimestampExt, EndTimestamp: r.srNewest.RtpTimestampExt, diff --git a/pkg/sfu/rtpstats/rtpstats_base_lite.go b/pkg/sfu/rtpstats/rtpstats_base_lite.go index 7fc20ddd9..b67ab245b 100644 --- a/pkg/sfu/rtpstats/rtpstats_base_lite.go +++ b/pkg/sfu/rtpstats/rtpstats_base_lite.go @@ -17,6 +17,7 @@ package rtpstats import ( "errors" "fmt" + "strings" "sync" "time" @@ -412,25 +413,18 @@ func (r *rtpStatsBaseLite) marshalLogObject(e zapcore.ObjectEncoder, packetsExpe e.AddFloat32("packetLostPercentage", float32(r.packetsLost)/float32(packetsExpected)*100.0) } - hasLoss := false - first := true - str := "[" + var sb strings.Builder for burst, count := range r.gapHistogram { if count == 0 { continue } - - hasLoss = true - - if !first { - str += ", " + if sb.Len() > 0 { + sb.WriteString(", ") } - first = false - str += fmt.Sprintf("%d:%d", burst+1, count) + sb.WriteString(fmt.Sprintf("%d:%d", burst+1, count)) } - str += "]" - if hasLoss { - e.AddString("gapHistogram", str) + if sb.Len() > 0 { + e.AddString("gapHistogram", "["+sb.String()+"]") } e.AddUint32("nacks", r.nacks) @@ -486,25 +480,14 @@ func (r *rtpStatsBaseLite) toProto(packetsExpected, packetsSeenMinusPadding, pac LastPli: timestamppb.New(time.Unix(0, r.lastPli)), } - gapsPresent := false for i := range len(r.gapHistogram) { if r.gapHistogram[i] == 0 { continue } - - gapsPresent = true - break - } - - if gapsPresent { - p.GapHistogram = make(map[int32]uint32, len(r.gapHistogram)) - for i := range len(r.gapHistogram) { - if r.gapHistogram[i] == 0 { - continue - } - - p.GapHistogram[int32(i+1)] = r.gapHistogram[i] + if p.GapHistogram == nil { + p.GapHistogram = make(map[int32]uint32, len(r.gapHistogram)) } + p.GapHistogram[int32(i+1)] = r.gapHistogram[i] } return p diff --git a/pkg/sfu/rtpstats/rtpstats_sender.go b/pkg/sfu/rtpstats/rtpstats_sender.go index e2a1d4590..d9ae74f77 100644 --- a/pkg/sfu/rtpstats/rtpstats_sender.go +++ b/pkg/sfu/rtpstats/rtpstats_sender.go @@ -21,12 +21,12 @@ import ( "time" "github.com/pion/rtcp" + "go.uber.org/zap" "go.uber.org/zap/zapcore" "google.golang.org/protobuf/types/known/timestamppb" "github.com/livekit/mediatransportutil" "github.com/livekit/protocol/livekit" - "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils/mono" ) @@ -40,6 +40,9 @@ const ( snInfoFlagOutOfOrder ) +// max number of receiver reports kept per sender snapshot for logging +const cMaxProcessedReceptionReports = 64 + type snInfo struct { pktSize uint16 hdrSize uint8 @@ -272,6 +275,70 @@ func (s *senderSnapshot) maybeUpdateMaxJitter(jitter float64) { // ------------------------------------------------------------------- +type senderUpdateLoggingFields struct { + packetTime int64 + extSequenceNumber uint64 + extTimestamp uint64 + marker bool + hdrSize int + payloadSize int + paddingSize int + gapSN int64 + gapTS int64 + timeSinceHighest int64 + rtpStats *RTPStatsSender +} + +func (sulf *senderUpdateLoggingFields) MarshalLogObject(e zapcore.ObjectEncoder) error { + if sulf != nil { + e.AddUint64("currSN", sulf.extSequenceNumber) + e.AddInt64("gapSN", sulf.gapSN) + e.AddUint64("currTS", sulf.extTimestamp) + e.AddInt64("gapTS", sulf.gapTS) + e.AddTime("packetTime", time.Unix(0, sulf.packetTime)) + e.AddDuration("timeSinceHighest", time.Duration(sulf.timeSinceHighest)) + e.AddBool("marker", sulf.marker) + e.AddInt("hdrSize", sulf.hdrSize) + e.AddInt("payloadSize", sulf.payloadSize) + e.AddInt("paddingSize", sulf.paddingSize) + e.AddObject("rtpStats", lockedRTPStatsSenderLogEncoder{sulf.rtpStats}) + } + return nil +} + +// ------------------------------------------------------------------- + +type senderReportLoggingFields struct { + srData *livekit.RTCPSenderReportState + publisherSRData *livekit.RTCPSenderReportState + tsOffset uint64 + reportTime int64 + reportTimeAdjusted int64 + nowNano int64 + nowRTPExt uint64 + rtpStats *RTPStatsSender +} + +func (srlf *senderReportLoggingFields) MarshalLogObject(e zapcore.ObjectEncoder) error { + if srlf != nil { + e.AddObject("curr", WrappedRTCPSenderReportStateLogger{srlf.srData}) + e.AddObject("feed", WrappedRTCPSenderReportStateLogger{srlf.publisherSRData}) + e.AddUint64("tsOffset", srlf.tsOffset) + e.AddTime("timeNow", time.Now()) + e.AddInt64("reportTime", srlf.reportTime) + e.AddInt64("reportTimeAdjusted", srlf.reportTimeAdjusted) + e.AddDuration("timeSinceHighest", time.Duration(srlf.nowNano-srlf.rtpStats.highestTime)) + e.AddDuration("timeSinceFirst", time.Duration(srlf.nowNano-srlf.rtpStats.firstTime)) + e.AddDuration("timeSincePublisherSRAdjusted", time.Duration(srlf.nowNano-srlf.publisherSRData.AtAdjusted)) + e.AddDuration("timeSincePublisherSR", time.Duration(srlf.nowNano-srlf.publisherSRData.At)) + e.AddUint64("nowRTPExt", srlf.nowRTPExt) + e.AddObject("rtpStats", lockedRTPStatsSenderLogEncoder{srlf.rtpStats}) + } + return nil +} + +// ------------------------------------------------------------------- + type rttMarker struct { ntpTime mediatransportutil.NtpTime sentAt time.Time @@ -444,21 +511,6 @@ func (r *RTPStatsSender) Update( pktSize := uint64(hdrSize + payloadSize + paddingSize) isDuplicate := false gapSN := int64(extSequenceNumber - r.extHighestSN) - ulgr := func() logger.UnlikelyLogger { - return r.logger.WithUnlikelyValues( - "currSN", extSequenceNumber, - "gapSN", gapSN, - "currTS", extTimestamp, - "gapTS", int64(extTimestamp-r.extHighestTS), - "packetTime", time.Unix(0, packetTime), - "timeSinceHighest", time.Duration(packetTime-r.highestTime), - "marker", marker, - "hdrSize", hdrSize, - "payloadSize", payloadSize, - "paddingSize", paddingSize, - "rtpStats", lockedRTPStatsSenderLogEncoder{r}, - ) - } if gapSN <= 0 { // duplicate OR out-of-order if payloadSize == 0 && extSequenceNumber < r.extStartSN && !r.params.IsRTX { // do not start on a padding only packet @@ -479,8 +531,22 @@ func (r *RTPStatsSender) Update( r.senderSnapshots[i].maybeReinit(r.extStartSN, extSequenceNumber) } - ulgr().Infow( + sulf := &senderUpdateLoggingFields{ + packetTime: packetTime, + extSequenceNumber: extSequenceNumber, + extTimestamp: extTimestamp, + marker: marker, + hdrSize: hdrSize, + payloadSize: payloadSize, + paddingSize: paddingSize, + gapSN: gapSN, + gapTS: int64(extTimestamp - r.extHighestTS), + timeSinceHighest: packetTime - r.highestTime, + rtpStats: r, + } + r.logger.Infow( "adjusting start sequence number", + zap.Inline(sulf), "snAfter", extSequenceNumber, "tsAfter", extTimestamp, ) @@ -504,8 +570,22 @@ func (r *RTPStatsSender) Update( if !isDuplicate && -gapSN >= cSequenceNumberLargeJumpThreshold { r.largeJumpNegativeCount++ if (r.largeJumpNegativeCount-1)%100 == 0 { - ulgr().Warnw( + sulf := &senderUpdateLoggingFields{ + packetTime: packetTime, + extSequenceNumber: extSequenceNumber, + extTimestamp: extTimestamp, + marker: marker, + hdrSize: hdrSize, + payloadSize: payloadSize, + paddingSize: paddingSize, + gapSN: gapSN, + gapTS: int64(extTimestamp - r.extHighestTS), + timeSinceHighest: packetTime - r.highestTime, + rtpStats: r, + } + r.logger.Warnw( "large sequence number gap negative", nil, + zap.Inline(sulf), "count", r.largeJumpNegativeCount, ) } @@ -514,8 +594,22 @@ func (r *RTPStatsSender) Update( if gapSN >= cSequenceNumberLargeJumpThreshold { r.largeJumpCount++ if (r.largeJumpCount-1)%100 == 0 { - ulgr().Warnw( + sulf := &senderUpdateLoggingFields{ + packetTime: packetTime, + extSequenceNumber: extSequenceNumber, + extTimestamp: extTimestamp, + marker: marker, + hdrSize: hdrSize, + payloadSize: payloadSize, + paddingSize: paddingSize, + gapSN: gapSN, + gapTS: int64(extTimestamp - r.extHighestTS), + timeSinceHighest: packetTime - r.highestTime, + rtpStats: r, + } + r.logger.Warnw( "large sequence number gap", nil, + zap.Inline(sulf), "count", r.largeJumpCount, ) } @@ -524,8 +618,22 @@ func (r *RTPStatsSender) Update( if extTimestamp < r.extHighestTS { r.timeReversedCount++ if (r.timeReversedCount-1)%100 == 0 { - ulgr().Warnw( + sulf := &senderUpdateLoggingFields{ + packetTime: packetTime, + extSequenceNumber: extSequenceNumber, + extTimestamp: extTimestamp, + marker: marker, + hdrSize: hdrSize, + payloadSize: payloadSize, + paddingSize: paddingSize, + gapSN: gapSN, + gapTS: int64(extTimestamp - r.extHighestTS), + timeSinceHighest: packetTime - r.highestTime, + rtpStats: r, + } + r.logger.Warnw( "time reversed", nil, + zap.Inline(sulf), "count", r.timeReversedCount, ) } @@ -544,8 +652,22 @@ func (r *RTPStatsSender) Update( } if extTimestamp < r.extStartTS { - ulgr().Infow( + sulf := &senderUpdateLoggingFields{ + packetTime: packetTime, + extSequenceNumber: extSequenceNumber, + extTimestamp: extTimestamp, + marker: marker, + hdrSize: hdrSize, + payloadSize: payloadSize, + paddingSize: paddingSize, + gapSN: gapSN, + gapTS: int64(extTimestamp - r.extHighestTS), + timeSinceHighest: packetTime - r.highestTime, + rtpStats: r, + } + r.logger.Infow( "adjusting start timestamp", + zap.Inline(sulf), "snAfter", extSequenceNumber, "tsAfter", extTimestamp, ) @@ -628,18 +750,16 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt r.lastRR = rr }() - timeSinceLastRR := func() time.Duration { - if r.lastRRTime != 0 { - return time.Duration(nowNano - r.lastRRTime) - } - return time.Duration(nowNano - r.startTime) + timeSinceLastRR := time.Duration(nowNano - r.startTime) + if r.lastRRTime != 0 { + timeSinceLastRR = time.Duration(nowNano - r.lastRRTime) } extReceivedRRSN := extHighestSNFromRR + (r.extStartSN & 0xFFFF_FFFF_FFFF_0000) if int64(r.extHighestSN-extReceivedRRSN) < 0 || int64(r.extHighestSN-extReceivedRRSN) > (1<<32) { r.logger.Infow( "receiver report runaway, dropping", - "timeSinceLastRR", timeSinceLastRR(), + "timeSinceLastRR", timeSinceLastRR, "receivedRR", rr, "extHighestSNFromRR", extHighestSNFromRR, "extReceivedRRSN", extReceivedRRSN, @@ -666,7 +786,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt // RTCP RR does not change. r.logger.Infow( "receiver report missed rollover, adjusting", - "timeSinceLastRR", timeSinceLastRR(), + "timeSinceLastRR", timeSinceLastRR, "receivedRR", rr, "extHighestSNFromRR", extHighestSNFromRR, "extReceivedRRSN", extReceivedRRSN, @@ -679,7 +799,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt } r.logger.Infow( "receiver report missed rollover, adjusted", - "timeSinceLastRR", timeSinceLastRR(), + "timeSinceLastRR", timeSinceLastRR, "receivedRR", rr, "extHighestSNFromRR", extHighestSNFromRR, "extReceivedRRSN", extReceivedRRSN, @@ -691,7 +811,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt // if remote adjusts somehow, roll back alignment r.logger.Infow( "receiver report caught up rollover, adjusting", - "timeSinceLastRR", timeSinceLastRR(), + "timeSinceLastRR", timeSinceLastRR, "receivedRR", rr, "extHighestSNFromRR", extHighestSNFromRR, "extReceivedRRSN", extReceivedRRSN, @@ -704,7 +824,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt } r.logger.Infow( "receiver report caught up rollover, adjusted", - "timeSinceLastRR", timeSinceLastRR(), + "timeSinceLastRR", timeSinceLastRR, "receivedRR", rr, "extHighestSNFromRR", extHighestSNFromRR, "extReceivedRRSN", extReceivedRRSN, @@ -715,7 +835,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt if r.extHighestSNFromRR > extHighestSNFromRR { r.logger.Debugw( "receiver report out-of-order, dropping", - "timeSinceLastRR", timeSinceLastRR(), + "timeSinceLastRR", timeSinceLastRR, "receivedRR", rr, "extHighestSNFromRR", extHighestSNFromRR, "extReceivedRRSN", extReceivedRRSN, @@ -780,7 +900,7 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt "metadata cache overflow", "senderSnapshotID", i+cFirstSnapshotID, "senderSnapshot", s, - "timeSinceLastRR", timeSinceLastRR(), + "timeSinceLastRR", timeSinceLastRR, "receivedRR", rr, "extReceivedRRSN", extReceivedRRSN, "packetsInInterval", extReceivedRRSN-s.receiverView.extLastRRSN, @@ -791,7 +911,13 @@ func (r *RTPStatsSender) UpdateFromReceiverReport(rr rtcp.ReceptionReport) (rtt } } s.receiverView.extLastRRSN = extReceivedRRSN - s.receiverView.processedReceptionReports = append(s.receiverView.processedReceptionReports, rr) + reports := &s.receiverView.processedReceptionReports + if len(*reports) >= cMaxProcessedReceptionReports { + copy(*reports, (*reports)[1:]) + (*reports)[len(*reports)-1] = rr + } else { + *reports = append(*reports, rr) + } } return @@ -874,32 +1000,28 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek Octets: octetCount, } - ulgr := func() logger.UnlikelyLogger { - return r.logger.WithUnlikelyValues( - "curr", WrappedRTCPSenderReportStateLogger{srData}, - "feed", WrappedRTCPSenderReportStateLogger{publisherSRData}, - "tsOffset", tsOffset, - "timeNow", mono.Now(), - "reportTime", time.Unix(0, reportTime), - "reportTimeAdjusted", time.Unix(0, reportTimeAdjusted), - "timeSinceHighest", time.Duration(nowNano-r.highestTime), - "timeSinceFirst", time.Duration(nowNano-r.firstTime), - "timeSincePublisherSRAdjusted", time.Duration(nowNano-publisherSRData.AtAdjusted), - "timeSincePublisherSR", time.Duration(nowNano-publisherSRData.At), - "nowRTPExt", nowRTPExt, - "rtpStats", lockedRTPStatsSenderLogEncoder{r}, - ) - } - if r.srNewest != nil && nowRTPExt >= r.srNewest.RtpTimestampExt && r.clockRate != 0 { - timeSinceLastReport := nowNTP.Time().Sub(mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time()) + nowTime := nowNTP.Time() + srNewestTime := mediatransportutil.NtpTime(r.srNewest.NtpTimestamp).Time() + timeSinceLastReport := nowTime.Sub(srNewestTime) rtpDiffSinceLastReport := nowRTPExt - r.srNewest.RtpTimestampExt windowClockRate := float64(rtpDiffSinceLastReport) / timeSinceLastReport.Seconds() if timeSinceLastReport.Seconds() > 0.2 && math.Abs(float64(r.clockRate)-windowClockRate) > 0.2*float64(r.clockRate) { r.clockSkewCount++ if (r.clockSkewCount-1)%100 == 0 { - ulgr().Infow( + srlf := &senderReportLoggingFields{ + srData: srData, + publisherSRData: publisherSRData, + tsOffset: tsOffset, + reportTime: reportTime, + reportTimeAdjusted: reportTimeAdjusted, + nowNano: nowNano, + nowRTPExt: nowRTPExt, + rtpStats: r, + } + r.logger.Infow( "sending sender report, clock skew", + zap.Inline(srlf), "timeSinceLastReport", timeSinceLastReport, "rtpDiffSinceLastReport", rtpDiffSinceLastReport, "windowClockRate", windowClockRate, @@ -912,7 +1034,20 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek if r.srNewest != nil && nowRTPExt < r.srNewest.RtpTimestampExt { // If report being generated is behind the last report, skip it. // Should not happen. - ulgr().Infow("sending sender report, out-of-order, skipping") + srlf := &senderReportLoggingFields{ + srData: srData, + publisherSRData: publisherSRData, + tsOffset: tsOffset, + reportTime: reportTime, + reportTimeAdjusted: reportTimeAdjusted, + nowNano: nowNano, + nowRTPExt: nowRTPExt, + rtpStats: r, + } + r.logger.Infow( + "sending sender report, out-of-order, skipping", + zap.Inline(srlf), + ) return nil } @@ -923,7 +1058,7 @@ func (r *RTPStatsSender) GetRtcpSenderReport(ssrc uint32, publisherSRData *livek r.rttMarker = rttMarker{ ntpTime: nowNTP, - sentAt: mono.Now(), + sentAt: time.Now(), } return &rtcp.SenderReport{