From 947e8f59099378bbc17fcc5c41eadccef4462498 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Fri, 23 Aug 2024 10:38:32 +0000 Subject: [PATCH] Speed up track publication (#2952) * speed up track publication Add metrics for track publication and subscription Return EnabledCodecs in JoinResponse so client can choose codec without server side codec fallback Cache remote webrtc track without AddTrackRequest to let client send publisher offer before AddTrackRequest response * go mod * clean code --- go.mod | 2 +- go.sum | 4 +- pkg/rtc/mediaengine.go | 12 ---- pkg/rtc/participant.go | 67 +++++++++++++++---- pkg/rtc/participant_internal_test.go | 2 +- pkg/rtc/participant_sdp.go | 6 +- pkg/rtc/room.go | 13 ++-- pkg/rtc/subscriptionmanager.go | 8 +++ pkg/rtc/types/interfaces.go | 1 + .../typesfakes/fake_local_participant.go | 65 ++++++++++++++++++ pkg/telemetry/prometheus/rooms.go | 17 +++++ 11 files changed, 159 insertions(+), 38 deletions(-) diff --git a/go.mod b/go.mod index 8664e1ab4..0565a8626 100644 --- a/go.mod +++ b/go.mod @@ -19,7 +19,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-20240730083616-559fa5ece598 - github.com/livekit/protocol v1.20.1-0.20240813123848-0072ee0c6e47 + github.com/livekit/protocol v1.20.1-0.20240823101247-81856d28076a github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a github.com/mackerelio/go-osstat v0.2.5 github.com/magefile/mage v1.15.0 diff --git a/go.sum b/go.sum index 26e3d61a3..8730877de 100644 --- a/go.sum +++ b/go.sum @@ -169,8 +169,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-20240730083616-559fa5ece598 h1:yLlkHk2feSLHstD9n4VKg7YEBR4rLODTI4WE8gNBEnQ= github.com/livekit/mediatransportutil v0.0.0-20240730083616-559fa5ece598/go.mod h1:jwKUCmObuiEDH0iiuJHaGMXwRs3RjrB4G6qqgkr/5oE= -github.com/livekit/protocol v1.20.1-0.20240813123848-0072ee0c6e47 h1:Hsur+//Q0Ll/JfWydKpXxjwDF0FnTzJYOsL0D7s8sAc= -github.com/livekit/protocol v1.20.1-0.20240813123848-0072ee0c6e47/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI= +github.com/livekit/protocol v1.20.1-0.20240823101247-81856d28076a h1:S3nK/EXOfKdKDZXSwLhzyH9w6gZj5CGeegC4UpuSz78= +github.com/livekit/protocol v1.20.1-0.20240823101247-81856d28076a/go.mod h1:AFuwk3+uIWFeO5ohKjx5w606Djl940+wktaZ441VoCI= github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a h1:EQAHmcYEGlc6V517cQ3Iy0+jHgP6+tM/B4l2vGuLpQo= github.com/livekit/psrpc v0.5.3-0.20240616012458-ac39c8549a0a/go.mod h1:CQUBSPfYYAaevg1TNCc6/aYsa8DJH4jSRFdCeSZk5u0= github.com/mackerelio/go-osstat v0.2.5 h1:+MqTbZUhoIt4m8qzkVoXUJg1EuifwlAJSk4Yl2GXh+o= diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index a9b1d623b..4dedc085a 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -19,7 +19,6 @@ import ( "strings" "github.com/pion/webrtc/v3" - "golang.org/x/exp/slices" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/protocol/livekit" @@ -204,17 +203,6 @@ func IsCodecEnabled(codecs []*livekit.Codec, cap webrtc.RTPCodecCapability) bool } func selectAlternativeVideoCodec(enabledCodecs []*livekit.Codec) string { - // sort these by compatibility, since we are looking for backups - if slices.ContainsFunc(enabledCodecs, func(c *livekit.Codec) bool { - return strings.EqualFold(c.Mime, webrtc.MimeTypeVP8) - }) { - return webrtc.MimeTypeVP8 - } - if slices.ContainsFunc(enabledCodecs, func(c *livekit.Codec) bool { - return strings.EqualFold(c.Mime, webrtc.MimeTypeH264) - }) { - return webrtc.MimeTypeH264 - } for _, c := range enabledCodecs { if strings.HasPrefix(c.Mime, "video/") { return c.Mime diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 0674d2274..5179660d2 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -19,6 +19,7 @@ import ( "fmt" "io" "os" + "slices" "strconv" "strings" "sync" @@ -69,6 +70,12 @@ const ( type pendingTrackInfo struct { trackInfos []*livekit.TrackInfo migrated bool + createdAt time.Time +} + +type pendingRemoteTrack struct { + track *webrtc.TrackRemote + receiver *webrtc.RTPReceiver } type downTrackState struct { @@ -183,6 +190,7 @@ type ParticipantImpl struct { pendingTracksLock utils.RWMutex pendingTracks map[string]*pendingTrackInfo pendingPublishingTracks map[livekit.TrackID]*pendingTrackInfo + pendingRemoteTracks []*pendingRemoteTrack // supported codecs enabledPublishCodecs []*livekit.Codec @@ -851,14 +859,15 @@ func (p *ParticipantImpl) AddTrack(req *livekit.AddTrackRequest) { } p.pendingTracksLock.Lock() - defer p.pendingTracksLock.Unlock() - ti := p.addPendingTrackLocked(req) + p.pendingTracksLock.Unlock() if ti == nil { return } p.sendTrackPublished(req.Cid, ti) + + p.handlePendingRemoteTracks() } func (p *ParticipantImpl) SetMigrateInfo( @@ -875,7 +884,7 @@ func (p *ParticipantImpl) SetMigrateInfo( p.supervisor.SetPublicationMute(livekit.TrackID(ti.Sid), ti.Muted) } - p.pendingTracks[t.GetCid()] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}, migrated: true} + p.pendingTracks[t.GetCid()] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}, migrated: true, createdAt: time.Now()} p.pubLogger.Infow("pending track added (migration)", "trackID", ti.Sid, "track", logger.Proto(ti)) } p.pendingTracksLock.Unlock() @@ -1513,7 +1522,10 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w publishedTrack, isNewTrack := p.mediaTrackReceived(track, rtpReceiver) if publishedTrack == nil { - p.pubLogger.Warnw("webrtc Track published but can't find MediaTrack", nil, + p.pendingTracksLock.Lock() + p.pendingRemoteTracks = append(p.pendingRemoteTracks, &pendingRemoteTrack{track: track, receiver: rtpReceiver}) + p.pendingTracksLock.Unlock() + p.pubLogger.Debugw("webrtc Track published but can't find MediaTrack, add to pendingTracks", "kind", track.Kind().String(), "webrtcTrackID", track.ID(), "rid", track.RID(), @@ -1551,6 +1563,16 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w } } +func (p *ParticipantImpl) handlePendingRemoteTracks() { + p.pendingTracksLock.Lock() + pendingTracks := p.pendingRemoteTracks + p.pendingRemoteTracks = nil + p.pendingTracksLock.Unlock() + for _, rt := range pendingTracks { + p.onMediaTrack(rt.track, rt.receiver) + } +} + func (p *ParticipantImpl) onDataMessage(kind livekit.DataPacket_Kind, data []byte) { if p.IsDisconnected() || !p.CanPublishData() { return @@ -1936,7 +1958,7 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l } if p.getPublishedTrackBySignalCid(req.Cid) != nil || p.getPublishedTrackBySdpCid(req.Cid) != nil || p.pendingTracks[req.Cid] != nil { if p.pendingTracks[req.Cid] == nil { - p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}} + p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}, createdAt: time.Now()} } else { p.pendingTracks[req.Cid].trackInfos = append(p.pendingTracks[req.Cid].trackInfos, ti) } @@ -1944,7 +1966,7 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l return nil } - p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}} + p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}, createdAt: time.Now()} p.pubLogger.Debugw("pending track added", "trackID", ti.Sid, "track", logger.Proto(ti), "request", logger.Proto(req)) return ti } @@ -2048,9 +2070,10 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei } // use existing media track to handle simulcast + var pubTime time.Duration mt, ok := p.getPublishedTrackBySdpCid(track.ID()).(*MediaTrack) if !ok { - signalCid, ti, migrated := p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind())) + signalCid, ti, migrated, createdAt := p.getPendingTrack(track.ID(), ToProtoTrackKind(track.Kind())) if ti == nil { p.pendingTracksLock.Unlock() return nil, false @@ -2083,6 +2106,7 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei } mt = p.addMediaTrack(signalCid, track.ID(), ti) newTrack = true + pubTime = time.Since(createdAt) p.dirty.Store(true) } @@ -2096,7 +2120,9 @@ func (p *ParticipantImpl) mediaTrackReceived(track *webrtc.TrackRemote, rtpRecei "track published", "trackID", mt.ID(), "track", logger.Proto(mt.ToProto()), + "cost", pubTime.Milliseconds(), ) + prometheus.RecordPublishTime(mt.Source(), mt.Kind(), pubTime) p.handleTrackPublished(mt) }() } @@ -2279,7 +2305,7 @@ func (p *ParticipantImpl) onUpTrackManagerClose() { p.pubRTCPQueue.Stop() } -func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackType) (string, *livekit.TrackInfo, bool) { +func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackType) (string, *livekit.TrackInfo, bool, time.Time) { signalCid := clientId pendingInfo := p.pendingTracks[clientId] if pendingInfo == nil { @@ -2314,11 +2340,10 @@ func (p *ParticipantImpl) getPendingTrack(clientId string, kind livekit.TrackTyp // if still not found, we are done if pendingInfo == nil { - p.pubLogger.Errorw("track info not published prior to track", nil, "clientId", clientId) - return signalCid, nil, false + return signalCid, nil, false, time.Time{} } - return signalCid, pendingInfo.trackInfos[0], pendingInfo.migrated + return signalCid, pendingInfo.trackInfos[0], pendingInfo.migrated, pendingInfo.createdAt } // setStableTrackID either generates a new TrackID or reuses a previously used one @@ -2668,9 +2693,21 @@ func (p *ParticipantImpl) setupEnabledCodecs(publishEnabledCodecs []*livekit.Cod if shouldDisable(c, disabledCodecs.GetCodecs()) || shouldDisable(c, disabledCodecs.GetPublish()) { continue } - publishCodecs = append(publishCodecs, c) + + // sort by compatibility, since we will look for backups in these. + if strings.EqualFold(c.Mime, webrtc.MimeTypeVP8) { + if len(p.enabledPublishCodecs) > 0 { + p.enabledPublishCodecs = slices.Insert(p.enabledPublishCodecs, 0, c) + } else { + p.enabledPublishCodecs = append(p.enabledPublishCodecs, c) + } + } else if strings.EqualFold(c.Mime, webrtc.MimeTypeH264) { + p.enabledPublishCodecs = append(p.enabledPublishCodecs, c) + } else { + publishCodecs = append(publishCodecs, c) + } } - p.enabledPublishCodecs = publishCodecs + p.enabledPublishCodecs = append(p.enabledPublishCodecs, publishCodecs...) subscribeCodecs := make([]*livekit.Codec, 0, len(subscribeEnabledCodecs)) for _, c := range subscribeEnabledCodecs { @@ -2682,6 +2719,10 @@ func (p *ParticipantImpl) setupEnabledCodecs(publishEnabledCodecs []*livekit.Cod p.enabledSubscribeCodecs = subscribeCodecs } +func (p *ParticipantImpl) GetEnabledPublishCodecs() []*livekit.Codec { + return p.enabledPublishCodecs +} + func (p *ParticipantImpl) UpdateAudioTrack(update *livekit.UpdateLocalAudioTrack) error { if track := p.UpTrackManager.UpdatePublishedAudioTrack(update); track != nil { return nil diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 217213029..42b7ee12f 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -286,7 +286,7 @@ func TestMuteSetting(t *testing.T) { Muted: true, }) - _, ti, _ := p.getPendingTrack("cid", livekit.TrackType_AUDIO) + _, ti, _, _ := p.getPendingTrack("cid", livekit.TrackType_AUDIO) require.NotNil(t, ti) require.True(t, ti.Muted) }) diff --git a/pkg/rtc/participant_sdp.go b/pkg/rtc/participant_sdp.go index 0ee4b60ae..a57bc84e7 100644 --- a/pkg/rtc/participant_sdp.go +++ b/pkg/rtc/participant_sdp.go @@ -45,7 +45,7 @@ func (p *ParticipantImpl) setCodecPreferencesOpusRedForPublisher(offer webrtc.Se } p.pendingTracksLock.RLock() - _, info, _ := p.getPendingTrack(streamID, livekit.TrackType_AUDIO) + _, info, _, _ := p.getPendingTrack(streamID, livekit.TrackType_AUDIO) // if RED is disabled for this track, don't prefer RED codec in offer disableRed := info != nil && info.DisableRed p.pendingTracksLock.RUnlock() @@ -131,7 +131,7 @@ func (p *ParticipantImpl) setCodecPreferencesVideoForPublisher(offer webrtc.Sess if mt != nil { info = mt.ToProto() } else { - _, info, _ = p.getPendingTrack(streamID, livekit.TrackType_VIDEO) + _, info, _, _ = p.getPendingTrack(streamID, livekit.TrackType_VIDEO) } if info == nil { @@ -227,7 +227,7 @@ func (p *ParticipantImpl) configurePublisherAnswer(answer webrtc.SessionDescript track, _ := p.getPublishedTrackBySdpCid(streamID).(*MediaTrack) if track == nil { p.pendingTracksLock.RLock() - _, ti, _ = p.getPendingTrack(streamID, livekit.TrackType_AUDIO) + _, ti, _, _ = p.getPendingTrack(streamID, livekit.TrackType_AUDIO) p.pendingTracksLock.RUnlock() } else { ti = track.ToProto() diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 8b2933842..329b6f1e8 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -1148,12 +1148,13 @@ func (r *Room) createJoinResponseLocked(participant types.LocalParticipant, iceS SubscriberPrimary: participant.SubscriberAsPrimary(), ClientConfiguration: participant.GetClientConfiguration(), // sane defaults for ping interval & timeout - PingInterval: PingIntervalSeconds, - PingTimeout: PingTimeoutSeconds, - ServerInfo: r.serverInfo, - ServerVersion: r.serverInfo.Version, - ServerRegion: r.serverInfo.Region, - SifTrailer: r.trailer, + PingInterval: PingIntervalSeconds, + PingTimeout: PingTimeoutSeconds, + ServerInfo: r.serverInfo, + ServerVersion: r.serverInfo.Version, + ServerRegion: r.serverInfo.Region, + SifTrailer: r.trailer, + EnabledPublishCodecs: participant.GetEnabledPublishCodecs(), } } diff --git a/pkg/rtc/subscriptionmanager.go b/pkg/rtc/subscriptionmanager.go index 37ca19561..6a3b27562 100644 --- a/pkg/rtc/subscriptionmanager.go +++ b/pkg/rtc/subscriptionmanager.go @@ -28,6 +28,7 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" "github.com/livekit/livekit-server/pkg/telemetry" + "github.com/livekit/livekit-server/pkg/telemetry/prometheus" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" ) @@ -758,6 +759,8 @@ type trackSubscription struct { // the later of when subscription was requested OR when the first failure was encountered OR when permission is granted // this timestamp determines when failures are reported subStartedAt atomic.Pointer[time.Time] + + createAt time.Time } func newTrackSubscription(subscriberID livekit.ParticipantID, trackID livekit.TrackID, l logger.Logger) *trackSubscription { @@ -765,6 +768,7 @@ func newTrackSubscription(subscriberID livekit.ParticipantID, trackID livekit.Tr subscriberID: subscriberID, trackID: trackID, logger: l, + createAt: time.Now(), } } @@ -993,6 +997,10 @@ func (s *trackSubscription) maybeRecordSuccess(ts telemetry.TelemetryService, pI return } + d := time.Since(s.createAt) + s.logger.Debugw("track subscribed", "cost", d.Milliseconds()) + prometheus.RecordSubscribeTime(mediaTrack.Source(), mediaTrack.Kind(), d) + eventSent := s.eventSent.Swap(true) pi := &livekit.ParticipantInfo{ diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 4e46791f9..7077420d9 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -323,6 +323,7 @@ type LocalParticipant interface { GetPendingTrack(trackID livekit.TrackID) *livekit.TrackInfo GetICEConnectionDetails() []*ICEConnectionDetails HasConnected() bool + GetEnabledPublishCodecs() []*livekit.Codec SetResponseSink(sink routing.MessageSink) CloseSignalConnection(reason SignallingCloseReason) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 94188dca6..e4bf3fa47 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -276,6 +276,16 @@ type FakeLocalParticipant struct { getDisableSenderReportPassThroughReturnsOnCall map[int]struct { result1 bool } + GetEnabledPublishCodecsStub func() []*livekit.Codec + getEnabledPublishCodecsMutex sync.RWMutex + getEnabledPublishCodecsArgsForCall []struct { + } + getEnabledPublishCodecsReturns struct { + result1 []*livekit.Codec + } + getEnabledPublishCodecsReturnsOnCall map[int]struct { + result1 []*livekit.Codec + } GetICEConnectionDetailsStub func() []*types.ICEConnectionDetails getICEConnectionDetailsMutex sync.RWMutex getICEConnectionDetailsArgsForCall []struct { @@ -2379,6 +2389,59 @@ func (fake *FakeLocalParticipant) GetDisableSenderReportPassThroughReturnsOnCall }{result1} } +func (fake *FakeLocalParticipant) GetEnabledPublishCodecs() []*livekit.Codec { + fake.getEnabledPublishCodecsMutex.Lock() + ret, specificReturn := fake.getEnabledPublishCodecsReturnsOnCall[len(fake.getEnabledPublishCodecsArgsForCall)] + fake.getEnabledPublishCodecsArgsForCall = append(fake.getEnabledPublishCodecsArgsForCall, struct { + }{}) + stub := fake.GetEnabledPublishCodecsStub + fakeReturns := fake.getEnabledPublishCodecsReturns + fake.recordInvocation("GetEnabledPublishCodecs", []interface{}{}) + fake.getEnabledPublishCodecsMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetEnabledPublishCodecsCallCount() int { + fake.getEnabledPublishCodecsMutex.RLock() + defer fake.getEnabledPublishCodecsMutex.RUnlock() + return len(fake.getEnabledPublishCodecsArgsForCall) +} + +func (fake *FakeLocalParticipant) GetEnabledPublishCodecsCalls(stub func() []*livekit.Codec) { + fake.getEnabledPublishCodecsMutex.Lock() + defer fake.getEnabledPublishCodecsMutex.Unlock() + fake.GetEnabledPublishCodecsStub = stub +} + +func (fake *FakeLocalParticipant) GetEnabledPublishCodecsReturns(result1 []*livekit.Codec) { + fake.getEnabledPublishCodecsMutex.Lock() + defer fake.getEnabledPublishCodecsMutex.Unlock() + fake.GetEnabledPublishCodecsStub = nil + fake.getEnabledPublishCodecsReturns = struct { + result1 []*livekit.Codec + }{result1} +} + +func (fake *FakeLocalParticipant) GetEnabledPublishCodecsReturnsOnCall(i int, result1 []*livekit.Codec) { + fake.getEnabledPublishCodecsMutex.Lock() + defer fake.getEnabledPublishCodecsMutex.Unlock() + fake.GetEnabledPublishCodecsStub = nil + if fake.getEnabledPublishCodecsReturnsOnCall == nil { + fake.getEnabledPublishCodecsReturnsOnCall = make(map[int]struct { + result1 []*livekit.Codec + }) + } + fake.getEnabledPublishCodecsReturnsOnCall[i] = struct { + result1 []*livekit.Codec + }{result1} +} + func (fake *FakeLocalParticipant) GetICEConnectionDetails() []*types.ICEConnectionDetails { fake.getICEConnectionDetailsMutex.Lock() ret, specificReturn := fake.getICEConnectionDetailsReturnsOnCall[len(fake.getICEConnectionDetailsArgsForCall)] @@ -6774,6 +6837,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getConnectionQualityMutex.RUnlock() fake.getDisableSenderReportPassThroughMutex.RLock() defer fake.getDisableSenderReportPassThroughMutex.RUnlock() + fake.getEnabledPublishCodecsMutex.RLock() + defer fake.getEnabledPublishCodecsMutex.RUnlock() fake.getICEConnectionDetailsMutex.RLock() defer fake.getICEConnectionDetailsMutex.RUnlock() fake.getLoggerMutex.RLock() diff --git a/pkg/telemetry/prometheus/rooms.go b/pkg/telemetry/prometheus/rooms.go index 7e1102026..cb4f3f989 100644 --- a/pkg/telemetry/prometheus/rooms.go +++ b/pkg/telemetry/prometheus/rooms.go @@ -46,6 +46,7 @@ var ( promTrackSubscribeCounter *prometheus.CounterVec promSessionStartTime *prometheus.HistogramVec promSessionDuration *prometheus.HistogramVec + promPubSubTime *prometheus.HistogramVec ) func initRoomStats(nodeID string, nodeType livekit.NodeType) { @@ -108,6 +109,13 @@ func initRoomStats(nodeID string, nodeType livekit.NodeType) { ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, Buckets: prometheus.ExponentialBucketsRange(100, 4*60*60*1000, 15), }, []string{"protocol_version"}) + promPubSubTime = prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Namespace: livekitNamespace, + Subsystem: "pubsubtime", + Name: "ms", + ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()}, + Buckets: []float64{100, 200, 500, 700, 1000, 5000, 10000}, + }, promStreamLabels) prometheus.MustRegister(promRoomCurrent) prometheus.MustRegister(promRoomDuration) @@ -118,6 +126,7 @@ func initRoomStats(nodeID string, nodeType livekit.NodeType) { prometheus.MustRegister(promTrackSubscribeCounter) prometheus.MustRegister(promSessionStartTime) prometheus.MustRegister(promSessionDuration) + prometheus.MustRegister(promPubSubTime) } func RoomStarted() { @@ -163,6 +172,14 @@ func AddPublishSuccess(kind string) { promTrackPublishCounter.WithLabelValues(kind, "success").Inc() } +func RecordPublishTime(source livekit.TrackSource, trackType livekit.TrackType, d time.Duration) { + promPubSubTime.WithLabelValues("publish", source.String(), trackType.String()).Observe(float64(d.Milliseconds())) +} + +func RecordSubscribeTime(source livekit.TrackSource, trackType livekit.TrackType, d time.Duration) { + promPubSubTime.WithLabelValues("subscribe", source.String(), trackType.String()).Observe(float64(d.Milliseconds())) +} + func RecordTrackSubscribeSuccess(kind string) { // modify both current and total counters promTrackSubscribedCurrent.WithLabelValues(kind).Add(1)