diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index eac717003..43e66cdba 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -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)) { diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index d75adea5d..0d834e409 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -491,6 +491,7 @@ func (r *Room) Join( p.ToProto(), meta, false, + participant.TelemetryGuard(), ) participant.GetReporter().Tx(func(tx roomobs.ParticipantSessionTx) { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 0cdb53445..2e4b5b82b 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -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 diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 4f850971e..7b433d11c 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -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)] diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index a98249fa6..6c2c164eb 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -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") diff --git a/pkg/telemetry/events.go b/pkg/telemetry/events.go index ef6ca0859..520b6f462 100644 --- a/pkg/telemetry/events.go +++ b/pkg/telemetry/events.go @@ -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() } } diff --git a/pkg/telemetry/events_test.go b/pkg/telemetry/events_test.go index e83b61ba9..0be7de176 100644 --- a/pkg/telemetry/events_test.go +++ b/pkg/telemetry/events_test.go @@ -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 diff --git a/pkg/telemetry/signalanddatastats.go b/pkg/telemetry/signalanddatastats.go index 83f6152eb..898aa9896 100644 --- a/pkg/telemetry/signalanddatastats.go +++ b/pkg/telemetry/signalanddatastats.go @@ -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) } diff --git a/pkg/telemetry/stats_test.go b/pkg/telemetry/stats_test.go index 7e33b4d9d..8db18df6a 100644 --- a/pkg/telemetry/stats_test.go +++ b/pkg/telemetry/stats_test.go @@ -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 diff --git a/pkg/telemetry/statsworker.go b/pkg/telemetry/statsworker.go index c5fcbe6e0..075a469af 100644 --- a/pkg/telemetry/statsworker.go +++ b/pkg/telemetry/statsworker.go @@ -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( diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index 3a1ca747f..5f1267fa9 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go @@ -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) { diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 431517e4e..02bb625a5 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -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()