OPC/mediamtx
2
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

api: rename WebRTC stats to match RTSP ones (#5581)

Browse Source
This commit is contained in:
Alessandro Ros
2026-03-16 23:29:25 +01:00
committed by GitHub
parent 02e05f1f16
commit adf45596a3
17 changed files with 350 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/openapi.yaml
+32
View File
@@ -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
internal/api/api_webrtc_test.go
+36
View File
@@ -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,
internal/core/api_test.go
+59 -2
View File
@@ -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)
})
}
internal/core/metrics_test.go
+16
View File
@@ -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"+
internal/defs/api_rtsp.go
+11 -10
View File
@@ -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.
internal/defs/api_webrtc.go
+17 -8
View File
@@ -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.
internal/metrics/metrics.go
+18
View File
@@ -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)
internal/metrics/metrics_test.go
+24
View File
@@ -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",`+
internal/protocols/webrtc/incoming_track.go
+7 -7
View File
@@ -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()
internal/servers/rtsp/session.go
+24 -24
View File
@@ -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 {
internal/servers/webrtc/server_test.go
+18 -2
View File
@@ -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,
internal/servers/webrtc/session.go
+19 -11
View File
@@ -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,
}
}
internal/stream/reader.go
+8 -8
View File
@@ -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()
}
}
internal/stream/stream.go
+26 -26
View File
@@ -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)
}
internal/stream/stream_format.go
+11 -11
View File
@@ -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
internal/stream/stream_media.go
+21 -21
View File
@@ -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 {
internal/stream/sub_stream_format.go
+3 -3
View File
@@ -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)
})