From b8a073cb68e02e80832e4ce2be0214ad109e0628 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 28 Jul 2026 02:18:14 +0530 Subject: [PATCH] A bit better counting for track publish. (#4707) - Count a publish attempt on a migrating in tarck as there is no AddTrack for that. - Add cancel publish only if the participant connection is canceled - Do not add publish counter for synthetic publish attempts which happens for migrating in tracks. It will be counted on migrating in node when the track is actually published, i. e. negotiated/packets flowing. --- pkg/rtc/participant.go | 43 +++++++++++-------- pkg/rtc/participant_internal_test.go | 4 +- pkg/rtc/room.go | 4 +- pkg/rtc/types/interfaces.go | 4 +- .../fake_participant_telemetry_listener.go | 18 ++++---- pkg/telemetry/events.go | 5 +++ .../telemetryfakes/fake_telemetry_service.go | 18 ++++---- pkg/telemetry/telemetryservice.go | 4 +- 8 files changed, 59 insertions(+), 41 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 2f35efb34..d6b4125aa 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1389,6 +1389,11 @@ func (p *ParticipantImpl) SetMigrateInfo( } p.pendingTracksLock.Unlock() + // for migrating in tracks, there is no AddTrack, so record a synthetic publish request + for _, t := range mediaTracks { + p.params.TelemetryListener.OnTrackPublishRequested(p.ID(), p.Identity(), t.GetTrack(), false) + } + for _, t := range dataTracks { dti := t.GetInfo() dt := NewDataTrack( @@ -1495,11 +1500,13 @@ func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseRea } p.pendingTracksLock.Lock() - for _, pti := range p.pendingTracks { - if len(pti.trackInfos) == 0 { - continue + if p.IsConnectionCanceled(reason) { + for _, pti := range p.pendingTracks { + if len(pti.trackInfos) == 0 { + continue + } + prometheus.RecordTrackPublishCancels(pti.trackInfos[0].Type.String(), int32(len(pti.trackInfos))) } - prometheus.RecordTrackPublishCancels(pti.trackInfos[0].Type.String(), int32(len(pti.trackInfos))) } p.pendingTracks = make(map[string]*pendingTrackInfo) p.pendingPublishingTracks = make(map[livekit.TrackID]*pendingTrackInfo) @@ -1687,7 +1694,7 @@ func (p *ParticipantImpl) SetMigrateState(s types.MigrateState) { // launch callbacks in goroutine since they could block. // callbacks handle webhooks as well as db persistence for _, t := range migratedTracks { - p.handleTrackPublished(t, true) + p.handleTrackPublished(t, true, true) } if s == types.MigrateStateComplete { @@ -3093,7 +3100,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit }, }) p.pendingTracksLock.Unlock() - p.params.TelemetryListener.OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti)) + p.params.TelemetryListener.OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti), true) return nil } @@ -3112,7 +3119,7 @@ func (p *ParticipantImpl) addPendingTrack(req *livekit.AddTrackRequest) *livekit "pendingTrack", p.pendingTracks[req.Cid], ) p.pendingTracksLock.Unlock() - p.params.TelemetryListener.OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti)) + p.params.TelemetryListener.OnTrackPublishRequested(p.ID(), p.Identity(), utils.CloneProto(ti), true) return ti } @@ -3321,7 +3328,7 @@ func (p *ParticipantImpl) mediaTrackReceived( p.GetClientInfo().GetSdk(), p.Kind(), ) - p.handleTrackPublished(mt, isMigrated) + p.handleTrackPublished(mt, isMigrated, false) }() } @@ -3476,17 +3483,19 @@ func (p *ParticipantImpl) addMediaTrack(signalCid string, ti *livekit.TrackInfo) return mt } -func (p *ParticipantImpl) handleTrackPublished(track types.MediaTrack, isMigrated bool) { +func (p *ParticipantImpl) handleTrackPublished(track types.MediaTrack, isMigrated bool, isSynthetic bool) { p.listener().OnTrackPublished(p, track) - // send webhook after callbacks are complete, persistence and state handling happens - // in `onTrackPublished` cb - p.params.TelemetryListener.OnTrackPublished( - p.ID(), - p.Identity(), - track.ToProto(), - !isMigrated, - ) + if !isSynthetic { + // send webhook after callbacks are complete, persistence and state handling happens + // in `onTrackPublished` cb + p.params.TelemetryListener.OnTrackPublished( + p.ID(), + p.Identity(), + track.ToProto(), + !isMigrated, + ) + } p.pendingTracksLock.Lock() delete(p.pendingPublishingTracks, track.ID()) diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 781747a90..68c7277e4 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -91,7 +91,7 @@ func TestTrackPublishing(t *testing.T) { published = true }) p.UpTrackManager.AddPublishedTrack(track) - p.handleTrackPublished(track, false) + p.handleTrackPublished(track, false, false) require.True(t, published) require.False(t, updated) require.Len(t, p.UpTrackManager.publishedTracks, 1) @@ -264,7 +264,7 @@ func TestDisconnectTiming(t *testing.T) { }() track := &typesfakes.FakeMediaTrack{} p.UpTrackManager.AddPublishedTrack(track) - p.handleTrackPublished(track, false) + p.handleTrackPublished(track, false, false) // close channel and then try to Negotiate msg.Close() diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 5bde1a5be..72e18eb48 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -2028,8 +2028,8 @@ type participantTelemetryListener struct { room *Room } -func (l participantTelemetryListener) OnTrackPublishRequested(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo) { - l.room.telemetry.TrackPublishRequested(context.Background(), l.room.ID(), l.room.Name(), pID, identity, ti) +func (l participantTelemetryListener) OnTrackPublishRequested(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) { + l.room.telemetry.TrackPublishRequested(context.Background(), l.room.ID(), l.room.Name(), pID, identity, ti, shouldSendEvent) } func (l participantTelemetryListener) OnTrackPublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 01c5d474b..92951e1ab 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -680,7 +680,7 @@ func (*NullLocalParticipantListener) OnLeave(LocalParticipant, ParticipantCloseR //counterfeiter:generate . ParticipantTelemetryListener type ParticipantTelemetryListener interface { - OnTrackPublishRequested(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo) + OnTrackPublishRequested(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) OnTrackPublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) OnTrackUnpublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) OnTrackSubscribeRequested(pID livekit.ParticipantID, ti *livekit.TrackInfo) @@ -702,7 +702,7 @@ var _ ParticipantTelemetryListener = (*NullParticipantTelemetryListener)(nil) type NullParticipantTelemetryListener struct{} -func (NullParticipantTelemetryListener) OnTrackPublishRequested(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo) { +func (NullParticipantTelemetryListener) OnTrackPublishRequested(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) { } func (NullParticipantTelemetryListener) OnTrackPublished(pID livekit.ParticipantID, identity livekit.ParticipantIdentity, ti *livekit.TrackInfo, shouldSendEvent bool) { } diff --git a/pkg/rtc/types/typesfakes/fake_participant_telemetry_listener.go b/pkg/rtc/types/typesfakes/fake_participant_telemetry_listener.go index 8d456adce..7b7bff499 100644 --- a/pkg/rtc/types/typesfakes/fake_participant_telemetry_listener.go +++ b/pkg/rtc/types/typesfakes/fake_participant_telemetry_listener.go @@ -35,12 +35,13 @@ type FakeParticipantTelemetryListener struct { arg4 int arg5 *livekit.RTPStats } - OnTrackPublishRequestedStub func(livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo) + OnTrackPublishRequestedStub func(livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) onTrackPublishRequestedMutex sync.RWMutex onTrackPublishRequestedArgsForCall []struct { arg1 livekit.ParticipantID arg2 livekit.ParticipantIdentity arg3 *livekit.TrackInfo + arg4 bool } OnTrackPublishedStub func(livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) onTrackPublishedMutex sync.RWMutex @@ -228,18 +229,19 @@ func (fake *FakeParticipantTelemetryListener) OnTrackPublishRTPStatsArgsForCall( return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5 } -func (fake *FakeParticipantTelemetryListener) OnTrackPublishRequested(arg1 livekit.ParticipantID, arg2 livekit.ParticipantIdentity, arg3 *livekit.TrackInfo) { +func (fake *FakeParticipantTelemetryListener) OnTrackPublishRequested(arg1 livekit.ParticipantID, arg2 livekit.ParticipantIdentity, arg3 *livekit.TrackInfo, arg4 bool) { fake.onTrackPublishRequestedMutex.Lock() fake.onTrackPublishRequestedArgsForCall = append(fake.onTrackPublishRequestedArgsForCall, struct { arg1 livekit.ParticipantID arg2 livekit.ParticipantIdentity arg3 *livekit.TrackInfo - }{arg1, arg2, arg3}) + arg4 bool + }{arg1, arg2, arg3, arg4}) stub := fake.OnTrackPublishRequestedStub - fake.recordInvocation("OnTrackPublishRequested", []interface{}{arg1, arg2, arg3}) + fake.recordInvocation("OnTrackPublishRequested", []interface{}{arg1, arg2, arg3, arg4}) fake.onTrackPublishRequestedMutex.Unlock() if stub != nil { - fake.OnTrackPublishRequestedStub(arg1, arg2, arg3) + fake.OnTrackPublishRequestedStub(arg1, arg2, arg3, arg4) } } @@ -249,17 +251,17 @@ func (fake *FakeParticipantTelemetryListener) OnTrackPublishRequestedCallCount() return len(fake.onTrackPublishRequestedArgsForCall) } -func (fake *FakeParticipantTelemetryListener) OnTrackPublishRequestedCalls(stub func(livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo)) { +func (fake *FakeParticipantTelemetryListener) OnTrackPublishRequestedCalls(stub func(livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool)) { fake.onTrackPublishRequestedMutex.Lock() defer fake.onTrackPublishRequestedMutex.Unlock() fake.OnTrackPublishRequestedStub = stub } -func (fake *FakeParticipantTelemetryListener) OnTrackPublishRequestedArgsForCall(i int) (livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo) { +func (fake *FakeParticipantTelemetryListener) OnTrackPublishRequestedArgsForCall(i int) (livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) { fake.onTrackPublishRequestedMutex.RLock() defer fake.onTrackPublishRequestedMutex.RUnlock() argsForCall := fake.onTrackPublishRequestedArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 } func (fake *FakeParticipantTelemetryListener) OnTrackPublished(arg1 livekit.ParticipantID, arg2 livekit.ParticipantIdentity, arg3 *livekit.TrackInfo, arg4 bool) { diff --git a/pkg/telemetry/events.go b/pkg/telemetry/events.go index 55a21f6a1..1a674c91f 100644 --- a/pkg/telemetry/events.go +++ b/pkg/telemetry/events.go @@ -230,9 +230,14 @@ func (t *telemetryService) TrackPublishRequested( participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, + shouldSendEvent bool, ) { t.enqueue(func() { prometheus.RecordTrackPublishAttempt(track.Type.String()) + if !shouldSendEvent { + return + } + room := toMinimalRoomProto(roomID, roomName) ev := newTrackEvent(livekit.AnalyticsEventType_TRACK_PUBLISH_REQUESTED, room, participantID, track) if ev.Participant != nil { diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index f8eecd7e2..faa12befb 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go @@ -201,7 +201,7 @@ type FakeTelemetryService struct { arg7 int arg8 *livekit.RTPStats } - TrackPublishRequestedStub func(context.Context, livekit.RoomID, livekit.RoomName, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo) + TrackPublishRequestedStub func(context.Context, livekit.RoomID, livekit.RoomName, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) trackPublishRequestedMutex sync.RWMutex trackPublishRequestedArgsForCall []struct { arg1 context.Context @@ -210,6 +210,7 @@ type FakeTelemetryService struct { arg4 livekit.ParticipantID arg5 livekit.ParticipantIdentity arg6 *livekit.TrackInfo + arg7 bool } TrackPublishedStub func(context.Context, livekit.RoomID, livekit.RoomName, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) trackPublishedMutex sync.RWMutex @@ -1233,7 +1234,7 @@ func (fake *FakeTelemetryService) TrackPublishRTPStatsArgsForCall(i int) (contex return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5, argsForCall.arg6, argsForCall.arg7, argsForCall.arg8 } -func (fake *FakeTelemetryService) TrackPublishRequested(arg1 context.Context, arg2 livekit.RoomID, arg3 livekit.RoomName, arg4 livekit.ParticipantID, arg5 livekit.ParticipantIdentity, arg6 *livekit.TrackInfo) { +func (fake *FakeTelemetryService) TrackPublishRequested(arg1 context.Context, arg2 livekit.RoomID, arg3 livekit.RoomName, arg4 livekit.ParticipantID, arg5 livekit.ParticipantIdentity, arg6 *livekit.TrackInfo, arg7 bool) { fake.trackPublishRequestedMutex.Lock() fake.trackPublishRequestedArgsForCall = append(fake.trackPublishRequestedArgsForCall, struct { arg1 context.Context @@ -1242,12 +1243,13 @@ func (fake *FakeTelemetryService) TrackPublishRequested(arg1 context.Context, ar arg4 livekit.ParticipantID arg5 livekit.ParticipantIdentity arg6 *livekit.TrackInfo - }{arg1, arg2, arg3, arg4, arg5, arg6}) + arg7 bool + }{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) stub := fake.TrackPublishRequestedStub - fake.recordInvocation("TrackPublishRequested", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6}) + fake.recordInvocation("TrackPublishRequested", []interface{}{arg1, arg2, arg3, arg4, arg5, arg6, arg7}) fake.trackPublishRequestedMutex.Unlock() if stub != nil { - fake.TrackPublishRequestedStub(arg1, arg2, arg3, arg4, arg5, arg6) + fake.TrackPublishRequestedStub(arg1, arg2, arg3, arg4, arg5, arg6, arg7) } } @@ -1257,17 +1259,17 @@ func (fake *FakeTelemetryService) TrackPublishRequestedCallCount() int { return len(fake.trackPublishRequestedArgsForCall) } -func (fake *FakeTelemetryService) TrackPublishRequestedCalls(stub func(context.Context, livekit.RoomID, livekit.RoomName, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo)) { +func (fake *FakeTelemetryService) TrackPublishRequestedCalls(stub func(context.Context, livekit.RoomID, livekit.RoomName, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool)) { fake.trackPublishRequestedMutex.Lock() defer fake.trackPublishRequestedMutex.Unlock() fake.TrackPublishRequestedStub = stub } -func (fake *FakeTelemetryService) TrackPublishRequestedArgsForCall(i int) (context.Context, livekit.RoomID, livekit.RoomName, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo) { +func (fake *FakeTelemetryService) TrackPublishRequestedArgsForCall(i int) (context.Context, livekit.RoomID, livekit.RoomName, livekit.ParticipantID, livekit.ParticipantIdentity, *livekit.TrackInfo, bool) { fake.trackPublishRequestedMutex.RLock() defer fake.trackPublishRequestedMutex.RUnlock() argsForCall := fake.trackPublishRequestedArgsForCall[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) TrackPublished(arg1 context.Context, arg2 livekit.RoomID, arg3 livekit.RoomName, arg4 livekit.ParticipantID, arg5 livekit.ParticipantIdentity, arg6 *livekit.TrackInfo, arg7 bool) { diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 2ab77f5d6..8d122fbdd 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -46,7 +46,7 @@ type TelemetryService interface { // 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, guard *ReferenceGuard) // TrackPublishRequested - a publication attempt has been received - TrackPublishRequested(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo) + TrackPublishRequested(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) // TrackPublished - a publication attempt has been successful TrackPublished(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) // TrackUnpublished - a participant unpublished a track @@ -114,7 +114,7 @@ func (n NullTelemetryService) ParticipantResumed(ctx context.Context, room *live } func (n NullTelemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool, guard *ReferenceGuard) { } -func (n NullTelemetryService) TrackPublishRequested(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo) { +func (n NullTelemetryService) TrackPublishRequested(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) { } func (n NullTelemetryService) TrackPublished(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, shouldSendEvent bool) { }