diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index b3649744f..d6bf654bb 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -286,13 +286,13 @@ func (t *MediaTrack) AddSubscriber(sub types.Participant) error { go t.sendDownTrackBindingReports(sub) }) downTrack.OnPacketSent(func(_ *sfu.DownTrack, size int) { - t.params.Telemetry.OnDownstreamPacket(subscriberID, size) + t.params.Telemetry.OnDownstreamPacket(subscriberID, t.ID(), size) }) downTrack.OnPaddingSent(func(_ *sfu.DownTrack, size int) { - t.params.Telemetry.OnDownstreamPacket(subscriberID, size) + t.params.Telemetry.OnDownstreamPacket(subscriberID, t.ID(), size) }) downTrack.OnRTCP(func(pkts []rtcp.Packet) { - t.params.Telemetry.HandleRTCP(livekit.StreamType_DOWNSTREAM, subscriberID, pkts) + t.params.Telemetry.HandleRTCP(livekit.StreamType_DOWNSTREAM, subscriberID, t.ID(), pkts) }) downTrack.OnCloseHandler(func() { @@ -437,7 +437,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra } t.receiver.AddUpTrack(track, buff) - t.params.Telemetry.AddUpTrack(t.params.ParticipantID, buff) + t.params.Telemetry.AddUpTrack(t.params.ParticipantID, t.ID(), buff) atomic.AddUint32(&t.numUpTracks, 1) // LK-TODO: can remove this completely when VideoLayers protocol becomes the default as it has info from client or if we decide to use TrackInfo.Simulcast @@ -631,6 +631,9 @@ func (t *MediaTrack) handlePublisherFeedback(packets []rtcp.Packet) { var totalLost uint32 var maxSeqNum uint32 + //forward to telemetry + t.params.Telemetry.HandleRTCP(livekit.StreamType_UPSTREAM, t.params.ParticipantID, t.ID(), packets) + for _, p := range packets { switch pkt := p.(type) { // sfu.Buffer generates ReceiverReports for the publisher diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index c265ed024..3cdf2ccca 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -7,7 +7,6 @@ import ( "github.com/bep/debounce" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" - "github.com/pion/interceptor" "github.com/pion/webrtc/v3" "github.com/livekit/livekit-server/pkg/rtc/types" @@ -70,16 +69,9 @@ func newPeerConnection(params TransportParams) (*webrtc.PeerConnection, *webrtc. se := params.Config.SettingEngine se.DisableMediaEngineCopy(true) - ir := &interceptor.Registry{} - // intercept pub -> SFU rtcp for analytics - if params.Telemetry != nil && params.Target == livekit.SignalTarget_PUBLISHER { - f := params.Telemetry.NewStatsInterceptorFactory(params.ParticipantID, params.ParticipantIdentity) - ir.Add(f) - } api := webrtc.NewAPI( webrtc.WithMediaEngine(me), webrtc.WithSettingEngine(se), - webrtc.WithInterceptorRegistry(ir), ) pc, err := api.NewPeerConnection(params.Config.Configuration) return pc, me, err diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index e4aabb396..9aaef37fb 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/pkg/sfu/buffer/buffer.go @@ -618,6 +618,13 @@ func (b *Buffer) GetStats() (stats Stats) { return } +// Used only in tests +func (b *Buffer) SetStatsTestOnly(stats Stats) { + b.Lock() + b.stats = stats + b.Unlock() +} + // GetLatestTimestamp returns the latest RTP timestamp factoring in potential RTP timestamp wrap-around func (b *Buffer) GetLatestTimestamp() (latestTimestamp uint32, latestTimestampTimeInNanosSinceEpoch int64) { latestTimestamp = atomic.LoadUint32(&b.latestTimestamp) diff --git a/pkg/telemetry/interceptor.go b/pkg/telemetry/interceptor.go deleted file mode 100644 index 9cd5d93e0..000000000 --- a/pkg/telemetry/interceptor.go +++ /dev/null @@ -1,46 +0,0 @@ -package telemetry - -import ( - "github.com/livekit/protocol/livekit" - "github.com/pion/interceptor" - "github.com/pion/rtcp" -) - -func (t *telemetryServiceInternal) NewStatsInterceptorFactory(participantID, identity string) *StatsInterceptorFactory { - return &StatsInterceptorFactory{ - t: t, - participantID: participantID, - identity: identity, - } -} - -type StatsInterceptorFactory struct { - t TelemetryService - participantID string - identity string -} - -func (f *StatsInterceptorFactory) NewInterceptor(_ string) (interceptor.Interceptor, error) { - return &StatsInterceptor{ - t: f.t, - participantID: f.participantID, - identity: f.identity, - }, nil -} - -type StatsInterceptor struct { - interceptor.NoOp - - t TelemetryService - participantID string - identity string -} - -// BindRTCPWriter lets you modify any outgoing RTCP packets. It is called once per PeerConnection. The returned method -// will be called once per packet batch. -func (s *StatsInterceptor) BindRTCPWriter(writer interceptor.RTCPWriter) interceptor.RTCPWriter { - return interceptor.RTCPWriterFunc(func(pkts []rtcp.Packet, attributes interceptor.Attributes) (int, error) { - s.t.HandleRTCP(livekit.StreamType_UPSTREAM, s.participantID, pkts) - return writer.Write(pkts, attributes) - }) -} diff --git a/pkg/telemetry/statsworker.go b/pkg/telemetry/statsworker.go index facf86399..a52c72b2f 100644 --- a/pkg/telemetry/statsworker.go +++ b/pkg/telemetry/statsworker.go @@ -19,15 +19,14 @@ type StatsWorker struct { participantID string sync.RWMutex - buffers map[uint32]*buffer.Buffer - drain map[uint32]bool + upstreamBuffers map[string][]*buffer.Buffer + drainUpstreamBuffers map[string]bool - incoming *Stats - outgoing *Stats + outgoingPerTrack map[string]*Stats + incomingPerTrack map[string]*Stats } type Stats struct { - sync.Mutex next *livekit.AnalyticsStat totalPackets uint32 prevPackets uint32 @@ -43,48 +42,64 @@ func newStatsWorker(ctx context.Context, t TelemetryReporter, roomID, roomName, roomName: roomName, participantID: participantID, - buffers: make(map[uint32]*buffer.Buffer), - drain: make(map[uint32]bool), + upstreamBuffers: make(map[string][]*buffer.Buffer), + drainUpstreamBuffers: make(map[string]bool), - incoming: &Stats{next: &livekit.AnalyticsStat{ - Kind: livekit.StreamType_UPSTREAM, - RoomId: roomID, - ParticipantId: participantID, - RoomName: roomName, - }}, - outgoing: &Stats{next: &livekit.AnalyticsStat{ - Kind: livekit.StreamType_DOWNSTREAM, - RoomId: roomID, - ParticipantId: participantID, - RoomName: roomName, - }}, + outgoingPerTrack: make(map[string]*Stats), + incomingPerTrack: make(map[string]*Stats), } return s } -func (s *StatsWorker) AddBuffer(buffer *buffer.Buffer) { +func (s *StatsWorker) AddBuffer(trackID string, buffer *buffer.Buffer) { s.Lock() defer s.Unlock() - s.buffers[buffer.GetMediaSSRC()] = buffer + s.upstreamBuffers[trackID] = append(s.upstreamBuffers[trackID], buffer) } -func (s *StatsWorker) OnDownstreamPacket(bytes int) { - s.outgoing.Lock() - defer s.outgoing.Unlock() +func (s *StatsWorker) OnDownstreamPacket(trackID string, bytes int) { + s.Lock() + defer s.Unlock() - s.outgoing.totalPackets++ - s.outgoing.totalBytes += uint64(bytes) + s.getOrCreateOutgoingStatsIfEmpty(trackID).totalBytes += uint64(bytes) + s.getOrCreateOutgoingStatsIfEmpty(trackID).totalPackets++ } -func (s *StatsWorker) OnRTCP(direction livekit.StreamType, stats *livekit.AnalyticsStat) { - ds := s.incoming - if direction == livekit.StreamType_DOWNSTREAM { - ds = s.outgoing +func (s *StatsWorker) getOrCreateOutgoingStatsIfEmpty(trackID string) *Stats { + if s.outgoingPerTrack[trackID] == nil { + s.outgoingPerTrack[trackID] = &Stats{next: &livekit.AnalyticsStat{ + Kind: livekit.StreamType_DOWNSTREAM, + RoomId: s.roomID, + ParticipantId: s.participantID, + RoomName: s.roomName, + }} } + return s.outgoingPerTrack[trackID] +} - ds.Lock() - defer ds.Unlock() +func (s *StatsWorker) getOrCreateIncomingStatsIfEmpty(trackID string) *Stats { + if s.incomingPerTrack[trackID] == nil { + s.incomingPerTrack[trackID] = &Stats{next: &livekit.AnalyticsStat{ + Kind: livekit.StreamType_UPSTREAM, + RoomId: s.roomID, + ParticipantId: s.participantID, + RoomName: s.roomName, + }} + } + return s.incomingPerTrack[trackID] +} + +func (s *StatsWorker) OnRTCP(trackID string, direction livekit.StreamType, stats *livekit.AnalyticsStat) { + s.Lock() + defer s.Unlock() + + var ds *Stats + if direction == livekit.StreamType_DOWNSTREAM { + ds = s.getOrCreateOutgoingStatsIfEmpty(trackID) + } else { + ds = s.getOrCreateIncomingStatsIfEmpty(trackID) + } if stats.Delay > ds.next.Delay { ds.next.Delay = stats.Delay @@ -98,52 +113,70 @@ func (s *StatsWorker) OnRTCP(direction livekit.StreamType, stats *livekit.Analyt ds.next.FirCount += stats.FirCount } +func (s *StatsWorker) calculateTotalBytesPackets(allBuffers []*buffer.Buffer) (totalBytes uint64, totalPackets uint32) { + totalBytes = 0 + totalPackets = 0 + + for _, buffer := range allBuffers { + totalBytes += buffer.GetStats().TotalByte + totalPackets += buffer.GetStats().PacketCount + } + return totalBytes, totalPackets +} + func (s *StatsWorker) Update() { - var packetsIn uint32 - var bytesIn uint64 - s.Lock() + defer s.Unlock() + ts := timestamppb.Now() - for _, buff := range s.buffers { - stats := buff.GetStats() - packetsIn += stats.PacketCount - bytesIn += stats.TotalByte - } + stats := make([]*livekit.AnalyticsStat, 0) - if len(s.drain) > 0 { - for ssrc := range s.drain { - delete(s.buffers, ssrc) - } - s.drain = make(map[uint32]bool) - } - s.Unlock() - - s.incoming.Lock() - s.incoming.totalPackets = packetsIn - s.incoming.totalBytes = bytesIn - s.incoming.Unlock() - - stats := make([]*livekit.AnalyticsStat, 0, 2) - upstream := s.update(s.incoming, ts) - if upstream != nil { - stats = append(stats, upstream) - } - downstream := s.update(s.outgoing, ts) - if downstream != nil { - stats = append(stats, downstream) - } + stats = s.collectUpstreamStats(ts, stats) + stats = s.collectDownstreamStats(ts, stats) 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 = trackID + stats = append(stats, analyticsStat) + } + } + return stats +} + +func (s *StatsWorker) collectUpstreamStats(ts *timestamppb.Timestamp, stats []*livekit.AnalyticsStat) []*livekit.AnalyticsStat { + for trackID, buffers := range s.upstreamBuffers { + totalBytes, totalPackets := s.calculateTotalBytesPackets(buffers) + + s.getOrCreateIncomingStatsIfEmpty(trackID).totalBytes = totalBytes + s.getOrCreateIncomingStatsIfEmpty(trackID).totalPackets = totalPackets + + analyticsStats := s.update(s.incomingPerTrack[trackID], ts) + if analyticsStats != nil { + analyticsStats.TrackId = trackID + stats = append(stats, analyticsStats) + } + } + + if len(s.drainUpstreamBuffers) > 0 { + for trackID := range s.drainUpstreamBuffers { + delete(s.upstreamBuffers, trackID) + delete(s.incomingPerTrack, trackID) + } + s.drainUpstreamBuffers = make(map[string]bool) + } + return stats +} + func (s *StatsWorker) update(stats *Stats, ts *timestamppb.Timestamp) *livekit.AnalyticsStat { if stats.totalBytes == 0 { return nil } - stats.Lock() - defer stats.Unlock() - next := stats.next stats.next = &livekit.AnalyticsStat{ Kind: next.Kind, @@ -162,9 +195,9 @@ func (s *StatsWorker) update(stats *Stats, ts *timestamppb.Timestamp) *livekit.A return next } -func (s *StatsWorker) RemoveBuffer(ssrc uint32) { +func (s *StatsWorker) RemoveBuffer(trackID string) { s.Lock() - s.drain[ssrc] = true + s.drainUpstreamBuffers[trackID] = true s.Unlock() } diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 16dad1050..85a7c9e24 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -15,10 +15,9 @@ const updateFrequency = time.Second * 10 type TelemetryService interface { // stats - NewStatsInterceptorFactory(participantID, identity string) *StatsInterceptorFactory - AddUpTrack(participantID string, buff *buffer.Buffer) - OnDownstreamPacket(participantID string, bytes int) - HandleRTCP(streamType livekit.StreamType, participantID string, pkts []rtcp.Packet) + AddUpTrack(participantID string, trackID string, buff *buffer.Buffer) + OnDownstreamPacket(participantID string, trackID string, bytes int) + HandleRTCP(streamType livekit.StreamType, participantID string, trackID string, pkts []rtcp.Packet) // events RoomStarted(ctx context.Context, room *livekit.Room) @@ -56,16 +55,16 @@ func (t *telemetryService) run() { } } -func (t *telemetryService) AddUpTrack(participantID string, buff *buffer.Buffer) { - t.internalService.AddUpTrack(participantID, buff) +func (t *telemetryService) AddUpTrack(participantID string, trackID string, buff *buffer.Buffer) { + t.internalService.AddUpTrack(participantID, trackID, buff) } -func (t *telemetryService) OnDownstreamPacket(participantID string, bytes int) { - t.internalService.OnDownstreamPacket(participantID, bytes) +func (t *telemetryService) OnDownstreamPacket(participantID string, trackID string, bytes int) { + t.internalService.OnDownstreamPacket(participantID, trackID, bytes) } -func (t *telemetryService) HandleRTCP(streamType livekit.StreamType, participantID string, pkts []rtcp.Packet) { - t.internalService.HandleRTCP(streamType, participantID, pkts) +func (t *telemetryService) HandleRTCP(streamType livekit.StreamType, participantID string, trackID string, pkts []rtcp.Packet) { + t.internalService.HandleRTCP(streamType, participantID, trackID, pkts) } func (t *telemetryService) RoomStarted(ctx context.Context, room *livekit.Room) { @@ -107,7 +106,3 @@ func (t *telemetryService) RecordingStarted(ctx context.Context, ri *livekit.Rec func (t *telemetryService) RecordingEnded(ctx context.Context, ri *livekit.RecordingInfo) { t.internalService.RecordingEnded(ctx, ri) } - -func (t *telemetryService) NewStatsInterceptorFactory(participantID, identity string) *StatsInterceptorFactory { - return t.internalService.NewStatsInterceptorFactory(participantID, identity) -} diff --git a/pkg/telemetry/telemetryserviceinternal.go b/pkg/telemetry/telemetryserviceinternal.go index 964ad7709..3d11e9469 100644 --- a/pkg/telemetry/telemetryserviceinternal.go +++ b/pkg/telemetry/telemetryserviceinternal.go @@ -42,25 +42,25 @@ func NewTelemetryServiceInternal(notifier webhook.Notifier, analytics AnalyticsS } } -func (t *telemetryServiceInternal) AddUpTrack(participantID string, buff *buffer.Buffer) { +func (t *telemetryServiceInternal) AddUpTrack(participantID string, trackID string, buff *buffer.Buffer) { t.RLock() w := t.workers[participantID] t.RUnlock() if w != nil { - w.AddBuffer(buff) + w.AddBuffer(trackID, buff) } } -func (t *telemetryServiceInternal) OnDownstreamPacket(participantID string, bytes int) { +func (t *telemetryServiceInternal) OnDownstreamPacket(participantID string, trackID string, bytes int) { t.RLock() w := t.workers[participantID] t.RUnlock() if w != nil { - w.OnDownstreamPacket(bytes) + w.OnDownstreamPacket(trackID, bytes) } } -func (t *telemetryServiceInternal) HandleRTCP(streamType livekit.StreamType, participantID string, pkts []rtcp.Packet) { +func (t *telemetryServiceInternal) HandleRTCP(streamType livekit.StreamType, participantID string, trackID string, pkts []rtcp.Packet) { stats := &livekit.AnalyticsStat{} for _, pkt := range pkts { switch pkt := pkt.(type) { @@ -94,7 +94,7 @@ func (t *telemetryServiceInternal) HandleRTCP(streamType livekit.StreamType, par w := t.workers[participantID] t.RUnlock() if w != nil { - w.OnRTCP(streamType, stats) + w.OnRTCP(trackID, streamType, stats) } } diff --git a/pkg/telemetry/telemetryserviceinternalevents.go b/pkg/telemetry/telemetryserviceinternalevents.go index d212f76de..50b0db942 100644 --- a/pkg/telemetry/telemetryserviceinternalevents.go +++ b/pkg/telemetry/telemetryserviceinternalevents.go @@ -59,12 +59,13 @@ func (t *telemetryServiceInternal) ParticipantJoined(ctx context.Context, room * }) t.analytics.SendEvent(ctx, &livekit.AnalyticsEvent{ - Type: livekit.AnalyticsEventType_PARTICIPANT_JOINED, - Timestamp: timestamppb.Now(), - RoomSid: room.Sid, - Participant: participant, - Room: room, - SdkType: clientInfo.GetSdk(), + Type: livekit.AnalyticsEventType_PARTICIPANT_JOINED, + Timestamp: timestamppb.Now(), + RoomSid: room.Sid, + ParticipantId: participant.Sid, + Participant: participant, + Room: room, + SdkType: clientInfo.GetSdk(), }) } @@ -115,7 +116,7 @@ func (t *telemetryServiceInternal) TrackUnpublished(ctx context.Context, partici t.RUnlock() if w != nil { roomID = w.roomID - w.RemoveBuffer(ssrc) + w.RemoveBuffer(track.GetSid()) roomName = w.roomName } diff --git a/pkg/telemetry/test/telemetry_service_events_test.go b/pkg/telemetry/test/telemetry_service_events_test.go new file mode 100644 index 000000000..fecaa24d4 --- /dev/null +++ b/pkg/telemetry/test/telemetry_service_events_test.go @@ -0,0 +1,52 @@ +package telemetrytest + +import ( + "context" + "testing" + + "github.com/livekit/protocol/livekit" + "github.com/stretchr/testify/require" +) + +func Test_OnParticipantJoin_EventIsSent(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"} + partSID := "part1" + clientInfo := &livekit.ClientInfo{Sdk: 2} + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + + //do + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo) + + //test + require.Equal(t, 1, fixture.analytics.SendEventCallCount()) + _, event := fixture.analytics.SendEventArgsForCall(0) + require.Equal(t, livekit.AnalyticsEventType_PARTICIPANT_JOINED, event.Type) + require.Equal(t, partSID, event.ParticipantId) + require.Equal(t, participantInfo, event.Participant) + require.Equal(t, room.Sid, event.RoomSid) + require.Equal(t, room, event.Room) + require.Equal(t, clientInfo.Sdk, event.SdkType) +} + +func Test_OnParticipantLeft_EventIsSent(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"} + partSID := "part1" + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + + //do + fixture.sut.ParticipantLeft(context.Background(), room, participantInfo) + + //test + require.Equal(t, 1, fixture.analytics.SendEventCallCount()) + _, event := fixture.analytics.SendEventArgsForCall(0) + require.Equal(t, livekit.AnalyticsEventType_PARTICIPANT_LEFT, event.Type) + require.Equal(t, partSID, event.ParticipantId) + require.Equal(t, room.Sid, event.RoomSid) + require.Equal(t, room, event.Room) +} diff --git a/pkg/telemetry/test/telemetry_service_test.go b/pkg/telemetry/test/telemetry_service_test.go index 9c4980b89..839c82151 100644 --- a/pkg/telemetry/test/telemetry_service_test.go +++ b/pkg/telemetry/test/telemetry_service_test.go @@ -5,8 +5,10 @@ import ( "testing" "github.com/livekit/protocol/livekit" + "github.com/pion/rtcp" "github.com/stretchr/testify/require" + "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/telemetry" "github.com/livekit/livekit-server/pkg/telemetry/telemetryfakes" ) @@ -23,7 +25,32 @@ func createFixture() *telemetryServiceFixture { return fixture } -func Test_OnDownstreamPacket(t *testing.T) { +func Test_ParticipantAndRoomDataAreSentWithAnalytics(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"} + partSID := "part1" + clientInfo := &livekit.ClientInfo{Sdk: 2} + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo) + + //do + packet := 33 + fixture.sut.OnDownstreamPacket(partSID, "", packet) + fixture.sut.SendAnalytics() + + //test + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, stats := fixture.analytics.SendStatsArgsForCall(0) + require.Equal(t, 1, len(stats)) + require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[0].Kind) + require.Equal(t, partSID, stats[0].ParticipantId) + require.Equal(t, room.Sid, stats[0].RoomId) + require.Equal(t, room.Name, stats[0].RoomName) +} + +func Test_OnDownstreamPackets(t *testing.T) { fixture := createFixture() //prepare @@ -37,8 +64,9 @@ func Test_OnDownstreamPacket(t *testing.T) { packets := []int{33, 23} totalBytes := packets[0] + packets[1] totalPackets := len(packets) + trackID := "trackID" for i := range packets { - fixture.sut.OnDownstreamPacket(partSID, packets[i]) + fixture.sut.OnDownstreamPacket(partSID, trackID, packets[i]) } fixture.sut.SendAnalytics() @@ -49,6 +77,248 @@ func Test_OnDownstreamPacket(t *testing.T) { 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, trackID, stats[0].TrackId) +} + +func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{} + partSID := "part1" + clientInfo := &livekit.ClientInfo{Sdk: 2} + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo) + + //do + packet1 := 33 + trackID1 := "trackID1" + packet2 := 23 + trackID2 := "trackID2" + fixture.sut.OnDownstreamPacket(partSID, trackID1, packet1) + fixture.sut.OnDownstreamPacket(partSID, trackID2, packet2) + fixture.sut.SendAnalytics() + + //test + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, stats := fixture.analytics.SendStatsArgsForCall(0) + require.Equal(t, 2, len(stats)) + + found1 := false + found2 := false + for _, sentStat := range stats { + if sentStat.TrackId == trackID1 { + found1 = true + require.Equal(t, packet1, int(sentStat.TotalBytes)) + require.Equal(t, 1, int(sentStat.TotalPackets)) + } else if sentStat.TrackId == trackID2 { + found2 = true + require.Equal(t, packet2, int(sentStat.TotalBytes)) + require.Equal(t, 1, int(sentStat.TotalPackets)) + } + } + require.True(t, found1) + require.True(t, found2) +} + +func Test_OnDownStreamRTCP(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{} + partSID := "part1" + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + + //do + pkts := []rtcp.Packet{ + &rtcp.TransportLayerNack{}, + &rtcp.PictureLossIndication{}, + &rtcp.FullIntraRequest{}, + &rtcp.ReceiverReport{ + Reports: []rtcp.ReceptionReport{ + {Delay: 1, Jitter: 5, TotalLost: 3}, + {Delay: 4, Jitter: 2, TotalLost: 4}, + }, + }, + } + trackID := "trackID1" + fixture.sut.OnDownstreamPacket(partSID, trackID, 1) // there should be bytes reported so that stats are sent + fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID, pkts) + fixture.sut.SendAnalytics() + + //test + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, 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, 4, int(stats[0].Delay)) // max of delay, see list of rtcp.ReceptionReport above + require.Equal(t, 5, int(stats[0].Jitter)) // max of jitter, see list of rtcp.ReceptionReport above + require.Equal(t, 7, int(stats[0].PacketLost)) // sum of lost packets, see list of rtcp.ReceptionReport above + require.Equal(t, trackID, stats[0].TrackId) +} + +func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{} + partSID := "part1" + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + + //do + pkts1 := []rtcp.Packet{ + &rtcp.TransportLayerNack{}, + } + pkts2 := []rtcp.Packet{ + &rtcp.FullIntraRequest{}, + } + trackID1 := "trackID1" + trackID2 := "trackID2" + fixture.sut.OnDownstreamPacket(partSID, trackID1, 1) // there should be bytes reported so that stats are sent + fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID1, pkts1) + fixture.sut.OnDownstreamPacket(partSID, trackID2, 1) // there should be bytes reported so that stats are sent + fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID2, pkts2) + fixture.sut.SendAnalytics() + + //test + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, stats := fixture.analytics.SendStatsArgsForCall(0) + require.Equal(t, 2, len(stats)) + + found1 := false + found2 := false + for _, sentStat := range stats { + if sentStat.TrackId == trackID1 { + found1 = true + require.Equal(t, livekit.StreamType_DOWNSTREAM, sentStat.Kind) + require.Equal(t, 1, int(sentStat.NackCount)) // see pkts1 above + } else if sentStat.TrackId == trackID2 { + found2 = true + require.Equal(t, livekit.StreamType_DOWNSTREAM, sentStat.Kind) + require.Equal(t, 1, int(sentStat.FirCount)) // see pkts2 above + } + } + require.True(t, found1) + require.True(t, found2) +} +func Test_OnUpstreamRTCP(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{} + partSID := "part1" + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + + //do + pkts := []rtcp.Packet{ + &rtcp.TransportLayerNack{}, + &rtcp.PictureLossIndication{}, + &rtcp.FullIntraRequest{}, + &rtcp.ReceiverReport{ + Reports: []rtcp.ReceptionReport{ + {Delay: 1, Jitter: 5, TotalLost: 3}, + {Delay: 4, Jitter: 2, TotalLost: 4}, + }, + }, + } + // there should be bytes reported so that stats are sent + buf := &buffer.Buffer{} + buf.SetStatsTestOnly(buffer.Stats{ + PacketCount: 1, + TotalByte: 1, + }) + trackID := "trackID" + fixture.sut.AddUpTrack(partSID, trackID, buf) + fixture.sut.HandleRTCP(livekit.StreamType_UPSTREAM, partSID, trackID, pkts) + fixture.sut.SendAnalytics() + + //test + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, 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, 4, int(stats[0].Delay)) // max of delay, see list of rtcp.ReceptionReport above + require.Equal(t, 5, int(stats[0].Jitter)) // max of jitter, see list of rtcp.ReceptionReport above + require.Equal(t, 7, int(stats[0].PacketLost)) // sum of lost packets, see list of rtcp.ReceptionReport above + require.Equal(t, trackID, stats[0].TrackId) +} + +func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{} + partSID := "part1" + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + + // there should be bytes reported so that stats are sent + buf := &buffer.Buffer{} + totalBytes := 1 + tolalPackets := 1 + buf.SetStatsTestOnly(buffer.Stats{ + PacketCount: uint32(tolalPackets), + TotalByte: uint64(totalBytes), + }) + trackID1 := "trackID1" + trackID2 := "trackID2" + fixture.sut.AddUpTrack(partSID, trackID1, buf) + fixture.sut.AddUpTrack(partSID, trackID2, buf) //using same buffer is not correct but for test it is fine + pkts1 := []rtcp.Packet{ + &rtcp.TransportLayerNack{}, + } + pkts2 := []rtcp.Packet{ + &rtcp.FullIntraRequest{}, + } + //do + fixture.sut.HandleRTCP(livekit.StreamType_UPSTREAM, partSID, trackID1, pkts1) + fixture.sut.HandleRTCP(livekit.StreamType_UPSTREAM, partSID, trackID2, pkts2) + fixture.sut.SendAnalytics() + + //test + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, stats := fixture.analytics.SendStatsArgsForCall(0) + require.Equal(t, 2, len(stats)) + + found1 := false + found2 := false + for _, sentStat := range stats { + if sentStat.TrackId == trackID1 { + found1 = true + require.Equal(t, livekit.StreamType_UPSTREAM, sentStat.Kind) + require.Equal(t, 1, int(sentStat.NackCount)) // see pkts1 above + } else if 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, totalBytes, int(sentStat.TotalBytes)) + require.Equal(t, tolalPackets, int(sentStat.TotalPackets)) + } + require.True(t, found1) + require.True(t, found2) + + //remove 1 buffer + fixture.sut.TrackUnpublished(context.Background(), partSID, &livekit.TrackInfo{Sid: 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 } func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) { @@ -66,3 +336,101 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) { //test require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) } + +func Test_AddUpTrack(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{} + partSID := "part1" + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + + //do + var totalBytes uint64 = 3 + var totalPackets uint32 = 3 + buf := &buffer.Buffer{} + bufferStats := buffer.Stats{ + PacketCount: totalPackets, + TotalByte: totalBytes, + } + buf.SetStatsTestOnly(bufferStats) + trackID := "trackID" + fixture.sut.AddUpTrack(partSID, trackID, buf) + fixture.sut.SendAnalytics() + + //test + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, 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, trackID, stats[0].TrackId) +} + +func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{} + partSID := "part1" + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + //do + trackID := "trackID" + //buffer 1 + buf1 := &buffer.Buffer{} + buf1.SetStatsTestOnly(buffer.Stats{ + PacketCount: 1, + TotalByte: 1, + }) + fixture.sut.AddUpTrack(partSID, trackID, buf1) + //buffer 2 + buf2 := &buffer.Buffer{} + buf2.SetStatsTestOnly(buffer.Stats{ + PacketCount: 2, + TotalByte: 2, + }) + fixture.sut.AddUpTrack(partSID, trackID, buf2) + fixture.sut.SendAnalytics() + //test + totalBytes := buf1.GetStats().TotalByte + buf2.GetStats().TotalByte + totalPackets := buf1.GetStats().PacketCount + buf2.GetStats().PacketCount + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, 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, trackID, stats[0].TrackId) +} + +func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) { + fixture := createFixture() + + //prepare + room := &livekit.Room{} + partSID := "part1" + participantInfo := &livekit.ParticipantInfo{Sid: partSID} + fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil) + + //do + //upstream bytes + buf := &buffer.Buffer{} + buf.SetStatsTestOnly(buffer.Stats{ + PacketCount: 3, + TotalByte: 3, + }) + fixture.sut.AddUpTrack(partSID, "trackID", buf) + //downstream bytes + fixture.sut.OnDownstreamPacket(partSID, "trackID1", 1) + fixture.sut.SendAnalytics() + + //test + require.Equal(t, 1, fixture.analytics.SendStatsCallCount()) + _, stats := fixture.analytics.SendStatsArgsForCall(0) + require.Equal(t, 2, len(stats)) + require.Equal(t, livekit.StreamType_UPSTREAM, stats[0].Kind) + require.Equal(t, livekit.StreamType_DOWNSTREAM, stats[1].Kind) +}