Report publish & subscribe RTPStats as Telemetry events (#1506)

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/jxskiss/base62 v1.1.0
 	github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1
 	github.com/livekit/mediatransportutil v0.0.0-20230130133657-96cfb115473a
-	github.com/livekit/protocol v1.4.3-0.20230306094455-176d6d15b6c0
+	github.com/livekit/protocol v1.5.0
 	github.com/livekit/psrpc v0.2.10-0.20230310095745-5cd63568998d
 	github.com/mackerelio/go-osstat v0.2.3
 	github.com/magefile/mage v1.14.0

go.sum

@@ -233,8 +233,8 @@ github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1 h1:jm09419p0lqTkD
 github.com/livekit/mageutil v0.0.0-20230125210925-54e8a70427c1/go.mod h1:Rs3MhFwutWhGwmY1VQsygw28z5bWcnEYmS1OG9OxjOQ=
 github.com/livekit/mediatransportutil v0.0.0-20230130133657-96cfb115473a h1:5UkGQpskXp7HcBmyrCwWtO7ygDWbqtjN09Yva4l/nyE=
 github.com/livekit/mediatransportutil v0.0.0-20230130133657-96cfb115473a/go.mod h1:1Dlx20JPoIKGP45eo+yuj0HjeE25zmyeX/EWHiPCjFw=
-github.com/livekit/protocol v1.4.3-0.20230306094455-176d6d15b6c0 h1:GiHYcnREXQdA3Mt0c/asU6RaHFMIZ5avXVlnK/VE2k0=
-github.com/livekit/protocol v1.4.3-0.20230306094455-176d6d15b6c0/go.mod h1:hkK/G0wwFiLUGp9F5kxeQxq2CQuIzkmfBwKhTsc71us=
+github.com/livekit/protocol v1.5.0 h1:jFGSkSEv0PTjUlrW/WnmERejwxyHOSE9If4VU33PYgk=
+github.com/livekit/protocol v1.5.0/go.mod h1:hkK/G0wwFiLUGp9F5kxeQxq2CQuIzkmfBwKhTsc71us=
 github.com/livekit/psrpc v0.2.10-0.20230310095745-5cd63568998d h1:3wfbd8zi7zGQCR+xfG3r2k9m2RwXUiIzR0SN4BHewwU=
 github.com/livekit/psrpc v0.2.10-0.20230310095745-5cd63568998d/go.mod h1:K0j8f1PgLShR7Lx80KbmwFkDH2BvOnycXGV0OSRURKc=
 github.com/mackerelio/go-osstat v0.2.3 h1:jAMXD5erlDE39kdX2CU7YwCGRcxIO33u/p8+Fhe5dJw=

pkg/rtc/mediatrack.go

@@ -220,6 +220,7 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 
 	t.lock.Lock()
 	mime := strings.ToLower(track.Codec().MimeType)
+	layer := buffer.RidToSpatialLayer(track.RID(), t.trackInfo)
 	t.params.Logger.Debugw("AddReceiver", "mime", track.Codec().MimeType)
 	wr := t.MediaTrackReceiver.Receiver(mime)
 	if wr == nil {
@@ -310,6 +311,17 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra
 	buff.OnFpsChanged(func() {
 		t.MediaTrackSubscriptions.UpdateVideoLayers()
 	})
+
+	buff.OnFinalRtpStats(func(stats *buffer.RTPStats) {
+		t.params.Telemetry.TrackPublishRTPStats(
+			context.Background(),
+			t.params.ParticipantID,
+			t.ID(),
+			mime,
+			int(layer),
+			stats.ToProto(),
+		)
+	})
 	return newCodec
 }
 
@@ -357,7 +369,6 @@ func (t *MediaTrack) Close(willBeResumed bool) {
 	if t.dynacastManager != nil {
 		t.dynacastManager.Close()
 	}
-
 	t.MediaTrackReceiver.ClearAllReceivers(willBeResumed)
 	t.MediaTrackReceiver.Close()
 }

pkg/rtc/mediatrackreceiver.go

@@ -85,14 +85,13 @@ type MediaTrackReceiver struct {
 	muted       atomic.Bool
 	simulcasted atomic.Bool
 
-	lock               sync.RWMutex
-	receivers          []*simulcastReceiver
-	receiversShadow    []*simulcastReceiver
-	trackInfo          *livekit.TrackInfo
-	layerDimensions    map[livekit.VideoQuality]*livekit.VideoLayer
-	potentialCodecs    []webrtc.RTPCodecParameters
-	pendingSubscribeOp map[livekit.ParticipantID]int
-	state              mediaTrackReceiverState
+	lock            sync.RWMutex
+	receivers       []*simulcastReceiver
+	receiversShadow []*simulcastReceiver
+	trackInfo       *livekit.TrackInfo
+	layerDimensions map[livekit.VideoQuality]*livekit.VideoLayer
+	potentialCodecs []webrtc.RTPCodecParameters
+	state           mediaTrackReceiverState
 
 	onSetupReceiver     func(mime string)
 	onMediaLossFeedback func(dt *sfu.DownTrack, report *rtcp.ReceiverReport)
@@ -104,11 +103,10 @@ type MediaTrackReceiver struct {
 
 func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver {
 	t := &MediaTrackReceiver{
-		params:             params,
-		trackInfo:          proto.Clone(params.TrackInfo).(*livekit.TrackInfo),
-		layerDimensions:    make(map[livekit.VideoQuality]*livekit.VideoLayer),
-		pendingSubscribeOp: make(map[livekit.ParticipantID]int),
-		state:              mediaTrackReceiverStateOpen,
+		params:          params,
+		trackInfo:       proto.Clone(params.TrackInfo).(*livekit.TrackInfo),
+		layerDimensions: make(map[livekit.VideoQuality]*livekit.VideoLayer),
+		state:           mediaTrackReceiverStateOpen,
 	}
 
 	t.MediaTrackSubscriptions = NewMediaTrackSubscriptions(MediaTrackSubscriptionsParams{
@@ -312,7 +310,6 @@ func (t *MediaTrackReceiver) IsOpen() bool {
 }
 
 func (t *MediaTrackReceiver) SetClosing() {
-	t.params.Logger.Infow("setting track to closing")
 	t.lock.Lock()
 	defer t.lock.Unlock()
 	if t.state == mediaTrackReceiverStateOpen {

pkg/rtc/subscriptionmanager.go

@@ -543,6 +543,20 @@ func (m *SubscriptionManager) handleSubscribedTrackClose(s *trackSubscription, w
 			&livekit.TrackInfo{Sid: string(s.trackID), Type: subTrack.MediaTrack().Kind()},
 			!willBeResumed && !m.params.Participant.IsClosed(),
 		)
+
+		dt := subTrack.DownTrack()
+		if dt != nil {
+			stats := dt.GetTrackStats()
+			if stats != nil {
+				m.params.Telemetry.TrackSubscribeRTPStats(
+					context.Background(),
+					m.params.Participant.ID(),
+					s.trackID,
+					subTrack.DownTrack().Codec().MimeType,
+					stats,
+				)
+			}
+		}
 	}
 
 	if !willBeResumed {
@@ -564,9 +578,7 @@ func (m *SubscriptionManager) handleSubscribedTrackClose(s *trackSubscription, w
 				}
 			}
 		}
-	}
 
-	if !willBeResumed {
 		m.params.Participant.Negotiate(false)
 	}
 	m.queueReconcile(s.trackID)

pkg/sfu/buffer/buffer.go

@@ -87,6 +87,7 @@ type Buffer struct {
 	onRtcpFeedback     func([]rtcp.Packet)
 	onRtcpSenderReport func(*RTCPSenderReportData)
 	onFpsChanged       func()
+	onFinalRtpStats    func(*RTPStats)
 
 	// logger
 	logger logger.Logger
@@ -321,6 +322,9 @@ func (b *Buffer) Close() error {
 		if b.rtpStats != nil {
 			b.rtpStats.Stop()
 			b.logger.Infow("rtp stats", "direction", "upstream", "stats", b.rtpStats.ToString())
+			if b.onFinalRtpStats != nil {
+				b.onFinalRtpStats(b.rtpStats)
+			}
 		}
 
 		if b.onClose != nil {
@@ -693,6 +697,10 @@ func (b *Buffer) OnRtcpSenderReport(fn func(srData *RTCPSenderReportData)) {
 	b.onRtcpSenderReport = fn
 }
 
+func (b *Buffer) OnFinalRtpStats(fn func(*RTPStats)) {
+	b.onFinalRtpStats = fn
+}
+
 // GetMediaSSRC returns the associated SSRC of the RTP stream
 func (b *Buffer) GetMediaSSRC() uint32 {
 	return b.mediaSSRC

pkg/telemetry/events.go

@@ -362,6 +362,44 @@ func (t *telemetryService) TrackUnmuted(
 	})
 }
 
+func (t *telemetryService) TrackPublishRTPStats(
+	ctx context.Context,
+	participantID livekit.ParticipantID,
+	trackID livekit.TrackID,
+	mimeType string,
+	layer int,
+	stats *livekit.RTPStats,
+) {
+	t.enqueue(func() {
+		room := t.getRoomDetails(participantID)
+		ev := newRoomEvent(livekit.AnalyticsEventType_TRACK_PUBLISH_STATS, room)
+		ev.ParticipantId = string(participantID)
+		ev.TrackId = string(trackID)
+		ev.Mime = mimeType
+		ev.VideoLayer = int32(layer)
+		ev.RtpStats = stats
+		t.SendEvent(ctx, ev)
+	})
+}
+
+func (t *telemetryService) TrackSubscribeRTPStats(
+	ctx context.Context,
+	participantID livekit.ParticipantID,
+	trackID livekit.TrackID,
+	mimeType string,
+	stats *livekit.RTPStats,
+) {
+	t.enqueue(func() {
+		room := t.getRoomDetails(participantID)
+		ev := newRoomEvent(livekit.AnalyticsEventType_TRACK_SUBSCRIBE_STATS, room)
+		ev.ParticipantId = string(participantID)
+		ev.TrackId = string(trackID)
+		ev.Mime = mimeType
+		ev.RtpStats = stats
+		t.SendEvent(ctx, ev)
+	})
+}
+
 func (t *telemetryService) EgressStarted(ctx context.Context, info *livekit.EgressInfo) {
 	t.enqueue(func() {
 		t.NotifyEvent(ctx, &livekit.WebhookEvent{

pkg/telemetry/statsworker.go

@@ -80,11 +80,6 @@ func (s *StatsWorker) Flush() {
 	ts := timestamppb.Now()
 
 	s.lock.Lock()
-	if !s.closedAt.IsZero() {
-		s.lock.Unlock()
-		return
-	}
-
 	stats := make([]*livekit.AnalyticsStat, 0, len(s.incomingPerTrack)+len(s.outgoingPerTrack))
 
 	incomingPerTrack := s.incomingPerTrack

pkg/telemetry/telemetryfakes/fake_telemetry_service.go

@@ -107,6 +107,16 @@ type FakeTelemetryService struct {
 		arg2 livekit.ParticipantID
 		arg3 *livekit.TrackInfo
 	}
+	TrackPublishRTPStatsStub        func(context.Context, livekit.ParticipantID, livekit.TrackID, string, int, *livekit.RTPStats)
+	trackPublishRTPStatsMutex       sync.RWMutex
+	trackPublishRTPStatsArgsForCall []struct {
+		arg1 context.Context
+		arg2 livekit.ParticipantID
+		arg3 livekit.TrackID
+		arg4 string
+		arg5 int
+		arg6 *livekit.RTPStats
+	}
 	TrackPublishRequestedStub        func(context.Context, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo)
 	trackPublishRequestedMutex       sync.RWMutex
 	trackPublishRequestedArgsForCall []struct {
@@ -145,6 +155,15 @@ type FakeTelemetryService struct {
 		arg4 error
 		arg5 bool
 	}
+	TrackSubscribeRTPStatsStub        func(context.Context, livekit.ParticipantID, livekit.TrackID, string, *livekit.RTPStats)
+	trackSubscribeRTPStatsMutex       sync.RWMutex
+	trackSubscribeRTPStatsArgsForCall []struct {
+		arg1 context.Context
+		arg2 livekit.ParticipantID
+		arg3 livekit.TrackID
+		arg4 string
+		arg5 *livekit.RTPStats
+	}
 	TrackSubscribeRequestedStub        func(context.Context, livekit.ParticipantID, *livekit.TrackInfo)
 	trackSubscribeRequestedMutex       sync.RWMutex
 	trackSubscribeRequestedArgsForCall []struct {
@@ -662,6 +681,43 @@ func (fake *FakeTelemetryService) TrackMutedArgsForCall(i int) (context.Context,
 	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
 }
 
+func (fake *FakeTelemetryService) TrackPublishRTPStats(arg1 context.Context, arg2 livekit.ParticipantID, arg3 livekit.TrackID, arg4 string, arg5 int, arg6 *livekit.RTPStats) {
+	fake.trackPublishRTPStatsMutex.Lock()
+	fake.trackPublishRTPStatsArgsForCall = append(fake.trackPublishRTPStatsArgsForCall, struct {
+		arg1 context.Context
+		arg2 livekit.ParticipantID
+		arg3 livekit.TrackID
+		arg4 string
+		arg5 int
+		arg6 *livekit.RTPStats
+	}{arg1, arg2, arg3, arg4, arg5, arg6})
+	stub := fake.TrackPublishRTPStatsStub
+	fake.recordInvocation("TrackPublishRTPStats", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6})
+	fake.trackPublishRTPStatsMutex.Unlock()
+	if stub != nil {
+		fake.TrackPublishRTPStatsStub(arg1, arg2, arg3, arg4, arg5, arg6)
+	}
+}
+
+func (fake *FakeTelemetryService) TrackPublishRTPStatsCallCount() int {
+	fake.trackPublishRTPStatsMutex.RLock()
+	defer fake.trackPublishRTPStatsMutex.RUnlock()
+	return len(fake.trackPublishRTPStatsArgsForCall)
+}
+
+func (fake *FakeTelemetryService) TrackPublishRTPStatsCalls(stub func(context.Context, livekit.ParticipantID, livekit.TrackID, string, int, *livekit.RTPStats)) {
+	fake.trackPublishRTPStatsMutex.Lock()
+	defer fake.trackPublishRTPStatsMutex.Unlock()
+	fake.TrackPublishRTPStatsStub = stub
+}
+
+func (fake *FakeTelemetryService) TrackPublishRTPStatsArgsForCall(i int) (context.Context, livekit.ParticipantID, livekit.TrackID, string, int, *livekit.RTPStats) {
+	fake.trackPublishRTPStatsMutex.RLock()
+	defer fake.trackPublishRTPStatsMutex.RUnlock()
+	argsForCall := fake.trackPublishRTPStatsArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6
+}
+
 func (fake *FakeTelemetryService) TrackPublishRequested(arg1 context.Context, arg2 livekit.ParticipantID, arg3 livekit.ParticipantIdentity, arg4 *livekit.TrackInfo) {
 	fake.trackPublishRequestedMutex.Lock()
 	fake.trackPublishRequestedArgsForCall = append(fake.trackPublishRequestedArgsForCall, struct {
@@ -835,6 +891,42 @@ func (fake *FakeTelemetryService) TrackSubscribeFailedArgsForCall(i int) (contex
 	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5
 }
 
+func (fake *FakeTelemetryService) TrackSubscribeRTPStats(arg1 context.Context, arg2 livekit.ParticipantID, arg3 livekit.TrackID, arg4 string, arg5 *livekit.RTPStats) {
+	fake.trackSubscribeRTPStatsMutex.Lock()
+	fake.trackSubscribeRTPStatsArgsForCall = append(fake.trackSubscribeRTPStatsArgsForCall, struct {
+		arg1 context.Context
+		arg2 livekit.ParticipantID
+		arg3 livekit.TrackID
+		arg4 string
+		arg5 *livekit.RTPStats
+	}{arg1, arg2, arg3, arg4, arg5})
+	stub := fake.TrackSubscribeRTPStatsStub
+	fake.recordInvocation("TrackSubscribeRTPStats", []interface{}{arg1, arg2, arg3, arg4, arg5})
+	fake.trackSubscribeRTPStatsMutex.Unlock()
+	if stub != nil {
+		fake.TrackSubscribeRTPStatsStub(arg1, arg2, arg3, arg4, arg5)
+	}
+}
+
+func (fake *FakeTelemetryService) TrackSubscribeRTPStatsCallCount() int {
+	fake.trackSubscribeRTPStatsMutex.RLock()
+	defer fake.trackSubscribeRTPStatsMutex.RUnlock()
+	return len(fake.trackSubscribeRTPStatsArgsForCall)
+}
+
+func (fake *FakeTelemetryService) TrackSubscribeRTPStatsCalls(stub func(context.Context, livekit.ParticipantID, livekit.TrackID, string, *livekit.RTPStats)) {
+	fake.trackSubscribeRTPStatsMutex.Lock()
+	defer fake.trackSubscribeRTPStatsMutex.Unlock()
+	fake.TrackSubscribeRTPStatsStub = stub
+}
+
+func (fake *FakeTelemetryService) TrackSubscribeRTPStatsArgsForCall(i int) (context.Context, livekit.ParticipantID, livekit.TrackID, string, *livekit.RTPStats) {
+	fake.trackSubscribeRTPStatsMutex.RLock()
+	defer fake.trackSubscribeRTPStatsMutex.RUnlock()
+	argsForCall := fake.trackSubscribeRTPStatsArgsForCall[i]
+	return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5
+}
+
 func (fake *FakeTelemetryService) TrackSubscribeRequested(arg1 context.Context, arg2 livekit.ParticipantID, arg3 *livekit.TrackInfo) {
 	fake.trackSubscribeRequestedMutex.Lock()
 	fake.trackSubscribeRequestedArgsForCall = append(fake.trackSubscribeRequestedArgsForCall, struct {
@@ -1041,6 +1133,8 @@ func (fake *FakeTelemetryService) Invocations() map[string][][]interface{} {
 	defer fake.trackMaxSubscribedVideoQualityMutex.RUnlock()
 	fake.trackMutedMutex.RLock()
 	defer fake.trackMutedMutex.RUnlock()
+	fake.trackPublishRTPStatsMutex.RLock()
+	defer fake.trackPublishRTPStatsMutex.RUnlock()
 	fake.trackPublishRequestedMutex.RLock()
 	defer fake.trackPublishRequestedMutex.RUnlock()
 	fake.trackPublishedMutex.RLock()
@@ -1051,6 +1145,8 @@ func (fake *FakeTelemetryService) Invocations() map[string][][]interface{} {
 	defer fake.trackStatsMutex.RUnlock()
 	fake.trackSubscribeFailedMutex.RLock()
 	defer fake.trackSubscribeFailedMutex.RUnlock()
+	fake.trackSubscribeRTPStatsMutex.RLock()
+	defer fake.trackSubscribeRTPStatsMutex.RUnlock()
 	fake.trackSubscribeRequestedMutex.RLock()
 	defer fake.trackSubscribeRequestedMutex.RUnlock()
 	fake.trackSubscribedMutex.RLock()

pkg/telemetry/telemetryservice.go

@@ -51,6 +51,8 @@ type TelemetryService interface {
 	TrackPublishedUpdate(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo)
 	// TrackMaxSubscribedVideoQuality - publisher is notified of the max quality subscribers desire
 	TrackMaxSubscribedVideoQuality(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, mime string, maxQuality livekit.VideoQuality)
+	TrackPublishRTPStats(ctx context.Context, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType string, layer int, stats *livekit.RTPStats)
+	TrackSubscribeRTPStats(ctx context.Context, participantID livekit.ParticipantID, trackID livekit.TrackID, mimeType string, stats *livekit.RTPStats)
 	EgressStarted(ctx context.Context, info *livekit.EgressInfo)
 	EgressEnded(ctx context.Context, info *livekit.EgressInfo)