diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index f5f459e4e..1034c2275 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -1413,12 +1413,30 @@ func (p *ParticipantImpl) IsReconnect() bool { return p.params.Reconnect } +func (p *ParticipantImpl) maybeRecordRTCanceled(closeReason types.ParticipantCloseReason) { + if p.State() >= livekit.ParticipantInfo_ACTIVE { + return + } + + if closeReason == types.ParticipantCloseReasonClientRequestLeave || + closeReason == types.ParticipantCloseReasonDuplicateIdentity || + closeReason == types.ParticipantCloseReasonRoomClosed || + closeReason == types.ParticipantCloseReasonMigrationRequested || + closeReason == types.ParticipantCloseReasonMigrationComplete || + // client closing signal connection too quickly, there is a time check to handle clients timing out and leaving without sending a leave message + (time.Since(p.params.SessionStartTime) < 3*time.Second && closeReason == types.ParticipantCloseReasonSignalSourceClose) { + prometheus.IncrementParticipantRtcCanceled(1) + } +} + func (p *ParticipantImpl) Close(sendLeave bool, reason types.ParticipantCloseReason, isExpectedToResume bool) error { if p.isClosed.Swap(true) { // already closed return nil } + p.maybeRecordRTCanceled(reason) + var sessionDuration time.Duration if activeAt := p.ActiveAt(); !activeAt.IsZero() { sessionDuration = time.Since(activeAt) diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index 172a7752e..921c73e8c 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -296,6 +296,9 @@ func (r *RoomManager) StartSession( createRoom := pi.CreateRoom room, err := r.getOrCreateRoom(ctx, createRoom) if err != nil { + if pi.Identity != "" { + prometheus.IncrementParticipantRtcCanceled(1) + } return err } defer room.Release() @@ -371,9 +374,11 @@ func (r *RoomManager) StartSession( pi.ReconnectReason, ); err != nil { participant.GetLogger().Warnw("could not resume participant", err) + prometheus.IncrementParticipantRtcCanceled(1) return err } r.telemetry.ParticipantResumed(ctx, room.ToProto(), participant.ToProto(), r.currentNode.NodeID(), pi.ReconnectReason) + prometheus.IncrementParticipantRtcActive(1) go room.HandleSyncState(participant, pi.SyncState) @@ -527,6 +532,7 @@ func (r *RoomManager) StartSession( EnableRTPStreamRestartDetection: r.config.RTC.EnableRTPStreamRestartDetection, }) if err != nil { + prometheus.IncrementParticipantRtcCanceled(1) return err } iceConfig := r.setIceConfig(room.Name(), participant) @@ -542,6 +548,7 @@ func (r *RoomManager) StartSession( if err = room.Join(participant, requestSource, &opts, iceServers); err != nil { pLogger.Errorw("could not join room", err) _ = participant.Close(true, types.ParticipantCloseReasonJoinFailed, false) + prometheus.IncrementParticipantRtcCanceled(1) return err } @@ -553,6 +560,7 @@ func (r *RoomManager) StartSession( participantServerClosers.Close() pLogger.Errorw("could not join register participant topic", err) _ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false) + prometheus.IncrementParticipantRtcCanceled(1) return err } @@ -563,6 +571,7 @@ func (r *RoomManager) StartSession( participantServerClosers.Close() pLogger.Errorw("could not join register participant topic for rtc rest participant server", err) _ = participant.Close(true, types.ParticipantCloseReasonMessageBusFailed, false) + prometheus.IncrementParticipantRtcCanceled(1) return err } }