Add prom metrics for peer connectino state. (#4574)

* Add prom metrics for peer connectino state.

By direction (PUBLISHER vs SUBSCRIBER) and state ("started" ->
"connected"). This gives a way to track peer connections failing to
finish establishment.

The RTC active count can be useful for primary peer connection, but not
for non-primary. This counter can be used to track any and can generally
be used to understand success/failure rate of peer connection
establishment.

* add a couple of more states

* clean up and avoid duplicate reporting fully established

* staticcheck

pkg/rtc/transport.go

@@ -232,6 +232,8 @@ type PCTransport struct {
 	resetShortConnOnICERestart atomic.Bool
 	signalingRTT               atomic.Uint32 // milliseconds
 
+	hasFullyEstablishedRecorded bool
+
 	debouncedNegotiate *sfuutils.Debouncer
 	debouncePending    bool
 	lastNegotiate      time.Time
@@ -717,6 +719,8 @@ func (t *PCTransport) setICEConnectedAt(at time.Time) {
 			t.tcpICETimer.Stop()
 			t.tcpICETimer = nil
 		}
+
+		prometheus.RecordPeerConnectionState(t.params.Transport, "ice_connected")
 	}
 
 	if t.mayFailedICEStatsTimer != nil {
@@ -801,6 +805,7 @@ func (t *PCTransport) setConnectedAt(at time.Time) bool {
 
 	t.firstConnectedAt = at
 	prometheus.RecordServiceOperationSuccess("peer_connection")
+	prometheus.RecordPeerConnectionState(t.params.Transport, "connected")
 	t.lock.Unlock()
 	return true
 }
@@ -964,6 +969,13 @@ func (t *PCTransport) onDataChannel(dc *webrtc.DataChannel) {
 func (t *PCTransport) maybeNotifyFullyEstablished() {
 	if t.isFullyEstablished() {
 		t.params.Handler.OnFullyEstablished()
+
+		t.lock.Lock()
+		if !t.hasFullyEstablishedRecorded {
+			t.hasFullyEstablishedRecorded = true
+			prometheus.RecordPeerConnectionState(t.params.Transport, "fully_established")
+		}
+		t.lock.Unlock()
 	}
 }
 
@@ -2604,6 +2616,8 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error {
 		t.params.Logger.Debugw("local offer (unfiltered)", "sdp", offer.SDP)
 	}
 
+	isStartOfConnectionSequence := t.pc.LocalDescription() == nil
+
 	err = t.pc.SetLocalDescription(offer)
 	if err != nil {
 		if errors.Is(err, webrtc.ErrConnectionClosed) {
@@ -2615,6 +2629,10 @@ func (t *PCTransport) createAndSendOffer(options *webrtc.OfferOptions) error {
 		return errors.Wrap(err, "setting local description failed")
 	}
 
+	if isStartOfConnectionSequence {
+		prometheus.RecordPeerConnectionState(t.params.Transport, "started")
+	}
+
 	//
 	// Filter after setting local description as pion expects the offer
 	// to match between CreateOffer and SetLocalDescription.
@@ -2788,7 +2806,7 @@ func (t *PCTransport) createAndSendAnswer() error {
 		return errors.Wrap(err, "could not send answer")
 	}
 	t.localAnswerId.Store(answerId)
-	prometheus.RecordServiceOperationSuccess("asnwer")
+	prometheus.RecordServiceOperationSuccess("answer")
 
 	if err := t.sendUnmatchedMediaRequirement(false); err != nil {
 		return err
@@ -2860,9 +2878,16 @@ func (t *PCTransport) handleRemoteOfferReceived(sd *webrtc.SessionDescription, o
 		t.outputAndClearICEStats()
 	}
 
+	isStartOfConnectionSequence := t.pc.RemoteDescription() == nil
+
 	if err := t.setRemoteDescription(*sd); err != nil {
 		return err
 	}
+
+	if isStartOfConnectionSequence {
+		prometheus.RecordPeerConnectionState(t.params.Transport, "started")
+	}
+
 	t.params.Handler.OnSetRemoteDescriptionOffer()
 	t.processSendersPendingConfig()
 

pkg/telemetry/prometheus/rooms.go

@@ -49,6 +49,8 @@ var (
 	promSessionStartTime       *prometheus.HistogramVec
 	promSessionDuration        *prometheus.HistogramVec
 	promPubSubTime             *prometheus.HistogramVec
+
+	promPeerConnection *prometheus.CounterVec
 )
 
 func initRoomStats(nodeID string, nodeType livekit.NodeType) {
@@ -118,6 +120,12 @@ func initRoomStats(nodeID string, nodeType livekit.NodeType) {
 		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
 		Buckets:     []float64{100, 200, 500, 700, 1000, 5000, 10000},
 	}, append(promStreamLabels, "sdk", "kind", "count"))
+	promPeerConnection = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace:   livekitNamespace,
+		Subsystem:   "peer_connection",
+		Name:        "state",
+		ConstLabels: prometheus.Labels{"node_id": nodeID, "node_type": nodeType.String()},
+	}, []string{"transport", "state"})
 
 	prometheus.MustRegister(promRoomCurrent)
 	prometheus.MustRegister(promRoomDuration)
@@ -129,6 +137,7 @@ func initRoomStats(nodeID string, nodeType livekit.NodeType) {
 	prometheus.MustRegister(promSessionStartTime)
 	prometheus.MustRegister(promSessionDuration)
 	prometheus.MustRegister(promPubSubTime)
+	prometheus.MustRegister(promPeerConnection)
 }
 
 func RoomStarted() {
@@ -269,3 +278,7 @@ func RecordSessionStartTime(protocolVersion int, d time.Duration) {
 func RecordSessionDuration(protocolVersion int, d time.Duration) {
 	promSessionDuration.WithLabelValues(strconv.Itoa(protocolVersion)).Observe(float64(d.Milliseconds()))
 }
+
+func RecordPeerConnectionState(transport livekit.SignalTarget, state string) {
+	promPeerConnection.WithLabelValues(transport.String(), state).Inc()
+}