Fix handling of PacketLost (#296)

They come accumulated already.

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
 }

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)

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)
 }