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)