From 9928feefefcf80720633e4489a8940b6daa03828 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Miroslav=20=C5=A0ediv=C3=BD?= Date: Mon, 10 Apr 2023 21:33:53 +0200 Subject: [PATCH] refactor metrics. --- internal/webrtc/manager.go | 27 ++- internal/webrtc/metrics.go | 433 ++++++++++++++++++------------------- 2 files changed, 225 insertions(+), 235 deletions(-) diff --git a/internal/webrtc/manager.go b/internal/webrtc/manager.go index f0fd27cb..a6348783 100644 --- a/internal/webrtc/manager.go +++ b/internal/webrtc/manager.go @@ -69,7 +69,7 @@ func New(desktop types.DesktopManager, capture types.CaptureManager, config *con return &WebRTCManagerCtx{ logger: log.With().Str("module", "webrtc").Logger(), config: config, - metrics: newMetrics(), + metrics: newMetricsManager(), webrtcConfiguration: configuration, @@ -83,7 +83,7 @@ func New(desktop types.DesktopManager, capture types.CaptureManager, config *con type WebRTCManagerCtx struct { logger zerolog.Logger config *config.WebRTC - metrics *metricsCtx + metrics *metricsManager peerId int32 desktop types.DesktopManager @@ -262,7 +262,10 @@ func (manager *WebRTCManagerCtx) newPeerConnection(bitrate int, codecs []codec.R func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int, videoAuto bool) (*webrtc.SessionDescription, error) { id := atomic.AddInt32(&manager.peerId, 1) - manager.metrics.NewConnection(session) + + // get metrics for session + metrics := manager.metrics.getBySession(session) + metrics.NewConnection() // add session id to logger context logger := manager.logger.With().Str("session_id", session.ID()).Int32("peer_id", id).Logger() @@ -362,7 +365,7 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int, videoID := videoTrack.stream.ID() bitrate := videoTrack.stream.Bitrate() - manager.metrics.SetVideoID(session, videoID) + metrics.SetVideoID(videoID) manager.logger.Debug(). Int("peer_bitrate", peerBitrate). Int("video_bitrate", bitrate). @@ -425,7 +428,7 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int, for range ticker.C { targetBitrate := estimator.GetTargetBitrate() - manager.metrics.SetReceiverEstimatedTargetBitrate(session, float64(targetBitrate)) + metrics.SetReceiverEstimatedTargetBitrate(float64(targetBitrate)) if connection.ConnectionState() == webrtc.PeerConnectionStateClosed { break @@ -594,7 +597,7 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int, }) } - manager.metrics.SetState(session, state) + metrics.SetState(state) }) cursorImage := func(entry *cursor.ImageEntry) { @@ -682,13 +685,13 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int, for _, p := range packets { switch rtcpPacket := p.(type) { case *rtcp.ReceiverEstimatedMaximumBitrate: // TODO: Deprecated. - manager.metrics.SetReceiverEstimatedMaximumBitrate(session, rtcpPacket.Bitrate) + metrics.SetReceiverEstimatedMaximumBitrate(rtcpPacket.Bitrate) case *rtcp.ReceiverReport: l := len(rtcpPacket.Reports) if l > 0 { // use only last report - manager.metrics.SetReceiverReport(session, rtcpPacket.Reports[l-1]) + metrics.SetReceiverReport(rtcpPacket.Reports[l-1]) } } } @@ -707,12 +710,12 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int, stats := connection.GetStats() data, ok := stats["iceTransport"].(webrtc.TransportStats) if ok { - manager.metrics.SetIceTransportStats(session, data) + metrics.SetIceTransportStats(data) } data, ok = stats["sctpTransport"].(webrtc.TransportStats) if ok { - manager.metrics.SetSctpTransportStats(session, data) + metrics.SetSctpTransportStats(data) } remoteCandidates := map[string]webrtc.ICECandidateStats{} @@ -721,7 +724,7 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int, // only remote ice candidate stats candidate, ok := entry.(webrtc.ICECandidateStats) if ok && candidate.Type == webrtc.StatsTypeRemoteCandidate { - manager.metrics.NewICECandidate(session, candidate) + metrics.NewICECandidate(candidate) remoteCandidates[candidate.ID] = candidate } @@ -739,7 +742,7 @@ func (manager *WebRTCManagerCtx) CreatePeer(session types.Session, bitrate int, } } - manager.metrics.SetICECandidatesUsed(session, iceCandidatesUsed) + metrics.SetICECandidatesUsed(iceCandidatesUsed) } }() diff --git a/internal/webrtc/metrics.go b/internal/webrtc/metrics.go index 923b791e..f0b2b3fc 100644 --- a/internal/webrtc/metrics.go +++ b/internal/webrtc/metrics.go @@ -10,7 +10,204 @@ import ( "github.com/prometheus/client_golang/prometheus/promauto" ) +type metricsManager struct { + mu sync.Mutex + + sessions map[string]metrics +} + +func newMetricsManager() *metricsManager { + return &metricsManager{ + sessions: map[string]metrics{}, + } +} + +func (m *metricsManager) getBySession(session types.Session) metrics { + m.mu.Lock() + defer m.mu.Unlock() + + sessionId := session.ID() + + met, ok := m.sessions[sessionId] + if ok { + return met + } + + met = metrics{ + sessionId: sessionId, + + connectionState: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "connection_state", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Connection state of session.", + ConstLabels: map[string]string{ + "session_id": sessionId, + }, + }), + connectionStateCount: promauto.NewCounter(prometheus.CounterOpts{ + Name: "connection_state_count", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Count of connection state changes for a session.", + ConstLabels: map[string]string{ + "session_id": sessionId, + }, + }), + connectionCount: promauto.NewCounter(prometheus.CounterOpts{ + Name: "connection_count", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Connection count of a session.", + ConstLabels: map[string]string{ + "session_id": sessionId, + }, + }), + + iceCandidates: map[string]struct{}{}, + iceCandidatesMu: &sync.Mutex{}, + iceCandidatesUdpCount: promauto.NewCounter(prometheus.CounterOpts{ + Name: "ice_candidates_count", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Count of ICE candidates sent by a remote client.", + ConstLabels: map[string]string{ + "session_id": sessionId, + "protocol": "udp", + }, + }), + iceCandidatesTcpCount: promauto.NewCounter(prometheus.CounterOpts{ + Name: "ice_candidates_count", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Count of ICE candidates sent by a remote client.", + ConstLabels: map[string]string{ + "session_id": sessionId, + "protocol": "tcp", + }, + }), + + iceCandidatesUsedUdp: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "ice_candidates_used", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Used ICE candidates that are currently in use.", + ConstLabels: map[string]string{ + "session_id": sessionId, + "protocol": "udp", + }, + }), + iceCandidatesUsedTcp: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "ice_candidates_used", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Used ICE candidates that are currently in use.", + ConstLabels: map[string]string{ + "session_id": sessionId, + "protocol": "tcp", + }, + }), + + videoIds: map[string]prometheus.Gauge{}, + videoIdsMu: &sync.Mutex{}, + + receiverEstimatedMaximumBitrate: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "receiver_estimated_maximum_bitrate", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Receiver Estimated Maximum Bitrate from SCTP.", + ConstLabels: map[string]string{ + "session_id": sessionId, + }, + }), + receiverEstimatedTargetBitrate: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "receiver_estimated_target_bitrate", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Receiver Estimated Target Bitrate using Google's congestion control.", + ConstLabels: map[string]string{ + "session_id": sessionId, + }, + }), + + receiverReportDelay: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "receiver_report_delay", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Receiver Report Delay from SCTP, expressed in units of 1/65536 seconds.", + ConstLabels: map[string]string{ + "session_id": sessionId, + }, + }), + receiverReportJitter: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "receiver_report_jitter", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Receiver Report Jitter from SCTP.", + ConstLabels: map[string]string{ + "session_id": sessionId, + }, + }), + receiverReportTotalLost: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "receiver_report_total_lost", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Receiver Report Total Lost from SCTP.", + ConstLabels: map[string]string{ + "session_id": sessionId, + }, + }), + + iceBytesSent: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "bytes_sent", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Sent bytes to a session.", + ConstLabels: map[string]string{ + "session_id": sessionId, + "transport": "ice", + }, + }), + iceBytesReceived: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "bytes_received", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Received bytes from a session.", + ConstLabels: map[string]string{ + "session_id": sessionId, + "transport": "ice", + }, + }), + + sctpBytesSent: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "bytes_sent", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Sent bytes to a session.", + ConstLabels: map[string]string{ + "session_id": sessionId, + "transport": "sctp", + }, + }), + sctpBytesReceived: promauto.NewGauge(prometheus.GaugeOpts{ + Name: "bytes_received", + Namespace: "neko", + Subsystem: "webrtc", + Help: "Received bytes from a session.", + ConstLabels: map[string]string{ + "session_id": sessionId, + "transport": "sctp", + }, + }), + } + + m.sessions[sessionId] = met + return met +} + type metrics struct { + sessionId string + connectionState prometheus.Gauge connectionStateCount prometheus.Counter connectionCount prometheus.Counter @@ -39,198 +236,7 @@ type metrics struct { sctpBytesReceived prometheus.Gauge } -type metricsCtx struct { - mu sync.Mutex - - sessions map[string]metrics -} - -func newMetrics() *metricsCtx { - return &metricsCtx{ - sessions: map[string]metrics{}, - } -} - -func (m *metricsCtx) getBySession(session types.Session) metrics { - m.mu.Lock() - defer m.mu.Unlock() - - met, ok := m.sessions[session.ID()] - if ok { - return met - } - - met = metrics{ - connectionState: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "connection_state", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Connection state of session.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - }, - }), - connectionStateCount: promauto.NewCounter(prometheus.CounterOpts{ - Name: "connection_state_count", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Count of connection state changes for a session.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - }, - }), - connectionCount: promauto.NewCounter(prometheus.CounterOpts{ - Name: "connection_count", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Connection count of a session.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - }, - }), - - iceCandidates: map[string]struct{}{}, - iceCandidatesMu: &sync.Mutex{}, - iceCandidatesUdpCount: promauto.NewCounter(prometheus.CounterOpts{ - Name: "ice_candidates_count", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Count of ICE candidates sent by a remote client.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - "protocol": "udp", - }, - }), - iceCandidatesTcpCount: promauto.NewCounter(prometheus.CounterOpts{ - Name: "ice_candidates_count", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Count of ICE candidates sent by a remote client.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - "protocol": "tcp", - }, - }), - - iceCandidatesUsedUdp: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "ice_candidates_used", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Used ICE candidates that are currently in use.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - "protocol": "udp", - }, - }), - iceCandidatesUsedTcp: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "ice_candidates_used", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Used ICE candidates that are currently in use.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - "protocol": "tcp", - }, - }), - - videoIds: map[string]prometheus.Gauge{}, - videoIdsMu: &sync.Mutex{}, - - receiverEstimatedMaximumBitrate: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "receiver_estimated_maximum_bitrate", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Receiver Estimated Maximum Bitrate from SCTP.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - }, - }), - receiverEstimatedTargetBitrate: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "receiver_estimated_target_bitrate", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Receiver Estimated Target Bitrate using Google's congestion control.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - }, - }), - - receiverReportDelay: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "receiver_report_delay", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Receiver Report Delay from SCTP, expressed in units of 1/65536 seconds.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - }, - }), - receiverReportJitter: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "receiver_report_jitter", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Receiver Report Jitter from SCTP.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - }, - }), - receiverReportTotalLost: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "receiver_report_total_lost", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Receiver Report Total Lost from SCTP.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - }, - }), - - iceBytesSent: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "bytes_sent", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Sent bytes to a session.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - "transport": "ice", - }, - }), - iceBytesReceived: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "bytes_received", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Received bytes from a session.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - "transport": "ice", - }, - }), - - sctpBytesSent: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "bytes_sent", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Sent bytes to a session.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - "transport": "sctp", - }, - }), - sctpBytesReceived: promauto.NewGauge(prometheus.GaugeOpts{ - Name: "bytes_received", - Namespace: "neko", - Subsystem: "webrtc", - Help: "Received bytes from a session.", - ConstLabels: map[string]string{ - "session_id": session.ID(), - "transport": "sctp", - }, - }), - } - - m.sessions[session.ID()] = met - return met -} - -func (m *metricsCtx) reset(met metrics) { +func (met *metrics) reset() { met.videoIdsMu.Lock() for _, entry := range met.videoIds { entry.Set(0) @@ -246,14 +252,11 @@ func (m *metricsCtx) reset(met metrics) { met.receiverReportJitter.Set(0) } -func (m *metricsCtx) NewConnection(session types.Session) { - met := m.getBySession(session) +func (met *metrics) NewConnection() { met.connectionCount.Add(1) } -func (m *metricsCtx) NewICECandidate(session types.Session, candidate webrtc.ICECandidateStats) { - met := m.getBySession(session) - +func (met *metrics) NewICECandidate(candidate webrtc.ICECandidateStats) { met.iceCandidatesMu.Lock() defer met.iceCandidatesMu.Unlock() @@ -269,9 +272,7 @@ func (m *metricsCtx) NewICECandidate(session types.Session, candidate webrtc.ICE } } -func (m *metricsCtx) SetICECandidatesUsed(session types.Session, candidates []webrtc.ICECandidateStats) { - met := m.getBySession(session) - +func (met *metrics) SetICECandidatesUsed(candidates []webrtc.ICECandidateStats) { udp, tcp := 0, 0 for _, candidate := range candidates { if candidate.Protocol == "udp" { @@ -285,9 +286,7 @@ func (m *metricsCtx) SetICECandidatesUsed(session types.Session, candidates []we met.iceCandidatesUsedTcp.Set(float64(tcp)) } -func (m *metricsCtx) SetState(session types.Session, state webrtc.PeerConnectionState) { - met := m.getBySession(session) - +func (met *metrics) SetState(state webrtc.PeerConnectionState) { switch state { case webrtc.PeerConnectionStateNew: met.connectionState.Set(0) @@ -301,7 +300,7 @@ func (m *metricsCtx) SetState(session types.Session, state webrtc.PeerConnection met.connectionState.Set(2) case webrtc.PeerConnectionStateClosed: met.connectionState.Set(1) - m.reset(met) + met.reset() default: met.connectionState.Set(-1) } @@ -309,9 +308,7 @@ func (m *metricsCtx) SetState(session types.Session, state webrtc.PeerConnection met.connectionStateCount.Add(1) } -func (m *metricsCtx) SetVideoID(session types.Session, videoId string) { - met := m.getBySession(session) - +func (met *metrics) SetVideoID(videoId string) { met.videoIdsMu.Lock() defer met.videoIdsMu.Unlock() @@ -322,7 +319,7 @@ func (m *metricsCtx) SetVideoID(session types.Session, videoId string) { Subsystem: "webrtc", Help: "Listeners for Video pipelines by a session.", ConstLabels: map[string]string{ - "session_id": session.ID(), + "session_id": met.sessionId, "video_id": videoId, }, }) @@ -337,36 +334,26 @@ func (m *metricsCtx) SetVideoID(session types.Session, videoId string) { } } -func (m *metricsCtx) SetReceiverEstimatedMaximumBitrate(session types.Session, bitrate float32) { - met := m.getBySession(session) - +func (met *metrics) SetReceiverEstimatedMaximumBitrate(bitrate float32) { met.receiverEstimatedMaximumBitrate.Set(float64(bitrate)) } -func (m *metricsCtx) SetReceiverEstimatedTargetBitrate(session types.Session, bitrate float64) { - met := m.getBySession(session) - +func (met *metrics) SetReceiverEstimatedTargetBitrate(bitrate float64) { met.receiverEstimatedTargetBitrate.Set(bitrate) } -func (m *metricsCtx) SetReceiverReport(session types.Session, report rtcp.ReceptionReport) { - met := m.getBySession(session) - +func (met *metrics) SetReceiverReport(report rtcp.ReceptionReport) { met.receiverReportDelay.Set(float64(report.Delay)) met.receiverReportJitter.Set(float64(report.Jitter)) met.receiverReportTotalLost.Set(float64(report.TotalLost)) } -func (m *metricsCtx) SetIceTransportStats(session types.Session, data webrtc.TransportStats) { - met := m.getBySession(session) - +func (met *metrics) SetIceTransportStats(data webrtc.TransportStats) { met.iceBytesSent.Set(float64(data.BytesSent)) met.iceBytesReceived.Set(float64(data.BytesReceived)) } -func (m *metricsCtx) SetSctpTransportStats(session types.Session, data webrtc.TransportStats) { - met := m.getBySession(session) - +func (met *metrics) SetSctpTransportStats(data webrtc.TransportStats) { met.sctpBytesSent.Set(float64(data.BytesSent)) met.sctpBytesReceived.Set(float64(data.BytesReceived)) }