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) { }