Record more RTC cancellation points. (#4600)

There are several places the participant can drop off after initiating a
connection attempt. Count those places as cancellation including when
participant is closed due to specific reasons.

Cancels should be discounted when determining RTC/ICE connectivity
success/failure percentage.

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)

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