dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-29 07:39:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

add idempotent reference count to telemetry stats worker (#3964)

Browse Source 
* add idempotent reference guard to telemetry stats worker

* tidy

* sync

* tidy
This commit is contained in:
Paul Wells
2025-09-29 02:35:16 -07:00
committed by GitHub
parent 735c663adc
commit 3d73703152
12 changed files with 198 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/rtc/participant.go
+8
View File
@@ -283,6 +283,9 @@ type ParticipantImpl struct {
rttUpdatedAt time.Time
lastRTT uint32
// idempotent reference guard for telemetry stats worker
telemetryGuard *telemetry.ReferenceGuard
lock utils.RWMutex
dirty atomic.Bool
@@ -369,6 +372,7 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
rpcPendingAcks: make(map[string]*utils.DataChannelRpcPendingAckHandler),
rpcPendingResponses: make(map[string]*utils.DataChannelRpcPendingResponseHandler),
onClose: make(map[string]func(types.LocalParticipant)),
telemetryGuard: &telemetry.ReferenceGuard{},
}
p.setupSignalling()
@@ -884,6 +888,10 @@ func (p *ParticipantImpl) ToProto() *livekit.ParticipantInfo {
return pi
}
func (p *ParticipantImpl) TelemetryGuard() *telemetry.ReferenceGuard {
return p.telemetryGuard
}
// callbacks for clients
func (p *ParticipantImpl) OnTrackPublished(callback func(types.LocalParticipant, types.MediaTrack)) {
pkg/rtc/room.go
+1
View File
@@ -491,6 +491,7 @@ func (r *Room) Join(
p.ToProto(),
meta,
false,
participant.TelemetryGuard(),
)
participant.GetReporter().Tx(func(tx roomobs.ParticipantSessionTx) {
pkg/rtc/types/interfaces.go
+2
View File
@@ -32,6 +32,7 @@ import (
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/livekit-server/pkg/sfu/mime"
"github.com/livekit/livekit-server/pkg/sfu/pacer"
"github.com/livekit/livekit-server/pkg/telemetry"
"google.golang.org/protobuf/proto"
)
@@ -353,6 +354,7 @@ type LocalParticipantHelper interface {
type LocalParticipant interface {
Participant
TelemetryGuard() *telemetry.ReferenceGuard
ToProtoWithVersion() (*livekit.ParticipantInfo, utils.TimedVersion)
// getters
pkg/rtc/types/typesfakes/fake_local_participant.go
+64
View File
@@ -10,6 +10,7 @@ import (
"github.com/livekit/livekit-server/pkg/sfu"
"github.com/livekit/livekit-server/pkg/sfu/buffer"
"github.com/livekit/livekit-server/pkg/sfu/pacer"
"github.com/livekit/livekit-server/pkg/telemetry"
"github.com/livekit/protocol/auth"
"github.com/livekit/protocol/livekit"
"github.com/livekit/protocol/logger"
@@ -1257,6 +1258,16 @@ type FakeLocalParticipant struct {
supportsTransceiverReuseReturnsOnCall map[int]struct {
result1 bool
}
TelemetryGuardStub func() *telemetry.ReferenceGuard
telemetryGuardMutex sync.RWMutex
telemetryGuardArgsForCall []struct {
}
telemetryGuardReturns struct {
result1 *telemetry.ReferenceGuard
}
telemetryGuardReturnsOnCall map[int]struct {
result1 *telemetry.ReferenceGuard
}
ToProtoStub func() *livekit.ParticipantInfo
toProtoMutex sync.RWMutex
toProtoArgsForCall []struct {
@@ -8161,6 +8172,59 @@ func (fake *FakeLocalParticipant) SupportsTransceiverReuseReturnsOnCall(i int, r
}{result1}
}
func (fake *FakeLocalParticipant) TelemetryGuard() *telemetry.ReferenceGuard {
fake.telemetryGuardMutex.Lock()
ret, specificReturn := fake.telemetryGuardReturnsOnCall[len(fake.telemetryGuardArgsForCall)]
fake.telemetryGuardArgsForCall = append(fake.telemetryGuardArgsForCall, struct {
}{})
stub := fake.TelemetryGuardStub
fakeReturns := fake.telemetryGuardReturns
fake.recordInvocation("TelemetryGuard", []interface{}{})
fake.telemetryGuardMutex.Unlock()
if stub != nil {
return stub()
}
if specificReturn {
return ret.result1
}
return fakeReturns.result1
}
func (fake *FakeLocalParticipant) TelemetryGuardCallCount() int {
fake.telemetryGuardMutex.RLock()
defer fake.telemetryGuardMutex.RUnlock()
return len(fake.telemetryGuardArgsForCall)
}
func (fake *FakeLocalParticipant) TelemetryGuardCalls(stub func() *telemetry.ReferenceGuard) {
fake.telemetryGuardMutex.Lock()
defer fake.telemetryGuardMutex.Unlock()
fake.TelemetryGuardStub = stub
}
func (fake *FakeLocalParticipant) TelemetryGuardReturns(result1 *telemetry.ReferenceGuard) {
fake.telemetryGuardMutex.Lock()
defer fake.telemetryGuardMutex.Unlock()
fake.TelemetryGuardStub = nil
fake.telemetryGuardReturns = struct {
result1 *telemetry.ReferenceGuard
}{result1}
}
func (fake *FakeLocalParticipant) TelemetryGuardReturnsOnCall(i int, result1 *telemetry.ReferenceGuard) {
fake.telemetryGuardMutex.Lock()
defer fake.telemetryGuardMutex.Unlock()
fake.TelemetryGuardStub = nil
if fake.telemetryGuardReturnsOnCall == nil {
fake.telemetryGuardReturnsOnCall = make(map[int]struct {
result1 *telemetry.ReferenceGuard
})
}
fake.telemetryGuardReturnsOnCall[i] = struct {
result1 *telemetry.ReferenceGuard
}{result1}
}
func (fake *FakeLocalParticipant) ToProto() *livekit.ParticipantInfo {
fake.toProtoMutex.Lock()
ret, specificReturn := fake.toProtoReturnsOnCall[len(fake.toProtoArgsForCall)]
pkg/service/roommanager.go
+2 -2
View File
@@ -555,7 +555,7 @@ func (r *RoomManager) StartSession(
persistRoomForParticipantCount(room.ToProto())
clientMeta := &livekit.AnalyticsClientMeta{Region: r.currentNode.Region(), Node: string(r.currentNode.NodeID())}
r.telemetry.ParticipantJoined(ctx, protoRoom, participant.ToProto(), pi.Client, clientMeta, true)
r.telemetry.ParticipantJoined(ctx, protoRoom, participant.ToProto(), pi.Client, clientMeta, true, participant.TelemetryGuard())
participant.AddOnClose(types.ParticipantCloseKeyNormal, func(p types.LocalParticipant) {
participantServerClosers.Close()
@@ -566,7 +566,7 @@ func (r *RoomManager) StartSession(
// update room store with new numParticipants
proto := room.ToProto()
persistRoomForParticipantCount(proto)
r.telemetry.ParticipantLeft(ctx, proto, p.ToProto(), true)
r.telemetry.ParticipantLeft(ctx, proto, p.ToProto(), true, participant.TelemetryGuard())
})
participant.OnClaimsChanged(func(participant types.LocalParticipant) {
pLogger.Debugw("refreshing client token after claims change")
pkg/telemetry/events.go
+7 -1
View File
@@ -80,6 +80,7 @@ func (t *telemetryService) ParticipantJoined(
clientInfo *livekit.ClientInfo,
clientMeta *livekit.AnalyticsClientMeta,
shouldSendEvent bool,
guard *ReferenceGuard,
) {
t.enqueue(func() {
_, found := t.getOrCreateWorker(
@@ -88,6 +89,7 @@ func (t *telemetryService) ParticipantJoined(
livekit.RoomName(room.Name),
livekit.ParticipantID(participant.Sid),
livekit.ParticipantIdentity(participant.Identity),
guard,
)
if !found {
prometheus.IncrementParticipantRtcConnected(1)
@@ -109,6 +111,7 @@ func (t *telemetryService) ParticipantActive(
participant *livekit.ParticipantInfo,
clientMeta *livekit.AnalyticsClientMeta,
isMigration bool,
guard *ReferenceGuard,
) {
t.enqueue(func() {
if !isMigration {
@@ -126,6 +129,7 @@ func (t *telemetryService) ParticipantActive(
livekit.RoomName(room.Name),
livekit.ParticipantID(participant.Sid),
livekit.ParticipantIdentity(participant.Identity),
guard,
)
if !found {
// need to also account for participant count
@@ -162,6 +166,7 @@ func (t *telemetryService) ParticipantResumed(
livekit.RoomName(room.Name),
livekit.ParticipantID(participant.Sid),
livekit.ParticipantIdentity(participant.Identity),
nil,
)
if !found {
prometheus.AddParticipant()
@@ -180,12 +185,13 @@ func (t *telemetryService) ParticipantLeft(ctx context.Context,
room *livekit.Room,
participant *livekit.ParticipantInfo,
shouldSendEvent bool,
guard *ReferenceGuard,
) {
t.enqueue(func() {
isConnected := false
if worker, ok := t.getWorker(livekit.ParticipantID(participant.Sid)); ok {
isConnected = worker.IsConnected()
if worker.Close() {
if worker.Close(guard) {
prometheus.SubParticipant()
}
}
pkg/telemetry/events_test.go
+11 -6
View File
@@ -21,6 +21,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/livekit/livekit-server/pkg/telemetry"
"github.com/livekit/protocol/livekit"
)
@@ -46,9 +47,10 @@ func Test_OnParticipantJoin_EventIsSent(t *testing.T) {
ClientConnectTime: 420,
}
participantInfo := &livekit.ParticipantInfo{Sid: partSID}
guard := &telemetry.ReferenceGuard{}
// do
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true)
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true, guard)
time.Sleep(time.Millisecond * 500)
// test
@@ -81,10 +83,11 @@ func Test_OnParticipantLeft_EventIsSent(t *testing.T) {
room := &livekit.Room{Sid: "RoomSid", Name: "RoomName"}
partSID := "part1"
participantInfo := &livekit.ParticipantInfo{Sid: partSID}
guard := &telemetry.ReferenceGuard{}
// do
fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false)
fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true)
fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}, false, guard)
fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard)
time.Sleep(time.Millisecond * 500)
// test
@@ -158,9 +161,10 @@ func Test_OnParticipantActive_EventIsSent(t *testing.T) {
ClientAddr: "127.0.0.1",
}
participantInfo := &livekit.ParticipantInfo{Sid: partSID}
guard := &telemetry.ReferenceGuard{}
// do
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true)
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true, guard)
time.Sleep(time.Millisecond * 500)
// test
@@ -173,7 +177,7 @@ func Test_OnParticipantActive_EventIsSent(t *testing.T) {
ClientConnectTime: 420,
}
fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false)
fixture.sut.ParticipantActive(context.Background(), room, participantInfo, clientMetaConnect, false, guard)
time.Sleep(time.Millisecond * 500)
require.Equal(t, 2, fixture.analytics.SendEventCallCount())
@@ -210,9 +214,10 @@ func Test_OnTrackSubscribed_EventIsSent(t *testing.T) {
ClientAddr: "127.0.0.1",
}
participantInfo := &livekit.ParticipantInfo{Sid: partSID}
guard := &telemetry.ReferenceGuard{}
// do
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true)
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, clientMeta, true, guard)
time.Sleep(time.Millisecond * 500)
// test
pkg/telemetry/signalanddatastats.go
+4 -2
View File
@@ -174,6 +174,8 @@ type BytesSignalStats struct {
BytesTrackStats
ctx context.Context
guard ReferenceGuard
participantResolver roomobs.ParticipantReporterResolver
trackResolver roomobs.KeyResolver
@@ -257,14 +259,14 @@ func (s *BytesSignalStats) maybeStart() {
)
s.trackResolver.Resolve(string(s.trackID))
s.telemetry.ParticipantJoined(s.ctx, s.ri, s.pi, nil, nil, false)
s.telemetry.ParticipantJoined(s.ctx, s.ri, s.pi, nil, nil, false, &s.guard)
s.stopped = make(chan struct{})
go s.worker()
}
func (s *BytesSignalStats) worker() {
s.BytesTrackStats.worker()
s.telemetry.ParticipantLeft(s.ctx, s.ri, s.pi, false)
s.telemetry.ParticipantLeft(s.ctx, s.ri, s.pi, false, &s.guard)
close(s.stopped)
}
pkg/telemetry/stats_test.go
+25 -13
View File
@@ -52,7 +52,8 @@ func Test_ParticipantAndRoomDataAreSentWithAnalytics(t *testing.T) {
partSID := livekit.ParticipantID("part1")
clientInfo := &livekit.ClientInfo{Sdk: 2}
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true, guard)
// do
packet := 33
@@ -80,7 +81,8 @@ func Test_OnDownstreamPackets(t *testing.T) {
partSID := livekit.ParticipantID("part1")
clientInfo := &livekit.ClientInfo{Sdk: 2}
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true, guard)
// do
packets := []int{33, 23}
@@ -113,7 +115,8 @@ func Test_OnDownstreamPackets_SeveralTracks(t *testing.T) {
partSID := livekit.ParticipantID("part1")
clientInfo := &livekit.ClientInfo{Sdk: 2}
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, clientInfo, nil, true, guard)
// do
packet1 := 33
@@ -158,7 +161,8 @@ func Test_OnDownStreamStat(t *testing.T) {
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
// do
stat1 := &livekit.AnalyticsStat{
@@ -217,7 +221,8 @@ func Test_PacketLostDiffShouldBeSentToTelemetry(t *testing.T) {
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
// do
trackID := livekit.TrackID("trackID1")
@@ -269,7 +274,8 @@ func Test_OnDownStreamRTCP_SeveralTracks(t *testing.T) {
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
// do
trackID1 := livekit.TrackID("trackID1")
@@ -338,7 +344,8 @@ func Test_OnUpstreamStat(t *testing.T) {
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
// do
stat1 := &livekit.AnalyticsStat{
@@ -400,7 +407,8 @@ func Test_OnUpstreamRTCP_SeveralTracks(t *testing.T) {
partSID := livekit.ParticipantID("part1")
identity := livekit.ParticipantIdentity("part1Identity")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID), Identity: string(identity)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
// there should be bytes reported so that stats are sent
totalBytes := 1
@@ -486,10 +494,11 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) {
room := &livekit.Room{}
partSID := "part1"
participantInfo := &livekit.ParticipantInfo{Sid: partSID}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
// do
fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true)
fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true, guard)
// should not be called if there are no track stats
time.Sleep(time.Millisecond * 500)
@@ -503,7 +512,8 @@ func Test_AddUpTrack(t *testing.T) {
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
// do
var totalBytes uint64 = 3
@@ -540,7 +550,8 @@ func Test_AddUpTrack_SeveralBuffers_Simulcast(t *testing.T) {
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
// do
trackID := livekit.TrackID("trackID")
@@ -579,7 +590,8 @@ func Test_BothDownstreamAndUpstreamStatsAreSentTogether(t *testing.T) {
room := &livekit.Room{}
partSID := livekit.ParticipantID("part1")
participantInfo := &livekit.ParticipantInfo{Sid: string(partSID)}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true)
guard := &telemetry.ReferenceGuard{}
fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil, true, guard)
// do
// upstream bytes
pkg/telemetry/statsworker.go
+38 -3
View File
@@ -27,6 +27,30 @@ import (
protoutils "github.com/livekit/protocol/utils"
)
type ReferenceGuard struct {
activated, released bool
}
type ReferenceCount struct {
count int
}
func (s *ReferenceCount) Activate(guard *ReferenceGuard) {
if guard != nil && !guard.activated {
guard.activated = true
s.count++
}
}
func (s *ReferenceCount) Release(guard *ReferenceGuard) bool {
if guard == nil || !guard.activated || guard.released {
return false
}
guard.released = true
s.count--
return s.count == 0
}
// StatsWorker handles participant stats
type StatsWorker struct {
next *StatsWorker
@@ -42,6 +66,7 @@ type StatsWorker struct {
lock sync.RWMutex
outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
incomingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat
refCount ReferenceCount
closedAt time.Time
}
@@ -52,6 +77,7 @@ func newStatsWorker(
roomName livekit.RoomName,
participantID livekit.ParticipantID,
identity livekit.ParticipantIdentity,
guard *ReferenceGuard,
) *StatsWorker {
s := &StatsWorker{
ctx: ctx,
@@ -63,6 +89,7 @@ func newStatsWorker(
outgoingPerTrack: make(map[livekit.TrackID][]*livekit.AnalyticsStat),
incomingPerTrack: make(map[livekit.TrackID][]*livekit.AnalyticsStat),
}
s.refCount.Activate(guard)
return s
}
@@ -117,10 +144,14 @@ func (s *StatsWorker) Flush(now time.Time) bool {
return closed
}
func (s *StatsWorker) Close() bool {
func (s *StatsWorker) Close(guard *ReferenceGuard) bool {
s.lock.Lock()
defer s.lock.Unlock()
if !s.refCount.Release(guard) {
return false
}
ok := s.closedAt.IsZero()
if ok {
s.closedAt = time.Now()
@@ -128,10 +159,14 @@ func (s *StatsWorker) Close() bool {
return ok
}
func (s *StatsWorker) Closed() bool {
func (s *StatsWorker) Closed(guard *ReferenceGuard) bool {
s.lock.Lock()
defer s.lock.Unlock()
return !s.closedAt.IsZero()
if !s.closedAt.IsZero() {
s.refCount.Activate(guard)
return false
}
return true
}
func (s *StatsWorker) collectStats(
pkg/telemetry/telemetryfakes/fake_telemetry_service.go
+30 -24
View File
@@ -83,7 +83,7 @@ type FakeTelemetryService struct {
arg2 string
arg3 *livekit.EgressInfo
}
ParticipantActiveStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool)
ParticipantActiveStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard)
participantActiveMutex sync.RWMutex
participantActiveArgsForCall []struct {
arg1 context.Context
@@ -91,8 +91,9 @@ type FakeTelemetryService struct {
arg3 *livekit.ParticipantInfo
arg4 *livekit.AnalyticsClientMeta
arg5 bool
arg6 *telemetry.ReferenceGuard
}
ParticipantJoinedStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.ClientInfo, *livekit.AnalyticsClientMeta, bool)
ParticipantJoinedStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.ClientInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard)
participantJoinedMutex sync.RWMutex
participantJoinedArgsForCall []struct {
arg1 context.Context
@@ -101,14 +102,16 @@ type FakeTelemetryService struct {
arg4 *livekit.ClientInfo
arg5 *livekit.AnalyticsClientMeta
arg6 bool
arg7 *telemetry.ReferenceGuard
}
ParticipantLeftStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool)
ParticipantLeftStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool, *telemetry.ReferenceGuard)
participantLeftMutex sync.RWMutex
participantLeftArgsForCall []struct {
arg1 context.Context
arg2 *livekit.Room
arg3 *livekit.ParticipantInfo
arg4 bool
arg5 *telemetry.ReferenceGuard
}
ParticipantResumedStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, livekit.NodeID, livekit.ReconnectReason)
participantResumedMutex sync.RWMutex
@@ -677,7 +680,7 @@ func (fake *FakeTelemetryService) NotifyEgressEventArgsForCall(i int) (context.C
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3
}
func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.AnalyticsClientMeta, arg5 bool) {
func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.AnalyticsClientMeta, arg5 bool, arg6 *telemetry.ReferenceGuard) {
fake.participantActiveMutex.Lock()
fake.participantActiveArgsForCall = append(fake.participantActiveArgsForCall, struct {
arg1 context.Context
@@ -685,12 +688,13 @@ func (fake *FakeTelemetryService) ParticipantActive(arg1 context.Context, arg2 *
arg3 *livekit.ParticipantInfo
arg4 *livekit.AnalyticsClientMeta
arg5 bool
}{arg1, arg2, arg3, arg4, arg5})
arg6 *telemetry.ReferenceGuard
}{arg1, arg2, arg3, arg4, arg5, arg6})
stub := fake.ParticipantActiveStub
fake.recordInvocation("ParticipantActive", []interface{}{arg1, arg2, arg3, arg4, arg5})
fake.recordInvocation("ParticipantActive", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6})
fake.participantActiveMutex.Unlock()
if stub != nil {
fake.ParticipantActiveStub(arg1, arg2, arg3, arg4, arg5)
fake.ParticipantActiveStub(arg1, arg2, arg3, arg4, arg5, arg6)
}
}
@@ -700,20 +704,20 @@ func (fake *FakeTelemetryService) ParticipantActiveCallCount() int {
return len(fake.participantActiveArgsForCall)
}
func (fake *FakeTelemetryService) ParticipantActiveCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool)) {
func (fake *FakeTelemetryService) ParticipantActiveCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard)) {
fake.participantActiveMutex.Lock()
defer fake.participantActiveMutex.Unlock()
fake.ParticipantActiveStub = stub
}
func (fake *FakeTelemetryService) ParticipantActiveArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool) {
func (fake *FakeTelemetryService) ParticipantActiveArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard) {
fake.participantActiveMutex.RLock()
defer fake.participantActiveMutex.RUnlock()
argsForCall := fake.participantActiveArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6
}
func (fake *FakeTelemetryService) ParticipantJoined(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.ClientInfo, arg5 *livekit.AnalyticsClientMeta, arg6 bool) {
func (fake *FakeTelemetryService) ParticipantJoined(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 *livekit.ClientInfo, arg5 *livekit.AnalyticsClientMeta, arg6 bool, arg7 *telemetry.ReferenceGuard) {
fake.participantJoinedMutex.Lock()
fake.participantJoinedArgsForCall = append(fake.participantJoinedArgsForCall, struct {
arg1 context.Context
@@ -722,12 +726,13 @@ func (fake *FakeTelemetryService) ParticipantJoined(arg1 context.Context, arg2 *
arg4 *livekit.ClientInfo
arg5 *livekit.AnalyticsClientMeta
arg6 bool
}{arg1, arg2, arg3, arg4, arg5, arg6})
arg7 *telemetry.ReferenceGuard
}{arg1, arg2, arg3, arg4, arg5, arg6, arg7})
stub := fake.ParticipantJoinedStub
fake.recordInvocation("ParticipantJoined", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6})
fake.recordInvocation("ParticipantJoined", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7})
fake.participantJoinedMutex.Unlock()
if stub != nil {
fake.ParticipantJoinedStub(arg1, arg2, arg3, arg4, arg5, arg6)
fake.ParticipantJoinedStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7)
}
}
@@ -737,32 +742,33 @@ func (fake *FakeTelemetryService) ParticipantJoinedCallCount() int {
return len(fake.participantJoinedArgsForCall)
}
func (fake *FakeTelemetryService) ParticipantJoinedCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.ClientInfo, *livekit.AnalyticsClientMeta, bool)) {
func (fake *FakeTelemetryService) ParticipantJoinedCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.ClientInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard)) {
fake.participantJoinedMutex.Lock()
defer fake.participantJoinedMutex.Unlock()
fake.ParticipantJoinedStub = stub
}
func (fake *FakeTelemetryService) ParticipantJoinedArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.ClientInfo, *livekit.AnalyticsClientMeta, bool) {
func (fake *FakeTelemetryService) ParticipantJoinedArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, *livekit.ClientInfo, *livekit.AnalyticsClientMeta, bool, *telemetry.ReferenceGuard) {
fake.participantJoinedMutex.RLock()
defer fake.participantJoinedMutex.RUnlock()
argsForCall := fake.participantJoinedArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7
}
func (fake *FakeTelemetryService) ParticipantLeft(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 bool) {
func (fake *FakeTelemetryService) ParticipantLeft(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 bool, arg5 *telemetry.ReferenceGuard) {
fake.participantLeftMutex.Lock()
fake.participantLeftArgsForCall = append(fake.participantLeftArgsForCall, struct {
arg1 context.Context
arg2 *livekit.Room
arg3 *livekit.ParticipantInfo
arg4 bool
}{arg1, arg2, arg3, arg4})
arg5 *telemetry.ReferenceGuard
}{arg1, arg2, arg3, arg4, arg5})
stub := fake.ParticipantLeftStub
fake.recordInvocation("ParticipantLeft", []interface{}{arg1, arg2, arg3, arg4})
fake.recordInvocation("ParticipantLeft", []interface{}{arg1, arg2, arg3, arg4, arg5})
fake.participantLeftMutex.Unlock()
if stub != nil {
fake.ParticipantLeftStub(arg1, arg2, arg3, arg4)
fake.ParticipantLeftStub(arg1, arg2, arg3, arg4, arg5)
}
}
@@ -772,17 +778,17 @@ func (fake *FakeTelemetryService) ParticipantLeftCallCount() int {
return len(fake.participantLeftArgsForCall)
}
func (fake *FakeTelemetryService) ParticipantLeftCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool)) {
func (fake *FakeTelemetryService) ParticipantLeftCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool, *telemetry.ReferenceGuard)) {
fake.participantLeftMutex.Lock()
defer fake.participantLeftMutex.Unlock()
fake.ParticipantLeftStub = stub
}
func (fake *FakeTelemetryService) ParticipantLeftArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, bool) {
func (fake *FakeTelemetryService) ParticipantLeftArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, bool, *telemetry.ReferenceGuard) {
fake.participantLeftMutex.RLock()
defer fake.participantLeftMutex.RUnlock()
argsForCall := fake.participantLeftArgsForCall[i]
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4
return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5
}
func (fake *FakeTelemetryService) ParticipantResumed(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 livekit.NodeID, arg5 livekit.ReconnectReason) {
pkg/telemetry/telemetryservice.go
+6 -4
View File
@@ -37,13 +37,13 @@ type TelemetryService interface {
RoomStarted(ctx context.Context, room *livekit.Room)
RoomEnded(ctx context.Context, room *livekit.Room)
// ParticipantJoined - a participant establishes signal connection to a room
ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool)
ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta, shouldSendEvent bool, guard *ReferenceGuard)
// ParticipantActive - a participant establishes media connection
ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool)
ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta, isMigration bool, guard *ReferenceGuard)
// ParticipantResumed - there has been an ICE restart or connection resume attempt, and we've received their signal connection
ParticipantResumed(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, nodeID livekit.NodeID, reason livekit.ReconnectReason)
// ParticipantLeft - the participant leaves the room, only sent if ParticipantActive has been called before
ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool)
ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard)
// TrackPublishRequested - a publication attempt has been received
TrackPublishRequested(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo)
// TrackPublished - a publication attempt has been successful
@@ -204,12 +204,13 @@ func (t *telemetryService) getOrCreateWorker(
roomName livekit.RoomName,
participantID livekit.ParticipantID,
participantIdentity livekit.ParticipantIdentity,
guard *ReferenceGuard,
) (*StatsWorker, bool) {
t.workersMu.Lock()
defer t.workersMu.Unlock()
worker, ok := t.workers[participantID]
if ok && !worker.Closed() {
if ok && !worker.Closed(guard) {
return worker, true
}
@@ -225,6 +226,7 @@ func (t *telemetryService) getOrCreateWorker(
roomName,
participantID,
participantIdentity,
guard,
)
if existingIsConnected {
worker.SetConnected()