dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-27 22:34:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Speed up track publication (#2952)

Browse Source 
* 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
This commit is contained in:
cnderrauber
2024-08-23 18:38:32 +08:00
committed by GitHub
parent 37c6f95fe4
commit 947e8f5909
11 changed files with 159 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+1 -1
View File
@@ -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
go.sum
+2 -2
View File
@@ -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=
pkg/rtc/mediaengine.go
-12
View File
@@ -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
pkg/rtc/participant.go
+54 -13
View File
@@ -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
pkg/rtc/participant_internal_test.go
+1 -1
View File
@@ -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)
})
pkg/rtc/participant_sdp.go
+3 -3
View File
@@ -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()
pkg/rtc/room.go
+7 -6
View File
@@ -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(),
}
}
pkg/rtc/subscriptionmanager.go
+8
View File
@@ -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{
pkg/rtc/types/interfaces.go
+1
View File
@@ -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)
pkg/rtc/types/typesfakes/fake_local_participant.go
+65
View File
@@ -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()
pkg/telemetry/prometheus/rooms.go
+17
View File
@@ -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)