Repository: pion/interceptor (go)

Issue:
Stats recorder SSRC handling for PLI/NACK/FIR/...
The stats recorder handles incoming receiver report blocks by filtering them by SSRC. The same should be done for PLI/NACK/FIR (and maybe SRs?). See also these code lines:

Correct RR handling:
https://github.com/pion/interceptor/blob/master/pkg/stats/stats_recorder.go#L190

Incorrect handling for PLI/NACK/FIR (includes RTCP packets from other SSRCs in the same stats):
https://github.com/pion/interceptor/blob/master/pkg/stats/stats_recorder.go#L241

Pull request:
Filter packets and reports by SSRC
Otherwise, we would record packets and reports which do not belong to the stream that is monitored by this recorder.

#### Reference issue
Fixes #166 #168


Patch:
diff --git a/AUTHORS.txt b/AUTHORS.txt
index b28c901..c6a2fc3 100644
--- a/AUTHORS.txt
+++ b/AUTHORS.txt
@@ -16,10 +16,10 @@ David Zhao <david@davidzhao.com>
 Jonathan Müller <jonathan@fotokite.com>
 Kevin Caffrey <kcaffrey@gmail.com>
 Maksim Nesterov <msnesterov@avito.ru>
+Mathis <mathis.engelbart@gmail.com>
 Mathis Engelbart <mathis.engelbart@gmail.com>
 Quentin Renard <contact@asticode.com>
 Rayleigh Li <rayleigh.li@zoom.us>
-Sean <sean@siobud.com>
 Sean DuBois <sean@siobud.com>
 Steffen Vogel <post@steffenvogel.de>
 XLPolar <guangjin_pan@163.com>
diff --git a/pkg/stats/stats_recorder.go b/pkg/stats/stats_recorder.go
index 339dbdd..17b4d8a 100644
--- a/pkg/stats/stats_recorder.go
+++ b/pkg/stats/stats_recorder.go
@@ -113,6 +113,9 @@ func (r *recorder) GetStats() Stats {
 }
 
 func (r *recorder) recordIncomingRTP(latestStats internalStats, v *incomingRTP) internalStats {
+	if v.header.SSRC != r.ssrc {
+		return latestStats
+	}
 	sequenceNumber := latestStats.inboundSequencerNumber.Unwrap(v.header.SequenceNumber)
 	if !latestStats.inboundSequenceNumberInitialized {
 		latestStats.inboundFirstSequenceNumber = sequenceNumber
@@ -149,14 +152,33 @@ func (r *recorder) recordIncomingRTP(latestStats internalStats, v *incomingRTP)
 
 func (r *recorder) recordOutgoingRTCP(latestStats internalStats, v *outgoingRTCP) internalStats {
 	for _, pkt := range v.pkts {
+		// The SSRC check is performed for most of the cases but not all. The
+		// reason is that ReceiverReferenceTimeReportBlocks don't have
+		// destination SSRCs but must still be recorded.
 		switch rtcpPkt := pkt.(type) {
 		case *rtcp.FullIntraRequest:
+			if !contains(pkt.DestinationSSRC(), r.ssrc) {
+				r.logger.Debugf("skipping outgoing RTCP pkt: %v\n", pkt)
+				continue
+			}
 			latestStats.InboundRTPStreamStats.FIRCount++
 		case *rtcp.PictureLossIndication:
+			if !contains(pkt.DestinationSSRC(), r.ssrc) {
+				r.logger.Debugf("skipping outgoing RTCP pkt: %v\n", pkt)
+				continue
+			}
 			latestStats.InboundRTPStreamStats.PLICount++
 		case *rtcp.TransportLayerNack:
+			if !contains(pkt.DestinationSSRC(), r.ssrc) {
+				r.logger.Debugf("skipping outgoing RTCP pkt: %v\n", pkt)
+				continue
+			}
 			latestStats.InboundRTPStreamStats.NACKCount++
 		case *rtcp.SenderReport:
+			if !contains(pkt.DestinationSSRC(), r.ssrc) {
+				r.logger.Debugf("skipping outgoing RTCP pkt: %v\n", pkt)
+				continue
+			}
 			latestStats.lastSenderReports = append(latestStats.lastSenderReports, rtcpPkt.NTPTime)
 			if len(latestStats.lastSenderReports) > r.maxLastSenderReports {
 				latestStats.lastSenderReports = latestStats.lastSenderReports[len(latestStats.lastSenderReports)-r.maxLastSenderReports:]
@@ -176,6 +198,9 @@ func (r *recorder) recordOutgoingRTCP(latestStats internalStats, v *outgoingRTCP
 }
 
 func (r *recorder) recordOutgoingRTP(latestStats internalStats, v *outgoingRTP) internalStats {
+	if v.header.SSRC != r.ssrc {
+		return latestStats
+	}
 	headerSize := v.header.MarshalSize()
 	latestStats.OutboundRTPStreamStats.PacketsSent++
 	latestStats.OutboundRTPStreamStats.BytesSent += uint64(headerSize + v.payloadLen)
@@ -189,30 +214,28 @@ func (r *recorder) recordOutgoingRTP(latestStats internalStats, v *outgoingRTP)
 
 func (r *recorder) recordIncomingRR(latestStats internalStats, pkt *rtcp.ReceiverReport, ts time.Time) internalStats {
 	for _, report := range pkt.Reports {
-		if report.SSRC == r.ssrc {
-			if latestStats.remoteInboundFirstSequenceNumberInitialized {
-				cycles := uint64(report.LastSequenceNumber & 0xFFFF0000)
-				nr := uint64(report.LastSequenceNumber & 0x0000FFFF)
-				highest := cycles*0xFFFF + nr
-				latestStats.RemoteInboundRTPStreamStats.PacketsReceived = highest - uint64(report.TotalLost) - uint64(latestStats.remoteInboundFirstSequenceNumber) + 1
-			}
-			latestStats.RemoteInboundRTPStreamStats.PacketsLost = int64(report.TotalLost)
-			latestStats.RemoteInboundRTPStreamStats.Jitter = float64(report.Jitter) / r.clockRate
-
-			if report.Delay != 0 && report.LastSenderReport != 0 {
-				for i := min(r.maxLastSenderReports, len(latestStats.lastSenderReports)) - 1; i >= 0; i-- {
-					lastReport := latestStats.lastSenderReports[i]
-					if (lastReport&0x0000FFFFFFFF0000)>>16 == uint64(report.LastSenderReport) {
-						dlsr := time.Duration(float64(report.Delay) / 65536.0 * float64(time.Second))
-						latestStats.RemoteInboundRTPStreamStats.RoundTripTime = (ts.Add(-dlsr)).Sub(ntp.ToTime(lastReport))
-						latestStats.RemoteInboundRTPStreamStats.TotalRoundTripTime += latestStats.RemoteInboundRTPStreamStats.RoundTripTime
-						latestStats.RemoteInboundRTPStreamStats.RoundTripTimeMeasurements++
-						break
-					}
+		if latestStats.remoteInboundFirstSequenceNumberInitialized {
+			cycles := uint64(report.LastSequenceNumber & 0xFFFF0000)
+			nr := uint64(report.LastSequenceNumber & 0x0000FFFF)
+			highest := cycles*0xFFFF + nr
+			latestStats.RemoteInboundRTPStreamStats.PacketsReceived = highest - uint64(report.TotalLost) - uint64(latestStats.remoteInboundFirstSequenceNumber) + 1
+		}
+		latestStats.RemoteInboundRTPStreamStats.PacketsLost = int64(report.TotalLost)
+		latestStats.RemoteInboundRTPStreamStats.Jitter = float64(report.Jitter) / r.clockRate
+
+		if report.Delay != 0 && report.LastSenderReport != 0 {
+			for i := min(r.maxLastSenderReports, len(latestStats.lastSenderReports)) - 1; i >= 0; i-- {
+				lastReport := latestStats.lastSenderReports[i]
+				if (lastReport&0x0000FFFFFFFF0000)>>16 == uint64(report.LastSenderReport) {
+					dlsr := time.Duration(float64(report.Delay) / 65536.0 * float64(time.Second))
+					latestStats.RemoteInboundRTPStreamStats.RoundTripTime = (ts.Add(-dlsr)).Sub(ntp.ToTime(lastReport))
+					latestStats.RemoteInboundRTPStreamStats.TotalRoundTripTime += latestStats.RemoteInboundRTPStreamStats.RoundTripTime
+					latestStats.RemoteInboundRTPStreamStats.RoundTripTimeMeasurements++
+					break
 				}
 			}
-			latestStats.FractionLost = float64(report.FractionLost) / 256.0
 		}
+		latestStats.FractionLost = float64(report.FractionLost) / 256.0
 	}
 	return latestStats
 }
@@ -238,8 +261,21 @@ func (r *recorder) recordIncomingXR(latestStats internalStats, pkt *rtcp.Extende
 	return latestStats
 }
 
+func contains(ls []uint32, e uint32) bool {
+	for _, x := range ls {
+		if x == e {
+			return true
+		}
+	}
+	return false
+}
+
 func (r *recorder) recordIncomingRTCP(latestStats internalStats, v *incomingRTCP) internalStats {
 	for _, pkt := range v.pkts {
+		if !contains(pkt.DestinationSSRC(), r.ssrc) {
+			r.logger.Debugf("skipping incoming RTCP pkt: %v\n", pkt)
+			continue
+		}
 		switch pkt := pkt.(type) {
 		case *rtcp.TransportLayerNack:
 			latestStats.OutboundRTPStreamStats.NACKCount++
@@ -248,7 +284,7 @@ func (r *recorder) recordIncomingRTCP(latestStats internalStats, v *incomingRTCP
 		case *rtcp.PictureLossIndication:
 			latestStats.OutboundRTPStreamStats.PLICount++
 		case *rtcp.ReceiverReport:
-			return r.recordIncomingRR(latestStats, pkt, v.ts)
+			latestStats = r.recordIncomingRR(latestStats, pkt, v.ts)
 		case *rtcp.SenderReport:
 			latestStats.RemoteOutboundRTPStreamStats.PacketsSent = uint64(pkt.PacketCount)
 			latestStats.RemoteOutboundRTPStreamStats.BytesSent = uint64(pkt.OctetCount)


Test patch:
diff --git a/pkg/stats/stats_recorder_test.go b/pkg/stats/stats_recorder_test.go
index 0fcf2eb..5b1f543 100644
--- a/pkg/stats/stats_recorder_test.go
+++ b/pkg/stats/stats_recorder_test.go
@@ -213,7 +213,12 @@ func TestStatsRecorder(t *testing.T) {
 					ts: now,
 					content: outgoingRTCP{
 						pkts: []rtcp.Packet{
-							&rtcp.ReceiverReport{},
+							&rtcp.ReceiverReport{
+								SSRC: 9999,
+								Reports: []rtcp.ReceptionReport{
+									{SSRC: 0},
+								},
+							},
 							&rtcp.ExtendedReport{
 								SenderSSRC: 0,
 								Reports: []rtcp.ReportBlock{
@@ -259,6 +264,466 @@ func TestStatsRecorder(t *testing.T) {
 				RoundTripTimeMeasurements: 1,
 			},
 		},
+		{
+			name: "RecordIncomingNACKAfterRR",
+			records: []record{
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+						},
+					},
+				},
+				{
+					ts: now.Add(time.Second),
+					content: incomingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{},
+							cname,
+							&rtcp.TransportLayerNack{
+								SenderSSRC: 9999,
+								MediaSSRC:  0,
+								Nacks:      rtcp.NackPairsFromSequenceNumbers([]uint16{2}),
+							},
+						},
+					},
+				},
+			},
+			expectedOutboundRTPStreamStats: OutboundRTPStreamStats{
+				SentRTPStreamStats: SentRTPStreamStats{
+					PacketsSent: 3,
+					BytesSent:   36,
+				},
+				HeaderBytesSent: 36,
+				NACKCount:       1,
+			},
+		},
+		{
+			name: "IgnoreUnknownOutgoingSSRCs",
+			records: []record{
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+							SSRC:           0,
+						},
+					},
+				},
+				{
+					ts: now.Add(33 * time.Millisecond),
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+							SSRC:           0,
+						},
+					},
+				},
+				{
+					ts: now.Add(66 * time.Millisecond),
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+							SSRC:           0,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+							SSRC:           1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+							SSRC:           1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+							SSRC:           1,
+						},
+					},
+				},
+				{
+					ts: now.Add(time.Second),
+					content: incomingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{
+								SSRC:    9999,
+								Reports: []rtcp.ReceptionReport{},
+							},
+							cname,
+							&rtcp.TransportLayerNack{
+								SenderSSRC: 9999,
+								MediaSSRC:  0,
+								Nacks:      rtcp.NackPairsFromSequenceNumbers([]uint16{2}),
+							},
+							&rtcp.TransportLayerNack{
+								SenderSSRC: 9999,
+								MediaSSRC:  1,
+								Nacks:      rtcp.NackPairsFromSequenceNumbers([]uint16{2}),
+							},
+						},
+					},
+				},
+			},
+			expectedOutboundRTPStreamStats: OutboundRTPStreamStats{
+				SentRTPStreamStats: SentRTPStreamStats{
+					PacketsSent: 3,
+					BytesSent:   36,
+				},
+				HeaderBytesSent: 36,
+				NACKCount:       1,
+			},
+		},
+		{
+			name: "IgnoreIncomingNACKForUnknownSSRC",
+			records: []record{
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+						},
+					},
+				},
+				{
+					ts: now.Add(time.Second),
+					content: incomingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{},
+							cname,
+							&rtcp.TransportLayerNack{
+								SenderSSRC: 9999,
+								MediaSSRC:  1,
+								Nacks:      rtcp.NackPairsFromSequenceNumbers([]uint16{2}),
+							},
+						},
+					},
+				},
+			},
+			expectedOutboundRTPStreamStats: OutboundRTPStreamStats{
+				SentRTPStreamStats: SentRTPStreamStats{
+					PacketsSent: 3,
+					BytesSent:   36,
+				},
+				HeaderBytesSent: 36,
+				NACKCount:       0,
+			},
+		},
+		{
+			name: "IgnoreIncomingFIRForUnknownSSRC",
+			records: []record{
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+						},
+					},
+				},
+				{
+					ts: now.Add(time.Second),
+					content: incomingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{},
+							cname,
+							&rtcp.FullIntraRequest{
+								SenderSSRC: 9999,
+								MediaSSRC:  1,
+							},
+						},
+					},
+				},
+			},
+			expectedOutboundRTPStreamStats: OutboundRTPStreamStats{
+				SentRTPStreamStats: SentRTPStreamStats{
+					PacketsSent: 3,
+					BytesSent:   36,
+				},
+				HeaderBytesSent: 36,
+				FIRCount:        0,
+			},
+		},
+		{
+			name: "IgnoreIncomingPLIForUnknownSSRC",
+			records: []record{
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: outgoingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+						},
+					},
+				},
+				{
+					ts: now.Add(time.Second),
+					content: incomingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{},
+							cname,
+							&rtcp.PictureLossIndication{
+								SenderSSRC: 9999,
+								MediaSSRC:  1,
+							},
+						},
+					},
+				},
+			},
+			expectedOutboundRTPStreamStats: OutboundRTPStreamStats{
+				SentRTPStreamStats: SentRTPStreamStats{
+					PacketsSent: 3,
+					BytesSent:   36,
+				},
+				HeaderBytesSent: 36,
+				PLICount:        0,
+			},
+		},
+		{
+			name: "IgnoreUnknownIncomingSSRCs",
+			records: []record{
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+							SSRC:           0,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+							SSRC:           0,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+							SSRC:           0,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+							SSRC:           1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+							SSRC:           1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+							SSRC:           1,
+						},
+					},
+				},
+			},
+			expectedInboundRTPStreamStats: InboundRTPStreamStats{
+				ReceivedRTPStreamStats: ReceivedRTPStreamStats{
+					PacketsReceived: 3,
+				},
+				LastPacketReceivedTimestamp: now,
+				HeaderBytesReceived:         36,
+				BytesReceived:               36,
+			},
+		},
+		{
+			name: "IgnoreOutgoingNACKForUnknownSSRC",
+			records: []record{
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+						},
+					},
+				},
+				{
+					ts: now.Add(time.Second),
+					content: outgoingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{},
+							cname,
+							&rtcp.PictureLossIndication{
+								SenderSSRC: 9999,
+								MediaSSRC:  1,
+							},
+						},
+					},
+				},
+			},
+			expectedInboundRTPStreamStats: InboundRTPStreamStats{
+				ReceivedRTPStreamStats: ReceivedRTPStreamStats{
+					PacketsReceived: 3,
+				},
+				LastPacketReceivedTimestamp: now,
+				HeaderBytesReceived:         36,
+				BytesReceived:               36,
+				PLICount:                    0,
+			},
+		},
+		{
+			name: "IgnoreOutgoingFIRForUnknownSSRC",
+			records: []record{
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 1,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 2,
+						},
+					},
+				},
+				{
+					ts: now,
+					content: incomingRTP{
+						header: rtp.Header{
+							SequenceNumber: 3,
+						},
+					},
+				},
+				{
+					ts: now.Add(time.Second),
+					content: outgoingRTCP{
+						pkts: []rtcp.Packet{
+							&rtcp.ReceiverReport{},
+							cname,
+							&rtcp.FullIntraRequest{
+								SenderSSRC: 9999,
+								MediaSSRC:  1,
+							},
+						},
+					},
+				},
+			},
+			expectedInboundRTPStreamStats: InboundRTPStreamStats{
+				ReceivedRTPStreamStats: ReceivedRTPStreamStats{
+					PacketsReceived: 3,
+				},
+				LastPacketReceivedTimestamp: now,
+				HeaderBytesReceived:         36,
+				BytesReceived:               36,
+				FIRCount:                    0,
+			},
+		},
 	} {
 		t.Run(fmt.Sprintf("%v:%v", i, cc.name), func(t *testing.T) {
 			r := newRecorder(0, 90_000)
@@ -389,3 +854,20 @@ func TestQueueNotBlocking(t *testing.T) {
 		})
 	}
 }
+
+func TestContains(t *testing.T) {
+	for i, tc := range []struct {
+		list     []uint32
+		element  uint32
+		expected bool
+	}{
+		{list: []uint32{}, element: 0, expected: false},
+		{list: []uint32{0}, element: 0, expected: true},
+		{list: []uint32{0, 1, 2}, element: 1, expected: true},
+		{list: []uint32{0, 1, 2}, element: 3, expected: false},
+	} {
+		t.Run(fmt.Sprintf("%v", i), func(t *testing.T) {
+			assert.Equal(t, tc.expected, contains(tc.list, tc.element))
+		})
+	}
+}