From 47b89c155fe20534dc555c85e49f9c6c2eb6dc16 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Wed, 22 Jun 2022 10:32:22 +0530 Subject: [PATCH] Log reason for participant close (#776) * Log reason for participant close Please suggest better naming for different scenarios if something comes to mind. * group service requests * incorporate feedback * Change names in tests --- pkg/rtc/participant.go | 6 +- pkg/rtc/room.go | 17 +++-- pkg/rtc/room_test.go | 8 +- pkg/rtc/signalhandler.go | 2 +- pkg/rtc/types/interfaces.go | 73 +++++++++++++++++-- .../typesfakes/fake_local_participant.go | 18 +++-- pkg/rtc/types/typesfakes/fake_participant.go | 18 +++-- pkg/rtc/types/typesfakes/fake_room.go | 18 +++-- pkg/service/roommanager.go | 13 ++-- 9 files changed, 122 insertions(+), 51 deletions(-) diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index f0149c457..cd3edcfb6 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -648,13 +648,13 @@ func (p *ParticipantImpl) Start() { }) } -func (p *ParticipantImpl) Close(sendLeave bool) error { +func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseReason) error { if p.isClosed.Swap(true) { // already closed return nil } - p.params.Logger.Infow("closing participant", "sendLeave", sendLeave) + p.params.Logger.Infow("closing participant", "sendLeave", sendLeave, "reason", reason) // send leave message if sendLeave { _ = p.writeMessage(&livekit.SignalResponse{ @@ -1207,7 +1207,7 @@ func (p *ParticipantImpl) handlePrimaryStateChange(state webrtc.PeerConnectionSt } if primaryPC.ConnectionState() != webrtc.PeerConnectionStateConnected { p.params.Logger.Infow("closing disconnected participant") - p.Close(true) + p.Close(true, types.ParticipantCloseReasonPeerConnectionDisconnected) } }) p.lock.Unlock() diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index d8ce33f0e..f6169afa4 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -246,7 +246,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions r.telemetry.ParticipantActive(context.Background(), r.ToProto(), p.ToProto(), &livekit.AnalyticsClientMeta{ClientConnectTime: uint32(time.Since(p.ConnectedAt()).Milliseconds())}) } else if state == livekit.ParticipantInfo_DISCONNECTED { // remove participant from room - go r.RemoveParticipant(p.Identity()) + go r.RemoveParticipant(p.Identity(), types.ParticipantCloseReasonStateDisconnected) } }) participant.OnTrackUpdated(r.onTrackUpdated) @@ -302,7 +302,7 @@ func (r *Room) Join(participant types.LocalParticipant, opts *ParticipantOptions time.AfterFunc(time.Minute, func() { state := participant.State() if state == livekit.ParticipantInfo_JOINING || state == livekit.ParticipantInfo_JOINED { - r.RemoveParticipant(participant.Identity()) + r.RemoveParticipant(participant.Identity(), types.ParticipantCloseReasonJoinTimeout) } }) @@ -348,7 +348,7 @@ func (r *Room) ResumeParticipant(p types.LocalParticipant, responseSink routing. return nil } -func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity) { +func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity, reason types.ParticipantCloseReason) { r.lock.Lock() p, ok := r.participants[identity] if ok { @@ -391,7 +391,7 @@ func (r *Room) RemoveParticipant(identity livekit.ParticipantIdentity) { // close participant as well r.Logger.Infow("closing participant for removal", "pID", p.ID(), "participant", p.Identity()) - _ = p.Close(true) + _ = p.Close(true, reason) r.lock.RLock() if len(r.participants) == 0 { @@ -617,15 +617,20 @@ func (r *Room) SimulateScenario(participant types.LocalParticipant, simulateScen Level: 0.9, }}) case *livekit.SimulateScenario_Migration: + r.Logger.Infow("simulating migration", "participant", participant.Identity()) + // drop participant without necessarily cleaning up + if err := participant.Close(false, types.ParticipantCloseReasonSimulateMigration); err != nil { + return err + } case *livekit.SimulateScenario_NodeFailure: r.Logger.Infow("simulating node failure", "participant", participant.Identity()) // drop participant without necessarily cleaning up - if err := participant.Close(false); err != nil { + if err := participant.Close(false, types.ParticipantCloseReasonSimulateNodeFailure); err != nil { return err } case *livekit.SimulateScenario_ServerLeave: r.Logger.Infow("simulating server leave", "participant", participant.Identity()) - if err := participant.Close(true); err != nil { + if err := participant.Close(true, types.ParticipantCloseReasonSimulateServerLeave); err != nil { return err } diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index ea8f48f34..ba2cc8a7c 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -54,14 +54,14 @@ func TestJoinedState(t *testing.T) { rm := newRoomWithParticipants(t, testRoomOpts{num: 1}) p0 := rm.GetParticipants()[0] s := time.Now().Unix() - rm.RemoveParticipant(p0.Identity()) + rm.RemoveParticipant(p0.Identity(), types.ParticipantCloseReasonClientRequestLeave) require.Equal(t, s, rm.LastLeftAt()) }) t.Run("LastLeftAt should not be set when there are still participants in the room", func(t *testing.T) { rm := newRoomWithParticipants(t, testRoomOpts{num: 2}) p0 := rm.GetParticipants()[0] - rm.RemoveParticipant(p0.Identity()) + rm.RemoveParticipant(p0.Identity(), types.ParticipantCloseReasonClientRequestLeave) require.EqualValues(t, 0, rm.LastLeftAt()) }) } @@ -119,7 +119,7 @@ func TestRoomJoin(t *testing.T) { disconnectedParticipant := participants[1].(*typesfakes.FakeLocalParticipant) disconnectedParticipant.StateReturns(livekit.ParticipantInfo_DISCONNECTED) - rm.RemoveParticipant(p.Identity()) + rm.RemoveParticipant(p.Identity(), types.ParticipantCloseReasonStateDisconnected) time.Sleep(defaultDelay) require.Equal(t, p, changedParticipant) @@ -332,7 +332,7 @@ func TestRoomClosure(t *testing.T) { p := rm.GetParticipants()[0] // allows immediate close after rm.protoRoom.EmptyTimeout = 0 - rm.RemoveParticipant(p.Identity()) + rm.RemoveParticipant(p.Identity(), types.ParticipantCloseReasonClientRequestLeave) time.Sleep(defaultDelay) diff --git a/pkg/rtc/signalhandler.go b/pkg/rtc/signalhandler.go index b5f095b1c..2aa8e36aa 100644 --- a/pkg/rtc/signalhandler.go +++ b/pkg/rtc/signalhandler.go @@ -76,7 +76,7 @@ func HandleParticipantSignal(room types.Room, participant types.LocalParticipant } case *livekit.SignalRequest_Leave: pLogger.Infow("client leaving room") - room.RemoveParticipant(participant.Identity()) + room.RemoveParticipant(participant.Identity(), types.ParticipantCloseReasonClientRequestLeave) case *livekit.SignalRequest_SubscriptionPermission: err := room.UpdateSubscriptionPermission(participant, msg.SubscriptionPermission) if err != nil { diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 369e8eecd..182cea8b6 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -28,6 +28,8 @@ type AddSubscriberParams struct { TrackIDs []livekit.TrackID } +// --------------------------------------------- + type MigrateState int32 const ( @@ -36,11 +38,6 @@ const ( MigrateStateComplete ) -type SubscribedCodecQuality struct { - CodecMime string - Quality livekit.VideoQuality -} - func (m MigrateState) String() string { switch m { case MigrateStateInit: @@ -54,6 +51,68 @@ func (m MigrateState) String() string { } } +// --------------------------------------------- + +type SubscribedCodecQuality struct { + CodecMime string + Quality livekit.VideoQuality +} + +// --------------------------------------------- + +type ParticipantCloseReason int + +const ( + ParticipantCloseReasonClientRequestLeave ParticipantCloseReason = iota + ParticipantCloseReasonRoomManagerStop + ParticipantCloseReasonJoinFailed + ParticipantCloseReasonJoinTimeout + ParticipantCloseReasonRTCSessionFinish + ParticipantCloseReasonStateDisconnected + ParticipantCloseReasonPeerConnectionDisconnected + ParticipantCloseReasonDuplicateIdentity + ParticipantCloseReasonServiceRequestRemoveParticipant + ParticipantCloseReasonServiceRequestDeleteRoom + ParticipantCloseReasonSimulateMigration + ParticipantCloseReasonSimulateNodeFailure + ParticipantCloseReasonSimulateServerLeave +) + +func (p ParticipantCloseReason) String() string { + switch p { + case ParticipantCloseReasonClientRequestLeave: + return "CLIENT_REQUEST_LEAVE" + case ParticipantCloseReasonRoomManagerStop: + return "ROOM_MANAGER_STOP" + case ParticipantCloseReasonJoinFailed: + return "JOIN_FAILED" + case ParticipantCloseReasonJoinTimeout: + return "JOIN_TIMEOUT" + case ParticipantCloseReasonRTCSessionFinish: + return "RTC_SESSION_FINISH" + case ParticipantCloseReasonStateDisconnected: + return "STATE_DISCONNECTED" + case ParticipantCloseReasonPeerConnectionDisconnected: + return "PEER_CONNECTION_DISCONNECTED" + case ParticipantCloseReasonDuplicateIdentity: + return "DUPLICATE_IDENTITY" + case ParticipantCloseReasonServiceRequestRemoveParticipant: + return "SERVICE_REQUEST_REMOVE_PARTICIPANT" + case ParticipantCloseReasonServiceRequestDeleteRoom: + return "SERVICE_REQUEST_DELETE_ROOM" + case ParticipantCloseReasonSimulateMigration: + return "SIMULATE_MIGRATION" + case ParticipantCloseReasonSimulateNodeFailure: + return "SIMULATE_NODE_FAILURE" + case ParticipantCloseReasonSimulateServerLeave: + return "SIMULATE_SERVER_LEAVE" + default: + return fmt.Sprintf("%d", int(p)) + } +} + +// --------------------------------------------- + //counterfeiter:generate . Participant type Participant interface { ID() livekit.ParticipantID @@ -74,7 +133,7 @@ type Participant interface { IsRecorder() bool Start() - Close(sendLeave bool) error + Close(sendLeave bool, reason ParticipantCloseReason) error SubscriptionPermission() *livekit.SubscriptionPermission @@ -182,7 +241,7 @@ type LocalParticipant interface { type Room interface { Name() livekit.RoomName ID() livekit.RoomID - RemoveParticipant(identity livekit.ParticipantIdentity) + RemoveParticipant(identity livekit.ParticipantIdentity, reason ParticipantCloseReason) UpdateSubscriptions(participant LocalParticipant, trackIDs []livekit.TrackID, participantTracks []*livekit.ParticipantTracks, subscribe bool) error UpdateSubscriptionPermission(participant LocalParticipant, permissions *livekit.SubscriptionPermission) error SyncState(participant LocalParticipant, state *livekit.SyncState) error diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index 490ce827c..50993aa37 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -90,10 +90,11 @@ type FakeLocalParticipant struct { claimGrantsReturnsOnCall map[int]struct { result1 *auth.ClaimGrants } - CloseStub func(bool) error + CloseStub func(bool, types.ParticipantCloseReason) error closeMutex sync.RWMutex closeArgsForCall []struct { arg1 bool + arg2 types.ParticipantCloseReason } closeReturns struct { result1 error @@ -1062,18 +1063,19 @@ func (fake *FakeLocalParticipant) ClaimGrantsReturnsOnCall(i int, result1 *auth. }{result1} } -func (fake *FakeLocalParticipant) Close(arg1 bool) error { +func (fake *FakeLocalParticipant) Close(arg1 bool, arg2 types.ParticipantCloseReason) error { fake.closeMutex.Lock() ret, specificReturn := fake.closeReturnsOnCall[len(fake.closeArgsForCall)] fake.closeArgsForCall = append(fake.closeArgsForCall, struct { arg1 bool - }{arg1}) + arg2 types.ParticipantCloseReason + }{arg1, arg2}) stub := fake.CloseStub fakeReturns := fake.closeReturns - fake.recordInvocation("Close", []interface{}{arg1}) + fake.recordInvocation("Close", []interface{}{arg1, arg2}) fake.closeMutex.Unlock() if stub != nil { - return stub(arg1) + return stub(arg1, arg2) } if specificReturn { return ret.result1 @@ -1087,17 +1089,17 @@ func (fake *FakeLocalParticipant) CloseCallCount() int { return len(fake.closeArgsForCall) } -func (fake *FakeLocalParticipant) CloseCalls(stub func(bool) error) { +func (fake *FakeLocalParticipant) CloseCalls(stub func(bool, types.ParticipantCloseReason) error) { fake.closeMutex.Lock() defer fake.closeMutex.Unlock() fake.CloseStub = stub } -func (fake *FakeLocalParticipant) CloseArgsForCall(i int) bool { +func (fake *FakeLocalParticipant) CloseArgsForCall(i int) (bool, types.ParticipantCloseReason) { fake.closeMutex.RLock() defer fake.closeMutex.RUnlock() argsForCall := fake.closeArgsForCall[i] - return argsForCall.arg1 + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeLocalParticipant) CloseReturns(result1 error) { diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index 2e391a39c..ecb3359ca 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -23,10 +23,11 @@ type FakeParticipant struct { result1 int result2 error } - CloseStub func(bool) error + CloseStub func(bool, types.ParticipantCloseReason) error closeMutex sync.RWMutex closeArgsForCall []struct { arg1 bool + arg2 types.ParticipantCloseReason } closeReturns struct { result1 error @@ -260,18 +261,19 @@ func (fake *FakeParticipant) AddSubscriberReturnsOnCall(i int, result1 int, resu }{result1, result2} } -func (fake *FakeParticipant) Close(arg1 bool) error { +func (fake *FakeParticipant) Close(arg1 bool, arg2 types.ParticipantCloseReason) error { fake.closeMutex.Lock() ret, specificReturn := fake.closeReturnsOnCall[len(fake.closeArgsForCall)] fake.closeArgsForCall = append(fake.closeArgsForCall, struct { arg1 bool - }{arg1}) + arg2 types.ParticipantCloseReason + }{arg1, arg2}) stub := fake.CloseStub fakeReturns := fake.closeReturns - fake.recordInvocation("Close", []interface{}{arg1}) + fake.recordInvocation("Close", []interface{}{arg1, arg2}) fake.closeMutex.Unlock() if stub != nil { - return stub(arg1) + return stub(arg1, arg2) } if specificReturn { return ret.result1 @@ -285,17 +287,17 @@ func (fake *FakeParticipant) CloseCallCount() int { return len(fake.closeArgsForCall) } -func (fake *FakeParticipant) CloseCalls(stub func(bool) error) { +func (fake *FakeParticipant) CloseCalls(stub func(bool, types.ParticipantCloseReason) error) { fake.closeMutex.Lock() defer fake.closeMutex.Unlock() fake.CloseStub = stub } -func (fake *FakeParticipant) CloseArgsForCall(i int) bool { +func (fake *FakeParticipant) CloseArgsForCall(i int) (bool, types.ParticipantCloseReason) { fake.closeMutex.RLock() defer fake.closeMutex.RUnlock() argsForCall := fake.closeArgsForCall[i] - return argsForCall.arg1 + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeParticipant) CloseReturns(result1 error) { diff --git a/pkg/rtc/types/typesfakes/fake_room.go b/pkg/rtc/types/typesfakes/fake_room.go index a4599b55a..5c0985c91 100644 --- a/pkg/rtc/types/typesfakes/fake_room.go +++ b/pkg/rtc/types/typesfakes/fake_room.go @@ -29,10 +29,11 @@ type FakeRoom struct { nameReturnsOnCall map[int]struct { result1 livekit.RoomName } - RemoveParticipantStub func(livekit.ParticipantIdentity) + RemoveParticipantStub func(livekit.ParticipantIdentity, types.ParticipantCloseReason) removeParticipantMutex sync.RWMutex removeParticipantArgsForCall []struct { arg1 livekit.ParticipantIdentity + arg2 types.ParticipantCloseReason } SetParticipantPermissionStub func(types.LocalParticipant, *livekit.ParticipantPermission) error setParticipantPermissionMutex sync.RWMutex @@ -218,16 +219,17 @@ func (fake *FakeRoom) NameReturnsOnCall(i int, result1 livekit.RoomName) { }{result1} } -func (fake *FakeRoom) RemoveParticipant(arg1 livekit.ParticipantIdentity) { +func (fake *FakeRoom) RemoveParticipant(arg1 livekit.ParticipantIdentity, arg2 types.ParticipantCloseReason) { fake.removeParticipantMutex.Lock() fake.removeParticipantArgsForCall = append(fake.removeParticipantArgsForCall, struct { arg1 livekit.ParticipantIdentity - }{arg1}) + arg2 types.ParticipantCloseReason + }{arg1, arg2}) stub := fake.RemoveParticipantStub - fake.recordInvocation("RemoveParticipant", []interface{}{arg1}) + fake.recordInvocation("RemoveParticipant", []interface{}{arg1, arg2}) fake.removeParticipantMutex.Unlock() if stub != nil { - fake.RemoveParticipantStub(arg1) + fake.RemoveParticipantStub(arg1, arg2) } } @@ -237,17 +239,17 @@ func (fake *FakeRoom) RemoveParticipantCallCount() int { return len(fake.removeParticipantArgsForCall) } -func (fake *FakeRoom) RemoveParticipantCalls(stub func(livekit.ParticipantIdentity)) { +func (fake *FakeRoom) RemoveParticipantCalls(stub func(livekit.ParticipantIdentity, types.ParticipantCloseReason)) { fake.removeParticipantMutex.Lock() defer fake.removeParticipantMutex.Unlock() fake.RemoveParticipantStub = stub } -func (fake *FakeRoom) RemoveParticipantArgsForCall(i int) livekit.ParticipantIdentity { +func (fake *FakeRoom) RemoveParticipantArgsForCall(i int) (livekit.ParticipantIdentity, types.ParticipantCloseReason) { fake.removeParticipantMutex.RLock() defer fake.removeParticipantMutex.RUnlock() argsForCall := fake.removeParticipantArgsForCall[i] - return argsForCall.arg1 + return argsForCall.arg1, argsForCall.arg2 } func (fake *FakeRoom) SetParticipantPermission(arg1 types.LocalParticipant, arg2 *livekit.ParticipantPermission) error { diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index f9ca84b7a..113349cdc 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -166,7 +166,7 @@ func (r *RoomManager) Stop() { for _, room := range rooms { for _, p := range room.GetParticipants() { - _ = p.Close(true) + _ = p.Close(true, types.ParticipantCloseReasonRoomManagerStop) } room.Close() } @@ -207,8 +207,9 @@ func (r *RoomManager) StartSession( ) return room.ResumeParticipant(participant, responseSink) } else { + participant.GetLogger().Infow("removing duplicate participant") // we need to clean up the existing participant, so a new one can join - room.RemoveParticipant(participant.Identity()) + room.RemoveParticipant(participant.Identity(), types.ParticipantCloseReasonDuplicateIdentity) } } else if pi.Reconnect { // send leave request if participant is trying to reconnect without keep subscribe state @@ -269,7 +270,7 @@ func (r *RoomManager) StartSession( } if err = room.Join(participant, &opts, r.iceServersForRoom(protoRoom), r.currentNode.Region); err != nil { pLogger.Errorw("could not join room", err) - _ = participant.Close(true) + _ = participant.Close(true, types.ParticipantCloseReasonJoinFailed) return err } if err = r.roomStore.StoreParticipant(ctx, roomName, participant.ToProto()); err != nil { @@ -389,7 +390,7 @@ func (r *RoomManager) rtcSessionWorker(room *rtc.Room, participant types.LocalPa "room", room.Name(), "roomID", room.ID(), ) - _ = participant.Close(true) + _ = participant.Close(true, types.ParticipantCloseReasonRTCSessionFinish) requestSource.Close() }() defer rtc.Recover() @@ -472,7 +473,7 @@ func (r *RoomManager) handleRTCMessage(ctx context.Context, roomName livekit.Roo return } pLogger.Infow("removing participant") - room.RemoveParticipant(identity) + room.RemoveParticipant(identity, types.ParticipantCloseReasonServiceRequestRemoveParticipant) case *livekit.RTCNodeMessage_MuteTrack: if participant == nil { return @@ -502,7 +503,7 @@ func (r *RoomManager) handleRTCMessage(ctx context.Context, roomName livekit.Roo case *livekit.RTCNodeMessage_DeleteRoom: room.Logger.Infow("deleting room") for _, p := range room.GetParticipants() { - _ = p.Close(true) + _ = p.Close(true, types.ParticipantCloseReasonServiceRequestDeleteRoom) } room.Close() case *livekit.RTCNodeMessage_UpdateSubscriptions: