From 36289bbca7858044c39853d3d400ba76b1a8632a Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 8 Feb 2022 12:53:14 +0530 Subject: [PATCH] FPS (#410) * WIP commit * WIP commit * WIP commit * WIP commit * WIP commit * WIP commit * Clean up * Clean up * Store RTT in stats * spelling mistake * Make tests compile * Fix test compilation error * fix tests * clone * latest protocol --- go.mod | 2 +- go.sum | 4 +- pkg/rtc/mediatrack.go | 66 +--- pkg/rtc/mediatrackreceiver.go | 2 +- pkg/rtc/participant.go | 6 +- pkg/rtc/participant_internal_test.go | 10 +- pkg/rtc/types/interfaces.go | 2 +- .../typesfakes/fake_local_media_track.go | 20 +- pkg/sfu/buffer/buffer.go | 192 +++++++----- pkg/sfu/buffer/buffer_test.go | 56 ---- pkg/sfu/buffer/nack.go | 10 +- pkg/sfu/buffer/nack_test.go | 64 ++-- pkg/sfu/buffer/streamstats.go | 28 ++ pkg/sfu/connectionquality/connectionstats.go | 283 ++++++++++-------- pkg/sfu/connectionquality/mos.go | 25 +- pkg/sfu/downtrack.go | 144 +++++---- pkg/sfu/receiver.go | 113 +++++-- pkg/sfu/streamtrackermanager.go | 7 + pkg/telemetry/prometheus/packets.go | 2 +- pkg/telemetry/statsworker.go | 219 +++++++------- pkg/telemetry/telemetryserviceinternal.go | 27 +- .../telemetryserviceinternalevents.go | 1 - pkg/telemetry/test/telemetry_service_test.go | 269 +++++++++++++---- 23 files changed, 913 insertions(+), 639 deletions(-) create mode 100644 pkg/sfu/buffer/streamstats.go diff --git a/go.mod b/go.mod index 87cbeb4d6..704c10c5d 100644 --- a/go.mod +++ b/go.mod @@ -14,7 +14,7 @@ require ( github.com/google/wire v0.5.0 github.com/gorilla/websocket v1.4.2 github.com/hashicorp/golang-lru v0.5.4 - github.com/livekit/protocol v0.11.13-0.20220207221412-40df25091bcb + github.com/livekit/protocol v0.11.13-0.20220208070440-7f8399010124 github.com/magefile/mage v1.11.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 github.com/mitchellh/go-homedir v1.1.0 diff --git a/go.sum b/go.sum index 5bf565f8e..88e183bfd 100644 --- a/go.sum +++ b/go.sum @@ -132,8 +132,8 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/lithammer/shortuuid/v3 v3.0.6 h1:pr15YQyvhiSX/qPxncFtqk+v4xLEpOZObbsY/mKrcvA= github.com/lithammer/shortuuid/v3 v3.0.6/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts= -github.com/livekit/protocol v0.11.13-0.20220207221412-40df25091bcb h1:oAx83edonLEOfsZardevXlHqzut6dEBIH9fOU/ZxDfU= -github.com/livekit/protocol v0.11.13-0.20220207221412-40df25091bcb/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= +github.com/livekit/protocol v0.11.13-0.20220208070440-7f8399010124 h1:nZor+yFxIBKLdQGwwx6GvsnjZhmG9gcz9I2K2zdLiVI= +github.com/livekit/protocol v0.11.13-0.20220208070440-7f8399010124/go.mod h1:YoHW9YbWbPnuVsgwBB4hAINKT+V68jmfh9zXBSSn6Wg= github.com/magefile/mage v1.11.0 h1:C/55Ywp9BpgVVclD3lRnSYCwXTYxmSppIgLeDYlNuls= github.com/magefile/mage v1.11.0/go.mod h1:z5UZb/iS3GoOSn0JgWuiw7dxlurVYTu+/jHXqQg881A= github.com/mattn/go-runewidth v0.0.9 h1:Lm995f3rfxdpd6TSmuVCHVb/QhupuXlYr8sCI/QdE+0= diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 160318b01..2bde9016a 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -6,8 +6,6 @@ import ( "sync/atomic" "time" - "github.com/livekit/livekit-server/pkg/sfu/connectionquality" - "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/pion/rtcp" @@ -36,8 +34,6 @@ type MediaTrack struct { audioLevelMu sync.RWMutex audioLevel *AudioLevel - connectionStats *connectionquality.ConnectionStats - done chan struct{} *MediaTrackReceiver @@ -97,30 +93,6 @@ func NewMediaTrack(track *webrtc.TrackRemote, params MediaTrackParams) *MediaTra } }) - t.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{ - UpdateInterval: connectionQualityUpdateInterval, - CodecType: track.Kind(), - GetTotalBytes: func() uint64 { - receiver := t.Receiver() - if receiver != nil { - return receiver.(*sfu.WebRTCReceiver).GetTotalBytes() - } - - return 0 - }, - GetIsReducedQuality: func() bool { - publishing, expected := t.getNumUpTracks() - return publishing < expected - }, - Logger: t.params.Logger, - }) - t.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) { - t.params.Telemetry.TrackStats(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), stat) - }) - - t.AddOnClose(func() { - t.connectionStats.Close() - }) return t } @@ -147,23 +119,6 @@ func (t *MediaTrack) ToProto() *livekit.TrackInfo { return info } -func (t *MediaTrack) getNumUpTracks() (uint32, uint32) { - numExpected := atomic.LoadUint32(&t.numUpTracks) - - numSubscribedLayers := t.numSubscribedLayers() - if numSubscribedLayers < numExpected { - numExpected = numSubscribedLayers - } - - numPublishing := uint32(0) - receiver := t.Receiver() - if receiver != nil { - numPublishing = uint32(receiver.(*sfu.WebRTCReceiver).NumAvailableSpatialLayers()) - } - - return numPublishing, numExpected -} - // AddReceiver adds a new RTP receiver to the track func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.TrackRemote, twcc *twcc.Responder) { buff, rtcpReader := t.params.BufferFactory.GetBufferPair(uint32(track.SSRC())) @@ -171,7 +126,6 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra t.params.Logger.Errorw("could not retrieve buffer pair", nil) return } - buff.OnFeedback(t.handlePublisherFeedback) if t.Kind() == livekit.TrackType_AUDIO { t.audioLevelMu.Lock() @@ -225,13 +179,14 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra t.MediaTrackReceiver.Close() t.params.Telemetry.TrackUnpublished(context.Background(), t.PublisherID(), t.ToProto(), uint32(track.SSRC())) }) + wr.OnStatsUpdate(func(_ *sfu.WebRTCReceiver, stat *livekit.AnalyticsStat) { + t.params.Telemetry.TrackStats(livekit.StreamType_UPSTREAM, t.PublisherID(), t.ID(), stat) + }) t.params.Telemetry.TrackPublished(context.Background(), t.PublisherID(), t.ToProto()) t.buffer = buff t.MediaTrackReceiver.SetupReceiver(wr) - - t.connectionStats.Start() } t.lock.Unlock() @@ -272,14 +227,11 @@ func (t *MediaTrack) GetAudioLevel() (level uint8, active bool) { return t.audioLevel.GetLevel() } -func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) { - t.connectionStats.RTCPFeedback(packets, 0) +func (t *MediaTrack) GetConnectionScore() float32 { + receiver := t.Receiver() + if receiver == nil { + return 0.0 + } - // also look for sender reports - // feedback for the source RTCP - t.params.RTCPChan <- packets -} - -func (t *MediaTrack) GetConnectionScore() float64 { - return t.connectionStats.GetScore() + return receiver.(*sfu.WebRTCReceiver).GetConnectionScore() } diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index 280098a15..887b895b3 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -291,7 +291,7 @@ func (t *MediaTrackReceiver) GetQualityForDimension(width, height uint32) liveki return quality } -// handles max loss for audio packets +// handles max loss for audio streams func (t *MediaTrackReceiver) handleMaxLossFeedback(_ *sfu.DownTrack, report *rtcp.ReceiverReport) { t.downFracLostLock.Lock() for _, rr := range report.Reports { diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 541c8798d..d2ce97517 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -762,9 +762,9 @@ func (p *ParticipantImpl) GetConnectionQuality() *livekit.ConnectionQualityInfo } p.lock.RUnlock() - avgScore := 5.0 + avgScore := float32(5.0) if numTracks > 0 { - avgScore = totalScore / float64(numTracks) + avgScore = totalScore / float32(numTracks) } rating := connectionquality.Score2Rating(avgScore) @@ -1348,7 +1348,7 @@ func (p *ParticipantImpl) setTrackMuted(trackID livekit.TrackID, muted bool) { } } -func (p *ParticipantImpl) getPublisherConnectionQuality() (totalScore float64, numTracks int) { +func (p *ParticipantImpl) getPublisherConnectionQuality() (totalScore float32, numTracks int) { for _, pt := range p.GetPublishedTracks() { if pt.IsMuted() { continue diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index ded70e38a..61c4e5261 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -244,7 +244,7 @@ func TestMuteSetting(t *testing.T) { func TestConnectionQuality(t *testing.T) { // loss based score is currently a publisher method. - videoScore := func(loss float64, numPublishing, numRegistered uint32) float64 { + videoScore := func(loss float32, numPublishing, numRegistered uint32) float32 { var reducedQuality bool if numRegistered > 0 && numPublishing != numRegistered { reducedQuality = true @@ -252,7 +252,7 @@ func TestConnectionQuality(t *testing.T) { return connectionquality.VideoConnectionScore(loss, reducedQuality) } - testPublishedVideoTrack := func(loss float64, numPublishing, numRegistered uint32) *typesfakes.FakeLocalMediaTrack { + testPublishedVideoTrack := func(loss float32, numPublishing, numRegistered uint32) *typesfakes.FakeLocalMediaTrack { tr := &typesfakes.FakeLocalMediaTrack{} score := videoScore(loss, numPublishing, numRegistered) t.Log("video score: ", score) @@ -263,11 +263,11 @@ func TestConnectionQuality(t *testing.T) { testPublishedAudioTrack := func(totalPackets, packetsLost uint32) *typesfakes.FakeLocalMediaTrack { tr := &typesfakes.FakeLocalMediaTrack{} - pctLoss := float64(0.0) + pctLoss := float32(0.0) if totalPackets > 0 { - pctLoss = (float64(packetsLost) / float64(totalPackets)) * 100.0 + pctLoss = (float32(packetsLost) / float32(totalPackets)) * 100.0 } - score := connectionquality.AudioConnectionScore(pctLoss, 0) + score := connectionquality.AudioConnectionScore(pctLoss, 0, 0.0) t.Log("audio score: ", score) tr.GetConnectionScoreReturns(score) return tr diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 9482b61fe..fdfad3dac 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -211,7 +211,7 @@ type LocalMediaTrack interface { SdpCid() string GetAudioLevel() (level uint8, active bool) - GetConnectionScore() float64 + GetConnectionScore() float32 } // MediaTrack is the main interface representing a track published to the room diff --git a/pkg/rtc/types/typesfakes/fake_local_media_track.go b/pkg/rtc/types/typesfakes/fake_local_media_track.go index d91159979..085ca9eb9 100644 --- a/pkg/rtc/types/typesfakes/fake_local_media_track.go +++ b/pkg/rtc/types/typesfakes/fake_local_media_track.go @@ -38,15 +38,15 @@ type FakeLocalMediaTrack struct { result1 uint8 result2 bool } - GetConnectionScoreStub func() float64 + GetConnectionScoreStub func() float32 getConnectionScoreMutex sync.RWMutex getConnectionScoreArgsForCall []struct { } getConnectionScoreReturns struct { - result1 float64 + result1 float32 } getConnectionScoreReturnsOnCall map[int]struct { - result1 float64 + result1 float32 } GetQualityForDimensionStub func(uint32, uint32) livekit.VideoQuality getQualityForDimensionMutex sync.RWMutex @@ -387,7 +387,7 @@ func (fake *FakeLocalMediaTrack) GetAudioLevelReturnsOnCall(i int, result1 uint8 }{result1, result2} } -func (fake *FakeLocalMediaTrack) GetConnectionScore() float64 { +func (fake *FakeLocalMediaTrack) GetConnectionScore() float32 { fake.getConnectionScoreMutex.Lock() ret, specificReturn := fake.getConnectionScoreReturnsOnCall[len(fake.getConnectionScoreArgsForCall)] fake.getConnectionScoreArgsForCall = append(fake.getConnectionScoreArgsForCall, struct { @@ -411,32 +411,32 @@ func (fake *FakeLocalMediaTrack) GetConnectionScoreCallCount() int { return len(fake.getConnectionScoreArgsForCall) } -func (fake *FakeLocalMediaTrack) GetConnectionScoreCalls(stub func() float64) { +func (fake *FakeLocalMediaTrack) GetConnectionScoreCalls(stub func() float32) { fake.getConnectionScoreMutex.Lock() defer fake.getConnectionScoreMutex.Unlock() fake.GetConnectionScoreStub = stub } -func (fake *FakeLocalMediaTrack) GetConnectionScoreReturns(result1 float64) { +func (fake *FakeLocalMediaTrack) GetConnectionScoreReturns(result1 float32) { fake.getConnectionScoreMutex.Lock() defer fake.getConnectionScoreMutex.Unlock() fake.GetConnectionScoreStub = nil fake.getConnectionScoreReturns = struct { - result1 float64 + result1 float32 }{result1} } -func (fake *FakeLocalMediaTrack) GetConnectionScoreReturnsOnCall(i int, result1 float64) { +func (fake *FakeLocalMediaTrack) GetConnectionScoreReturnsOnCall(i int, result1 float32) { fake.getConnectionScoreMutex.Lock() defer fake.getConnectionScoreMutex.Unlock() fake.GetConnectionScoreStub = nil if fake.getConnectionScoreReturnsOnCall == nil { fake.getConnectionScoreReturnsOnCall = make(map[int]struct { - result1 float64 + result1 float32 }) } fake.getConnectionScoreReturnsOnCall[i] = struct { - result1 float64 + result1 float32 }{result1} } diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index 145f38af7..1a116aa6f 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -3,6 +3,7 @@ package buffer import ( "encoding/binary" "io" + "math/rand" "strings" "sync" "sync/atomic" @@ -36,7 +37,7 @@ type ExtPacket struct { // Buffer contains all packets type Buffer struct { - sync.Mutex + sync.RWMutex bucket *Bucket nacker *NackQueue videoPool *sync.Pool @@ -63,21 +64,20 @@ type Buffer struct { latestTSForAudioLevelInitialized bool latestTSForAudioLevel uint32 - lastPacketRead int - bitrate atomic.Value - bitrateHelper [4]int64 - lastSRNTPTime uint64 - lastSRRTPTime uint32 - lastSRRecv int64 // Represents wall clock of the most recent sender report arrival - cycle uint16 - lastRtcpPacketTime int64 // Time the last RTCP packet was received. - lastRtcpSrTime int64 // Time the last RTCP SR was received. Required for DLSR computation. - lastTransit uint32 + lastPacketRead int + bitrate atomic.Value + bitrateHelper [4]int64 + lastSRNTPTime uint64 + lastSRRTPTime uint32 + lastSRRecv int64 // Represents wall clock of the most recent sender report arrival + lastTransit uint32 - stats Stats + started bool + stats StreamStats rrSnapshot *receiverReportSnapshot highestSN uint16 + cycle uint16 lastFractionLostToReport uint8 // Last fraction lost from subscribers, should report to publisher; Audio only @@ -91,17 +91,10 @@ type Buffer struct { logger logger.Logger } -type Stats struct { - PacketCount uint32 // Number of packets received from this source. - TotalBytes uint64 - Jitter float64 // An estimate of the statistical variance of the RTP data packet inter-arrival time. -} - type receiverReportSnapshot struct { - extSeqNum uint32 - packetsReceived uint32 - packetsLost uint32 - lastLossRate float32 + extHighestSeqNum uint32 + packetsLost uint32 + lastLossRate float32 } // BufferOptions provides configuration options for the buffer @@ -164,7 +157,7 @@ func (b *Buffer) Bind(params webrtc.RTPParameters, codec webrtc.RTPCodecCapabili case webrtc.TypeRTCPFBNACK: b.logger.Debugw("Setting feedback", "type", webrtc.TypeRTCPFBNACK) b.nacker = NewNACKQueue() - b.nacker.SetRTT(70) // LK-TODO: sane default till we get better data + b.nacker.SetRTT(70) // default till it is updated b.nack = true } } @@ -271,6 +264,29 @@ func (b *Buffer) OnClose(fn func()) { b.onClose = fn } +func (b *Buffer) SendPLI() { + b.Lock() + b.stats.TotalPLIs++ + b.Unlock() + + pli := []rtcp.Packet{ + &rtcp.PictureLossIndication{SenderSSRC: rand.Uint32(), MediaSSRC: b.mediaSSRC}, + } + + b.feedbackCB(pli) +} + +func (b *Buffer) SetRTT(rtt uint32) { + b.Lock() + defer b.Unlock() + + b.stats.RTT = rtt + + if b.nacker != nil { + b.nacker.SetRTT(rtt) + } +} + func (b *Buffer) calc(pkt []byte, arrivalTime int64) { isRTX := false @@ -328,41 +344,58 @@ func (b *Buffer) calc(pkt []byte, arrivalTime int64) { func (b *Buffer) updateStreamState(p *rtp.Packet, pktSize int, arrivalTime int64, isRTX bool) { sn := p.SequenceNumber - if b.stats.PacketCount == 0 { - b.highestSN = sn - 1 + if !b.started { + b.started = true + b.highestSN = sn b.lastReport = arrivalTime b.rrSnapshot = &receiverReportSnapshot{ - extSeqNum: uint32(sn) - 1, - packetsReceived: 0, - packetsLost: 0, - lastLossRate: 0.0, - } - } - - diff := sn - b.highestSN - if diff > (1 << 15) { - // out-of-order, remove it from nack queue - if b.nacker != nil { - b.nacker.Remove(sn) + extHighestSeqNum: uint32(sn) - 1, + packetsLost: 0, + lastLossRate: 0.0, } } else { - if b.nacker != nil && diff > 1 { - for lost := b.highestSN + 1; lost != sn; lost++ { - b.nacker.Push(lost) + diff := sn - b.highestSN + if diff > (1 << 15) { + if !isRTX && b.stats.TotalPacketsLost != 0 { + b.stats.TotalPacketsLost-- } - } - if sn < b.highestSN && b.stats.PacketCount > 0 { - b.cycle++ - } + // out-of-order, remove it from nack queue + if b.nacker != nil { + b.nacker.Remove(sn) + } + } else { + b.stats.TotalPacketsLost += (uint32(diff) - 1) + if b.nacker != nil && diff > 1 { + for lost := b.highestSN + 1; lost != sn; lost++ { + b.nacker.Push(lost) + } + } - b.highestSN = sn + if sn < b.highestSN { + b.cycle++ + } + + b.highestSN = sn + } } - b.stats.PacketCount++ - b.stats.TotalBytes += uint64(pktSize) + switch { + case isRTX: + b.stats.TotalRetransmitPackets++ + b.stats.TotalRetransmitBytes += uint64(pktSize) + case len(p.Payload) == 0: + b.stats.TotalPaddingPackets++ + b.stats.TotalPaddingBytes += uint64(pktSize) + default: + b.stats.TotalPrimaryPackets++ + b.stats.TotalPrimaryBytes += uint64(pktSize) + if p.Marker { + b.stats.TotalFrames++ + } + } if !isRTX { // jitter @@ -445,8 +478,9 @@ func (b *Buffer) doNACKs() { return } - if r := b.buildNACKPacket(); r != nil { - b.feedbackCB(r) + if r, numSeqNumsNacked := b.buildNACKPacket(); r != nil { + go b.feedbackCB(r) + b.stats.TotalNACKs += uint32(numSeqNumsNacked) } } @@ -476,11 +510,11 @@ func (b *Buffer) doReports(arrivalTime int64) { b.bitrate.Store(bitrates) // RTCP reports - b.feedbackCB(b.getRTCP()) + go b.feedbackCB(b.getRTCP()) } -func (b *Buffer) buildNACKPacket() []rtcp.Packet { - if nacks := b.nacker.Pairs(); len(nacks) > 0 { +func (b *Buffer) buildNACKPacket() ([]rtcp.Packet, int) { + if nacks, numSeqNumsNacked := b.nacker.Pairs(); len(nacks) > 0 { var pkts []rtcp.Packet if len(nacks) > 0 { pkts = []rtcp.Packet{&rtcp.TransportLayerNack{ @@ -489,9 +523,9 @@ func (b *Buffer) buildNACKPacket() []rtcp.Packet { }} } - return pkts + return pkts, numSeqNumsNacked } - return nil + return nil, 0 } func (b *Buffer) buildREMBPacket() *rtcp.ReceiverEstimatedMaximumBitrate { @@ -514,7 +548,6 @@ func (b *Buffer) buildREMBPacket() *rtcp.ReceiverEstimatedMaximumBitrate { if br < 100000 { br = 100000 } - b.stats.TotalBytes = 0 return &rtcp.ReceiverEstimatedMaximumBitrate{ Bitrate: float32(br), @@ -527,14 +560,13 @@ func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport { return nil } - extMaxSeq := (uint32(b.cycle) << 16) | uint32(b.highestSN) - expectedInInterval := extMaxSeq - b.rrSnapshot.extSeqNum + extHighestSeqNum := (uint32(b.cycle) << 16) | uint32(b.highestSN) + expectedInInterval := extHighestSeqNum - b.rrSnapshot.extHighestSeqNum if expectedInInterval == 0 { return nil } - receivedInInterval := b.stats.PacketCount - b.rrSnapshot.packetsReceived - lostInInterval := expectedInInterval - receivedInInterval + lostInInterval := b.stats.TotalPacketsLost - b.rrSnapshot.packetsLost if int32(lostInInterval) < 0 { // could happen if retransmitted packets arrive and make received greater than expected lostInInterval = 0 @@ -547,8 +579,6 @@ func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport { fracLost = b.lastFractionLostToReport } - totalLost := b.rrSnapshot.packetsLost + lostInInterval - var dlsr uint32 if b.lastSRRecv != 0 { delayMS := uint32((time.Now().UnixNano() - b.lastSRRecv) / 1e6) @@ -557,17 +587,16 @@ func (b *Buffer) buildReceptionReport() *rtcp.ReceptionReport { } b.rrSnapshot = &receiverReportSnapshot{ - extSeqNum: extMaxSeq, - packetsReceived: b.stats.PacketCount, - packetsLost: totalLost, - lastLossRate: lossRate, + extHighestSeqNum: extHighestSeqNum, + packetsLost: b.stats.TotalPacketsLost, + lastLossRate: lossRate, } return &rtcp.ReceptionReport{ SSRC: b.mediaSSRC, FractionLost: fracLost, - TotalLost: totalLost, - LastSequenceNumber: extMaxSeq, + TotalLost: b.stats.TotalPacketsLost, + LastSequenceNumber: extHighestSeqNum, Jitter: uint32(b.stats.Jitter), LastSenderReport: uint32(b.lastSRNTPTime >> 16), Delay: dlsr, @@ -677,22 +706,19 @@ func (b *Buffer) GetSenderReportData() (rtpTime uint32, ntpTime uint64, lastRece return rtpTime, ntpTime, lastReceivedTimeInNanosSinceEpoch } -// GetStats returns the raw statistics about a particular buffer state -func (b *Buffer) GetStats() (stats Stats) { - b.Lock() - stats = b.stats - b.Unlock() - return -} +func (b *Buffer) GetStats() *StreamStatsWithLayers { + b.RLock() + defer b.RUnlock() -// Used only in tests -func (b *Buffer) SetStatsTestOnly(stats Stats) { - b.Lock() - b.stats = stats - b.Unlock() -} + layers := make(map[int]LayerStats) + layers[0] = LayerStats{ + TotalPackets: b.stats.TotalPrimaryPackets + b.stats.TotalRetransmitPackets + b.stats.TotalPaddingPackets, + TotalBytes: b.stats.TotalPrimaryBytes + b.stats.TotalRetransmitBytes + b.stats.TotalPaddingBytes, + TotalFrames: b.stats.TotalFrames, + } -// IsLaterTimestamp returns true if timestamp1 is later in time than timestamp2 factoring in timestamp wrap-around -func IsLaterTimestamp(timestamp1 uint32, timestamp2 uint32) bool { - return (timestamp1 - timestamp2) < (1 << 31) + return &StreamStatsWithLayers{ + StreamStats: b.stats, + Layers: layers, + } } diff --git a/pkg/sfu/buffer/buffer_test.go b/pkg/sfu/buffer/buffer_test.go index 0dd6376ac..e49ae93d5 100644 --- a/pkg/sfu/buffer/buffer_test.go +++ b/pkg/sfu/buffer/buffer_test.go @@ -244,59 +244,3 @@ func TestFractionLostReport(t *testing.T) { } wg.Wait() } - -/* -func TestSeqWrapHandler(t *testing.T) { - s := SeqWrapHandler{} - s.UpdateMaxSeq(1) - require.Equal(t, uint32(1), s.MaxSeqNo()) - - type caseInfo struct { - seqs []uint32 // {seq1, seq2, unwrap of seq2} - newer bool // seq2 is newer than seq1 - } - // test normal case, name -> {seq1, seq2, unwrap of seq2} - cases := map[string]caseInfo{ - "no wrap": {[]uint32{1, 4, 4}, true}, - "no wrap backward": {[]uint32{4, 1, 1}, false}, - "wrap around forward to zero": {[]uint32{65534, 0, 65536}, true}, - "wrap around forward": {[]uint32{65534, 10, 65546}, true}, - "wrap around forward 2": {[]uint32{65535 + 65536*2, 1, 1 + 65536*3}, true}, - "wrap around backward ": {[]uint32{5, 65534, 65534}, false}, - "wrap around backward less than zero": {[]uint32{5, 65534, 65534}, false}, - } - - for k, v := range cases { - t.Run(k, func(t *testing.T) { - s := SeqWrapHandler{} - s.UpdateMaxSeq(v.seqs[0]) - extsn, newer := s.Unwrap(uint16(v.seqs[1])) - require.Equal(t, v.newer, newer) - require.Equal(t, v.seqs[2], extsn) - }) - } - -} - -func TestIsTimestampWrap(t *testing.T) { - type caseInfo struct { - name string - ts1 uint32 - ts2 uint32 - later bool - } - - cases := []caseInfo{ - {"normal case 1 timestamp later ", 2, 1, true}, - {"normal case 2 timestamp later", 0x1c000000, 0x10000000, true}, - {"wrap case timestamp later", 0xffff, 0xfc000000, true}, - {"wrap case timestamp early", 0xfc000000, 0xffff, false}, - } - - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - require.Equal(t, c.later, IsLaterTimestamp(c.ts1, c.ts2)) - }) - } -} -*/ diff --git a/pkg/sfu/buffer/nack.go b/pkg/sfu/buffer/nack.go index 52e069d66..37cf6a30b 100644 --- a/pkg/sfu/buffer/nack.go +++ b/pkg/sfu/buffer/nack.go @@ -26,7 +26,7 @@ func NewNACKQueue() *NackQueue { } } -func (n *NackQueue) SetRTT(rtt int) { +func (n *NackQueue) SetRTT(rtt uint32) { n.rtt = time.Duration(rtt) * time.Millisecond } @@ -52,9 +52,9 @@ func (n *NackQueue) Push(sn uint16) { n.nacks = append(n.nacks, &nack{seqNum: sn, nacked: 0, lastNackTime: time.Now()}) } -func (n *NackQueue) Pairs() []rtcp.NackPair { +func (n *NackQueue) Pairs() ([]rtcp.NackPair, int) { if len(n.nacks) == 0 { - return nil + return nil, 0 } now := time.Now() @@ -64,6 +64,7 @@ func (n *NackQueue) Pairs() []rtcp.NackPair { snsToPurge := []uint16{} + numSeqNumsNacked := 0 isPairActive := false var np rtcp.NackPair var nps []rtcp.NackPair @@ -77,6 +78,7 @@ func (n *NackQueue) Pairs() []rtcp.NackPair { nack.nacked++ nack.lastNackTime = now + numSeqNumsNacked++ if (nack.seqNum - baseSN) > 16 { // need a new nack pair @@ -104,5 +106,5 @@ func (n *NackQueue) Pairs() []rtcp.NackPair { n.Remove(sn) } - return nps + return nps, numSeqNumsNacked } diff --git a/pkg/sfu/buffer/nack_test.go b/pkg/sfu/buffer/nack_test.go index 9c5af79f2..cb92d071e 100644 --- a/pkg/sfu/buffer/nack_test.go +++ b/pkg/sfu/buffer/nack_test.go @@ -8,44 +8,61 @@ import ( ) func Test_nackQueue_pairs(t *testing.T) { + type PairsResult struct { + pairs []rtcp.NackPair + numSeqNumsNacked int + } + tests := []struct { name string args []uint16 - want []rtcp.NackPair + want PairsResult }{ { name: "Must return correct single pairs pair", args: []uint16{1, 2, 4, 5}, - want: []rtcp.NackPair{{ - PacketID: 1, - LostPackets: 13, - }}, + want: PairsResult{ + pairs: []rtcp.NackPair{ + { + PacketID: 1, + LostPackets: 13, + }, + }, + numSeqNumsNacked: 4, + }, }, { name: "Must return correct pair wrap", args: []uint16{65533, 2, 4, 5, 30, 32}, - want: []rtcp.NackPair{ - { - PacketID: 65533, - LostPackets: 1<<7 + 1<<6 + 1<<4, + want: PairsResult{ + pairs: []rtcp.NackPair{ + { + PacketID: 65533, + LostPackets: 1<<7 + 1<<6 + 1<<4, + }, + { + PacketID: 30, + LostPackets: 1 << 1, + }, }, - { - PacketID: 30, - LostPackets: 1 << 1, - }}, + numSeqNumsNacked: 6, + }, }, { name: "Must return 2 pairs pair", args: []uint16{1, 2, 4, 5, 20, 22, 24, 27}, - want: []rtcp.NackPair{ - { - PacketID: 1, - LostPackets: 13, - }, - { - PacketID: 20, - LostPackets: 74, + want: PairsResult{ + pairs: []rtcp.NackPair{ + { + PacketID: 1, + LostPackets: 13, + }, + { + PacketID: 20, + LostPackets: 74, + }, }, + numSeqNumsNacked: 8, }, }, } @@ -56,8 +73,9 @@ func Test_nackQueue_pairs(t *testing.T) { for _, sn := range tt.args { n.Push(sn) } - got := n.Pairs() - require.EqualValues(t, tt.want, got) + got, numSeqNumsNacked := n.Pairs() + require.EqualValues(t, tt.want.pairs, got) + require.Equal(t, tt.want.numSeqNumsNacked, numSeqNumsNacked) }) } } diff --git a/pkg/sfu/buffer/streamstats.go b/pkg/sfu/buffer/streamstats.go new file mode 100644 index 000000000..ca717d11b --- /dev/null +++ b/pkg/sfu/buffer/streamstats.go @@ -0,0 +1,28 @@ +package buffer + +type LayerStats struct { + TotalPackets uint32 + TotalBytes uint64 + TotalFrames uint32 +} + +type StreamStats struct { + TotalPrimaryPackets uint32 + TotalPrimaryBytes uint64 + TotalRetransmitPackets uint32 + TotalRetransmitBytes uint64 + TotalPaddingPackets uint32 + TotalPaddingBytes uint64 + TotalPacketsLost uint32 + TotalFrames uint32 + RTT uint32 + Jitter float64 + TotalNACKs uint32 + TotalPLIs uint32 + TotalFIRs uint32 +} + +type StreamStatsWithLayers struct { + StreamStats StreamStats + Layers map[int]LayerStats +} diff --git a/pkg/sfu/connectionquality/connectionstats.go b/pkg/sfu/connectionquality/connectionstats.go index 4b9547888..9b55924f7 100644 --- a/pkg/sfu/connectionquality/connectionstats.go +++ b/pkg/sfu/connectionquality/connectionstats.go @@ -4,60 +4,54 @@ import ( "sync" "time" + "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" "github.com/livekit/protocol/utils" - "github.com/pion/rtcp" "github.com/pion/webrtc/v3" ) -type Snapshot struct { - Initialized bool - TotalPacketsLost uint32 - HighestSeqNum uint32 +const ( + connectionQualityUpdateInterval = 5 * time.Second +) + +type qualityWindow struct { + startSeqNum uint32 + endSeqNum uint32 + startPacketsLost uint32 + endPacketsLost uint32 + maxRTT uint32 + maxJitter uint32 } type ConnectionStatsParams struct { UpdateInterval time.Duration CodecType webrtc.RTPCodecType - GetTotalBytes func() uint64 + ClockRate uint32 + GetTrackStats func() map[uint32]*buffer.StreamStatsWithLayers GetIsReducedQuality func() bool Logger logger.Logger } type ConnectionStats struct { - lock sync.RWMutex - params ConnectionStatsParams - baseSeqNumInitialized bool - baseSeqNum uint32 - highestSeqNum uint32 - totalPacketsLost uint32 - totalBytes uint64 - - snapshot Snapshot - - maxDelay uint32 - maxJitter uint32 - - nackCount int32 - pliCount int32 - firCount int32 - - score float64 - onStatsUpdate func(cs *ConnectionStats, stat *livekit.AnalyticsStat) + lock sync.RWMutex + score float32 + qualityWindows map[uint32]*qualityWindow + done chan struct{} isClosed utils.AtomicFlag } func NewConnectionStats(params ConnectionStatsParams) *ConnectionStats { return &ConnectionStats{ - params: params, - score: 4.0, - done: make(chan struct{}), + params: params, + score: 4.0, + qualityWindows: make(map[uint32]*qualityWindow), + done: make(chan struct{}), } } @@ -77,7 +71,14 @@ func (cs *ConnectionStats) OnStatsUpdate(fn func(cs *ConnectionStats, stat *live cs.onStatsUpdate = fn } -func (cs *ConnectionStats) RTCPFeedback(packets []rtcp.Packet, expectedSSRC uint32) { +func (cs *ConnectionStats) GetScore() float32 { + cs.lock.RLock() + defer cs.lock.RUnlock() + + return cs.score +} + +func (cs *ConnectionStats) UpdateWindow(ssrc uint32, extHighestSeqNum uint32, packetsLost uint32, rtt uint32, jitter uint32) { if cs.isClosed.Get() { return } @@ -85,123 +86,169 @@ func (cs *ConnectionStats) RTCPFeedback(packets []rtcp.Packet, expectedSSRC uint cs.lock.Lock() defer cs.lock.Unlock() - for _, p := range packets { - switch pkt := p.(type) { - case *rtcp.ReceiverReport: - for _, r := range pkt.Reports { - if expectedSSRC != 0 && r.SSRC != expectedSSRC { - continue - } + qw := cs.qualityWindows[ssrc] + if qw == nil { + qw = &qualityWindow{} + cs.qualityWindows[ssrc] = qw + } - if r.Delay > cs.maxDelay { - cs.maxDelay = r.Delay - } + if qw.startSeqNum == 0 { + qw.startSeqNum = extHighestSeqNum + qw.startPacketsLost = packetsLost + } - if r.Jitter > cs.maxJitter { - cs.maxJitter = r.Jitter - } + if extHighestSeqNum > qw.endSeqNum { + qw.endSeqNum = extHighestSeqNum + qw.endPacketsLost = packetsLost + } - if !cs.baseSeqNumInitialized { - cs.baseSeqNumInitialized = true - cs.baseSeqNum = r.LastSequenceNumber - cs.highestSeqNum = r.LastSequenceNumber - cs.totalPacketsLost = r.TotalLost - } + if rtt > qw.maxRTT { + qw.maxRTT = rtt + } - if r.LastSequenceNumber > cs.highestSeqNum { - cs.highestSeqNum = r.LastSequenceNumber - cs.totalPacketsLost = r.TotalLost - } - } + if jitter > qw.maxJitter { + qw.maxJitter = jitter + } +} - case *rtcp.TransportLayerNack: - nackCount := 0 - for _, pair := range pkt.Nacks { - nackCount += len(pair.PacketList()) - } - cs.nackCount += int32(nackCount) - - case *rtcp.PictureLossIndication: - cs.pliCount += 1 - - case *rtcp.FullIntraRequest: - cs.firCount += 1 +func (cs *ConnectionStats) updateScore() { + expectedPacketsInInterval := uint32(0) + lostPacketsInInterval := uint32(0) + maxRTT := uint32(0) + maxJitter := uint32(0) + for _, qw := range cs.qualityWindows { + expectedPacketsInInterval += qw.endSeqNum - qw.endSeqNum + 1 + lostPacketsInInterval += qw.endPacketsLost - qw.startPacketsLost + if qw.maxRTT > maxRTT { + maxRTT = qw.maxRTT + } + if qw.maxJitter > maxJitter { + maxJitter = qw.maxJitter } - } -} -func (cs *ConnectionStats) GetScore() float64 { - cs.lock.RLock() - defer cs.lock.RUnlock() - - return cs.score -} - -func (cs *ConnectionStats) updateAndGetPercentageLoss() float64 { - if cs.params.GetTotalBytes != nil { - cs.totalBytes = cs.params.GetTotalBytes() + qw.startSeqNum = qw.endSeqNum + qw.startPacketsLost = qw.endPacketsLost + qw.maxRTT = 0 + qw.maxJitter = 0 } - if !cs.snapshot.Initialized { - cs.snapshot.Initialized = true - cs.snapshot.HighestSeqNum = cs.highestSeqNum - cs.snapshot.TotalPacketsLost = cs.totalPacketsLost + pctLoss := float32(0.0) + if int32(lostPacketsInInterval) < 0 { + lostPacketsInInterval = 0 } - - packetsLostInInterval := cs.totalPacketsLost - cs.snapshot.TotalPacketsLost - expectedPacketsInInterval := cs.highestSeqNum - cs.snapshot.HighestSeqNum - percentageLoss := float64(0.0) if expectedPacketsInInterval > 0 { - percentageLoss = (float64(packetsLostInInterval) / float64(expectedPacketsInInterval)) * 100 + pctLoss = (float32(lostPacketsInInterval) / float32(expectedPacketsInInterval)) * 100.0 } - cs.snapshot.HighestSeqNum = cs.highestSeqNum - cs.snapshot.TotalPacketsLost = cs.totalPacketsLost + if cs.params.CodecType == webrtc.RTPCodecTypeAudio { + // covert jitter (in media samples units) to milliseconds + cs.score = AudioConnectionScore(pctLoss, maxRTT, float32(maxJitter)*1000.0/float32(cs.params.ClockRate)) + } else { + isReducedQuality := false + if cs.params.GetIsReducedQuality != nil { + isReducedQuality = cs.params.GetIsReducedQuality() + } + cs.score = VideoConnectionScore(pctLoss, isReducedQuality) + } +} - return percentageLoss +func (cs *ConnectionStats) getStat() *livekit.AnalyticsStat { + cs.lock.Lock() + defer cs.lock.Unlock() + + if cs.params.GetTrackStats == nil { + return nil + } + + streams := cs.params.GetTrackStats() + if len(streams) == 0 { + return nil + } + + analyticsStreams := make([]*livekit.AnalyticsStream, 0, len(streams)) + for ssrc, stream := range streams { + maxRTT := stream.StreamStats.RTT + maxJitter := uint32(stream.StreamStats.Jitter) + + if qw := cs.qualityWindows[ssrc]; qw != nil { + maxRTT = qw.maxRTT + maxJitter = qw.maxJitter + } + + as := ToAnalyticsStream(ssrc, &stream.StreamStats, maxRTT, maxJitter, cs.params.ClockRate) + + // + // add video layer if either + // 1. Simulcast - even if there is only one layer per stream as it provides layer id + // 2. A stream has multiple layers + // + if cs.params.CodecType == webrtc.RTPCodecTypeVideo && (len(streams) > 1 || len(stream.Layers) > 1) { + for layer, layerStats := range stream.Layers { + as.VideoLayers = append(as.VideoLayers, ToAnalyticsVideoLayer(layer, &layerStats)) + } + } + + analyticsStreams = append(analyticsStreams, as) + } + + cs.updateScore() + + return &livekit.AnalyticsStat{ + Score: float32(cs.score), + Streams: analyticsStreams, + } } func (cs *ConnectionStats) updateStats() { - tk := time.NewTicker(cs.params.UpdateInterval) + interval := cs.params.UpdateInterval + if interval == 0 { + interval = connectionQualityUpdateInterval + } + tk := time.NewTicker(interval) for { select { case <-cs.done: return case <-tk.C: - cs.lock.Lock() - pctLoss := cs.updateAndGetPercentageLoss() - if cs.params.CodecType == webrtc.RTPCodecTypeAudio { - cs.score = AudioConnectionScore(pctLoss, cs.maxJitter) - } else { - isReducedQuality := false - if cs.params.GetIsReducedQuality != nil { - isReducedQuality = cs.params.GetIsReducedQuality() - } - cs.score = VideoConnectionScore(pctLoss, isReducedQuality) + stat := cs.getStat() + if stat == nil { + continue } - totalPacketsReceived := uint32(0) - if cs.baseSeqNumInitialized { - totalPacketsReceived = cs.highestSeqNum - cs.baseSeqNum - } - - stat := &livekit.AnalyticsStat{ - TotalPackets: uint64(totalPacketsReceived), - PacketLost: uint64(cs.totalPacketsLost), - TotalBytes: cs.totalBytes, - Delay: uint64(cs.maxDelay), - Jitter: float64(cs.maxJitter), - NackCount: cs.nackCount, - PliCount: cs.pliCount, - FirCount: cs.firCount, - ConnectionScore: float32(cs.score), - } - cs.lock.Unlock() - if cs.onStatsUpdate != nil { cs.onStatsUpdate(cs, stat) } } } } + +func ToAnalyticsStream(ssrc uint32, streamStats *buffer.StreamStats, maxRTT uint32, maxJitter uint32, clockRate uint32) *livekit.AnalyticsStream { + // convert jitter (from number of media samples to microseconds + jitter := uint32((float32(maxJitter) * 1e6) / float32(clockRate)) + return &livekit.AnalyticsStream{ + Ssrc: ssrc, + TotalPrimaryPackets: streamStats.TotalPrimaryPackets, + TotalPrimaryBytes: streamStats.TotalPrimaryBytes, + TotalRetransmitPackets: streamStats.TotalRetransmitPackets, + TotalRetransmitBytes: streamStats.TotalRetransmitBytes, + TotalPaddingPackets: streamStats.TotalPaddingPackets, + TotalPaddingBytes: streamStats.TotalPaddingBytes, + TotalPacketsLost: streamStats.TotalPacketsLost, + TotalFrames: streamStats.TotalFrames, + Rtt: maxRTT, + Jitter: jitter, + TotalNacks: streamStats.TotalNACKs, + TotalPlis: streamStats.TotalPLIs, + TotalFirs: streamStats.TotalFIRs, + } +} + +func ToAnalyticsVideoLayer(layer int, layerStats *buffer.LayerStats) *livekit.AnalyticsVideoLayer { + return &livekit.AnalyticsVideoLayer{ + Layer: int32(layer), + TotalPackets: layerStats.TotalPackets, + TotalBytes: layerStats.TotalBytes, + TotalFrames: layerStats.TotalFrames, + } +} diff --git a/pkg/sfu/connectionquality/mos.go b/pkg/sfu/connectionquality/mos.go index 0fcbf06c4..ef73a91d4 100644 --- a/pkg/sfu/connectionquality/mos.go +++ b/pkg/sfu/connectionquality/mos.go @@ -8,10 +8,10 @@ import ( // available @ https://github.com/oanguenot/webrtc-stats const ( - rtt = 70 + defaultRtt = uint32(70) ) -func Score2Rating(score float64) livekit.ConnectionQuality { +func Score2Rating(score float32) livekit.ConnectionQuality { if score > 3.9 { return livekit.ConnectionQuality_EXCELLENT } @@ -22,12 +22,15 @@ func Score2Rating(score float64) livekit.ConnectionQuality { return livekit.ConnectionQuality_POOR } -func mosAudioEmodel(pctLoss float64, jitter uint32) float64 { +func mosAudioEmodel(pctLoss float32, rtt uint32, jitter float32) float32 { rx := 93.2 - pctLoss ry := 0.18*rx*rx - 27.9*rx + 1126.62 - // Jitter is in MicroSecs (1/1e6) units. Convert it to MilliSecs - d := float64(rtt + (jitter / 1000)) + if rtt == 0 { + rtt = defaultRtt + } + // Jitter is in Milliseconds + d := float32(rtt) + jitter h := d - 177.3 if h < 0 { h = 0 @@ -47,13 +50,13 @@ func mosAudioEmodel(pctLoss float64, jitter uint32) float64 { return score } -func loss2Score(pctLoss float64, reducedQuality bool) float64 { +func loss2Score(pctLoss float32, reducedQuality bool) float32 { // No Loss, excellent if pctLoss == 0.0 && !reducedQuality { - return 5 + return 5.0 } // default when loss is minimal, but reducedQuality - score := 3.5 + score := float32(3.5) // loss is bad if pctLoss >= 4.0 { score = 2.0 @@ -64,10 +67,10 @@ func loss2Score(pctLoss float64, reducedQuality bool) float64 { return score } -func AudioConnectionScore(pctLoss float64, jitter uint32) float64 { - return mosAudioEmodel(pctLoss, jitter) +func AudioConnectionScore(pctLoss float32, rtt uint32, jitter float32) float32 { + return mosAudioEmodel(pctLoss, rtt, jitter) } -func VideoConnectionScore(pctLoss float64, reducedQuality bool) float64 { +func VideoConnectionScore(pctLoss float32, reducedQuality bool) float32 { return loss2Score(pctLoss, reducedQuality) } diff --git a/pkg/sfu/downtrack.go b/pkg/sfu/downtrack.go index 650c3feb2..d1c0040a0 100644 --- a/pkg/sfu/downtrack.go +++ b/pkg/sfu/downtrack.go @@ -7,7 +7,6 @@ import ( "io" "strings" "sync" - "sync/atomic" "time" "github.com/livekit/protocol/livekit" @@ -22,10 +21,6 @@ import ( "github.com/livekit/livekit-server/pkg/sfu/connectionquality" ) -const ( - connectionQualityUpdateInterval = 5 * time.Second -) - // TrackSender defines an interface send media to remote peer type TrackSender interface { UpTrackLayersChange(availableLayers []int32) @@ -66,12 +61,7 @@ var ( type ReceiverReportListener func(dt *DownTrack, report *rtcp.ReceiverReport) -type PacketStats struct { - octets uint64 - packets uint32 -} - -// DownTrack implements TrackLocal, is the track used to write packets +// DownTrack implements TrackLocal, is the track used to write packets // to SFU Subscriber, the track handle the packets for simple, simulcast // and SVC Publisher. type DownTrack struct { @@ -101,10 +91,8 @@ type DownTrack struct { listenerLock sync.RWMutex closeOnce sync.Once - // Report helpers - primaryStats atomic.Value // contains *PacketStats - rtxStats atomic.Value // contains *PacketStats - paddingStats atomic.Value // contains *PacketStats + statsLock sync.RWMutex + stats buffer.StreamStats connectionStats *connectionquality.ConnectionStats @@ -172,12 +160,9 @@ func NewDownTrack( } d.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{ - UpdateInterval: connectionQualityUpdateInterval, - CodecType: kind, - GetTotalBytes: func() uint64 { - octets, _ := d.getSRStats() - return octets - }, + CodecType: kind, + ClockRate: c.ClockRate, + GetTrackStats: d.getTrackStats, GetIsReducedQuality: func() bool { return d.GetForwardingStatus() != ForwardingStatusOptimal }, @@ -190,10 +175,6 @@ func NewDownTrack( }) d.connectionStats.Start() - d.primaryStats.Store(new(PacketStats)) - d.rtxStats.Store(new(PacketStats)) - d.paddingStats.Store(new(PacketStats)) - return d, nil } @@ -285,12 +266,12 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { } }() - d.lastRTP.set(time.Now().UnixNano()) - if !d.bound.get() { return nil } + d.lastRTP.set(time.Now().UnixNano()) + tp, err := d.forwarder.GetTranslationParams(extPkt, layer) if tp.shouldSendPLI { d.lastPli.set(time.Now().UnixNano()) @@ -343,7 +324,7 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { d.onMaxLayerChanged(d, layer) } - d.UpdatePrimaryStats(uint32(pktSize)) + d.updatePrimaryStats(pktSize, hdr.Marker) } else { d.logger.Errorw("writing rtp packet err", err) d.pktsDropped.add(1) @@ -355,10 +336,12 @@ func (d *DownTrack) WriteRTP(extPkt *buffer.ExtPacket, layer int32) error { // WritePaddingRTP tries to write as many padding only RTP packets as necessary // to satisfy given size to the DownTrack func (d *DownTrack) WritePaddingRTP(bytesToSend int) int { - primaryStats := d.primaryStats.Load().(*PacketStats) - if primaryStats.packets == 0 { + d.statsLock.RLock() + if d.stats.TotalPrimaryPackets == 0 { + d.statsLock.RUnlock() return 0 } + d.statsLock.RUnlock() // LK-TODO-START // Ideally should look at header extensions negotiated for @@ -428,7 +411,7 @@ func (d *DownTrack) WritePaddingRTP(bytesToSend int) int { } size := hdr.MarshalSize() + len(payload) - d.UpdatePaddingStats(uint32(size)) + d.updatePaddingStats(size) for _, f := range d.onPaddingSent { f(d, size) } @@ -717,39 +700,41 @@ func (d *DownTrack) CreateSenderReport() *rtcp.SenderReport { } } -func (d *DownTrack) UpdatePrimaryStats(packetLen uint32) { - primaryStats, _ := d.primaryStats.Load().(*PacketStats) +func (d *DownTrack) updatePrimaryStats(packetLen int, marker bool) { + d.statsLock.Lock() + defer d.statsLock.Unlock() - primaryStats.octets += uint64(packetLen) - primaryStats.packets += 1 - - d.primaryStats.Store(primaryStats) + d.stats.TotalPrimaryPackets++ + d.stats.TotalPrimaryBytes += uint64(packetLen) + if marker { + d.stats.TotalFrames++ + } } -func (d *DownTrack) UpdateRtxStats(packetLen uint32) { - rtxStats, _ := d.rtxStats.Load().(*PacketStats) +func (d *DownTrack) updateRtxStats(packetLen int) { + d.statsLock.Lock() + defer d.statsLock.Unlock() - rtxStats.octets += uint64(packetLen) - rtxStats.packets += 1 - - d.rtxStats.Store(rtxStats) + d.stats.TotalRetransmitPackets++ + d.stats.TotalRetransmitBytes += uint64(packetLen) } -func (d *DownTrack) UpdatePaddingStats(packetLen uint32) { - paddingStats, _ := d.paddingStats.Load().(*PacketStats) +func (d *DownTrack) updatePaddingStats(packetLen int) { + d.statsLock.Lock() + defer d.statsLock.Unlock() - paddingStats.octets += uint64(packetLen) - paddingStats.packets += 1 - - d.paddingStats.Store(paddingStats) + d.stats.TotalPaddingPackets++ + d.stats.TotalPaddingBytes += uint64(packetLen) } func (d *DownTrack) writeBlankFrameRTP() error { // don't send if nothing has been sent - primaryStats := d.primaryStats.Load().(*PacketStats) - if primaryStats.packets == 0 { + d.statsLock.RLock() + if d.stats.TotalPrimaryPackets == 0 { + d.statsLock.RUnlock() return nil } + d.statsLock.RUnlock() // LK-TODO: Support other video codecs if d.kind == webrtc.RTPCodecTypeAudio || (d.mime != "video/vp8" && d.mime != "video/h264") { @@ -798,7 +783,7 @@ func (d *DownTrack) writeBlankFrameRTP() error { f(d, pktSize) } - d.UpdatePrimaryStats(uint32(pktSize)) + d.updatePrimaryStats(pktSize, hdr.Marker) // only the first frame will need frameEndNeeded to close out the // previous picture, rest are small key frames @@ -865,12 +850,17 @@ func (d *DownTrack) handleRTCP(bytes []byte) { } } + var numNACKs uint32 + var numPLIs uint32 + var numFIRs uint32 for _, pkt := range pkts { switch p := pkt.(type) { case *rtcp.PictureLossIndication: + numPLIs++ sendPliOnce() case *rtcp.FullIntraRequest: + numFIRs++ sendPliOnce() case *rtcp.ReceiverEstimatedMaximumBitrate: @@ -889,6 +879,14 @@ func (d *DownTrack) handleRTCP(bytes []byte) { continue } rr.Reports = append(rr.Reports, r) + + d.statsLock.Lock() + d.stats.TotalPacketsLost = r.TotalLost + // RAJA-TODO - calculate RTT and update + d.stats.Jitter = float64(r.Jitter) + + d.connectionStats.UpdateWindow(r.SSRC, r.LastSequenceNumber, r.TotalLost, 0, r.Jitter) + d.statsLock.Unlock() } if len(rr.Reports) > 0 { d.listenerLock.RLock() @@ -904,6 +902,7 @@ func (d *DownTrack) handleRTCP(bytes []byte) { nackedPackets = append(nackedPackets, d.sequencer.getSeqNoPairs(pair.PacketList())...) } go d.retransmitPackets(nackedPackets) + numNACKs += uint32(len(nackedPackets)) case *rtcp.TransportLayerCC: if p.MediaSSRC == d.ssrc && d.onTransportCCFeedback != nil { @@ -912,7 +911,11 @@ func (d *DownTrack) handleRTCP(bytes []byte) { } } - d.connectionStats.RTCPFeedback(pkts, d.ssrc) + d.statsLock.Lock() + d.stats.TotalNACKs += numNACKs + d.stats.TotalPLIs += numPLIs + d.stats.TotalFIRs += numFIRs + d.statsLock.Unlock() } func (d *DownTrack) retransmitPackets(nackedPackets []packetMeta) { @@ -985,17 +988,19 @@ func (d *DownTrack) retransmitPackets(nackedPackets []packetMeta) { f(d, pktSize) } - d.UpdateRtxStats(uint32(pktSize)) + d.updateRtxStats(pktSize) } } } func (d *DownTrack) getSRStats() (uint64, uint32) { - primary := d.primaryStats.Load().(*PacketStats) - rtx := d.rtxStats.Load().(*PacketStats) - padding := d.paddingStats.Load().(*PacketStats) + d.statsLock.RLock() + defer d.statsLock.RUnlock() - return primary.octets + rtx.octets + padding.octets, primary.packets + rtx.packets + padding.packets + packets := d.stats.TotalPrimaryPackets + d.stats.TotalRetransmitPackets + d.stats.TotalPaddingPackets + octets := d.stats.TotalPrimaryBytes + d.stats.TotalRetransmitBytes + d.stats.TotalPaddingBytes + + return octets, packets } // writes RTP header extensions of track @@ -1091,6 +1096,27 @@ func (d *DownTrack) DebugInfo() map[string]interface{} { } } -func (d *DownTrack) GetConnectionScore() float64 { +func (d *DownTrack) GetConnectionScore() float32 { return d.connectionStats.GetScore() } + +func (d *DownTrack) getTrackStats() map[uint32]*buffer.StreamStatsWithLayers { + d.statsLock.RLock() + defer d.statsLock.RUnlock() + + stats := make(map[uint32]*buffer.StreamStatsWithLayers, 1) + + layers := make(map[int]buffer.LayerStats) + layers[0] = buffer.LayerStats{ + TotalPackets: d.stats.TotalPrimaryPackets + d.stats.TotalRetransmitPackets + d.stats.TotalPaddingPackets, + TotalBytes: d.stats.TotalPrimaryBytes + d.stats.TotalRetransmitBytes + d.stats.TotalPaddingBytes, + TotalFrames: d.stats.TotalFrames, + } + + stats[d.ssrc] = &buffer.StreamStatsWithLayers{ + StreamStats: d.stats, + Layers: layers, + } + + return stats +} diff --git a/pkg/sfu/receiver.go b/pkg/sfu/receiver.go index d15d2a3da..fe4d61dcb 100644 --- a/pkg/sfu/receiver.go +++ b/pkg/sfu/receiver.go @@ -2,7 +2,6 @@ package sfu import ( "io" - "math/rand" "runtime" "sync" "sync/atomic" @@ -15,6 +14,7 @@ import ( "github.com/rs/zerolog/log" "github.com/livekit/livekit-server/pkg/sfu/buffer" + "github.com/livekit/livekit-server/pkg/sfu/connectionquality" ) type AudioLevelHandle func(level uint8, duration uint32) @@ -76,6 +76,12 @@ type WebRTCReceiver struct { lbThreshold int streamTrackerManager *StreamTrackerManager + + rtt uint32 + connectionStats *connectionquality.ConnectionStats + + // update stats + onStatsUpdate func(w *WebRTCReceiver, stat *livekit.AnalyticsStat) } func RidToLayer(rid string) int32 { @@ -145,15 +151,52 @@ func NewWebRTCReceiver( streamTrackerManager: NewStreamTrackerManager(), } w.streamTrackerManager.OnAvailableLayersChanged(w.downTrackLayerChange) + if runtime.GOMAXPROCS(0) < w.numProcs { w.numProcs = runtime.GOMAXPROCS(0) } + for _, opt := range opts { w = opt(w) } + + w.connectionStats = connectionquality.NewConnectionStats(connectionquality.ConnectionStatsParams{ + CodecType: w.kind, + ClockRate: w.codec.ClockRate, + GetTrackStats: w.getTrackStats, + GetIsReducedQuality: func() bool { + return w.streamTrackerManager.IsReducedQuality() + }, + Logger: w.logger, + }) + w.connectionStats.OnStatsUpdate(func(_cs *connectionquality.ConnectionStats, stat *livekit.AnalyticsStat) { + if w.onStatsUpdate != nil { + w.onStatsUpdate(w, stat) + } + }) + w.connectionStats.Start() + return w } +func (w *WebRTCReceiver) OnStatsUpdate(fn func(w *WebRTCReceiver, stat *livekit.AnalyticsStat)) { + w.onStatsUpdate = fn +} + +func (w *WebRTCReceiver) GetConnectionScore() float32 { + return w.connectionStats.GetScore() +} + +func (w *WebRTCReceiver) SetRTT(rtt uint32) { + w.rtt = rtt + + w.bufferMu.RLock() + for _, buffer := range w.buffers { + buffer.SetRTT(rtt) + } + w.bufferMu.RUnlock() +} + func (w *WebRTCReceiver) SetTrackMeta(trackID livekit.TrackID, streamID string) { w.streamID = streamID w.trackID = trackID @@ -191,6 +234,7 @@ func (w *WebRTCReceiver) AddUpTrack(track *webrtc.TrackRemote, buff *buffer.Buff } buff.SetLogger(w.logger) + buff.OnFeedback(w.sendRTCP) layer := RidToLayer(track.RID()) @@ -242,10 +286,6 @@ func (w *WebRTCReceiver) SetMaxExpectedSpatialLayer(layer int32) { w.streamTrackerManager.SetMaxExpectedSpatialLayer(layer) } -func (w *WebRTCReceiver) NumAvailableSpatialLayers() int { - return len(w.streamTrackerManager.GetAvailableLayers()) -} - func (w *WebRTCReceiver) downTrackLayerChange(layers []int32) { w.downTrackMu.RLock() downTracks := w.downTracks @@ -301,26 +341,34 @@ func (w *WebRTCReceiver) DeleteDownTrack(peerID livekit.ParticipantID) { w.free[idx] = struct{}{} } -func (w *WebRTCReceiver) SendRTCP(p []rtcp.Packet) { - if _, ok := p[0].(*rtcp.PictureLossIndication); ok { - w.rtcpMu.Lock() - throttled := time.Now().UnixNano()-w.lastPli.get() < w.pliThrottle - w.rtcpMu.Unlock() - if throttled { - return - } - w.lastPli.set(time.Now().UnixNano()) - } +func (w *WebRTCReceiver) sendRTCP(packets []rtcp.Packet) { + w.rtcpCh <- packets - w.rtcpCh <- p + for _, p := range packets { + switch pkt := p.(type) { + case *rtcp.ReceiverReport: + for _, r := range pkt.Reports { + w.connectionStats.UpdateWindow(r.SSRC, r.LastSequenceNumber, r.TotalLost, w.rtt, r.Jitter) + } + } + } } func (w *WebRTCReceiver) SendPLI(layer int32) { - pli := []rtcp.Packet{ - &rtcp.PictureLossIndication{SenderSSRC: rand.Uint32(), MediaSSRC: w.SSRC(int(layer))}, + w.bufferMu.RLock() + buff := w.buffers[layer] + w.bufferMu.RUnlock() + if buff == nil { + return } - w.SendRTCP(pli) + throttled := time.Now().UnixNano()-w.lastPli.get() < w.pliThrottle + if throttled { + return + } + w.lastPli.set(time.Now().UnixNano()) + + buff.SendPLI() } func (w *WebRTCReceiver) SetRTCPCh(ch chan []rtcp.Packet) { @@ -343,19 +391,30 @@ func (w *WebRTCReceiver) ReadRTP(buf []byte, layer uint8, sn uint16) (int, error return buff.GetPacket(buf, sn) } -func (w *WebRTCReceiver) GetTotalBytes() uint64 { +func (w *WebRTCReceiver) getTrackStats() map[uint32]*buffer.StreamStatsWithLayers { w.bufferMu.RLock() defer w.bufferMu.RUnlock() - totalBytes := uint64(0) - for _, buff := range w.buffers { - if buff != nil { - stats := buff.GetStats() - totalBytes += stats.TotalBytes + stats := make(map[uint32]*buffer.StreamStatsWithLayers, len(w.buffers)) + + for layer, buff := range w.buffers { + if buff == nil { + continue } + + sswl := buff.GetStats() + + // if simulcast, patch buffer stats with correct layer + if w.isSimulcast { + patched := make(map[int]buffer.LayerStats, 1) + patched[layer] = sswl.Layers[0] + sswl.Layers = patched + } + + stats[w.SSRC(layer)] = sswl } - return totalBytes + return stats } func (w *WebRTCReceiver) forwardRTP(layer int32) { @@ -435,6 +494,8 @@ func (w *WebRTCReceiver) writeRTP(layer int32, dt TrackSender, pkt *buffer.ExtPa // closeTracks close all tracks from Receiver func (w *WebRTCReceiver) closeTracks() { + w.connectionStats.Close() + w.downTrackMu.Lock() for _, dt := range w.downTracks { if dt != nil { diff --git a/pkg/sfu/streamtrackermanager.go b/pkg/sfu/streamtrackermanager.go index 2900d4c60..37ab24a43 100644 --- a/pkg/sfu/streamtrackermanager.go +++ b/pkg/sfu/streamtrackermanager.go @@ -134,6 +134,13 @@ func (s *StreamTrackerManager) SetMaxExpectedSpatialLayer(layer int32) { } } +func (s *StreamTrackerManager) IsReducedQuality() bool { + s.lock.RLock() + defer s.lock.RUnlock() + + return int32(len(s.availableLayers)) < (s.maxExpectedLayer + 1) +} + func (s *StreamTrackerManager) GetAvailableLayers() []int32 { s.lock.RLock() defer s.lock.RUnlock() diff --git a/pkg/telemetry/prometheus/packets.go b/pkg/telemetry/prometheus/packets.go index 1e5ee205b..fa8d46bfd 100644 --- a/pkg/telemetry/prometheus/packets.go +++ b/pkg/telemetry/prometheus/packets.go @@ -86,7 +86,7 @@ func IncrementBytes(direction Direction, count uint64) { } } -func IncrementRTCP(direction Direction, nack, pli, fir int32) { +func IncrementRTCP(direction Direction, nack, pli, fir uint32) { if nack > 0 { promNackTotal.WithLabelValues(string(direction)).Add(float64(nack)) atomic.AddUint64(&atomicNackTotal, uint64(nack)) diff --git a/pkg/telemetry/statsworker.go b/pkg/telemetry/statsworker.go index 054ed7c95..ac2107bc3 100644 --- a/pkg/telemetry/statsworker.go +++ b/pkg/telemetry/statsworker.go @@ -4,6 +4,7 @@ import ( "context" "github.com/livekit/protocol/livekit" + "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/timestamppb" ) @@ -15,23 +16,17 @@ type StatsWorker struct { roomName livekit.RoomName participantID livekit.ParticipantID - drainStats map[livekit.TrackID]bool - outgoingPerTrack map[livekit.TrackID]*Stats - incomingPerTrack map[livekit.TrackID]*Stats + outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat + incomingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat } -type Stats struct { - next *livekit.AnalyticsStat - totalPackets uint32 - prevPackets uint32 - totalBytes uint64 - prevBytes uint64 - totalPacketsLost uint64 - prevPacketsLost uint64 - connectionScore float32 -} - -func newStatsWorker(ctx context.Context, t TelemetryReporter, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID) *StatsWorker { +func newStatsWorker( + ctx context.Context, + t TelemetryReporter, + roomID livekit.RoomID, + roomName livekit.RoomName, + participantID livekit.ParticipantID, +) *StatsWorker { s := &StatsWorker{ ctx: ctx, t: t, @@ -39,134 +34,146 @@ func newStatsWorker(ctx context.Context, t TelemetryReporter, roomID livekit.Roo roomName: roomName, participantID: participantID, - outgoingPerTrack: make(map[livekit.TrackID]*Stats), - incomingPerTrack: make(map[livekit.TrackID]*Stats), - drainStats: make(map[livekit.TrackID]bool), + outgoingPerTrack: make(map[livekit.TrackID][]*livekit.AnalyticsStat), + incomingPerTrack: make(map[livekit.TrackID][]*livekit.AnalyticsStat), } return s } -func (s *StatsWorker) getOrCreateOutgoingStatsIfEmpty(trackID livekit.TrackID) *Stats { - if s.outgoingPerTrack[trackID] == nil { - s.outgoingPerTrack[trackID] = &Stats{next: &livekit.AnalyticsStat{ - Kind: livekit.StreamType_DOWNSTREAM, - RoomId: string(s.roomID), - ParticipantId: string(s.participantID), - RoomName: string(s.roomName), - }} - } - return s.outgoingPerTrack[trackID] +func (s *StatsWorker) appendOutgoing(trackID livekit.TrackID, stat *livekit.AnalyticsStat) { + s.outgoingPerTrack[trackID] = append(s.outgoingPerTrack[trackID], stat) } -func (s *StatsWorker) getOrCreateIncomingStatsIfEmpty(trackID livekit.TrackID) *Stats { - if s.incomingPerTrack[trackID] == nil { - s.incomingPerTrack[trackID] = &Stats{next: &livekit.AnalyticsStat{ - Kind: livekit.StreamType_UPSTREAM, - RoomId: string(s.roomID), - ParticipantId: string(s.participantID), - RoomName: string(s.roomName), - }} - } - return s.incomingPerTrack[trackID] +func (s *StatsWorker) appendIncoming(trackID livekit.TrackID, stat *livekit.AnalyticsStat) { + s.incomingPerTrack[trackID] = append(s.incomingPerTrack[trackID], stat) } -func (s *StatsWorker) OnTrackStat(trackID livekit.TrackID, direction livekit.StreamType, stats *livekit.AnalyticsStat) { - var ds *Stats +func (s *StatsWorker) OnTrackStat(trackID livekit.TrackID, direction livekit.StreamType, stat *livekit.AnalyticsStat) { if direction == livekit.StreamType_DOWNSTREAM { - ds = s.getOrCreateOutgoingStatsIfEmpty(trackID) + s.appendOutgoing(trackID, stat) } else { - ds = s.getOrCreateIncomingStatsIfEmpty(trackID) - } - ds.totalPacketsLost = stats.PacketLost - ds.totalPackets = uint32(stats.TotalPackets) - ds.totalBytes = stats.TotalBytes - - if stats.Rtt > ds.next.Rtt { - ds.next.Rtt = stats.Rtt - } - if stats.Jitter > ds.next.Jitter { - ds.next.Jitter = stats.Jitter - } - ds.next.NackCount += stats.NackCount - ds.next.PliCount += stats.PliCount - ds.next.FirCount += stats.FirCount - // average out scores received in this interval - if ds.connectionScore == 0 { - ds.connectionScore = stats.ConnectionScore - } else { - ds.connectionScore = (ds.connectionScore + stats.ConnectionScore) / 2 + s.appendIncoming(trackID, stat) } } func (s *StatsWorker) Update() { ts := timestamppb.Now() - stats := make([]*livekit.AnalyticsStat, 0) + stats := make([]*livekit.AnalyticsStat, 0, len(s.incomingPerTrack)+len(s.outgoingPerTrack)) stats = s.collectUpstreamStats(ts, stats) stats = s.collectDownstreamStats(ts, stats) - - s.t.Report(s.ctx, stats) + if len(stats) > 0 { + s.t.Report(s.ctx, stats) + } } func (s *StatsWorker) collectDownstreamStats(ts *timestamppb.Timestamp, stats []*livekit.AnalyticsStat) []*livekit.AnalyticsStat { - for trackID, trackDownStreamStats := range s.outgoingPerTrack { - analyticsStat := s.update(trackDownStreamStats, ts) - if analyticsStat != nil { - analyticsStat.TrackId = string(trackID) - stats = append(stats, analyticsStat) + for trackID, analyticsStats := range s.outgoingPerTrack { + analyticsStat := coalesce(analyticsStats) + if analyticsStat == nil { + continue } + + s.patch(analyticsStat, ts, trackID, livekit.StreamType_DOWNSTREAM) + stats = append(stats, analyticsStat) } - if len(s.drainStats) > 0 { - for trackID := range s.drainStats { - delete(s.outgoingPerTrack, trackID) - delete(s.incomingPerTrack, trackID) - } - s.drainStats = make(map[livekit.TrackID]bool) - } + s.outgoingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat, 0) return stats } func (s *StatsWorker) collectUpstreamStats(ts *timestamppb.Timestamp, stats []*livekit.AnalyticsStat) []*livekit.AnalyticsStat { - for trackID, trackUpStreamStats := range s.incomingPerTrack { - analyticsStat := s.update(trackUpStreamStats, ts) - if analyticsStat != nil { - analyticsStat.TrackId = string(trackID) - stats = append(stats, analyticsStat) + for trackID, analyticsStats := range s.incomingPerTrack { + analyticsStat := coalesce(analyticsStats) + if analyticsStat == nil { + continue } + + s.patch(analyticsStat, ts, trackID, livekit.StreamType_UPSTREAM) + stats = append(stats, analyticsStat) } + s.incomingPerTrack = make(map[livekit.TrackID][]*livekit.AnalyticsStat, 0) + return stats } -func (s *StatsWorker) update(stats *Stats, ts *timestamppb.Timestamp) *livekit.AnalyticsStat { - if stats.totalBytes == 0 { - return nil - } - - next := stats.next - stats.next = &livekit.AnalyticsStat{ - Kind: next.Kind, - RoomId: string(s.roomID), - ParticipantId: string(s.participantID), - RoomName: string(s.roomName), - } - - next.TimeStamp = ts - next.TotalPackets = uint64(stats.totalPackets - stats.prevPackets) - next.TotalBytes = stats.totalBytes - stats.prevBytes - next.PacketLost = stats.totalPacketsLost - stats.prevPacketsLost - next.ConnectionScore = stats.connectionScore - - stats.prevPackets = stats.totalPackets - stats.prevBytes = stats.totalBytes - stats.prevPacketsLost = stats.totalPacketsLost - return next +func (s *StatsWorker) patch( + analyticsStat *livekit.AnalyticsStat, + ts *timestamppb.Timestamp, + trackID livekit.TrackID, + kind livekit.StreamType, +) { + analyticsStat.TimeStamp = ts + analyticsStat.TrackId = string(trackID) + analyticsStat.Kind = kind + analyticsStat.RoomId = string(s.roomID) + analyticsStat.ParticipantId = string(s.participantID) + analyticsStat.RoomName = string(s.roomName) } func (s *StatsWorker) Close() { s.Update() } -func (s *StatsWorker) RemoveStats(trackID livekit.TrackID) { - s.drainStats[trackID] = true +func coalesce(stats []*livekit.AnalyticsStat) *livekit.AnalyticsStat { + if len(stats) == 0 { + return nil + } + + // average score of all available stats + score := float32(0.0) + for _, stat := range stats { + score += stat.Score + } + score = score / float32(len(stats)) + + // aggregate streams across all stats + maxRTT := make(map[uint32]uint32) + maxJitter := make(map[uint32]uint32) + analyticsStreams := make(map[uint32]*livekit.AnalyticsStream) + for _, stat := range stats { + // + // For each stream (identified by SSRC) consolidate reports. + // For cumulative stats, take the latest report. + // For instantaneous stats, take maximum (or some other appropriate representation) + // + for _, stream := range stat.Streams { + ssrc := stream.Ssrc + analyticsStream := analyticsStreams[ssrc] + if analyticsStream == nil { + analyticsStreams[ssrc] = stream + maxRTT[ssrc] = stream.Rtt + maxJitter[ssrc] = stream.Jitter + continue + } + + if stream.TotalPrimaryPackets <= analyticsStream.TotalPrimaryPackets { + // total count should be monotonically increasing + continue + } + + analyticsStreams[ssrc] = stream + if stream.Rtt > maxRTT[ssrc] { + maxRTT[ssrc] = stream.Rtt + } + + if stream.Jitter > maxJitter[ssrc] { + maxJitter[ssrc] = stream.Jitter + } + } + } + + streams := make([]*livekit.AnalyticsStream, 0, len(analyticsStreams)) + for ssrc, analyticsStream := range analyticsStreams { + stream := proto.Clone(analyticsStream).(*livekit.AnalyticsStream) + stream.Rtt = maxRTT[ssrc] + stream.Jitter = maxJitter[ssrc] + + streams = append(streams, stream) + } + + return &livekit.AnalyticsStat{ + Score: score, + Streams: streams, + } } diff --git a/pkg/telemetry/telemetryserviceinternal.go b/pkg/telemetry/telemetryserviceinternal.go index a1e439950..fe8856c38 100644 --- a/pkg/telemetry/telemetryserviceinternal.go +++ b/pkg/telemetry/telemetryserviceinternal.go @@ -4,10 +4,9 @@ import ( "context" "github.com/gammazero/workerpool" + "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/webhook" - - "github.com/livekit/livekit-server/pkg/telemetry/prometheus" ) type TelemetryServiceInternal interface { @@ -45,7 +44,15 @@ func (t *telemetryServiceInternal) TrackStats(streamType livekit.StreamType, par direction = prometheus.Outgoing } - prometheus.IncrementRTCP(direction, stat.NackCount, stat.PliCount, stat.FirCount) + totalNACKs := uint32(0) + totalPLIs := uint32(0) + totalFIRs := uint32(0) + for _, stream := range stat.Streams { + totalNACKs += stream.TotalNacks + totalPLIs += stream.TotalPlis + totalFIRs += stream.TotalFirs + } + prometheus.IncrementRTCP(direction, totalNACKs, totalPLIs, totalFIRs) w := t.workers[participantID] if w != nil { @@ -55,13 +62,23 @@ func (t *telemetryServiceInternal) TrackStats(streamType livekit.StreamType, par func (t *telemetryServiceInternal) Report(ctx context.Context, stats []*livekit.AnalyticsStat) { for _, stat := range stats { + if len(stat.Streams) == 0 { + continue + } + direction := prometheus.Incoming if stat.Kind == livekit.StreamType_DOWNSTREAM { direction = prometheus.Outgoing } - prometheus.IncrementPackets(direction, stat.TotalPackets) - prometheus.IncrementBytes(direction, stat.TotalBytes) + totalPackets := uint32(0) + totalBytes := uint64(0) + for _, stream := range stat.Streams { + totalPackets += (stream.TotalPrimaryPackets + stream.TotalRetransmitPackets + stream.TotalPaddingPackets) + totalBytes += (stream.TotalPrimaryBytes + stream.TotalRetransmitBytes + stream.TotalPaddingBytes) + } + prometheus.IncrementPackets(direction, uint64(totalPackets)) + prometheus.IncrementBytes(direction, totalBytes) } t.analytics.SendStats(ctx, stats) diff --git a/pkg/telemetry/telemetryserviceinternalevents.go b/pkg/telemetry/telemetryserviceinternalevents.go index 37b230c3f..9913c3f0b 100644 --- a/pkg/telemetry/telemetryserviceinternalevents.go +++ b/pkg/telemetry/telemetryserviceinternalevents.go @@ -126,7 +126,6 @@ func (t *telemetryServiceInternal) TrackUnpublished(ctx context.Context, partici if w != nil { roomID = w.roomID roomName = w.roomName - w.RemoveStats(livekit.TrackID(track.GetSid())) } prometheus.SubPublishedTrack(track.Type.String()) diff --git a/pkg/telemetry/test/telemetry_service_test.go b/pkg/telemetry/test/telemetry_service_test.go index ee045d9b5..8e8a2a3ab 100644 --- a/pkg/telemetry/test/telemetry_service_test.go +++ b/pkg/telemetry/test/telemetry_service_test.go @@ -35,7 +35,7 @@ func Test_ParticipantAndRoomDataAreSentWithAnalytics(t *testing.T) { // do packet := 33 - stat := &livekit.AnalyticsStat{TotalBytes: uint64(packet)} + stat := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{TotalPrimaryBytes: uint64(packet)}}} fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, "", stat) fixture.sut.SendAnalytics() @@ -67,7 +67,7 @@ func Test_OnDownstreamPackets(t *testing.T) { var bytes int for i := range packets { bytes += packets[i] - stat := &livekit.AnalyticsStat{TotalBytes: uint64(bytes), TotalPackets: uint64(i + 1)} + stat := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{TotalPrimaryBytes: uint64(bytes), TotalPrimaryPackets: uint32(i + 1)}}} fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat) } fixture.sut.SendAnalytics() @@ -77,8 +77,8 @@ func Test_OnDownstreamPackets(t *testing.T) { _, stats := fixture.analytics.SendStatsArgsForCall(0) require.Equal(t, 1, len(stats)) require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind) - require.Equal(t, totalBytes, int(stats[0].TotalBytes)) - require.Equal(t, totalPackets, int(stats[0].TotalPackets)) + require.Equal(t, totalBytes, int(stats[0].Streams[0].TotalPrimaryBytes)) + require.Equal(t, totalPackets, int(stats[0].Streams[0].TotalPrimaryPackets)) require.Equal(t, string(trackID), stats[0].TrackId) } @@ -95,12 +95,12 @@ func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) { // do packet1 := 33 trackID1 := livekit.TrackID("trackID1") - stat1 := &livekit.AnalyticsStat{TotalBytes: uint64(packet1), TotalPackets: 1} + stat1 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{TotalPrimaryBytes: uint64(packet1), TotalPrimaryPackets: 1}}} fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID1, stat1) packet2 := 23 trackID2 := livekit.TrackID("trackID2") - stat2 := &livekit.AnalyticsStat{TotalBytes: uint64(packet2), TotalPackets: 1} + stat2 := &livekit.AnalyticsStat{Streams: []*livekit.AnalyticsStream{{TotalPrimaryBytes: uint64(packet2), TotalPrimaryPackets: 1}}} fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID2, stat2) fixture.sut.SendAnalytics() @@ -114,19 +114,19 @@ func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) { for _, sentStat := range stats { if livekit.TrackID(sentStat.TrackId) == trackID1 { found1 = true - require.Equal(t, packet1, int(sentStat.TotalBytes)) - require.Equal(t, 1, int(sentStat.TotalPackets)) + require.Equal(t, packet1, int(sentStat.Streams[0].TotalPrimaryBytes)) + require.Equal(t, 1, int(sentStat.Streams[0].TotalPrimaryPackets)) } else if livekit.TrackID(sentStat.TrackId) == trackID2 { found2 = true - require.Equal(t, packet2, int(sentStat.TotalBytes)) - require.Equal(t, 1, int(sentStat.TotalPackets)) + require.Equal(t, packet2, int(sentStat.Streams[0].TotalPrimaryBytes)) + require.Equal(t, 1, int(sentStat.Streams[0].TotalPrimaryPackets)) } } require.True(t, found1) require.True(t, found2) } -func Test_OnDownStreamRTCP(t *testing.T) { +func Test_OnDownStreamStat(t *testing.T) { fixture := createFixture() // prepare @@ -136,10 +136,34 @@ func Test_OnDownStreamRTCP(t *testing.T) { fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do - stat1 := &livekit.AnalyticsStat{NackCount: 1, PliCount: 1, Jitter: 3, PacketLost: 3, TotalBytes: 1, TotalPackets: 1} + stat1 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 1, + TotalPrimaryPackets: 1, + TotalPacketsLost: 3, + TotalNacks: 1, + TotalPlis: 1, + Jitter: 3, + }, + }, + } trackID := livekit.TrackID("trackID1") fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat1) - stat2 := &livekit.AnalyticsStat{FirCount: 1, Jitter: 5, PacketLost: 4, TotalBytes: 2, TotalPackets: 2} + + stat2 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 2, + TotalPrimaryPackets: 2, + TotalPacketsLost: 4, + TotalNacks: 1, + TotalPlis: 1, + TotalFirs: 1, + Jitter: 5, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat2) fixture.sut.SendAnalytics() @@ -149,12 +173,12 @@ func Test_OnDownStreamRTCP(t *testing.T) { _, stats := fixture.analytics.SendStatsArgsForCall(0) require.Equal(t, 1, len(stats)) require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind) - require.Equal(t, 1, int(stats[0].NackCount)) - require.Equal(t, 1, int(stats[0].PliCount)) - require.Equal(t, 1, int(stats[0].FirCount)) - require.Equal(t, 0, int(stats[0].Rtt)) // TODO: test for RTT - require.Equal(t, 5, int(stats[0].Jitter)) // max of jitter, see list of rtcp.ReceptionReport above - require.Equal(t, 4, int(stats[0].PacketLost)) // last reported packets lost, see list of rtcp.ReceptionReport above + require.Equal(t, 1, int(stats[0].Streams[0].TotalNacks)) + require.Equal(t, 1, int(stats[0].Streams[0].TotalPlis)) + require.Equal(t, 1, int(stats[0].Streams[0].TotalFirs)) + require.Equal(t, 0, int(stats[0].Streams[0].Rtt)) // TODO: test for RTT + require.Equal(t, 5, int(stats[0].Streams[0].Jitter)) // max of jitter, see list of rtcp.ReceptionReport above + require.Equal(t, 4, int(stats[0].Streams[0].TotalPacketsLost)) // last reported packets lost, see list of rtcp.ReceptionReport above require.Equal(t, string(trackID), stats[0].TrackId) } @@ -169,10 +193,26 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) { // do trackID := livekit.TrackID("trackID1") - stat1 := &livekit.AnalyticsStat{PacketLost: 1, TotalPackets: 1, TotalBytes: 1} + stat1 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 1, + TotalPrimaryPackets: 1, + TotalPacketsLost: 1, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat1) // there should be bytes reported so that stats are sent fixture.sut.SendAnalytics() - stat2 := &livekit.AnalyticsStat{PacketLost: 4, TotalPackets: 2, TotalBytes: 2} + stat2 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 2, + TotalPrimaryPackets: 2, + TotalPacketsLost: 4, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID, stat2) fixture.sut.SendAnalytics() @@ -181,12 +221,12 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) { _, stats := fixture.analytics.SendStatsArgsForCall(0) require.Equal(t, 1, len(stats)) require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind) - require.Equal(t, 1, int(stats[0].PacketLost)) // see pkts1 + require.Equal(t, 1, int(stats[0].Streams[0].TotalPacketsLost)) // see pkts1 _, stats = fixture.analytics.SendStatsArgsForCall(1) require.Equal(t, 1, len(stats)) require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind) - require.Equal(t, 3, int(stats[0].PacketLost)) // see diff of TotalLost between pkts2 and pkts1 + require.Equal(t, 4, int(stats[0].Streams[0].TotalPacketsLost)) // see diff of TotalLost between pkts2 and pkts1 } func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) { @@ -202,13 +242,36 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) { trackID1 := livekit.TrackID("trackID1") trackID2 := livekit.TrackID("trackID2") - stat1 := &livekit.AnalyticsStat{TotalBytes: 1, TotalPackets: 1} + stat1 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 1, + TotalPrimaryPackets: 1, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID1, stat1) // there should be bytes reported so that stats are sent - stat2 := &livekit.AnalyticsStat{NackCount: 1, TotalPackets: 2, TotalBytes: 2} + stat2 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 2, + TotalPrimaryPackets: 2, + TotalNacks: 1, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID1, stat2) - stat3 := &livekit.AnalyticsStat{FirCount: 1, TotalPackets: 3, TotalBytes: 3} + stat3 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 3, + TotalPrimaryPackets: 3, + TotalFirs: 1, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, trackID2, stat3) fixture.sut.SendAnalytics() @@ -223,17 +286,17 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) { if livekit.TrackID(sentStat.TrackId) == trackID1 { found1 = true require.Equal(t, livekit.StreamType_DOWNSTREAM, sentStat.Kind) - require.Equal(t, 1, int(sentStat.NackCount)) // see pkts1 above + require.Equal(t, 1, int(sentStat.Streams[0].TotalNacks)) // see pkts1 above } else if livekit.TrackID(sentStat.TrackId) == trackID2 { found2 = true require.Equal(t, livekit.StreamType_DOWNSTREAM, sentStat.Kind) - require.Equal(t, 1, int(sentStat.FirCount)) // see pkts2 above + require.Equal(t, 1, int(sentStat.Streams[0].TotalFirs)) // see pkts2 above } } require.True(t, found1) require.True(t, found2) } -func Test_OnUpstreamRTCP(t *testing.T) { +func Test_OnUpstreamStat(t *testing.T) { fixture := createFixture() // prepare @@ -243,13 +306,36 @@ func Test_OnUpstreamRTCP(t *testing.T) { fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do - - stat1 := &livekit.AnalyticsStat{NackCount: 1, PliCount: 1, FirCount: 1, Jitter: 5, PacketLost: 3, TotalPackets: 1, TotalBytes: 1} + stat1 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 1, + TotalPrimaryPackets: 1, + TotalPacketsLost: 3, + TotalNacks: 1, + TotalPlis: 1, + TotalFirs: 1, + Jitter: 5, + }, + }, + } trackID := livekit.TrackID("trackID") fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat1) - stat2 := &livekit.AnalyticsStat{Jitter: 2, PacketLost: 4, TotalPackets: 2, TotalBytes: 2} + stat2 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 2, + TotalPrimaryPackets: 2, + TotalPacketsLost: 4, + TotalNacks: 1, + TotalPlis: 1, + TotalFirs: 1, + Jitter: 2, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat2) fixture.sut.SendAnalytics() @@ -258,12 +344,12 @@ func Test_OnUpstreamRTCP(t *testing.T) { _, stats := fixture.analytics.SendStatsArgsForCall(0) require.Equal(t, 1, len(stats)) require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind) - require.Equal(t, 1, int(stats[0].NackCount)) - require.Equal(t, 1, int(stats[0].PliCount)) - require.Equal(t, 1, int(stats[0].FirCount)) - require.Equal(t, 0, int(stats[0].Rtt)) // TODO: test for RTT - require.Equal(t, 5, int(stats[0].Jitter)) // max of jitter, see list of rtcp.ReceptionReport above - require.Equal(t, 4, int(stats[0].PacketLost)) // last reported packets lost, see list of rtcp.ReceptionReport above + require.Equal(t, 1, int(stats[0].Streams[0].TotalNacks)) + require.Equal(t, 1, int(stats[0].Streams[0].TotalPlis)) + require.Equal(t, 1, int(stats[0].Streams[0].TotalFirs)) + require.Equal(t, 0, int(stats[0].Streams[0].Rtt)) // TODO: test for RTT + require.Equal(t, 5, int(stats[0].Streams[0].Jitter)) // max of jitter, see list of rtcp.ReceptionReport above + require.Equal(t, 4, int(stats[0].Streams[0].TotalPacketsLost)) // last reported packets lost, see list of rtcp.ReceptionReport above require.Equal(t, string(trackID), stats[0].TrackId) } @@ -282,16 +368,40 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) { trackID1 := livekit.TrackID("trackID1") trackID2 := livekit.TrackID("trackID2") - stat1 := &livekit.AnalyticsStat{TotalBytes: uint64(totalBytes), TotalPackets: uint64(totalPackets)} + stat1 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: uint64(totalBytes), + TotalPrimaryPackets: uint32(totalPackets), + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID1, stat1) fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID2, stat1) // using same buffer is not correct but for test it is fine // do totalBytes++ totalPackets++ - stat2 := &livekit.AnalyticsStat{NackCount: 1, TotalBytes: uint64(totalBytes), TotalPackets: uint64(totalPackets)} + stat2 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: uint64(totalBytes), + TotalPrimaryPackets: uint32(totalPackets), + TotalNacks: 1, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID1, stat2) - stat3 := &livekit.AnalyticsStat{FirCount: 1, TotalBytes: uint64(totalBytes), TotalPackets: uint64(totalPackets)} + + stat3 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: uint64(totalBytes), + TotalPrimaryPackets: uint32(totalPackets), + TotalFirs: 1, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID2, stat3) fixture.sut.SendAnalytics() @@ -306,30 +416,22 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) { if livekit.TrackID(sentStat.TrackId) == trackID1 { found1 = true require.Equal(t, livekit.StreamType_UPSTREAM, sentStat.Kind) - require.Equal(t, 1, int(sentStat.NackCount)) // see pkts1 above + require.Equal(t, 1, int(sentStat.Streams[0].TotalNacks)) // see pkts1 above } else if livekit.TrackID(sentStat.TrackId) == trackID2 { found2 = true require.Equal(t, livekit.StreamType_UPSTREAM, sentStat.Kind) - require.Equal(t, 1, int(sentStat.FirCount)) // see pkts2 above + require.Equal(t, 1, int(sentStat.Streams[0].TotalFirs)) // see pkts2 above } - require.Equal(t, totalBytes, int(sentStat.TotalBytes)) - require.Equal(t, totalPackets, int(sentStat.TotalPackets)) + require.Equal(t, totalBytes, int(sentStat.Streams[0].TotalPrimaryBytes)) + require.Equal(t, totalPackets, int(sentStat.Streams[0].TotalPrimaryPackets)) } require.True(t, found1) require.True(t, found2) - // remove 1 track + // remove 1 track - track stats were flushed above, so no more calls to SendStats fixture.sut.TrackUnpublished(context.Background(), partSID, &livekit.TrackInfo{Sid: string(trackID2)}, 0) fixture.sut.SendAnalytics() - require.Equal(t, 2, fixture.analytics.SendStatsCallCount()) - _, stats = fixture.analytics.SendStatsArgsForCall(1) - require.Equal(t, 2, len(stats)) // still 2 tracks, next call won't contain 1 track - - // now only 1 track stats remaining - fixture.sut.SendAnalytics() - require.Equal(t, 3, fixture.analytics.SendStatsCallCount()) - _, stats = fixture.analytics.SendStatsArgsForCall(2) - require.Equal(t, 1, len(stats)) // now only 1 track remaining + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) } func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) { @@ -344,8 +446,8 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) { // do fixture.sut.ParticipantLeft(context.Background(), room, participantInfo) - // test - require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + // should not be called if there are not track stats + require.Equal(t, 0, fixture.analytics.SendStatsCallCount()) } func Test_AddUpTrack(t *testing.T) { @@ -361,7 +463,14 @@ func Test_AddUpTrack(t *testing.T) { var totalBytes uint64 = 3 var totalPackets uint32 = 3 - stat := &livekit.AnalyticsStat{TotalPackets: uint64(totalPackets), TotalBytes: totalBytes} + stat := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: totalBytes, + TotalPrimaryPackets: totalPackets, + }, + }, + } trackID := livekit.TrackID("trackID") fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat) fixture.sut.SendAnalytics() @@ -371,8 +480,8 @@ func Test_AddUpTrack(t *testing.T) { _, stats := fixture.analytics.SendStatsArgsForCall(0) require.Equal(t, 1, len(stats)) require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind) - require.Equal(t, totalBytes, stats[0].TotalBytes) - require.Equal(t, totalPackets, uint32(stats[0].TotalPackets)) + require.Equal(t, totalBytes, stats[0].Streams[0].TotalPrimaryBytes) + require.Equal(t, totalPackets, uint32(stats[0].Streams[0].TotalPrimaryPackets)) require.Equal(t, string(trackID), stats[0].TrackId) } @@ -387,10 +496,24 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) { // do trackID := livekit.TrackID("trackID") - stat1 := &livekit.AnalyticsStat{TotalBytes: 1, TotalPackets: 1} + stat1 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 1, + TotalPrimaryPackets: 1, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat1) - stat2 := &livekit.AnalyticsStat{TotalPackets: 2, TotalBytes: 2} + stat2 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 2, + TotalPrimaryPackets: 2, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, trackID, stat2) fixture.sut.SendAnalytics() // test @@ -398,8 +521,8 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) { _, stats := fixture.analytics.SendStatsArgsForCall(0) require.Equal(t, 1, len(stats)) require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind) - require.Equal(t, stat2.TotalBytes, stats[0].TotalBytes) - require.Equal(t, stat2.TotalPackets, stats[0].TotalPackets) + require.Equal(t, stat2.Streams[0].TotalPrimaryBytes, stats[0].Streams[0].TotalPrimaryBytes) + require.Equal(t, stat2.Streams[0].TotalPrimaryPackets, stats[0].Streams[0].TotalPrimaryPackets) require.Equal(t, string(trackID), stats[0].TrackId) } @@ -414,10 +537,24 @@ func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) { // do // upstream bytes - stat1 := &livekit.AnalyticsStat{TotalPackets: 3, TotalBytes: 3} + stat1 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 3, + TotalPrimaryPackets: 3, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_UPSTREAM, partSID, "trackID", stat1) // downstream bytes - stat2 := &livekit.AnalyticsStat{TotalPackets: 1, TotalBytes: 1} + stat2 := &livekit.AnalyticsStat{ + Streams: []*livekit.AnalyticsStream{ + { + TotalPrimaryBytes: 1, + TotalPrimaryPackets: 1, + }, + }, + } fixture.sut.TrackStats(livekit.StreamType_DOWNSTREAM, partSID, "trackID1", stat2) fixture.sut.SendAnalytics()