diff --git a/pkg/telemetry/statsworker.go b/pkg/telemetry/statsworker.go index 01067e99b..c5e6879ae 100644 --- a/pkg/telemetry/statsworker.go +++ b/pkg/telemetry/statsworker.go @@ -25,11 +25,13 @@ type StatsWorker struct { } type Stats struct { - next *livekit.AnalyticsStat - totalPackets uint32 - prevPackets uint32 - totalBytes uint64 - prevBytes uint64 + next *livekit.AnalyticsStat + totalPackets uint32 + prevPackets uint32 + totalBytes uint64 + prevBytes uint64 + totalPacketsLost uint64 + prevPacketsLost uint64 } func newStatsWorker(ctx context.Context, t TelemetryReporter, roomID, roomName, participantID string) *StatsWorker { @@ -89,6 +91,7 @@ func (s *StatsWorker) OnRTCP(trackID string, direction livekit.StreamType, stats } else { ds = s.getOrCreateIncomingStatsIfEmpty(trackID) } + ds.totalPacketsLost = stats.PacketLost if stats.Delay > ds.next.Delay { ds.next.Delay = stats.Delay @@ -96,7 +99,6 @@ func (s *StatsWorker) OnRTCP(trackID string, direction livekit.StreamType, stats if stats.Jitter > ds.next.Jitter { ds.next.Jitter = stats.Jitter } - ds.next.PacketLost += stats.PacketLost ds.next.NackCount += stats.NackCount ds.next.PliCount += stats.PliCount ds.next.FirCount += stats.FirCount @@ -174,9 +176,11 @@ func (s *StatsWorker) update(stats *Stats, ts *timestamppb.Timestamp) *livekit.A next.TimeStamp = ts next.TotalPackets = uint64(stats.totalPackets - stats.prevPackets) next.TotalBytes = stats.totalBytes - stats.prevBytes + next.PacketLost = stats.totalPacketsLost - stats.prevPacketsLost stats.prevPackets = stats.totalPackets stats.prevBytes = stats.totalBytes + stats.prevPacketsLost = stats.totalPacketsLost return next } diff --git a/pkg/telemetry/telemetryserviceinternal.go b/pkg/telemetry/telemetryserviceinternal.go index a361fb188..192b70150 100644 --- a/pkg/telemetry/telemetryserviceinternal.go +++ b/pkg/telemetry/telemetryserviceinternal.go @@ -69,7 +69,9 @@ func (t *telemetryServiceInternal) HandleRTCP(streamType livekit.StreamType, par if jitter := float64(rr.Jitter); jitter > stats.Jitter { stats.Jitter = jitter } - stats.PacketLost += uint64(rr.TotalLost) + if totalLost := uint64(rr.TotalLost); totalLost > stats.PacketLost { + stats.PacketLost = totalLost + } } if streamType == livekit.StreamType_DOWNSTREAM { rtt := GetRttMs(pkt) diff --git a/pkg/telemetry/test/telemetry_service_test.go b/pkg/telemetry/test/telemetry_service_test.go index 36219a960..13efc0a6c 100644 --- a/pkg/telemetry/test/telemetry_service_test.go +++ b/pkg/telemetry/test/telemetry_service_test.go @@ -157,10 +157,54 @@ func Test_OnDownStreamRTCP(t *testing.T) { require.Equal(t, 1, int(stats[0].FirCount)) require.Equal(t, 0, int(stats[0].Delay)) // TODO: test for RTT 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, 4, int(stats[0].PacketLost)) // last reported packets lost, see list of rtcp.ReceptionReport above require.Equal(t, trackID, stats[0].TrackId) } +func Test_PacketLostDiffShouldBeSentToTelemetry(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.ReceiverReport{ + Reports: []rtcp.ReceptionReport{ + {Delay: 0, Jitter: 0, TotalLost: 1}, + }, + }, + } + pkts2 := []rtcp.Packet{ + &rtcp.ReceiverReport{ + Reports: []rtcp.ReceptionReport{ + {Delay: 0, Jitter: 0, TotalLost: 4}, // diff with previous is 3, so in second call to SendAnalytics, 3 should be sent + }, + }, + } + 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, pkts1) + fixture.sut.SendAnalytics() + fixture.sut.HandleRTCP(livekit.StreamType_DOWNSTREAM, partSID, trackID, pkts2) + fixture.sut.SendAnalytics() + + //test + require.Equal(t, 2, fixture.analytics.SendStatsCallCount()) // 2 calls to fixture.sut.SendAnalytics() + _, 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 + + _, 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 +} + func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) { fixture := createFixture() @@ -248,7 +292,7 @@ func Test_OnUpstreamRTCP(t *testing.T) { require.Equal(t, 1, int(stats[0].FirCount)) require.Equal(t, 0, int(stats[0].Delay)) // TODO: test for RTT 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, 4, int(stats[0].PacketLost)) // last reported packets lost, see list of rtcp.ReceptionReport above require.Equal(t, trackID, stats[0].TrackId) }