From 33902a9f2aec0e8241390bac686def66210498d1 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sun, 18 Dec 2022 17:37:55 -0800 Subject: [PATCH] Do not send ParticipantLeft webhook event unless connected successfully. (#1234) Fixes #1130 --- .../typesfakes/fake_local_participant.go | 76 ++++++++++--------- pkg/rtc/types/typesfakes/fake_participant.go | 76 ++++++++++--------- pkg/service/roommanager.go | 8 +- pkg/telemetry/events.go | 38 ++++++---- pkg/telemetry/events_test.go | 7 +- pkg/telemetry/stats_test.go | 2 +- pkg/telemetry/statsworker.go | 14 ++++ .../telemetryfakes/fake_telemetry_service.go | 18 +++-- pkg/telemetry/telemetryservice.go | 11 ++- 9 files changed, 141 insertions(+), 109 deletions(-) diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index b4ecf7911..9ecc6f686 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -612,11 +612,6 @@ type FakeLocalParticipant struct { setICEConfigArgsForCall []struct { arg1 types.IceConfig } - SetNameStub func(string) - setNameMutex sync.RWMutex - setNameArgsForCall []struct { - arg1 string - } SetMetadataStub func(string) setMetadataMutex sync.RWMutex setMetadataArgsForCall []struct { @@ -635,6 +630,11 @@ type FakeLocalParticipant struct { setMigrateStateArgsForCall []struct { arg1 types.MigrateState } + SetNameStub func(string) + setNameMutex sync.RWMutex + setNameArgsForCall []struct { + arg1 string + } SetPermissionStub func(*livekit.ParticipantPermission) bool setPermissionMutex sync.RWMutex setPermissionArgsForCall []struct { @@ -4011,38 +4011,6 @@ func (fake *FakeLocalParticipant) SetICEConfigArgsForCall(i int) types.IceConfig return argsForCall.arg1 } -func (fake *FakeLocalParticipant) SetName(arg1 string) { - fake.setNameMutex.Lock() - fake.setNameArgsForCall = append(fake.setNameArgsForCall, struct { - arg1 string - }{arg1}) - stub := fake.SetNameStub - fake.recordInvocation("SetName", []interface{}{arg1}) - fake.setNameMutex.Unlock() - if stub != nil { - fake.SetNameStub(arg1) - } -} - -func (fake *FakeLocalParticipant) SetNameCallCount() int { - fake.setNameMutex.RLock() - defer fake.setNameMutex.RUnlock() - return len(fake.setNameArgsForCall) -} - -func (fake *FakeLocalParticipant) SetNameCalls(stub func(string)) { - fake.setNameMutex.Lock() - defer fake.setNameMutex.Unlock() - fake.SetNameStub = stub -} - -func (fake *FakeLocalParticipant) SetNameArgsForCall(i int) string { - fake.setNameMutex.RLock() - defer fake.setNameMutex.RUnlock() - argsForCall := fake.setNameArgsForCall[i] - return argsForCall.arg1 -} - func (fake *FakeLocalParticipant) SetMetadata(arg1 string) { fake.setMetadataMutex.Lock() fake.setMetadataArgsForCall = append(fake.setMetadataArgsForCall, struct { @@ -4152,6 +4120,38 @@ func (fake *FakeLocalParticipant) SetMigrateStateArgsForCall(i int) types.Migrat return argsForCall.arg1 } +func (fake *FakeLocalParticipant) SetName(arg1 string) { + fake.setNameMutex.Lock() + fake.setNameArgsForCall = append(fake.setNameArgsForCall, struct { + arg1 string + }{arg1}) + stub := fake.SetNameStub + fake.recordInvocation("SetName", []interface{}{arg1}) + fake.setNameMutex.Unlock() + if stub != nil { + fake.SetNameStub(arg1) + } +} + +func (fake *FakeLocalParticipant) SetNameCallCount() int { + fake.setNameMutex.RLock() + defer fake.setNameMutex.RUnlock() + return len(fake.setNameArgsForCall) +} + +func (fake *FakeLocalParticipant) SetNameCalls(stub func(string)) { + fake.setNameMutex.Lock() + defer fake.setNameMutex.Unlock() + fake.SetNameStub = stub +} + +func (fake *FakeLocalParticipant) SetNameArgsForCall(i int) string { + fake.setNameMutex.RLock() + defer fake.setNameMutex.RUnlock() + argsForCall := fake.setNameArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeLocalParticipant) SetPermission(arg1 *livekit.ParticipantPermission) bool { fake.setPermissionMutex.Lock() ret, specificReturn := fake.setPermissionReturnsOnCall[len(fake.setPermissionArgsForCall)] @@ -5110,6 +5110,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.setMigrateInfoMutex.RUnlock() fake.setMigrateStateMutex.RLock() defer fake.setMigrateStateMutex.RUnlock() + fake.setNameMutex.RLock() + defer fake.setNameMutex.RUnlock() fake.setPermissionMutex.RLock() defer fake.setPermissionMutex.RUnlock() fake.setResponseSinkMutex.RLock() diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 52dc94b11..a68cffb1a 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -120,16 +120,16 @@ type FakeParticipant struct { arg2 livekit.TrackID arg3 bool } - SetNameStub func(string) - setNameMutex sync.RWMutex - setNameArgsForCall []struct { - arg1 string - } SetMetadataStub func(string) setMetadataMutex sync.RWMutex setMetadataArgsForCall []struct { arg1 string } + SetNameStub func(string) + setNameMutex sync.RWMutex + setNameArgsForCall []struct { + arg1 string + } StartStub func() startMutex sync.RWMutex startArgsForCall []struct { @@ -759,38 +759,6 @@ func (fake *FakeParticipant) RemoveSubscriberArgsForCall(i int) (types.LocalPart return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 } -func (fake *FakeParticipant) SetName(arg1 string) { - fake.setNameMutex.Lock() - fake.setNameArgsForCall = append(fake.setNameArgsForCall, struct { - arg1 string - }{arg1}) - stub := fake.SetNameStub - fake.recordInvocation("SetName", []interface{}{arg1}) - fake.setNameMutex.Unlock() - if stub != nil { - fake.SetNameStub(arg1) - } -} - -func (fake *FakeParticipant) SetNameCallCount() int { - fake.setNameMutex.RLock() - defer fake.setNameMutex.RUnlock() - return len(fake.setNameArgsForCall) -} - -func (fake *FakeParticipant) SetNameCalls(stub func(string)) { - fake.setNameMutex.Lock() - defer fake.setNameMutex.Unlock() - fake.SetNameStub = stub -} - -func (fake *FakeParticipant) SetNameArgsForCall(i int) string { - fake.setNameMutex.RLock() - defer fake.setNameMutex.RUnlock() - argsForCall := fake.setNameArgsForCall[i] - return argsForCall.arg1 -} - func (fake *FakeParticipant) SetMetadata(arg1 string) { fake.setMetadataMutex.Lock() fake.setMetadataArgsForCall = append(fake.setMetadataArgsForCall, struct { @@ -823,6 +791,38 @@ func (fake *FakeParticipant) SetMetadataArgsForCall(i int) string { return argsForCall.arg1 } +func (fake *FakeParticipant) SetName(arg1 string) { + fake.setNameMutex.Lock() + fake.setNameArgsForCall = append(fake.setNameArgsForCall, struct { + arg1 string + }{arg1}) + stub := fake.SetNameStub + fake.recordInvocation("SetName", []interface{}{arg1}) + fake.setNameMutex.Unlock() + if stub != nil { + fake.SetNameStub(arg1) + } +} + +func (fake *FakeParticipant) SetNameCallCount() int { + fake.setNameMutex.RLock() + defer fake.setNameMutex.RUnlock() + return len(fake.setNameArgsForCall) +} + +func (fake *FakeParticipant) SetNameCalls(stub func(string)) { + fake.setNameMutex.Lock() + defer fake.setNameMutex.Unlock() + fake.SetNameStub = stub +} + +func (fake *FakeParticipant) SetNameArgsForCall(i int) string { + fake.setNameMutex.RLock() + defer fake.setNameMutex.RUnlock() + argsForCall := fake.setNameArgsForCall[i] + return argsForCall.arg1 +} + func (fake *FakeParticipant) Start() { fake.startMutex.Lock() fake.startArgsForCall = append(fake.startArgsForCall, struct { @@ -1108,6 +1108,8 @@ func (fake *FakeParticipant) Invocations() map[string][][]interface{} { defer fake.removeSubscriberMutex.RUnlock() fake.setMetadataMutex.RLock() defer fake.setMetadataMutex.RUnlock() + fake.setNameMutex.RLock() + defer fake.setNameMutex.RUnlock() fake.startMutex.RLock() defer fake.startMutex.RUnlock() fake.subscriptionPermissionMutex.RLock() diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index a949fe1ca..f1eb58d53 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -332,7 +332,7 @@ func (r *RoomManager) StartSession( pLogger.Errorw("could not store participant", err) } - updateParticipantCount := func(proto *livekit.Room) { + persistRoomForParticipantCount := func(proto *livekit.Room) { if !participant.Hidden() { err = r.roomStore.StoreRoom(ctx, proto, room.Internal()) if err != nil { @@ -342,7 +342,7 @@ func (r *RoomManager) StartSession( } // update room store with new numParticipants - updateParticipantCount(room.ToProto()) + persistRoomForParticipantCount(room.ToProto()) clientMeta := &livekit.AnalyticsClientMeta{Region: r.currentNode.Region, Node: r.currentNode.Id} r.telemetry.ParticipantJoined(ctx, protoRoom, participant.ToProto(), pi.Client, clientMeta) @@ -353,8 +353,8 @@ func (r *RoomManager) StartSession( // update room store with new numParticipants proto := room.ToProto() - updateParticipantCount(proto) - r.telemetry.ParticipantLeft(ctx, proto, p.ToProto()) + persistRoomForParticipantCount(proto) + r.telemetry.ParticipantLeft(ctx, proto, p.ToProto(), true) room.RemoveDisallowedSubscriptions(p, disallowedSubscriptions) }) diff --git a/pkg/telemetry/events.go b/pkg/telemetry/events.go index e33aa5f66..cfd9d2f80 100644 --- a/pkg/telemetry/events.go +++ b/pkg/telemetry/events.go @@ -105,8 +105,9 @@ func (t *telemetryService) ParticipantActive( Participant: participant, }) - if _, ok := t.getWorker(livekit.ParticipantID(participant.Sid)); !ok { - t.createWorker( + worker, ok := t.getWorker(livekit.ParticipantID(participant.Sid)) + if !ok { + worker = t.createWorker( ctx, livekit.RoomID(room.Sid), livekit.RoomName(room.Name), @@ -114,6 +115,7 @@ func (t *telemetryService) ParticipantActive( livekit.ParticipantIdentity(participant.Identity), ) } + worker.SetConnected() t.SendEvent(ctx, &livekit.AnalyticsEvent{ Type: livekit.AnalyticsEventType_PARTICIPANT_ACTIVE, @@ -126,28 +128,32 @@ func (t *telemetryService) ParticipantActive( }) } -func (t *telemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo) { +func (t *telemetryService) ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool) { t.enqueue(func() { + isConnected := false if worker, ok := t.getWorker(livekit.ParticipantID(participant.Sid)); ok { + isConnected = worker.IsConnected() worker.Close() } prometheus.SubParticipant() - t.NotifyEvent(ctx, &livekit.WebhookEvent{ - Event: webhook.EventParticipantLeft, - Room: room, - Participant: participant, - }) + if isConnected && shouldSendEvent { + t.NotifyEvent(ctx, &livekit.WebhookEvent{ + Event: webhook.EventParticipantLeft, + Room: room, + Participant: participant, + }) - t.SendEvent(ctx, &livekit.AnalyticsEvent{ - Type: livekit.AnalyticsEventType_PARTICIPANT_LEFT, - Timestamp: timestamppb.Now(), - RoomId: room.Sid, - ParticipantId: participant.Sid, - Participant: participant, - Room: room, - }) + t.SendEvent(ctx, &livekit.AnalyticsEvent{ + Type: livekit.AnalyticsEventType_PARTICIPANT_LEFT, + Timestamp: timestamppb.Now(), + RoomId: room.Sid, + ParticipantId: participant.Sid, + Participant: participant, + Room: room, + }) + } }) } diff --git a/pkg/telemetry/events_test.go b/pkg/telemetry/events_test.go index bdb008dc3..4db264c3d 100644 --- a/pkg/telemetry/events_test.go +++ b/pkg/telemetry/events_test.go @@ -69,12 +69,13 @@ func Test_OnParticipantLeft_EventIsSent(t *testing.T) { participantInfo := &livekit.ParticipantInfo{Sid: partSID} // do - fixture.sut.ParticipantLeft(context.Background(), room, participantInfo) + fixture.sut.ParticipantActive(context.Background(), room, participantInfo, &livekit.AnalyticsClientMeta{}) + fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true) time.Sleep(time.Millisecond * 500) // test - require.Equal(t, 1, fixture.analytics.SendEventCallCount()) - _, event := fixture.analytics.SendEventArgsForCall(0) + require.Equal(t, 2, fixture.analytics.SendEventCallCount()) + _, event := fixture.analytics.SendEventArgsForCall(1) require.Equal(t, livekit.AnalyticsEventType_PARTICIPANT_LEFT, event.Type) require.Equal(t, partSID, event.ParticipantId) require.Equal(t, room.Sid, event.RoomId) diff --git a/pkg/telemetry/stats_test.go b/pkg/telemetry/stats_test.go index 86e6b60fb..0d5098aac 100644 --- a/pkg/telemetry/stats_test.go +++ b/pkg/telemetry/stats_test.go @@ -475,7 +475,7 @@ func Test_AnalyticsSentWhenParticipantLeaves(t *testing.T) { fixture.sut.ParticipantJoined(context.Background(), room, participantInfo, nil, nil) // do - fixture.sut.ParticipantLeft(context.Background(), room, participantInfo) + fixture.sut.ParticipantLeft(context.Background(), room, participantInfo, true) // should not be called if there are no track stats time.Sleep(time.Millisecond * 500) diff --git a/pkg/telemetry/statsworker.go b/pkg/telemetry/statsworker.go index 79f834cce..fd975a321 100644 --- a/pkg/telemetry/statsworker.go +++ b/pkg/telemetry/statsworker.go @@ -20,6 +20,7 @@ type StatsWorker struct { roomName livekit.RoomName participantID livekit.ParticipantID participantIdentity livekit.ParticipantIdentity + isConnected bool lock sync.RWMutex outgoingPerTrack map[livekit.TrackID][]*livekit.AnalyticsStat @@ -62,6 +63,19 @@ func (s *StatsWorker) ParticipantID() livekit.ParticipantID { return s.participantID } +func (s *StatsWorker) SetConnected() { + s.lock.Lock() + s.isConnected = true + s.lock.Unlock() +} + +func (s *StatsWorker) IsConnected() bool { + s.lock.RLock() + defer s.lock.RUnlock() + + return s.isConnected +} + func (s *StatsWorker) Flush() { ts := timestamppb.Now() diff --git a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go index c8eba10f1..069e3b64f 100644 --- a/pkg/telemetry/telemetryfakes/fake_telemetry_service.go +++ b/pkg/telemetry/telemetryfakes/fake_telemetry_service.go @@ -49,12 +49,13 @@ type FakeTelemetryService struct { arg4 *livekit.ClientInfo arg5 *livekit.AnalyticsClientMeta } - ParticipantLeftStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo) + ParticipantLeftStub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool) participantLeftMutex sync.RWMutex participantLeftArgsForCall []struct { arg1 context.Context arg2 *livekit.Room arg3 *livekit.ParticipantInfo + arg4 bool } RoomEndedStub func(context.Context, *livekit.Room) roomEndedMutex sync.RWMutex @@ -334,18 +335,19 @@ func (fake *FakeTelemetryService) ParticipantJoinedArgsForCall(i int) (context.C return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4, argsForCall.arg5 } -func (fake *FakeTelemetryService) ParticipantLeft(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo) { +func (fake *FakeTelemetryService) ParticipantLeft(arg1 context.Context, arg2 *livekit.Room, arg3 *livekit.ParticipantInfo, arg4 bool) { fake.participantLeftMutex.Lock() fake.participantLeftArgsForCall = append(fake.participantLeftArgsForCall, struct { arg1 context.Context arg2 *livekit.Room arg3 *livekit.ParticipantInfo - }{arg1, arg2, arg3}) + arg4 bool + }{arg1, arg2, arg3, arg4}) stub := fake.ParticipantLeftStub - fake.recordInvocation("ParticipantLeft", []interface{}{arg1, arg2, arg3}) + fake.recordInvocation("ParticipantLeft", []interface{}{arg1, arg2, arg3, arg4}) fake.participantLeftMutex.Unlock() if stub != nil { - fake.ParticipantLeftStub(arg1, arg2, arg3) + fake.ParticipantLeftStub(arg1, arg2, arg3, arg4) } } @@ -355,17 +357,17 @@ func (fake *FakeTelemetryService) ParticipantLeftCallCount() int { return len(fake.participantLeftArgsForCall) } -func (fake *FakeTelemetryService) ParticipantLeftCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo)) { +func (fake *FakeTelemetryService) ParticipantLeftCalls(stub func(context.Context, *livekit.Room, *livekit.ParticipantInfo, bool)) { fake.participantLeftMutex.Lock() defer fake.participantLeftMutex.Unlock() fake.ParticipantLeftStub = stub } -func (fake *FakeTelemetryService) ParticipantLeftArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo) { +func (fake *FakeTelemetryService) ParticipantLeftArgsForCall(i int) (context.Context, *livekit.Room, *livekit.ParticipantInfo, bool) { fake.participantLeftMutex.RLock() defer fake.participantLeftMutex.RUnlock() argsForCall := fake.participantLeftArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 + return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 } func (fake *FakeTelemetryService) RoomEnded(arg1 context.Context, arg2 *livekit.Room) { diff --git a/pkg/telemetry/telemetryservice.go b/pkg/telemetry/telemetryservice.go index 1b72e18f1..a0a2254ec 100644 --- a/pkg/telemetry/telemetryservice.go +++ b/pkg/telemetry/telemetryservice.go @@ -23,7 +23,7 @@ type TelemetryService interface { RoomEnded(ctx context.Context, room *livekit.Room) ParticipantJoined(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientInfo *livekit.ClientInfo, clientMeta *livekit.AnalyticsClientMeta) ParticipantActive(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, clientMeta *livekit.AnalyticsClientMeta) - ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo) + ParticipantLeft(ctx context.Context, room *livekit.Room, participant *livekit.ParticipantInfo, shouldSendEvent bool) TrackPublished(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo) TrackUnpublished(ctx context.Context, participantID livekit.ParticipantID, identity livekit.ParticipantIdentity, track *livekit.TrackInfo, ssrc uint32) TrackSubscribed(ctx context.Context, participantID livekit.ParticipantID, track *livekit.TrackInfo, publisher *livekit.ParticipantInfo) @@ -116,8 +116,12 @@ func (t *telemetryService) getWorker(participantID livekit.ParticipantID) (worke return } -func (t *telemetryService) createWorker(ctx context.Context, roomID livekit.RoomID, roomName livekit.RoomName, - participantID livekit.ParticipantID, participantIdentity livekit.ParticipantIdentity) { +func (t *telemetryService) createWorker(ctx context.Context, + roomID livekit.RoomID, + roomName livekit.RoomName, + participantID livekit.ParticipantID, + participantIdentity livekit.ParticipantIdentity, +) *StatsWorker { worker := newStatsWorker( ctx, t, @@ -130,6 +134,7 @@ func (t *telemetryService) createWorker(ctx context.Context, roomID livekit.Room t.lock.Lock() t.workers[participantID] = worker t.lock.Unlock() + return worker } func (t *telemetryService) cleanupWorkers() {