From 5ff72a99b9cec69cf65673ec102c7064f59e1ad7 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Fri, 10 Mar 2023 10:28:54 -0800 Subject: [PATCH] Report publish & subscribe RTPStats as Telemetry events (#1506) --- go.mod | 2 +- go.sum | 4 +- pkg/rtc/mediatrack.go | 13 ++- pkg/rtc/mediatrackreceiver.go | 25 +++-- pkg/rtc/subscriptionmanager.go | 16 +++- pkg/sfu/buffer/buffer.go | 8 ++ pkg/telemetry/events.go | 38 ++++++++ pkg/telemetry/statsworker.go | 5 - .../telemetryfakes/fake_telemetry_service.go | 96 +++++++++++++++++++ pkg/telemetry/telemetryservice.go | 2 + 10 files changed, 184 insertions(+), 25 deletions(-) diff --git a/go.mod b/go.mod index aad4f833d..2f3676438 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 95095de15..dbff2e09b 100644 --- a/go.sum +++ b/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= diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index ce117fbd3..f492930e3 100644 --- a/pkg/rtc/mediatrack.go +++ b/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() } diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index d06924e25..31a6dcf3c 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/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 { diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index da0d452ad..9e880aeb3 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/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) diff --git a/pkg/sfu/buffer/buffer.go b/pkg/sfu/buffer/buffer.go index e30633d11..dde05a51e 100644 --- a/pkg/sfu/buffer/buffer.go +++ b/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 diff --git a/pkg/telemetry/events.go b/pkg/telemetry/events.go index d1e81444f..1c78a5486 100644 --- a/pkg/telemetry/events.go +++ b/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{ diff --git a/pkg/telemetry/statsworker.go b/pkg/telemetry/statsworker.go index fd975a321..9447703b6 100644 --- a/pkg/telemetry/statsworker.go +++ b/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 diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index 72768e995..4e6cff95e 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/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() diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index c1048d5ed..bcef252ae 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/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)