From adf45596a319b27a0bbb8792b77d33d386e1f9d8 Mon Sep 17 00:00:00 2001 From: Alessandro Ros Date: Mon, 16 Mar 2026 23:29:25 +0100 Subject: [PATCH] api: rename WebRTC stats to match RTSP ones (#5581) --- api/openapi.yaml | 32 +++++++++++ internal/api/api_webrtc_test.go | 36 ++++++++++++ internal/core/api_test.go | 61 ++++++++++++++++++++- internal/core/metrics_test.go | 16 ++++++ internal/defs/api_rtsp.go | 21 +++---- internal/defs/api_webrtc.go | 25 ++++++--- internal/metrics/metrics.go | 18 ++++++ internal/metrics/metrics_test.go | 24 ++++++++ internal/protocols/webrtc/incoming_track.go | 14 ++--- internal/servers/rtsp/session.go | 48 ++++++++-------- internal/servers/webrtc/server_test.go | 20 ++++++- internal/servers/webrtc/session.go | 30 ++++++---- internal/stream/reader.go | 16 +++--- internal/stream/stream.go | 52 +++++++++--------- internal/stream/stream_format.go | 22 ++++---- internal/stream/stream_media.go | 42 +++++++------- internal/stream/sub_stream_format.go | 6 +- 17 files changed, 350 insertions(+), 133 deletions(-) diff --git a/api/openapi.yaml b/api/openapi.yaml index de49a9a7..1bcb72a3 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -1565,30 +1565,62 @@ components: type: string user: type: string + inboundBytes: + type: integer + format: uint64 + inboundRTPPackets: + type: integer + format: uint64 + inboundRTPPacketsLost: + type: integer + format: uint64 + inboundRTPPacketsJitter: + type: number + format: double + inboundRTCPPackets: + type: integer + format: uint64 + outboundBytes: + type: integer + format: uint64 + outboundRTPPackets: + type: integer + format: uint64 + outboundRTCPPackets: + type: integer + format: uint64 bytesReceived: type: integer format: uint64 + deprecated: true bytesSent: type: integer format: uint64 + deprecated: true rtpPacketsReceived: type: integer format: uint64 + deprecated: true rtpPacketsSent: type: integer format: uint64 + deprecated: true rtpPacketsLost: type: integer format: uint64 + deprecated: true rtpPacketsJitter: type: number format: double + deprecated: true rtcpPacketsReceived: type: integer format: uint64 + deprecated: true rtcpPacketsSent: type: integer format: uint64 + deprecated: true WebRTCSessionList: type: object diff --git a/internal/api/api_webrtc_test.go b/internal/api/api_webrtc_test.go index d0111f5e..817b201c 100644 --- a/internal/api/api_webrtc_test.go +++ b/internal/api/api_webrtc_test.go @@ -59,6 +59,14 @@ func TestWebRTCSessionsList(t *testing.T) { State: defs.APIWebRTCSessionStatePublish, Path: "stream1", Query: "token=abc", + InboundBytes: 1000, + InboundRTPPackets: 100, + InboundRTPPacketsLost: 5, + InboundRTPPacketsJitter: 0.5, + InboundRTCPPackets: 10, + OutboundBytes: 2000, + OutboundRTPPackets: 200, + OutboundRTCPPackets: 15, BytesReceived: 1000, BytesSent: 2000, RTPPacketsReceived: 100, @@ -78,6 +86,14 @@ func TestWebRTCSessionsList(t *testing.T) { State: defs.APIWebRTCSessionStateRead, Path: "stream2", Query: "", + InboundBytes: 500, + InboundRTPPackets: 50, + InboundRTPPacketsLost: 0, + InboundRTPPacketsJitter: 0.1, + InboundRTCPPackets: 5, + OutboundBytes: 1500, + OutboundRTPPackets: 150, + OutboundRTCPPackets: 10, BytesReceived: 500, BytesSent: 1500, RTPPacketsReceived: 50, @@ -130,6 +146,14 @@ func TestWebRTCSessionsGet(t *testing.T) { State: defs.APIWebRTCSessionStatePublish, Path: "mystream", Query: "key=value", + InboundBytes: 999999, + InboundRTPPackets: 10000, + InboundRTPPacketsLost: 50, + InboundRTPPacketsJitter: 1.5, + InboundRTCPPackets: 100, + OutboundBytes: 888888, + OutboundRTPPackets: 20000, + OutboundRTCPPackets: 200, BytesReceived: 999999, BytesSent: 888888, RTPPacketsReceived: 10000, @@ -168,6 +192,10 @@ func TestWebRTCSessionsGet(t *testing.T) { require.True(t, out.PeerConnectionEstablished) require.Equal(t, "192.168.1.200:8000", out.LocalCandidate) require.Equal(t, "192.168.1.100:5000", out.RemoteCandidate) + require.Equal(t, uint64(999999), out.InboundBytes) + require.Equal(t, uint64(888888), out.OutboundBytes) + require.Equal(t, uint64(10000), out.InboundRTPPackets) + require.Equal(t, uint64(20000), out.OutboundRTPPackets) require.Equal(t, uint64(999999), out.BytesReceived) require.Equal(t, uint64(888888), out.BytesSent) require.Equal(t, uint64(10000), out.RTPPacketsReceived) @@ -192,6 +220,14 @@ func TestWebRTCSessionsKick(t *testing.T) { State: defs.APIWebRTCSessionStatePublish, Path: "mystream", Query: "", + InboundBytes: 1000, + InboundRTPPackets: 100, + InboundRTPPacketsLost: 0, + InboundRTPPacketsJitter: 0.5, + InboundRTCPPackets: 10, + OutboundBytes: 2000, + OutboundRTPPackets: 200, + OutboundRTCPPackets: 15, BytesReceived: 1000, BytesSent: 2000, RTPPacketsReceived: 100, diff --git a/internal/core/api_test.go b/internal/core/api_test.go index cb1f0cb2..df523669 100644 --- a/internal/core/api_test.go +++ b/internal/core/api_test.go @@ -653,7 +653,7 @@ func TestAPIProtocolListGet(t *testing.T) { "outboundRTPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPackets"], "outboundRTPPacketsReportedLost": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPacketsReportedLost"], "outboundRTCPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTCPPackets"], - "bytesReceived": float64(0), + "bytesReceived": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesReceived"], "bytesSent": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesSent"], "created": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["created"], "id": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["id"], @@ -713,7 +713,7 @@ func TestAPIProtocolListGet(t *testing.T) { "outboundRTPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPackets"], "outboundRTPPacketsReportedLost": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPacketsReportedLost"], "outboundRTCPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTCPPackets"], - "bytesReceived": float64(0), + "bytesReceived": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesReceived"], "bytesSent": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesSent"], "created": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["created"], "id": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["id"], @@ -795,6 +795,14 @@ func TestAPIProtocolListGet(t *testing.T) { "pageCount": float64(1), "items": []any{ map[string]any{ + "inboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundBytes"], + "inboundRTPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPackets"], + "inboundRTPPacketsLost": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPacketsLost"], + "inboundRTPPacketsJitter": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPacketsJitter"], + "inboundRTCPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTCPPackets"], + "outboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"], + "outboundRTPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPackets"], + "outboundRTCPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTCPPackets"], "bytesReceived": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesReceived"], "bytesSent": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesSent"], "created": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["created"], @@ -900,6 +908,55 @@ func TestAPIProtocolListGet(t *testing.T) { nil, &out2) } + switch ca { + case "rtsp conns", "rtsps conns": + out2.(map[string]any)["inboundBytes"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundBytes"] + out2.(map[string]any)["outboundBytes"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"] + out2.(map[string]any)["bytesReceived"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesReceived"] + out2.(map[string]any)["bytesSent"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesSent"] + + case "rtsp sessions", "rtsps sessions": + out2.(map[string]any)["inboundBytes"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundBytes"] + out2.(map[string]any)["inboundRTPPackets"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPackets"] + out2.(map[string]any)["inboundRTPPacketsLost"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPacketsLost"] + out2.(map[string]any)["inboundRTPPacketsInError"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPacketsInError"] + out2.(map[string]any)["inboundRTPPacketsJitter"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPacketsJitter"] + out2.(map[string]any)["inboundRTCPPackets"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTCPPackets"] + out2.(map[string]any)["inboundRTCPPacketsInError"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTCPPacketsInError"] + out2.(map[string]any)["outboundBytes"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"] + out2.(map[string]any)["outboundRTPPackets"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPackets"] + out2.(map[string]any)["outboundRTPPacketsReportedLost"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPacketsReportedLost"] + out2.(map[string]any)["outboundRTCPPackets"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTCPPackets"] + out2.(map[string]any)["bytesReceived"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesReceived"] + out2.(map[string]any)["bytesSent"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesSent"] + out2.(map[string]any)["rtpPacketsReceived"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtpPacketsReceived"] + out2.(map[string]any)["rtpPacketsSent"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtpPacketsSent"] + out2.(map[string]any)["rtpPacketsLost"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtpPacketsLost"] + out2.(map[string]any)["rtpPacketsInError"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtpPacketsInError"] + out2.(map[string]any)["rtpPacketsJitter"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtpPacketsJitter"] + out2.(map[string]any)["rtcpPacketsReceived"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtcpPacketsReceived"] + out2.(map[string]any)["rtcpPacketsSent"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtcpPacketsSent"] + out2.(map[string]any)["rtcpPacketsInError"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtcpPacketsInError"] + + case "webrtc": + out2.(map[string]any)["inboundBytes"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundBytes"] + out2.(map[string]any)["inboundRTPPackets"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPackets"] + out2.(map[string]any)["inboundRTPPacketsLost"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPacketsLost"] + out2.(map[string]any)["inboundRTPPacketsJitter"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTPPacketsJitter"] + out2.(map[string]any)["inboundRTCPPackets"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundRTCPPackets"] + out2.(map[string]any)["outboundBytes"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"] + out2.(map[string]any)["outboundRTPPackets"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPackets"] + out2.(map[string]any)["outboundRTCPPackets"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTCPPackets"] + out2.(map[string]any)["bytesReceived"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesReceived"] + out2.(map[string]any)["bytesSent"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesSent"] + out2.(map[string]any)["rtpPacketsReceived"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtpPacketsReceived"] + out2.(map[string]any)["rtpPacketsSent"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtpPacketsSent"] + out2.(map[string]any)["rtpPacketsLost"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtpPacketsLost"] + out2.(map[string]any)["rtpPacketsJitter"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtpPacketsJitter"] + out2.(map[string]any)["rtcpPacketsReceived"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtcpPacketsReceived"] + out2.(map[string]any)["rtcpPacketsSent"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["rtcpPacketsSent"] + } + require.Equal(t, out1.(map[string]any)["items"].([]any)[0], out2) }) } diff --git a/internal/core/metrics_test.go b/internal/core/metrics_test.go index 9c53c544..d05bc770 100644 --- a/internal/core/metrics_test.go +++ b/internal/core/metrics_test.go @@ -196,6 +196,14 @@ srt_conns_packets_received_avg_belated_time 0 srt_conns_packets_send_loss_rate 0 srt_conns_packets_received_loss_rate 0 webrtc_sessions 0 +webrtc_sessions_inbound_bytes 0 +webrtc_sessions_inbound_rtp_packets 0 +webrtc_sessions_inbound_rtp_packets_lost 0 +webrtc_sessions_inbound_rtp_packets_jitter 0 +webrtc_sessions_inbound_rtcp_packets 0 +webrtc_sessions_outbound_bytes 0 +webrtc_sessions_outbound_rtp_packets 0 +webrtc_sessions_outbound_rtcp_packets 0 webrtc_sessions_bytes_received 0 webrtc_sessions_bytes_sent 0 webrtc_sessions_rtp_packets_received 0 @@ -539,6 +547,14 @@ webrtc_sessions_rtcp_packets_sent 0 `srt_conns_packets_send_loss_rate\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `srt_conns_packets_received_loss_rate\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `webrtc_sessions\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} 1`+"\n"+ + `webrtc_sessions_inbound_bytes\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `webrtc_sessions_inbound_rtp_packets\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `webrtc_sessions_inbound_rtp_packets_lost\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `webrtc_sessions_inbound_rtp_packets_jitter\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `webrtc_sessions_inbound_rtcp_packets\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `webrtc_sessions_outbound_bytes\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `webrtc_sessions_outbound_rtp_packets\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `webrtc_sessions_outbound_rtcp_packets\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `webrtc_sessions_bytes_received\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `webrtc_sessions_bytes_sent\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `webrtc_sessions_rtp_packets_received\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ diff --git a/internal/defs/api_rtsp.go b/internal/defs/api_rtsp.go index e04a03e0..acfb8c14 100644 --- a/internal/defs/api_rtsp.go +++ b/internal/defs/api_rtsp.go @@ -68,16 +68,17 @@ type APIRTSPSession struct { OutboundRTPPackets uint64 `json:"outboundRTPPackets"` OutboundRTPPacketsReportedLost uint64 `json:"outboundRTPPacketsReportedLost"` OutboundRTCPPackets uint64 `json:"outboundRTCPPackets"` - BytesReceived uint64 `json:"bytesReceived" deprecated:"true"` - BytesSent uint64 `json:"bytesSent" deprecated:"true"` - RTPPacketsReceived uint64 `json:"rtpPacketsReceived" deprecated:"true"` - RTPPacketsSent uint64 `json:"rtpPacketsSent" deprecated:"true"` - RTPPacketsLost uint64 `json:"rtpPacketsLost" deprecated:"true"` - RTPPacketsInError uint64 `json:"rtpPacketsInError" deprecated:"true"` - RTPPacketsJitter float64 `json:"rtpPacketsJitter" deprecated:"true"` - RTCPPacketsReceived uint64 `json:"rtcpPacketsReceived" deprecated:"true"` - RTCPPacketsSent uint64 `json:"rtcpPacketsSent" deprecated:"true"` - RTCPPacketsInError uint64 `json:"rtcpPacketsInError" deprecated:"true"` + // deprecated + BytesReceived uint64 `json:"bytesReceived" deprecated:"true"` + BytesSent uint64 `json:"bytesSent" deprecated:"true"` + RTPPacketsReceived uint64 `json:"rtpPacketsReceived" deprecated:"true"` + RTPPacketsSent uint64 `json:"rtpPacketsSent" deprecated:"true"` + RTPPacketsLost uint64 `json:"rtpPacketsLost" deprecated:"true"` + RTPPacketsInError uint64 `json:"rtpPacketsInError" deprecated:"true"` + RTPPacketsJitter float64 `json:"rtpPacketsJitter" deprecated:"true"` + RTCPPacketsReceived uint64 `json:"rtcpPacketsReceived" deprecated:"true"` + RTCPPacketsSent uint64 `json:"rtcpPacketsSent" deprecated:"true"` + RTCPPacketsInError uint64 `json:"rtcpPacketsInError" deprecated:"true"` } // APIRTSPSessionList is a list of RTSP sessions. diff --git a/internal/defs/api_webrtc.go b/internal/defs/api_webrtc.go index 746f8ffa..e0f145a5 100644 --- a/internal/defs/api_webrtc.go +++ b/internal/defs/api_webrtc.go @@ -34,14 +34,23 @@ type APIWebRTCSession struct { Path string `json:"path"` Query string `json:"query"` User string `json:"user"` - BytesReceived uint64 `json:"bytesReceived"` - BytesSent uint64 `json:"bytesSent"` - RTPPacketsReceived uint64 `json:"rtpPacketsReceived"` - RTPPacketsSent uint64 `json:"rtpPacketsSent"` - RTPPacketsLost uint64 `json:"rtpPacketsLost"` - RTPPacketsJitter float64 `json:"rtpPacketsJitter"` - RTCPPacketsReceived uint64 `json:"rtcpPacketsReceived"` - RTCPPacketsSent uint64 `json:"rtcpPacketsSent"` + InboundBytes uint64 `json:"inboundBytes"` + InboundRTPPackets uint64 `json:"inboundRTPPackets"` + InboundRTPPacketsLost uint64 `json:"inboundRTPPacketsLost"` + InboundRTPPacketsJitter float64 `json:"inboundRTPPacketsJitter"` + InboundRTCPPackets uint64 `json:"inboundRTCPPackets"` + OutboundBytes uint64 `json:"outboundBytes"` + OutboundRTPPackets uint64 `json:"outboundRTPPackets"` + OutboundRTCPPackets uint64 `json:"outboundRTCPPackets"` + // deprecated + BytesReceived uint64 `json:"bytesReceived" deprecated:"true"` + BytesSent uint64 `json:"bytesSent" deprecated:"true"` + RTPPacketsReceived uint64 `json:"rtpPacketsReceived" deprecated:"true"` + RTPPacketsSent uint64 `json:"rtpPacketsSent" deprecated:"true"` + RTPPacketsLost uint64 `json:"rtpPacketsLost" deprecated:"true"` + RTPPacketsJitter float64 `json:"rtpPacketsJitter" deprecated:"true"` + RTCPPacketsReceived uint64 `json:"rtcpPacketsReceived" deprecated:"true"` + RTCPPacketsSent uint64 `json:"rtcpPacketsSent" deprecated:"true"` } // APIWebRTCSessionList is a list of WebRTC sessions. diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 66db0116..30f5da3b 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -638,6 +638,15 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { "remoteAddr": i.RemoteAddr, }) out += metric("webrtc_sessions", ta, 1) + out += metric("webrtc_sessions_inbound_bytes", ta, int64(i.InboundBytes)) + out += metric("webrtc_sessions_inbound_rtp_packets", ta, int64(i.InboundRTPPackets)) + out += metric("webrtc_sessions_inbound_rtp_packets_lost", ta, int64(i.InboundRTPPacketsLost)) + out += metricFloat("webrtc_sessions_inbound_rtp_packets_jitter", ta, i.InboundRTPPacketsJitter) + out += metric("webrtc_sessions_inbound_rtcp_packets", ta, int64(i.InboundRTCPPackets)) + out += metric("webrtc_sessions_outbound_bytes", ta, int64(i.OutboundBytes)) + out += metric("webrtc_sessions_outbound_rtp_packets", ta, int64(i.OutboundRTPPackets)) + out += metric("webrtc_sessions_outbound_rtcp_packets", ta, int64(i.OutboundRTCPPackets)) + // deprecated out += metric("webrtc_sessions_bytes_received", ta, int64(i.BytesReceived)) out += metric("webrtc_sessions_bytes_sent", ta, int64(i.BytesSent)) out += metric("webrtc_sessions_rtp_packets_received", ta, int64(i.RTPPacketsReceived)) @@ -650,6 +659,15 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { } } else if webrtcSessionFilter == "" { out += metric("webrtc_sessions", "", 0) + out += metric("webrtc_sessions_inbound_bytes", "", 0) + out += metric("webrtc_sessions_inbound_rtp_packets", "", 0) + out += metric("webrtc_sessions_inbound_rtp_packets_lost", "", 0) + out += metricFloat("webrtc_sessions_inbound_rtp_packets_jitter", "", 0) + out += metric("webrtc_sessions_inbound_rtcp_packets", "", 0) + out += metric("webrtc_sessions_outbound_bytes", "", 0) + out += metric("webrtc_sessions_outbound_rtp_packets", "", 0) + out += metric("webrtc_sessions_outbound_rtcp_packets", "", 0) + // deprecated out += metric("webrtc_sessions_bytes_received", "", 0) out += metric("webrtc_sessions_bytes_sent", "", 0) out += metric("webrtc_sessions_rtp_packets_received", "", 0) diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index c3e15346..cb5ba044 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -184,6 +184,14 @@ func (dummyWebRTCServer) APISessionsList() (*defs.APIWebRTCSessionList, error) { State: defs.APIWebRTCSessionStateRead, Path: "mypath", Query: "myquery", + InboundBytes: 123, + InboundRTPPackets: 789, + InboundRTPPacketsLost: 456, + InboundRTPPacketsJitter: 789, + InboundRTCPPackets: 123, + OutboundBytes: 456, + OutboundRTPPackets: 123, + OutboundRTCPPackets: 456, BytesReceived: 123, BytesSent: 456, RTPPacketsReceived: 789, @@ -405,6 +413,22 @@ func TestMetrics(t *testing.T) { `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 456`+"\n"+ `webrtc_sessions{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 1`+"\n"+ + `webrtc_sessions_inbound_bytes{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 123`+"\n"+ + `webrtc_sessions_inbound_rtp_packets{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 789`+"\n"+ + `webrtc_sessions_inbound_rtp_packets_lost{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 456`+"\n"+ + `webrtc_sessions_inbound_rtp_packets_jitter{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 789`+"\n"+ + `webrtc_sessions_inbound_rtcp_packets{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 123`+"\n"+ + `webrtc_sessions_outbound_bytes{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 456`+"\n"+ + `webrtc_sessions_outbound_rtp_packets{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 123`+"\n"+ + `webrtc_sessions_outbound_rtcp_packets{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 456`+"\n"+ `webrtc_sessions_bytes_received{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 123`+"\n"+ `webrtc_sessions_bytes_sent{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ diff --git a/internal/protocols/webrtc/incoming_track.go b/internal/protocols/webrtc/incoming_track.go index 0cd1345a..1ec8e436 100644 --- a/internal/protocols/webrtc/incoming_track.go +++ b/internal/protocols/webrtc/incoming_track.go @@ -244,8 +244,8 @@ type IncomingTrack struct { writeRTCP func([]rtcp.Packet) error log logger.Writer - packetsLost *counterdumper.Dumper - rtpReceiver *rtpreceiver.Receiver + inboundRTPPacketsLost *counterdumper.Dumper + rtpReceiver *rtpreceiver.Receiver } func (t *IncomingTrack) initialize() { @@ -268,7 +268,7 @@ func (*IncomingTrack) PTSEqualsDTS(*rtp.Packet) bool { } func (t *IncomingTrack) start() { - t.packetsLost = &counterdumper.Dumper{ + t.inboundRTPPacketsLost = &counterdumper.Dumper{ OnReport: func(val uint64) { t.log.Log(logger.Warn, "%d RTP %s lost", val, @@ -280,7 +280,7 @@ func (t *IncomingTrack) start() { }()) }, } - t.packetsLost.Start() + t.inboundRTPPacketsLost.Start() t.rtpReceiver = &rtpreceiver.Receiver{ ClockRate: int(t.track.Codec().ClockRate), @@ -348,7 +348,7 @@ func (t *IncomingTrack) start() { packets, lost := t.rtpReceiver.ProcessPacket2(pkt, time.Now(), true) if lost != 0 { - t.packetsLost.Add(lost) + t.inboundRTPPacketsLost.Add(lost) // do not return } @@ -370,8 +370,8 @@ func (t *IncomingTrack) PacketNTP(pkt *rtp.Packet) (time.Time, bool) { } func (t *IncomingTrack) close() { - if t.packetsLost != nil { - t.packetsLost.Stop() + if t.inboundRTPPacketsLost != nil { + t.inboundRTPPacketsLost.Stop() } if t.rtpReceiver != nil { t.rtpReceiver.Close() diff --git a/internal/servers/rtsp/session.go b/internal/servers/rtsp/session.go index 6b69ca15..cc83071a 100644 --- a/internal/servers/rtsp/session.go +++ b/internal/servers/rtsp/session.go @@ -52,25 +52,25 @@ type session struct { pathManager serverPathManager parent sessionParent - uuid uuid.UUID - created time.Time - pathConf *conf.Path // record only - path defs.Path - stream *stream.Stream - subStream *stream.SubStream - onUnreadHook func() - packetsLost *counterdumper.Dumper - decodeErrors *errordumper.Dumper - discardedFrames *counterdumper.Dumper - mutex sync.RWMutex - user string + uuid uuid.UUID + created time.Time + pathConf *conf.Path // record only + path defs.Path + stream *stream.Stream + subStream *stream.SubStream + onUnreadHook func() + inboundRTPPacketsLost *counterdumper.Dumper + inboundRTPPacketsInError *errordumper.Dumper + outboundRTPPacketsDiscarded *counterdumper.Dumper + mutex sync.RWMutex + user string } func (s *session) initialize() { s.uuid = uuid.New() s.created = time.Now() - s.packetsLost = &counterdumper.Dumper{ + s.inboundRTPPacketsLost = &counterdumper.Dumper{ OnReport: func(val uint64) { s.Log(logger.Warn, "%d RTP %s lost", val, @@ -82,9 +82,9 @@ func (s *session) initialize() { }()) }, } - s.packetsLost.Start() + s.inboundRTPPacketsLost.Start() - s.decodeErrors = &errordumper.Dumper{ + s.inboundRTPPacketsInError = &errordumper.Dumper{ OnReport: func(val uint64, last error) { if val == 1 { s.Log(logger.Warn, "decode error: %v", last) @@ -93,9 +93,9 @@ func (s *session) initialize() { } }, } - s.decodeErrors.Start() + s.inboundRTPPacketsInError.Start() - s.discardedFrames = &counterdumper.Dumper{ + s.outboundRTPPacketsDiscarded = &counterdumper.Dumper{ OnReport: func(val uint64) { s.Log(logger.Warn, "reader is too slow, discarding %d %s", val, @@ -107,7 +107,7 @@ func (s *session) initialize() { }()) }, } - s.discardedFrames.Start() + s.outboundRTPPacketsDiscarded.Start() s.Log(logger.Info, "created by %v", s.rconn.NetConn().RemoteAddr()) } @@ -147,9 +147,9 @@ func (s *session) onClose(err error) { s.stream = nil s.subStream = nil - s.discardedFrames.Stop() - s.decodeErrors.Stop() - s.packetsLost.Stop() + s.outboundRTPPacketsDiscarded.Stop() + s.inboundRTPPacketsInError.Stop() + s.inboundRTPPacketsLost.Stop() s.Log(logger.Info, "destroyed: %v", err) } @@ -405,18 +405,18 @@ func (s *session) APISourceDescribe() *defs.APIPathSource { // onPacketLost is called by rtspServer. func (s *session) onPacketsLost(ctx *gortsplib.ServerHandlerOnPacketsLostCtx) { - s.packetsLost.Add(ctx.Lost) + s.inboundRTPPacketsLost.Add(ctx.Lost) } // onDecodeError is called by rtspServer. func (s *session) onDecodeError(ctx *gortsplib.ServerHandlerOnDecodeErrorCtx) { - s.decodeErrors.Add(ctx.Error) + s.inboundRTPPacketsInError.Add(ctx.Error) } // onStreamWriteError is called by rtspServer. func (s *session) onStreamWriteError(_ *gortsplib.ServerHandlerOnStreamWriteErrorCtx) { // currently the only error returned by OnStreamWriteError is ErrServerWriteQueueFull - s.discardedFrames.Increase() + s.outboundRTPPacketsDiscarded.Increase() } func (s *session) apiItem() *defs.APIRTSPSession { diff --git a/internal/servers/webrtc/server_test.go b/internal/servers/webrtc/server_test.go index a8e49da3..67ed9d64 100644 --- a/internal/servers/webrtc/server_test.go +++ b/internal/servers/webrtc/server_test.go @@ -324,7 +324,7 @@ func TestServerPublish(t *testing.T) { list, err := s.APISessionsList() require.NoError(t, err) - require.Equal(t, &defs.APIWebRTCSessionList{ + require.Equal(t, &defs.APIWebRTCSessionList{ //nolint:dupl Items: []defs.APIWebRTCSession{ { ID: list.Items[0].ID, @@ -334,6 +334,14 @@ func TestServerPublish(t *testing.T) { Path: "teststream", Query: "param=value", User: "myuser", + InboundBytes: list.Items[0].InboundBytes, + InboundRTPPackets: list.Items[0].InboundRTPPackets, + InboundRTPPacketsLost: list.Items[0].InboundRTPPacketsLost, + InboundRTPPacketsJitter: list.Items[0].InboundRTPPacketsJitter, + InboundRTCPPackets: list.Items[0].InboundRTCPPackets, + OutboundBytes: list.Items[0].OutboundBytes, + OutboundRTPPackets: list.Items[0].OutboundRTPPackets, + OutboundRTCPPackets: list.Items[0].OutboundRTCPPackets, BytesReceived: list.Items[0].BytesReceived, BytesSent: list.Items[0].BytesSent, RTPPacketsReceived: list.Items[0].RTPPacketsReceived, @@ -609,7 +617,7 @@ func TestServerRead(t *testing.T) { list, err := s.APISessionsList() require.NoError(t, err) - require.Equal(t, &defs.APIWebRTCSessionList{ + require.Equal(t, &defs.APIWebRTCSessionList{ //nolint:dupl Items: []defs.APIWebRTCSession{ { ID: list.Items[0].ID, @@ -619,6 +627,14 @@ func TestServerRead(t *testing.T) { Path: "teststream", Query: "param=value", User: "myuser", + InboundBytes: list.Items[0].InboundBytes, + InboundRTPPackets: list.Items[0].InboundRTPPackets, + InboundRTPPacketsLost: list.Items[0].InboundRTPPacketsLost, + InboundRTPPacketsJitter: list.Items[0].InboundRTPPacketsJitter, + InboundRTCPPackets: list.Items[0].InboundRTCPPackets, + OutboundBytes: list.Items[0].OutboundBytes, + OutboundRTPPackets: list.Items[0].OutboundRTPPackets, + OutboundRTCPPackets: list.Items[0].OutboundRTCPPackets, BytesReceived: list.Items[0].BytesReceived, BytesSent: list.Items[0].BytesSent, RTPPacketsReceived: list.Items[0].RTPPacketsReceived, diff --git a/internal/servers/webrtc/session.go b/internal/servers/webrtc/session.go index 37315397..6454b7ba 100644 --- a/internal/servers/webrtc/session.go +++ b/internal/servers/webrtc/session.go @@ -508,16 +508,24 @@ func (s *session) apiItem() *defs.APIWebRTCSession { } return defs.APIWebRTCSessionStateRead }(), - Path: s.req.pathName, - Query: s.req.httpRequest.URL.RawQuery, - User: s.user, - BytesReceived: bytesReceived, - BytesSent: bytesSent, - RTPPacketsReceived: rtpPacketsReceived, - RTPPacketsSent: rtpPacketsSent, - RTPPacketsLost: rtpPacketsLost, - RTPPacketsJitter: rtpPacketsJitter, - RTCPPacketsReceived: rtcpPacketsReceived, - RTCPPacketsSent: rtcpPacketsSent, + Path: s.req.pathName, + Query: s.req.httpRequest.URL.RawQuery, + User: s.user, + InboundBytes: bytesReceived, + InboundRTPPackets: rtpPacketsReceived, + InboundRTPPacketsLost: rtpPacketsLost, + InboundRTPPacketsJitter: rtpPacketsJitter, + InboundRTCPPackets: rtcpPacketsReceived, + OutboundBytes: bytesSent, + OutboundRTPPackets: rtpPacketsSent, + OutboundRTCPPackets: rtcpPacketsSent, + BytesReceived: bytesReceived, + BytesSent: bytesSent, + RTPPacketsReceived: rtpPacketsReceived, + RTPPacketsSent: rtpPacketsSent, + RTPPacketsLost: rtpPacketsLost, + RTPPacketsJitter: rtpPacketsJitter, + RTCPPacketsReceived: rtcpPacketsReceived, + RTCPPacketsSent: rtcpPacketsSent, } } diff --git a/internal/stream/reader.go b/internal/stream/reader.go index e1767072..5ad92e75 100644 --- a/internal/stream/reader.go +++ b/internal/stream/reader.go @@ -19,10 +19,10 @@ type Reader struct { SkipBytesSent bool Parent logger.Writer - onDatas map[*description.Media]map[format.Format]OnDataFunc - queueSize int - buffer *ringbuffer.RingBuffer - discardedFrames *counterdumper.Dumper + onDatas map[*description.Media]map[format.Format]OnDataFunc + queueSize int + buffer *ringbuffer.RingBuffer + outboundFramesDiscarded *counterdumper.Dumper // out err chan error @@ -76,7 +76,7 @@ func (r *Reader) start() { r.buffer = buffer r.err = make(chan error) - r.discardedFrames = &counterdumper.Dumper{ + r.outboundFramesDiscarded = &counterdumper.Dumper{ OnReport: func(val uint64) { r.Parent.Log(logger.Warn, "reader is too slow, discarding %d %s", val, @@ -88,14 +88,14 @@ func (r *Reader) start() { }()) }, } - r.discardedFrames.Start() + r.outboundFramesDiscarded.Start() go r.run() } func (r *Reader) stop() { r.buffer.Close() - r.discardedFrames.Stop() + r.outboundFramesDiscarded.Stop() <-r.err } @@ -121,6 +121,6 @@ func (r *Reader) runInner() error { func (r *Reader) push(cb func() error) { ok := r.buffer.Push(cb) if !ok { - r.discardedFrames.Increase() + r.outboundFramesDiscarded.Increase() } } diff --git a/internal/stream/stream.go b/internal/stream/stream.go index c7a1aaa9..5c2f5d69 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -321,17 +321,17 @@ type Stream struct { ReplaceNTP bool Parent logger.Writer - offlineDesc *description.Session - mutex sync.RWMutex - subStream *SubStream - offlineSubStream *offlineSubStream - inboundBytes *uint64 - outboundBytes *uint64 - medias map[*description.Media]*streamMedia - rtspStream *gortsplib.ServerStream - rtspsStream *gortsplib.ServerStream - readers map[*Reader]struct{} - processingErrors *errordumper.Dumper + offlineDesc *description.Session + mutex sync.RWMutex + subStream *SubStream + offlineSubStream *offlineSubStream + inboundBytes *uint64 + outboundBytes *uint64 + medias map[*description.Media]*streamMedia + rtspStream *gortsplib.ServerStream + rtspsStream *gortsplib.ServerStream + readers map[*Reader]struct{} + inboundFramesInError *errordumper.Dumper timeMutex sync.Mutex firstTimeReceived bool @@ -377,7 +377,7 @@ func (s *Stream) Initialize() error { s.readers = make(map[*Reader]struct{}) s.hasReaders = make(chan struct{}) - s.processingErrors = &errordumper.Dumper{ + s.inboundFramesInError = &errordumper.Dumper{ OnReport: func(val uint64, last error) { if val == 1 { s.Parent.Log(logger.Warn, "processing error: %v", last) @@ -386,22 +386,22 @@ func (s *Stream) Initialize() error { } }, } - s.processingErrors.Start() + s.inboundFramesInError.Start() s.lastSystemTime = time.Now() for _, media := range s.Desc.Medias { sm := &streamMedia{ - media: media, - alwaysAvailable: s.AlwaysAvailable, - rtpMaxPayloadSize: s.RTPMaxPayloadSize, - replaceNTP: s.ReplaceNTP, - addBytesReceived: s.addBytesReceived, - addBytesSent: s.addBytesSent, - updateLastTime: s.updateLastTime, - writeRTSP: s.writeRTSP, - processingErrors: s.processingErrors, - parent: s.Parent, + media: media, + alwaysAvailable: s.AlwaysAvailable, + rtpMaxPayloadSize: s.RTPMaxPayloadSize, + replaceNTP: s.ReplaceNTP, + addInboundBytes: s.addInboundBytes, + addOutboundBytes: s.addOutboundBytes, + updateLastTime: s.updateLastTime, + writeRTSP: s.writeRTSP, + inboundFramesInError: s.inboundFramesInError, + parent: s.Parent, } err := sm.initialize() if err != nil { @@ -426,7 +426,7 @@ func (s *Stream) Close() { s.offlineSubStream.close(false) } - s.processingErrors.Stop() + s.inboundFramesInError.Stop() if s.rtspStream != nil { s.rtspStream.Close() @@ -571,11 +571,11 @@ func (s *Stream) WaitForReaders() { <-s.hasReaders } -func (s *Stream) addBytesReceived(v uint64) { +func (s *Stream) addInboundBytes(v uint64) { atomic.AddUint64(s.inboundBytes, v) } -func (s *Stream) addBytesSent(v uint64) { +func (s *Stream) addOutboundBytes(v uint64) { atomic.AddUint64(s.outboundBytes, v) } diff --git a/internal/stream/stream_format.go b/internal/stream/stream_format.go index 8f76ed44..522b7dc4 100644 --- a/internal/stream/stream_format.go +++ b/internal/stream/stream_format.go @@ -38,17 +38,17 @@ func randUint32() (uint32, error) { } type streamFormat struct { - format format.Format - media *description.Media - alwaysAvailable bool - rtpMaxPayloadSize int - replaceNTP bool - processingErrors *errordumper.Dumper - addBytesReceived func(uint64) - addBytesSent func(uint64) - updateLastTime func(time.Duration) - writeRTSP func(*description.Media, []*rtp.Packet, time.Time) - parent logger.Writer + format format.Format + media *description.Media + alwaysAvailable bool + rtpMaxPayloadSize int + replaceNTP bool + inboundFramesInError *errordumper.Dumper + addInboundBytes func(uint64) + addOutboundBytes func(uint64) + updateLastTime func(time.Duration) + writeRTSP func(*description.Media, []*rtp.Packet, time.Time) + parent logger.Writer ptsOffset int64 formatUpdater formatUpdater diff --git a/internal/stream/stream_media.go b/internal/stream/stream_media.go index 90e314c4..c7e4dff0 100644 --- a/internal/stream/stream_media.go +++ b/internal/stream/stream_media.go @@ -11,16 +11,16 @@ import ( ) type streamMedia struct { - media *description.Media - alwaysAvailable bool - rtpMaxPayloadSize int - replaceNTP bool - addBytesReceived func(uint64) - addBytesSent func(uint64) - updateLastTime func(time.Duration) - writeRTSP func(*description.Media, []*rtp.Packet, time.Time) - processingErrors *errordumper.Dumper - parent logger.Writer + media *description.Media + alwaysAvailable bool + rtpMaxPayloadSize int + replaceNTP bool + addInboundBytes func(uint64) + addOutboundBytes func(uint64) + updateLastTime func(time.Duration) + writeRTSP func(*description.Media, []*rtp.Packet, time.Time) + inboundFramesInError *errordumper.Dumper + parent logger.Writer formats map[format.Format]*streamFormat } @@ -30,17 +30,17 @@ func (sm *streamMedia) initialize() error { for _, forma := range sm.media.Formats { sf := &streamFormat{ - format: forma, - media: sm.media, - alwaysAvailable: sm.alwaysAvailable, - rtpMaxPayloadSize: sm.rtpMaxPayloadSize, - replaceNTP: sm.replaceNTP, - processingErrors: sm.processingErrors, - addBytesReceived: sm.addBytesReceived, - addBytesSent: sm.addBytesSent, - updateLastTime: sm.updateLastTime, - writeRTSP: sm.writeRTSP, - parent: sm.parent, + format: forma, + media: sm.media, + alwaysAvailable: sm.alwaysAvailable, + rtpMaxPayloadSize: sm.rtpMaxPayloadSize, + replaceNTP: sm.replaceNTP, + inboundFramesInError: sm.inboundFramesInError, + addInboundBytes: sm.addInboundBytes, + addOutboundBytes: sm.addOutboundBytes, + updateLastTime: sm.updateLastTime, + writeRTSP: sm.writeRTSP, + parent: sm.parent, } err := sf.initialize() if err != nil { diff --git a/internal/stream/sub_stream_format.go b/internal/stream/sub_stream_format.go index 0651f627..4cd45611 100644 --- a/internal/stream/sub_stream_format.go +++ b/internal/stream/sub_stream_format.go @@ -94,7 +94,7 @@ func (ssf *subStreamFormat) initialize2(firstTimeReceived bool, lastPTS time.Dur func (ssf *subStreamFormat) writeUnit(u *unit.Unit) { err := ssf.writeUnitInner(u) if err != nil { - ssf.streamFormat.processingErrors.Add(err) + ssf.streamFormat.inboundFramesInError.Add(err) return } } @@ -170,7 +170,7 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { } size := unitSize(u) - ssf.streamFormat.addBytesReceived(size) + ssf.streamFormat.addInboundBytes(size) ssf.streamFormat.writeRTSP(ssf.streamFormat.media, u.RTPPackets, u.NTP) @@ -179,7 +179,7 @@ func (ssf *subStreamFormat) writeUnitInner(u *unit.Unit) error { cOnData := onData sr.push(func() error { if !csr.SkipBytesSent { - ssf.streamFormat.addBytesSent(size) + ssf.streamFormat.addOutboundBytes(size) } return cOnData(u) })