diff --git a/api/openapi.yaml b/api/openapi.yaml index 1bcb72a3..2fa803e3 100644 --- a/api/openapi.yaml +++ b/api/openapi.yaml @@ -973,12 +973,23 @@ components: type: array items: $ref: '#/components/schemas/PathTrackCodec' + inboundBytes: + type: integer + format: uint64 + outboundBytes: + type: integer + format: uint64 + inboundFramesInError: + type: integer + format: uint64 bytesReceived: type: integer format: uint64 + deprecated: true bytesSent: type: integer format: uint64 + deprecated: true readers: type: array items: @@ -1023,9 +1034,16 @@ components: type: string lastRequest: type: string + outboundBytes: + type: integer + format: uint64 + outboundFramesDiscarded: + type: integer + format: uint64 bytesSent: type: integer format: uint64 + deprecated: true HLSMuxerList: type: object @@ -1089,12 +1107,23 @@ components: type: string user: type: string + inboundBytes: + type: integer + format: uint64 + outboundBytes: + type: integer + format: uint64 + outboundFramesDiscarded: + type: integer + format: uint64 bytesReceived: type: integer format: uint64 + deprecated: true bytesSent: type: integer format: uint64 + deprecated: true RTMPConnList: type: object @@ -1214,6 +1243,9 @@ components: outboundRTPPacketsReportedLost: type: integer format: uint64 + outboundRTPPacketsDiscarded: + type: integer + format: uint64 outboundRTCPPackets: type: integer format: uint64 @@ -1500,6 +1532,9 @@ components: type: number format: double description: Percentage of retransmitted data vs. received data + outboundFramesDiscarded: + type: integer + format: uint64 SRTConnList: type: object @@ -1589,6 +1624,9 @@ components: outboundRTCPPackets: type: integer format: uint64 + outboundFramesDiscarded: + type: integer + format: uint64 bytesReceived: type: integer format: uint64 diff --git a/internal/api/api_hls_test.go b/internal/api/api_hls_test.go index a358a1d1..71d57000 100644 --- a/internal/api/api_hls_test.go +++ b/internal/api/api_hls_test.go @@ -37,16 +37,20 @@ func TestHLSMuxersList(t *testing.T) { hlsServer := &testHLSServer{ muxers: map[string]*defs.APIHLSMuxer{ "test1": { - Path: "test1", - Created: now, - LastRequest: now.Add(5 * time.Second), - BytesSent: 1234, + Path: "test1", + Created: now, + LastRequest: now.Add(5 * time.Second), + OutboundBytes: 1234, + OutboundFramesDiscarded: 10, + BytesSent: 1234, }, "test2": { - Path: "test2", - Created: now.Add(time.Minute), - LastRequest: now.Add(time.Minute + 10*time.Second), - BytesSent: 5678, + Path: "test2", + Created: now.Add(time.Minute), + LastRequest: now.Add(time.Minute + 10*time.Second), + OutboundBytes: 5678, + OutboundFramesDiscarded: 20, + BytesSent: 5678, }, }, } @@ -80,10 +84,12 @@ func TestHLSMuxersGet(t *testing.T) { hlsServer := &testHLSServer{ muxers: map[string]*defs.APIHLSMuxer{ "mypath": { - Path: "mypath", - Created: now, - LastRequest: now.Add(5 * time.Second), - BytesSent: 9999, + Path: "mypath", + Created: now, + LastRequest: now.Add(5 * time.Second), + OutboundBytes: 9999, + OutboundFramesDiscarded: 12, + BytesSent: 9999, }, }, } @@ -108,5 +114,7 @@ func TestHLSMuxersGet(t *testing.T) { httpRequest(t, hc, http.MethodGet, "http://localhost:9997/v3/hlsmuxers/get/mypath", nil, &out) require.Equal(t, "mypath", out.Path) + require.Equal(t, uint64(9999), out.OutboundBytes) + require.Equal(t, uint64(12), out.OutboundFramesDiscarded) require.Equal(t, uint64(9999), out.BytesSent) } diff --git a/internal/api/api_paths_test.go b/internal/api/api_paths_test.go index 122e3826..0f1a36c9 100644 --- a/internal/api/api_paths_test.go +++ b/internal/api/api_paths_test.go @@ -36,26 +36,32 @@ func TestPathsList(t *testing.T) { pathManager := &testPathManager{ paths: map[string]*defs.APIPath{ "test1": { - Name: "test1", - ConfName: "test1", - Source: &defs.APIPathSource{Type: defs.APIPathSourceTypeRTSPSession, ID: "pub1"}, - Ready: true, - ReadyTime: &now, - Tracks: []defs.APIPathTrackCodec{defs.APIPathTrackCodecH264, defs.APIPathTrackCodecOpus}, - BytesReceived: 1000, - BytesSent: 2000, + Name: "test1", + ConfName: "test1", + Source: &defs.APIPathSource{Type: defs.APIPathSourceTypeRTSPSession, ID: "pub1"}, + Ready: true, + ReadyTime: &now, + Tracks: []defs.APIPathTrackCodec{defs.APIPathTrackCodecH264, defs.APIPathTrackCodecOpus}, + InboundBytes: 1000, + OutboundBytes: 2000, + InboundFramesInError: 3, + BytesReceived: 1000, + BytesSent: 2000, Readers: []defs.APIPathReader{ {Type: defs.APIPathReaderTypeRTSPSession, ID: "reader1"}, }, }, "test2": { - Name: "test2", - ConfName: "test2", - Ready: false, - Tracks: []defs.APIPathTrackCodec{}, - BytesReceived: 500, - BytesSent: 100, - Readers: []defs.APIPathReader{}, + Name: "test2", + ConfName: "test2", + Ready: false, + Tracks: []defs.APIPathTrackCodec{}, + InboundBytes: 500, + OutboundBytes: 100, + InboundFramesInError: 1, + BytesReceived: 500, + BytesSent: 100, + Readers: []defs.APIPathReader{}, }, }, } @@ -89,14 +95,17 @@ func TestPathsGet(t *testing.T) { pathManager := &testPathManager{ paths: map[string]*defs.APIPath{ "mystream": { - Name: "mystream", - ConfName: "mystream", - Source: &defs.APIPathSource{Type: defs.APIPathSourceTypeRTSPSession, ID: "session123"}, - Ready: true, - ReadyTime: &now, - Tracks: []defs.APIPathTrackCodec{defs.APIPathTrackCodecH264, defs.APIPathTrackCodecOpus}, - BytesReceived: 123456, - BytesSent: 789012, + Name: "mystream", + ConfName: "mystream", + Source: &defs.APIPathSource{Type: defs.APIPathSourceTypeRTSPSession, ID: "session123"}, + Ready: true, + ReadyTime: &now, + Tracks: []defs.APIPathTrackCodec{defs.APIPathTrackCodecH264, defs.APIPathTrackCodecOpus}, + InboundBytes: 123456, + OutboundBytes: 789012, + InboundFramesInError: 12, + BytesReceived: 123456, + BytesSent: 789012, Readers: []defs.APIPathReader{ {Type: defs.APIPathReaderTypeHLSMuxer, ID: "muxer1"}, {Type: defs.APIPathReaderTypeWebRTCSession, ID: "session456"}, @@ -131,6 +140,9 @@ func TestPathsGet(t *testing.T) { require.Equal(t, defs.APIPathSourceTypeRTSPSession, out.Source.Type) require.Len(t, out.Tracks, 2) require.Len(t, out.Readers, 2) + require.Equal(t, uint64(123456), out.InboundBytes) + require.Equal(t, uint64(789012), out.OutboundBytes) + require.Equal(t, uint64(12), out.InboundFramesInError) require.Equal(t, uint64(123456), out.BytesReceived) require.Equal(t, uint64(789012), out.BytesSent) } diff --git a/internal/api/api_rtmp_test.go b/internal/api/api_rtmp_test.go index 7cc831f8..754aa075 100644 --- a/internal/api/api_rtmp_test.go +++ b/internal/api/api_rtmp_test.go @@ -67,24 +67,30 @@ func TestRTMPConnsList(t *testing.T) { rtmpServer := &testRTMPServer{ conns: map[uuid.UUID]*defs.APIRTMPConn{ id1: { - ID: id1, - Created: now, - RemoteAddr: "192.168.1.1:5000", - State: defs.APIRTMPConnStatePublish, - Path: "stream1", - Query: "token=abc", - BytesReceived: 1000, - BytesSent: 2000, + ID: id1, + Created: now, + RemoteAddr: "192.168.1.1:5000", + State: defs.APIRTMPConnStatePublish, + Path: "stream1", + Query: "token=abc", + InboundBytes: 1000, + OutboundBytes: 2000, + OutboundFramesDiscarded: 11, + BytesReceived: 1000, + BytesSent: 2000, }, id2: { - ID: id2, - Created: now.Add(time.Minute), - RemoteAddr: "192.168.1.2:5001", - State: defs.APIRTMPConnStateRead, - Path: "stream2", - Query: "", - BytesReceived: 500, - BytesSent: 1500, + ID: id2, + Created: now.Add(time.Minute), + RemoteAddr: "192.168.1.2:5001", + State: defs.APIRTMPConnStateRead, + Path: "stream2", + Query: "", + InboundBytes: 500, + OutboundBytes: 1500, + OutboundFramesDiscarded: 22, + BytesReceived: 500, + BytesSent: 1500, }, }, } @@ -148,14 +154,17 @@ func TestRTMPConnsGet(t *testing.T) { rtmpServer := &testRTMPServer{ conns: map[uuid.UUID]*defs.APIRTMPConn{ id: { - ID: id, - Created: now, - RemoteAddr: "192.168.1.100:5000", - State: defs.APIRTMPConnStatePublish, - Path: ca.path, - Query: "key=value", - BytesReceived: 999999, - BytesSent: 888888, + ID: id, + Created: now, + RemoteAddr: "192.168.1.100:5000", + State: defs.APIRTMPConnStatePublish, + Path: ca.path, + Query: "key=value", + InboundBytes: 999999, + OutboundBytes: 888888, + OutboundFramesDiscarded: 33, + BytesReceived: 999999, + BytesSent: 888888, }, }, } @@ -189,6 +198,9 @@ func TestRTMPConnsGet(t *testing.T) { require.Equal(t, "192.168.1.100:5000", out.RemoteAddr) require.Equal(t, defs.APIRTMPConnStatePublish, out.State) require.Equal(t, ca.path, out.Path) + require.Equal(t, uint64(999999), out.InboundBytes) + require.Equal(t, uint64(888888), out.OutboundBytes) + require.Equal(t, uint64(33), out.OutboundFramesDiscarded) require.Equal(t, uint64(999999), out.BytesReceived) }) } diff --git a/internal/api/api_srt_test.go b/internal/api/api_srt_test.go index c059b3b0..b16d35e9 100644 --- a/internal/api/api_srt_test.go +++ b/internal/api/api_srt_test.go @@ -50,38 +50,40 @@ func TestSRTConnsList(t *testing.T) { srtServer := &testSRTServer{ conns: map[uuid.UUID]*defs.APISRTConn{ id1: { - ID: id1, - Created: now, - RemoteAddr: "192.168.1.1:5000", - State: defs.APISRTConnStatePublish, - Path: "stream1", - Query: "token=abc", - PacketsSent: 1000, - PacketsReceived: 2000, - PacketsSentUnique: 950, - PacketsReceivedUnique: 1950, - BytesReceived: 100000, - BytesSent: 200000, - MsRTT: 10.5, - MbpsSendRate: 5.2, - MbpsReceiveRate: 4.8, + ID: id1, + Created: now, + RemoteAddr: "192.168.1.1:5000", + State: defs.APISRTConnStatePublish, + Path: "stream1", + Query: "token=abc", + PacketsSent: 1000, + PacketsReceived: 2000, + PacketsSentUnique: 950, + PacketsReceivedUnique: 1950, + BytesReceived: 100000, + BytesSent: 200000, + OutboundFramesDiscarded: 5, + MsRTT: 10.5, + MbpsSendRate: 5.2, + MbpsReceiveRate: 4.8, }, id2: { - ID: id2, - Created: now.Add(time.Minute), - RemoteAddr: "192.168.1.2:5001", - State: defs.APISRTConnStateRead, - Path: "stream2", - Query: "", - PacketsSent: 500, - PacketsReceived: 1500, - PacketsSentUnique: 480, - PacketsReceivedUnique: 1470, - BytesReceived: 50000, - BytesSent: 150000, - MsRTT: 15.2, - MbpsSendRate: 3.5, - MbpsReceiveRate: 3.2, + ID: id2, + Created: now.Add(time.Minute), + RemoteAddr: "192.168.1.2:5001", + State: defs.APISRTConnStateRead, + Path: "stream2", + Query: "", + PacketsSent: 500, + PacketsReceived: 1500, + PacketsSentUnique: 480, + PacketsReceivedUnique: 1470, + BytesReceived: 50000, + BytesSent: 150000, + OutboundFramesDiscarded: 6, + MsRTT: 15.2, + MbpsSendRate: 3.5, + MbpsReceiveRate: 3.2, }, }, } @@ -176,6 +178,7 @@ func TestSRTConnsGet(t *testing.T) { PacketsReceivedAvgBelatedTime: 50, PacketsSendLossRate: 0.5, PacketsReceivedLossRate: 0.6, + OutboundFramesDiscarded: 7, }, }, } @@ -205,6 +208,7 @@ func TestSRTConnsGet(t *testing.T) { require.Equal(t, "mystream", out.Path) require.Equal(t, uint64(999999), out.BytesReceived) require.Equal(t, uint64(888888), out.BytesSent) + require.Equal(t, uint64(7), out.OutboundFramesDiscarded) require.Equal(t, 25.5, out.MsRTT) require.Equal(t, 10.5, out.MbpsSendRate) require.Equal(t, 9.8, out.MbpsReceiveRate) diff --git a/internal/api/api_webrtc_test.go b/internal/api/api_webrtc_test.go index 817b201c..1dc2b21d 100644 --- a/internal/api/api_webrtc_test.go +++ b/internal/api/api_webrtc_test.go @@ -67,6 +67,7 @@ func TestWebRTCSessionsList(t *testing.T) { OutboundBytes: 2000, OutboundRTPPackets: 200, OutboundRTCPPackets: 15, + OutboundFramesDiscarded: 11, BytesReceived: 1000, BytesSent: 2000, RTPPacketsReceived: 100, @@ -94,6 +95,7 @@ func TestWebRTCSessionsList(t *testing.T) { OutboundBytes: 1500, OutboundRTPPackets: 150, OutboundRTCPPackets: 10, + OutboundFramesDiscarded: 22, BytesReceived: 500, BytesSent: 1500, RTPPacketsReceived: 50, @@ -154,6 +156,7 @@ func TestWebRTCSessionsGet(t *testing.T) { OutboundBytes: 888888, OutboundRTPPackets: 20000, OutboundRTCPPackets: 200, + OutboundFramesDiscarded: 33, BytesReceived: 999999, BytesSent: 888888, RTPPacketsReceived: 10000, @@ -196,6 +199,7 @@ func TestWebRTCSessionsGet(t *testing.T) { 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(33), out.OutboundFramesDiscarded) require.Equal(t, uint64(999999), out.BytesReceived) require.Equal(t, uint64(888888), out.BytesSent) require.Equal(t, uint64(10000), out.RTPPacketsReceived) diff --git a/internal/core/api_test.go b/internal/core/api_test.go index df523669..87d717b6 100644 --- a/internal/core/api_test.go +++ b/internal/core/api_test.go @@ -80,12 +80,15 @@ func TestAPIPathsList(t *testing.T) { } type path struct { - Name string `json:"name"` - Source pathSource `json:"source"` - Ready bool `json:"ready"` - Tracks []defs.APIPathTrackCodec `json:"tracks"` - BytesReceived uint64 `json:"bytesReceived"` - BytesSent uint64 `json:"bytesSent"` + Name string `json:"name"` + Source pathSource `json:"source"` + Ready bool `json:"ready"` + Tracks []defs.APIPathTrackCodec `json:"tracks"` + InboundBytes uint64 `json:"inboundBytes"` + OutboundBytes uint64 `json:"outboundBytes"` + InboundFramesInError uint64 `json:"inboundFramesInError"` + BytesReceived uint64 `json:"bytesReceived"` + BytesSent uint64 `json:"bytesSent"` } type pathList struct { @@ -136,9 +139,11 @@ func TestAPIPathsList(t *testing.T) { Source: pathSource{ Type: "rtspSession", }, - Ready: true, - Tracks: []defs.APIPathTrackCodec{defs.APIPathTrackCodecH264, defs.APIPathTrackCodecMPEG4Audio}, - BytesReceived: 17, + Ready: true, + Tracks: []defs.APIPathTrackCodec{defs.APIPathTrackCodecH264, defs.APIPathTrackCodecMPEG4Audio}, + InboundBytes: 17, + InboundFramesInError: 0, + BytesReceived: 17, }}, }, out) }) @@ -296,12 +301,15 @@ func TestAPIPathsGet(t *testing.T) { } type path struct { - Name string `json:"name"` - Source pathSource `json:"source"` - Ready bool `json:"Ready"` - Tracks []defs.APIPathTrackCodec `json:"tracks"` - BytesReceived uint64 `json:"bytesReceived"` - BytesSent uint64 `json:"bytesSent"` + Name string `json:"name"` + Source pathSource `json:"source"` + Ready bool `json:"Ready"` + Tracks []defs.APIPathTrackCodec `json:"tracks"` + InboundBytes uint64 `json:"inboundBytes"` + OutboundBytes uint64 `json:"outboundBytes"` + InboundFramesInError uint64 `json:"inboundFramesInError"` + BytesReceived uint64 `json:"bytesReceived"` + BytesSent uint64 `json:"bytesSent"` } var pathName string @@ -652,6 +660,7 @@ func TestAPIProtocolListGet(t *testing.T) { "outboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"], "outboundRTPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPackets"], "outboundRTPPacketsReportedLost": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPacketsReportedLost"], + "outboundRTPPacketsDiscarded": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPacketsDiscarded"], "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"], @@ -712,6 +721,7 @@ func TestAPIProtocolListGet(t *testing.T) { "outboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"], "outboundRTPPackets": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPackets"], "outboundRTPPacketsReportedLost": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPacketsReportedLost"], + "outboundRTPPacketsDiscarded": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPacketsDiscarded"], "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"], @@ -743,15 +753,18 @@ func TestAPIProtocolListGet(t *testing.T) { "itemCount": float64(1), "items": []any{ map[string]any{ - "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"], - "path": "mypath", - "query": "key=val", - "user": "", - "remoteAddr": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["remoteAddr"], - "state": "publish", + "inboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundBytes"], + "outboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"], + "outboundFramesDiscarded": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundFramesDiscarded"], + "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"], + "path": "mypath", + "query": "key=val", + "user": "", + "remoteAddr": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["remoteAddr"], + "state": "publish", }, }, }, out1) @@ -762,15 +775,18 @@ func TestAPIProtocolListGet(t *testing.T) { "itemCount": float64(1), "items": []any{ map[string]any{ - "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"], - "path": "mypath", - "query": "key=val", - "user": "", - "remoteAddr": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["remoteAddr"], - "state": "publish", + "inboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["inboundBytes"], + "outboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"], + "outboundFramesDiscarded": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundFramesDiscarded"], + "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"], + "path": "mypath", + "query": "key=val", + "user": "", + "remoteAddr": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["remoteAddr"], + "state": "publish", }, }, }, out1) @@ -781,10 +797,12 @@ func TestAPIProtocolListGet(t *testing.T) { "pageCount": float64(1), "items": []any{ map[string]any{ - "bytesSent": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesSent"], - "created": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["created"], - "lastRequest": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["lastRequest"], - "path": "mypath", + "outboundBytes": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundBytes"], + "outboundFramesDiscarded": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundFramesDiscarded"], + "bytesSent": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["bytesSent"], + "created": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["created"], + "lastRequest": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["lastRequest"], + "path": "mypath", }, }, }, out1) @@ -803,6 +821,7 @@ func TestAPIProtocolListGet(t *testing.T) { "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"], + "outboundFramesDiscarded": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundFramesDiscarded"], "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"], @@ -841,6 +860,7 @@ func TestAPIProtocolListGet(t *testing.T) { "bytesReceivedLoss": float64(0), "bytesReceivedRetrans": float64(0), "bytesReceivedUndecrypt": float64(0), + "outboundFramesDiscarded": out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundFramesDiscarded"], "bytesReceivedUnique": float64(628), "bytesRetrans": float64(0), "bytesSendBuf": float64(0), @@ -926,6 +946,7 @@ func TestAPIProtocolListGet(t *testing.T) { 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)["outboundRTPPacketsDiscarded"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundRTPPacketsDiscarded"] 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"] @@ -947,6 +968,7 @@ func TestAPIProtocolListGet(t *testing.T) { 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)["outboundFramesDiscarded"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["outboundFramesDiscarded"] 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"] @@ -955,6 +977,9 @@ func TestAPIProtocolListGet(t *testing.T) { 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"] + + case "hls": + out2.(map[string]any)["lastRequest"] = out1.(map[string]any)["items"].([]any)[0].(map[string]any)["lastRequest"] } 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 d05bc770..c3e6ae34 100644 --- a/internal/core/metrics_test.go +++ b/internal/core/metrics_test.go @@ -76,10 +76,15 @@ func TestMetrics(t *testing.T) { bo := httpPullFile(t, hc, "http://localhost:9998/metrics") require.Equal(t, `paths 0 +paths_inbound_bytes 0 +paths_outbound_bytes 0 +paths_inbound_frames_in_error 0 paths_bytes_received 0 paths_bytes_sent 0 paths_readers 0 hls_muxers 0 +hls_muxers_outbound_bytes 0 +hls_muxers_outbound_frames_discarded 0 hls_muxers_bytes_sent 0 rtsp_conns 0 rtsp_conns_inbound_bytes 0 @@ -97,6 +102,7 @@ rtsp_sessions_inbound_rtcp_packets_in_error 0 rtsp_sessions_outbound_bytes 0 rtsp_sessions_outbound_rtp_packets 0 rtsp_sessions_outbound_rtp_packets_reported_lost 0 +rtsp_sessions_outbound_rtp_packets_discarded 0 rtsp_sessions_outbound_rtcp_packets 0 rtsp_sessions_bytes_received 0 rtsp_sessions_bytes_sent 0 @@ -124,6 +130,7 @@ rtsps_sessions_inbound_rtcp_packets_in_error 0 rtsps_sessions_outbound_bytes 0 rtsps_sessions_outbound_rtp_packets 0 rtsps_sessions_outbound_rtp_packets_reported_lost 0 +rtsps_sessions_outbound_rtp_packets_discarded 0 rtsps_sessions_outbound_rtcp_packets 0 rtsps_sessions_bytes_received 0 rtsps_sessions_bytes_sent 0 @@ -136,9 +143,15 @@ rtsps_sessions_rtcp_packets_received 0 rtsps_sessions_rtcp_packets_sent 0 rtsps_sessions_rtcp_packets_in_error 0 rtmp_conns 0 +rtmp_conns_inbound_bytes 0 +rtmp_conns_outbound_bytes 0 +rtmp_conns_outbound_frames_discarded 0 rtmp_conns_bytes_received 0 rtmp_conns_bytes_sent 0 rtmps_conns 0 +rtmps_conns_inbound_bytes 0 +rtmps_conns_outbound_bytes 0 +rtmps_conns_outbound_frames_discarded 0 rtmps_conns_bytes_received 0 rtmps_conns_bytes_sent 0 srt_conns 0 @@ -195,6 +208,7 @@ srt_conns_packets_reorder_tolerance 0 srt_conns_packets_received_avg_belated_time 0 srt_conns_packets_send_loss_rate 0 srt_conns_packets_received_loss_rate 0 +srt_conns_outbound_frames_discarded 0 webrtc_sessions 0 webrtc_sessions_inbound_bytes 0 webrtc_sessions_inbound_rtp_packets 0 @@ -204,6 +218,7 @@ 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_outbound_frames_discarded 0 webrtc_sessions_bytes_received 0 webrtc_sessions_bytes_sent 0 webrtc_sessions_rtp_packets_received 0 @@ -395,40 +410,70 @@ webrtc_sessions_rtcp_packets_sent 0 require.Regexp(t, `^paths\{name=".*?",state="ready"\} 1`+"\n"+ + `paths_inbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_outbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_inbound_frames_in_error\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_readers\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths\{name=".*?",state="ready"\} 1`+"\n"+ + `paths_inbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_outbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_inbound_frames_in_error\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_readers\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths\{name=".*?",state="ready"\} 1`+"\n"+ + `paths_inbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_outbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_inbound_frames_in_error\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_readers\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths\{name=".*?",state="ready"\} 1`+"\n"+ + `paths_inbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_outbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_inbound_frames_in_error\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_readers\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths\{name=".*?",state="ready"\} 1`+"\n"+ + `paths_inbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_outbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_inbound_frames_in_error\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_readers\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths\{name=".*?",state="ready"\} 1`+"\n"+ + `paths_inbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_outbound_bytes\{name=".*?",state="ready"\} [0-9]+`+"\n"+ + `paths_inbound_frames_in_error\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_readers\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+ + `hls_muxers_outbound_bytes\{name=".*?"\} 0`+"\n"+ + `hls_muxers_outbound_frames_discarded\{name=".*?"\} 0`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+ + `hls_muxers_outbound_bytes\{name=".*?"\} 0`+"\n"+ + `hls_muxers_outbound_frames_discarded\{name=".*?"\} 0`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+ + `hls_muxers_outbound_bytes\{name=".*?"\} 0`+"\n"+ + `hls_muxers_outbound_frames_discarded\{name=".*?"\} 0`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+ + `hls_muxers_outbound_bytes\{name=".*?"\} 0`+"\n"+ + `hls_muxers_outbound_frames_discarded\{name=".*?"\} 0`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+ + `hls_muxers_outbound_bytes\{name=".*?"\} 0`+"\n"+ + `hls_muxers_outbound_frames_discarded\{name=".*?"\} 0`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+ + `hls_muxers_outbound_bytes\{name=".*?"\} 0`+"\n"+ + `hls_muxers_outbound_frames_discarded\{name=".*?"\} 0`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+ `rtsp_conns\{id=".*?"\} 1`+"\n"+ `rtsp_conns_inbound_bytes\{id=".*?"\} [0-9]+`+"\n"+ @@ -447,6 +492,8 @@ webrtc_sessions_rtcp_packets_sent 0 `rtsp_sessions_outbound_rtp_packets\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsp_sessions_outbound_rtp_packets_reported_lost\{id=".*?",path=".*?",`+ `remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `rtsp_sessions_outbound_rtp_packets_discarded\{id=".*?",path=".*?",`+ + `remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsp_sessions_outbound_rtcp_packets\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsp_sessions_bytes_received\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} 0`+"\n"+ `rtsp_sessions_bytes_sent\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ @@ -475,6 +522,8 @@ webrtc_sessions_rtcp_packets_sent 0 `rtsps_sessions_outbound_rtp_packets\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsps_sessions_outbound_rtp_packets_reported_lost\{id=".*?",path=".*?",`+ `remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `rtsps_sessions_outbound_rtp_packets_discarded\{id=".*?",path=".*?",`+ + `remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsps_sessions_outbound_rtcp_packets\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsps_sessions_bytes_received\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} 0`+"\n"+ `rtsps_sessions_bytes_sent\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ @@ -487,9 +536,15 @@ webrtc_sessions_rtcp_packets_sent 0 `rtsps_sessions_rtcp_packets_sent\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsps_sessions_rtcp_packets_in_error\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtmp_conns\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} 1`+"\n"+ + `rtmp_conns_inbound_bytes\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `rtmp_conns_outbound_bytes\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `rtmp_conns_outbound_frames_discarded\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtmp_conns_bytes_received\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtmp_conns_bytes_sent\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtmps_conns\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} 1`+"\n"+ + `rtmps_conns_inbound_bytes\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `rtmps_conns_outbound_bytes\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ + `rtmps_conns_outbound_frames_discarded\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtmps_conns_bytes_received\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `rtmps_conns_bytes_sent\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `srt_conns\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} 1`+"\n"+ @@ -546,6 +601,7 @@ webrtc_sessions_rtcp_packets_sent 0 `srt_conns_packets_received_avg_belated_time\{id=".*?",path=".*?",remoteAddr=".*?",state="publish"\} [0-9]+`+"\n"+ `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"+ + `srt_conns_outbound_frames_discarded\{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"+ @@ -555,6 +611,7 @@ webrtc_sessions_rtcp_packets_sent 0 `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_outbound_frames_discarded\{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"+ @@ -584,6 +641,9 @@ webrtc_sessions_rtcp_packets_sent 0 bo := httpPullFile(t, hc, "http://localhost:9998/metrics") require.Equal(t, "paths 0\n"+ + "paths_inbound_bytes 0\n"+ + "paths_outbound_bytes 0\n"+ + "paths_inbound_frames_in_error 0\n"+ "paths_bytes_received 0\n"+ "paths_bytes_sent 0\n"+ "paths_readers 0\n", diff --git a/internal/core/path.go b/internal/core/path.go index c25b25aa..ecaadc09 100644 --- a/internal/core/path.go +++ b/internal/core/path.go @@ -661,6 +661,24 @@ func (pa *path) doAPIPathsGet(req pathAPIPathsGetReq) { } return defs.MediasToCodecs(pa.stream.Desc.Medias) }(), + InboundBytes: func() uint64 { + if !pa.isAvailable() { + return 0 + } + return pa.stream.InboundBytes() + }(), + OutboundBytes: func() uint64 { + if !pa.isAvailable() { + return 0 + } + return pa.stream.OutboundBytes() + }(), + InboundFramesInError: func() uint64 { + if !pa.isAvailable() { + return 0 + } + return pa.stream.InboundFramesInError() + }(), BytesReceived: func() uint64 { if !pa.isAvailable() { return 0 diff --git a/internal/counterdumper/dumper.go b/internal/counterdumper/dumper.go index eb58ca6c..21093446 100644 --- a/internal/counterdumper/dumper.go +++ b/internal/counterdumper/dumper.go @@ -2,7 +2,7 @@ package counterdumper import ( - "sync/atomic" + "sync" "time" ) @@ -14,7 +14,9 @@ const ( type Dumper struct { OnReport func(v uint64) - counter *uint64 + mutex sync.Mutex + counter uint64 + absCounter uint64 terminate chan struct{} done chan struct{} @@ -22,7 +24,6 @@ type Dumper struct { // Start starts the counter. func (c *Dumper) Start() { - c.counter = new(uint64) c.terminate = make(chan struct{}) c.done = make(chan struct{}) @@ -37,12 +38,25 @@ func (c *Dumper) Stop() { // Increase increases the counter value by 1. func (c *Dumper) Increase() { - atomic.AddUint64(c.counter, 1) + c.mutex.Lock() + defer c.mutex.Unlock() + c.counter++ + c.absCounter++ } // Add adds value to the counter. func (c *Dumper) Add(v uint64) { - atomic.AddUint64(c.counter, v) + c.mutex.Lock() + defer c.mutex.Unlock() + c.counter += v + c.absCounter += v +} + +// Get returns the counter value. +func (c *Dumper) Get() uint64 { + c.mutex.Lock() + defer c.mutex.Unlock() + return c.absCounter } func (c *Dumper) run() { @@ -57,7 +71,11 @@ func (c *Dumper) run() { return case <-t.C: - v := atomic.SwapUint64(c.counter, 0) + c.mutex.Lock() + var v uint64 + v, c.counter = c.counter, 0 + c.mutex.Unlock() + if v != 0 { c.OnReport(v) } diff --git a/internal/defs/api_hls.go b/internal/defs/api_hls.go index 0b3f0806..be12489a 100644 --- a/internal/defs/api_hls.go +++ b/internal/defs/api_hls.go @@ -10,10 +10,13 @@ type APIHLSServer interface { // APIHLSMuxer is an HLS muxer. type APIHLSMuxer struct { - Path string `json:"path"` - Created time.Time `json:"created"` - LastRequest time.Time `json:"lastRequest"` - BytesSent uint64 `json:"bytesSent"` + Path string `json:"path"` + Created time.Time `json:"created"` + LastRequest time.Time `json:"lastRequest"` + OutboundBytes uint64 `json:"outboundBytes"` + OutboundFramesDiscarded uint64 `json:"outboundFramesDiscarded"` + // deprecated + BytesSent uint64 `json:"bytesSent" deprecated:"true"` } // APIHLSMuxerList is a list of HLS muxers. diff --git a/internal/defs/api_path.go b/internal/defs/api_path.go index 096f440b..36733231 100644 --- a/internal/defs/api_path.go +++ b/internal/defs/api_path.go @@ -63,19 +63,23 @@ type APIPathReader struct { // APIPath is a path. type APIPath struct { - Name string `json:"name"` - ConfName string `json:"confName"` - Ready bool `json:"ready" deprecated:"true"` - ReadyTime *time.Time `json:"readyTime" deprecated:"true"` - Available bool `json:"available"` - AvailableTime *time.Time `json:"availableTime"` - Online bool `json:"online"` - OnlineTime *time.Time `json:"onlineTime"` - Source *APIPathSource `json:"source"` - Tracks []APIPathTrackCodec `json:"tracks"` - BytesReceived uint64 `json:"bytesReceived"` - BytesSent uint64 `json:"bytesSent"` - Readers []APIPathReader `json:"readers"` + Name string `json:"name"` + ConfName string `json:"confName"` + Ready bool `json:"ready" deprecated:"true"` + ReadyTime *time.Time `json:"readyTime" deprecated:"true"` + Available bool `json:"available"` + AvailableTime *time.Time `json:"availableTime"` + Online bool `json:"online"` + OnlineTime *time.Time `json:"onlineTime"` + Source *APIPathSource `json:"source"` + Tracks []APIPathTrackCodec `json:"tracks"` + Readers []APIPathReader `json:"readers"` + InboundBytes uint64 `json:"inboundBytes"` + OutboundBytes uint64 `json:"outboundBytes"` + InboundFramesInError uint64 `json:"inboundFramesInError"` + // deprecated + BytesReceived uint64 `json:"bytesReceived" deprecated:"true"` + BytesSent uint64 `json:"bytesSent" deprecated:"true"` } // APIPathList is a list of paths. diff --git a/internal/defs/api_rtmp.go b/internal/defs/api_rtmp.go index 242f7b37..07fa0a2a 100644 --- a/internal/defs/api_rtmp.go +++ b/internal/defs/api_rtmp.go @@ -25,15 +25,19 @@ const ( // APIRTMPConn is a RTMP connection. type APIRTMPConn struct { - ID uuid.UUID `json:"id"` - Created time.Time `json:"created"` - RemoteAddr string `json:"remoteAddr"` - State APIRTMPConnState `json:"state"` - Path string `json:"path"` - Query string `json:"query"` - User string `json:"user"` - BytesReceived uint64 `json:"bytesReceived"` - BytesSent uint64 `json:"bytesSent"` + ID uuid.UUID `json:"id"` + Created time.Time `json:"created"` + RemoteAddr string `json:"remoteAddr"` + State APIRTMPConnState `json:"state"` + Path string `json:"path"` + Query string `json:"query"` + User string `json:"user"` + InboundBytes uint64 `json:"inboundBytes"` + OutboundBytes uint64 `json:"outboundBytes"` + OutboundFramesDiscarded uint64 `json:"outboundFramesDiscarded"` + // deprecated + BytesReceived uint64 `json:"bytesReceived" deprecated:"true"` + BytesSent uint64 `json:"bytesSent" deprecated:"true"` } // APIRTMPConnList is a list of RTMP connections. diff --git a/internal/defs/api_rtsp.go b/internal/defs/api_rtsp.go index acfb8c14..2e8521ea 100644 --- a/internal/defs/api_rtsp.go +++ b/internal/defs/api_rtsp.go @@ -67,6 +67,7 @@ type APIRTSPSession struct { OutboundBytes uint64 `json:"outboundBytes"` OutboundRTPPackets uint64 `json:"outboundRTPPackets"` OutboundRTPPacketsReportedLost uint64 `json:"outboundRTPPacketsReportedLost"` + OutboundRTPPacketsDiscarded uint64 `json:"outboundRTPPacketsDiscarded"` OutboundRTCPPackets uint64 `json:"outboundRTCPPackets"` // deprecated BytesReceived uint64 `json:"bytesReceived" deprecated:"true"` diff --git a/internal/defs/api_srt.go b/internal/defs/api_srt.go index f6f094d1..8531eaf6 100644 --- a/internal/defs/api_srt.go +++ b/internal/defs/api_srt.go @@ -153,6 +153,8 @@ type APISRTConn struct { PacketsSendLossRate float64 `json:"packetsSendLossRate"` // Percentage of retransmitted data vs. received data PacketsReceivedLossRate float64 `json:"packetsReceivedLossRate"` + + OutboundFramesDiscarded uint64 `json:"outboundFramesDiscarded"` } // APISRTConnList is a list of SRT connections. diff --git a/internal/defs/api_webrtc.go b/internal/defs/api_webrtc.go index e0f145a5..23f61099 100644 --- a/internal/defs/api_webrtc.go +++ b/internal/defs/api_webrtc.go @@ -42,6 +42,7 @@ type APIWebRTCSession struct { OutboundBytes uint64 `json:"outboundBytes"` OutboundRTPPackets uint64 `json:"outboundRTPPackets"` OutboundRTCPPackets uint64 `json:"outboundRTCPPackets"` + OutboundFramesDiscarded uint64 `json:"outboundFramesDiscarded"` // deprecated BytesReceived uint64 `json:"bytesReceived" deprecated:"true"` BytesSent uint64 `json:"bytesSent" deprecated:"true"` diff --git a/internal/errordumper/dumper.go b/internal/errordumper/dumper.go index 0075154a..44ac3c7e 100644 --- a/internal/errordumper/dumper.go +++ b/internal/errordumper/dumper.go @@ -14,9 +14,10 @@ const ( type Dumper struct { OnReport func(v uint64, last error) - mutex sync.Mutex - counter uint64 - last error + mutex sync.Mutex + counter uint64 + absCounter uint64 + last error terminate chan struct{} done chan struct{} @@ -41,9 +42,17 @@ func (c *Dumper) Add(err error) { c.mutex.Lock() defer c.mutex.Unlock() c.counter++ + c.absCounter++ c.last = err } +// Get returns the total number of errors. +func (c *Dumper) Get() uint64 { + c.mutex.Lock() + defer c.mutex.Unlock() + return c.absCounter +} + func (c *Dumper) run() { defer close(c.done) diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go index 30f5da3b..3521cfc3 100644 --- a/internal/metrics/metrics.go +++ b/internal/metrics/metrics.go @@ -223,6 +223,10 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { "state": state, }) out += metric("paths", ta, 1) + out += metric("paths_inbound_bytes", ta, int64(i.InboundBytes)) + out += metric("paths_outbound_bytes", ta, int64(i.OutboundBytes)) + out += metric("paths_inbound_frames_in_error", ta, int64(i.InboundFramesInError)) + // deprecated out += metric("paths_bytes_received", ta, int64(i.BytesReceived)) out += metric("paths_bytes_sent", ta, int64(i.BytesSent)) out += metric("paths_readers", ta, int64(len(i.Readers))) @@ -230,6 +234,10 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { } } else if pathFilter == "" { out += metric("paths", "", 0) + out += metric("paths_inbound_bytes", "", 0) + out += metric("paths_outbound_bytes", "", 0) + out += metric("paths_inbound_frames_in_error", "", 0) + // deprecated out += metric("paths_bytes_received", "", 0) out += metric("paths_bytes_sent", "", 0) out += metric("paths_readers", "", 0) @@ -248,11 +256,17 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { "name": i.Path, }) out += metric("hls_muxers", ta, 1) + out += metric("hls_muxers_outbound_bytes", ta, int64(i.OutboundBytes)) + out += metric("hls_muxers_outbound_frames_discarded", ta, int64(i.OutboundFramesDiscarded)) + // deprecated out += metric("hls_muxers_bytes_sent", ta, int64(i.BytesSent)) } } } else if hlsMuxerFilter == "" { out += metric("hls_muxers", "", 0) + out += metric("hls_muxers_outbound_bytes", "", 0) + out += metric("hls_muxers_outbound_frames_discarded", "", 0) + // deprecated out += metric("hls_muxers_bytes_sent", "", 0) } } @@ -308,6 +322,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { out += metric("rtsp_sessions_outbound_bytes", ta, int64(i.OutboundBytes)) out += metric("rtsp_sessions_outbound_rtp_packets", ta, int64(i.OutboundRTPPackets)) out += metric("rtsp_sessions_outbound_rtp_packets_reported_lost", ta, int64(i.OutboundRTPPacketsReportedLost)) + out += metric("rtsp_sessions_outbound_rtp_packets_discarded", ta, int64(i.OutboundRTPPacketsDiscarded)) out += metric("rtsp_sessions_outbound_rtcp_packets", ta, int64(i.OutboundRTCPPackets)) // deprecated out += metric("rtsp_sessions_bytes_received", ta, int64(i.BytesReceived)) @@ -334,6 +349,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { out += metric("rtsp_sessions_outbound_bytes", "", 0) out += metric("rtsp_sessions_outbound_rtp_packets", "", 0) out += metric("rtsp_sessions_outbound_rtp_packets_reported_lost", "", 0) + out += metric("rtsp_sessions_outbound_rtp_packets_discarded", "", 0) out += metric("rtsp_sessions_outbound_rtcp_packets", "", 0) // deprecated out += metric("rtsp_sessions_bytes_received", "", 0) @@ -401,6 +417,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { out += metric("rtsps_sessions_outbound_bytes", ta, int64(i.OutboundBytes)) out += metric("rtsps_sessions_outbound_rtp_packets", ta, int64(i.OutboundRTPPackets)) out += metric("rtsps_sessions_outbound_rtp_packets_reported_lost", ta, int64(i.OutboundRTPPacketsReportedLost)) + out += metric("rtsps_sessions_outbound_rtp_packets_discarded", ta, int64(i.OutboundRTPPacketsDiscarded)) out += metric("rtsps_sessions_outbound_rtcp_packets", ta, int64(i.OutboundRTCPPackets)) // deprecated out += metric("rtsps_sessions_bytes_received", ta, int64(i.BytesReceived)) @@ -427,6 +444,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { out += metric("rtsps_sessions_outbound_bytes", "", 0) out += metric("rtsps_sessions_outbound_rtp_packets", "", 0) out += metric("rtsps_sessions_outbound_rtp_packets_reported_lost", "", 0) + out += metric("rtsps_sessions_outbound_rtp_packets_discarded", "", 0) out += metric("rtsps_sessions_outbound_rtcp_packets", "", 0) // deprecated out += metric("rtsps_sessions_bytes_received", "", 0) @@ -458,12 +476,20 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { "remoteAddr": i.RemoteAddr, }) out += metric("rtmp_conns", ta, 1) + out += metric("rtmp_conns_inbound_bytes", ta, int64(i.InboundBytes)) + out += metric("rtmp_conns_outbound_bytes", ta, int64(i.OutboundBytes)) + out += metric("rtmp_conns_outbound_frames_discarded", ta, int64(i.OutboundFramesDiscarded)) + // deprecated out += metric("rtmp_conns_bytes_received", ta, int64(i.BytesReceived)) out += metric("rtmp_conns_bytes_sent", ta, int64(i.BytesSent)) } } } else if rtmpConnFilter == "" { out += metric("rtmp_conns", "", 0) + out += metric("rtmp_conns_inbound_bytes", "", 0) + out += metric("rtmp_conns_outbound_bytes", "", 0) + out += metric("rtmp_conns_outbound_frames_discarded", "", 0) + // deprecated out += metric("rtmp_conns_bytes_received", "", 0) out += metric("rtmp_conns_bytes_sent", "", 0) } @@ -484,12 +510,20 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { "remoteAddr": i.RemoteAddr, }) out += metric("rtmps_conns", ta, 1) + out += metric("rtmps_conns_inbound_bytes", ta, int64(i.InboundBytes)) + out += metric("rtmps_conns_outbound_bytes", ta, int64(i.OutboundBytes)) + out += metric("rtmps_conns_outbound_frames_discarded", ta, int64(i.OutboundFramesDiscarded)) + // deprecated out += metric("rtmps_conns_bytes_received", ta, int64(i.BytesReceived)) out += metric("rtmps_conns_bytes_sent", ta, int64(i.BytesSent)) } } } else if rtmpsConnFilter == "" { out += metric("rtmps_conns", "", 0) + out += metric("rtmps_conns_inbound_bytes", "", 0) + out += metric("rtmps_conns_outbound_bytes", "", 0) + out += metric("rtmps_conns_outbound_frames_discarded", "", 0) + // deprecated out += metric("rtmps_conns_bytes_received", "", 0) out += metric("rtmps_conns_bytes_sent", "", 0) } @@ -563,6 +597,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { out += metric("srt_conns_packets_received_avg_belated_time", ta, int64(i.PacketsReceivedAvgBelatedTime)) out += metricFloat("srt_conns_packets_send_loss_rate", ta, i.PacketsSendLossRate) out += metricFloat("srt_conns_packets_received_loss_rate", ta, i.PacketsReceivedLossRate) + out += metric("srt_conns_outbound_frames_discarded", ta, int64(i.OutboundFramesDiscarded)) } } } else if srtConnFilter == "" { @@ -620,6 +655,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { out += metric("srt_conns_packets_received_avg_belated_time", "", 0) out += metricFloat("srt_conns_packets_send_loss_rate", "", 0) out += metricFloat("srt_conns_packets_received_loss_rate", "", 0) + out += metric("srt_conns_outbound_frames_discarded", "", 0) } } @@ -646,6 +682,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { 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)) + out += metric("webrtc_sessions_outbound_frames_discarded", ta, int64(i.OutboundFramesDiscarded)) // deprecated out += metric("webrtc_sessions_bytes_received", ta, int64(i.BytesReceived)) out += metric("webrtc_sessions_bytes_sent", ta, int64(i.BytesSent)) @@ -667,6 +704,7 @@ func (m *Metrics) onMetrics(ctx *gin.Context) { out += metric("webrtc_sessions_outbound_bytes", "", 0) out += metric("webrtc_sessions_outbound_rtp_packets", "", 0) out += metric("webrtc_sessions_outbound_rtcp_packets", "", 0) + out += metric("webrtc_sessions_outbound_frames_discarded", "", 0) // deprecated out += metric("webrtc_sessions_bytes_received", "", 0) out += metric("webrtc_sessions_bytes_sent", "", 0) diff --git a/internal/metrics/metrics_test.go b/internal/metrics/metrics_test.go index cb5ba044..a00ac47f 100644 --- a/internal/metrics/metrics_test.go +++ b/internal/metrics/metrics_test.go @@ -4,6 +4,7 @@ import ( "fmt" "io" "net/http" + "strings" "testing" "time" @@ -22,6 +23,18 @@ func ptrOf[T any](v T) *T { return p } +func requireMetricsLines(t *testing.T, body []byte, lines []string) { + t.Helper() + + got := string(body) + for _, line := range lines { + require.Contains(t, got, line+"\n") + } + + nonEmptyLines := strings.Count(strings.TrimSpace(got), "\n") + 1 + require.GreaterOrEqual(t, nonEmptyLines, len(lines)) +} + type dummyPathManager struct{} func (dummyPathManager) APIPathsList() (*defs.APIPathList, error) { @@ -35,11 +48,14 @@ func (dummyPathManager) APIPathsList() (*defs.APIPathList, error) { Type: defs.APIPathSourceTypeRTSPSession, ID: "123324354", }, - Ready: true, - ReadyTime: ptrOf(time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC)), - Tracks: []defs.APIPathTrackCodec{defs.APIPathTrackCodecH264, defs.APIPathTrackCodecH265}, - BytesReceived: 123, - BytesSent: 456, + Ready: true, + ReadyTime: ptrOf(time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC)), + Tracks: []defs.APIPathTrackCodec{defs.APIPathTrackCodecH264, defs.APIPathTrackCodecH265}, + InboundBytes: 123, + OutboundBytes: 456, + InboundFramesInError: 7, + BytesReceived: 123, + BytesSent: 456, Readers: []defs.APIPathReader{ { Type: defs.APIPathReaderTypeRTSPSession, @@ -61,10 +77,12 @@ func (dummyHLSServer) APIMuxersList() (*defs.APIHLSMuxerList, error) { ItemCount: 1, PageCount: 1, Items: []defs.APIHLSMuxer{{ - Path: "mypath", - Created: time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC), - LastRequest: time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC), - BytesSent: 789, + Path: "mypath", + Created: time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC), + LastRequest: time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC), + OutboundBytes: 789, + OutboundFramesDiscarded: 12, + BytesSent: 789, }}, }, nil } @@ -118,6 +136,7 @@ func (dummyRTSPServer) APISessionsList() (*defs.APIRTSPSessionList, error) { OutboundBytes: 456, OutboundRTPPackets: 123, OutboundRTPPacketsReportedLost: 321, + OutboundRTPPacketsDiscarded: 111, OutboundRTCPPackets: 789, BytesReceived: 123, BytesSent: 456, @@ -148,14 +167,17 @@ func (dummyRTMPServer) APIConnsList() (*defs.APIRTMPConnList, error) { ItemCount: 1, PageCount: 1, Items: []defs.APIRTMPConn{{ - ID: uuid.MustParse("9a07afe4-fc07-4c9b-be6e-6255720c36d0"), - Created: time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC), - RemoteAddr: "3.3.3.3:5678", - State: defs.APIRTMPConnStateRead, - Path: "mypath", - Query: "myquery", - BytesReceived: 123, - BytesSent: 456, + ID: uuid.MustParse("9a07afe4-fc07-4c9b-be6e-6255720c36d0"), + Created: time.Date(2003, 11, 4, 23, 15, 7, 0, time.UTC), + RemoteAddr: "3.3.3.3:5678", + State: defs.APIRTMPConnStateRead, + Path: "mypath", + Query: "myquery", + InboundBytes: 123, + OutboundBytes: 456, + OutboundFramesDiscarded: 12, + BytesReceived: 123, + BytesSent: 456, }}, }, nil } @@ -192,6 +214,7 @@ func (dummyWebRTCServer) APISessionsList() (*defs.APIWebRTCSessionList, error) { OutboundBytes: 456, OutboundRTPPackets: 123, OutboundRTCPPackets: 456, + OutboundFramesDiscarded: 12, BytesReceived: 123, BytesSent: 456, RTPPacketsReceived: 789, @@ -294,158 +317,35 @@ func TestMetrics(t *testing.T) { byts, err := io.ReadAll(res.Body) require.NoError(t, err) - require.Equal(t, - `paths{name="mypath",state="ready"} 1`+"\n"+ - `paths_bytes_received{name="mypath",state="ready"} 123`+"\n"+ - `paths_bytes_sent{name="mypath",state="ready"} 456`+"\n"+ - `paths_readers{name="mypath",state="ready"} 1`+"\n"+ - `hls_muxers{name="mypath"} 1`+"\n"+ - `hls_muxers_bytes_sent{name="mypath"} 789`+"\n"+ - `rtsp_conns{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 1`+"\n"+ - `rtsp_conns_inbound_bytes{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 123`+"\n"+ - `rtsp_conns_outbound_bytes{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 456`+"\n"+ - `rtsp_conns_bytes_received{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 123`+"\n"+ - `rtsp_conns_bytes_sent{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 456`+"\n"+ - `rtsp_sessions{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 1`+"\n"+ - `rtsp_sessions_inbound_bytes{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsp_sessions_inbound_rtp_packets{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsp_sessions_inbound_rtp_packets_lost{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsp_sessions_inbound_rtp_packets_in_error{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsp_sessions_inbound_rtp_packets_jitter{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsp_sessions_inbound_rtcp_packets{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsp_sessions_inbound_rtcp_packets_in_error{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsp_sessions_outbound_bytes{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsp_sessions_outbound_rtp_packets{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsp_sessions_outbound_rtp_packets_reported_lost{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 321`+"\n"+ - `rtsp_sessions_outbound_rtcp_packets{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsp_sessions_bytes_received{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsp_sessions_bytes_sent{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsp_sessions_rtp_packets_received{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsp_sessions_rtp_packets_sent{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsp_sessions_rtp_packets_lost{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsp_sessions_rtp_packets_in_error{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsp_sessions_rtp_packets_jitter{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsp_sessions_rtcp_packets_received{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsp_sessions_rtcp_packets_sent{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsp_sessions_rtcp_packets_in_error{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsps_conns{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 1`+"\n"+ - `rtsps_conns_inbound_bytes{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 123`+"\n"+ - `rtsps_conns_outbound_bytes{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 456`+"\n"+ - `rtsps_conns_bytes_received{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 123`+"\n"+ - `rtsps_conns_bytes_sent{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 456`+"\n"+ - `rtsps_sessions{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 1`+"\n"+ - `rtsps_sessions_inbound_bytes{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsps_sessions_inbound_rtp_packets{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsps_sessions_inbound_rtp_packets_lost{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsps_sessions_inbound_rtp_packets_in_error{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsps_sessions_inbound_rtp_packets_jitter{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsps_sessions_inbound_rtcp_packets{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsps_sessions_inbound_rtcp_packets_in_error{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsps_sessions_outbound_bytes{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsps_sessions_outbound_rtp_packets{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsps_sessions_outbound_rtp_packets_reported_lost{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 321`+"\n"+ - `rtsps_sessions_outbound_rtcp_packets{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsps_sessions_bytes_received{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsps_sessions_bytes_sent{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsps_sessions_rtp_packets_received{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsps_sessions_rtp_packets_sent{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsps_sessions_rtp_packets_lost{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsps_sessions_rtp_packets_in_error{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsps_sessions_rtp_packets_jitter{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ - `rtsps_sessions_rtcp_packets_received{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtsps_sessions_rtcp_packets_sent{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ - `rtsps_sessions_rtcp_packets_in_error{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ - `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 456`+"\n"+ - `rtmp_conns{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",`+ - `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 1`+"\n"+ - `rtmp_conns_bytes_received{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",`+ - `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 123`+"\n"+ - `rtmp_conns_bytes_sent{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",`+ - `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 456`+"\n"+ - `rtmps_conns{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",`+ - `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 1`+"\n"+ - `rtmps_conns_bytes_received{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",`+ - `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 123`+"\n"+ - `rtmps_conns_bytes_sent{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",`+ - `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",`+ - `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 456`+"\n"+ - `webrtc_sessions_rtp_packets_received{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ - `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 789`+"\n"+ - `webrtc_sessions_rtp_packets_sent{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ - `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 123`+"\n"+ - `webrtc_sessions_rtp_packets_lost{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ - `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 456`+"\n"+ - `webrtc_sessions_rtp_packets_jitter{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ - `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 789`+"\n"+ - `webrtc_sessions_rtcp_packets_received{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ - `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 123`+"\n"+ - `webrtc_sessions_rtcp_packets_sent{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",`+ - `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 456`+"\n", - string(byts)) + requireMetricsLines(t, byts, []string{ + `paths{name="mypath",state="ready"} 1`, + `paths_inbound_bytes{name="mypath",state="ready"} 123`, + `paths_outbound_bytes{name="mypath",state="ready"} 456`, + `paths_inbound_frames_in_error{name="mypath",state="ready"} 7`, + `paths_bytes_received{name="mypath",state="ready"} 123`, + `paths_bytes_sent{name="mypath",state="ready"} 456`, + `paths_readers{name="mypath",state="ready"} 1`, + `hls_muxers{name="mypath"} 1`, + `hls_muxers_outbound_bytes{name="mypath"} 789`, + `hls_muxers_outbound_frames_discarded{name="mypath"} 12`, + `rtsp_conns_inbound_bytes{id="18294761-f9d1-4ea9-9a35-fe265b62eb41"} 123`, + `rtsp_sessions_outbound_rtp_packets_discarded{id="124b22ce-9c34-4387-b045-44caf98049f7",` + + `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 111`, + `rtsps_sessions_outbound_rtp_packets_discarded{id="124b22ce-9c34-4387-b045-44caf98049f7",` + + `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 111`, + `rtmp_conns_inbound_bytes{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",` + + `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 123`, + `rtmp_conns_outbound_bytes{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",` + + `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 456`, + `rtmp_conns_outbound_frames_discarded{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",` + + `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 12`, + `rtmp_conns_bytes_received{id="9a07afe4-fc07-4c9b-be6e-6255720c36d0",` + + `path="mypath",remoteAddr="3.3.3.3:5678",state="read"} 123`, + `webrtc_sessions_outbound_bytes{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",` + + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 456`, + `webrtc_sessions_outbound_frames_discarded{id="f47ac10b-58cc-4372-a567-0e02b2c3d479",` + + `path="mypath",remoteAddr="127.0.0.1:3455",state="read"} 12`, + }) require.True(t, checked) } @@ -558,6 +458,9 @@ func TestFilter(t *testing.T) { case "path": require.Equal(t, `paths{name="mypath",state="ready"} 1`+"\n"+ + `paths_inbound_bytes{name="mypath",state="ready"} 123`+"\n"+ + `paths_outbound_bytes{name="mypath",state="ready"} 456`+"\n"+ + `paths_inbound_frames_in_error{name="mypath",state="ready"} 7`+"\n"+ `paths_bytes_received{name="mypath",state="ready"} 123`+"\n"+ `paths_bytes_sent{name="mypath",state="ready"} 456`+"\n"+ `paths_readers{name="mypath",state="ready"} 1`+"\n", @@ -566,6 +469,8 @@ func TestFilter(t *testing.T) { case "hls_muxer": require.Equal(t, `hls_muxers{name="mypath"} 1`+"\n"+ + `hls_muxers_outbound_bytes{name="mypath"} 789`+"\n"+ + `hls_muxers_outbound_frames_discarded{name="mypath"} 12`+"\n"+ `hls_muxers_bytes_sent{name="mypath"} 789`+"\n", string(byts)) @@ -602,6 +507,8 @@ func TestFilter(t *testing.T) { `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 123`+"\n"+ `rtsp_sessions_outbound_rtp_packets_reported_lost{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 321`+"\n"+ + `rtsp_sessions_outbound_rtp_packets_discarded{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ + `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 111`+"\n"+ `rtsp_sessions_outbound_rtcp_packets{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ `path="mypath",remoteAddr="124.5.5.5:34542",state="publish"} 789`+"\n"+ `rtsp_sessions_bytes_received{id="124b22ce-9c34-4387-b045-44caf98049f7",`+ diff --git a/internal/servers/hls/http_server.go b/internal/servers/hls/http_server.go index 86204976..863e9cea 100644 --- a/internal/servers/hls/http_server.go +++ b/internal/servers/hls/http_server.go @@ -203,13 +203,7 @@ func (s *httpServer) onRequest(ctx *gin.Context) { return } - mi := mux.getInstance() - if mi == nil { - ctx.Writer.WriteHeader(http.StatusNotFound) - return - } - ctx.Request.URL.Path = fname - mi.handleRequest(ctx) + mux.handleRequest(ctx) } } diff --git a/internal/servers/hls/muxer.go b/internal/servers/hls/muxer.go index 752368ad..d173ab79 100644 --- a/internal/servers/hls/muxer.go +++ b/internal/servers/hls/muxer.go @@ -13,6 +13,8 @@ import ( "github.com/bluenviron/mediamtx/internal/defs" "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/hls" + "github.com/bluenviron/mediamtx/internal/stream" + "github.com/gin-gonic/gin" ) const ( @@ -43,8 +45,13 @@ func (w *responseWriterWithCounter) Write(p []byte) (int, error) { return n, err } +type muxerGetInstanceRes struct { + instance *muxerInstance + cumulatedOutboundFramesDiscarded uint64 +} + type muxerGetInstanceReq struct { - res chan *muxerInstance + res chan muxerGetInstanceRes } type muxer struct { @@ -138,22 +145,7 @@ func (m *muxer) runInner() error { defer m.path.RemoveReader(defs.PathRemoveReaderReq{Author: m}) - var instanceError chan error - var recreateTimer *time.Timer - - mi := &muxerInstance{ - variant: m.variant, - segmentCount: m.segmentCount, - segmentDuration: m.segmentDuration, - partDuration: m.partDuration, - segmentMaxSize: m.segmentMaxSize, - directory: m.directory, - pathName: m.pathName, - stream: res.Stream, - bytesSent: m.bytesSent, - parent: m, - } - err = mi.initialize() + mi, err := m.createInstance(res.Stream) if err != nil { if m.remoteAddr != "" || errors.Is(err, hls.ErrNoSupportedCodecs) { return err @@ -161,11 +153,6 @@ func (m *muxer) runInner() error { m.Log(logger.Error, err.Error()) mi = nil - instanceError = make(chan error) - recreateTimer = time.NewTimer(recreatePause) - } else { - instanceError = mi.errorChan() - recreateTimer = emptyTimer() } defer func() { @@ -174,6 +161,17 @@ func (m *muxer) runInner() error { } }() + var instanceError chan error + var recreateTimer *time.Timer + + if mi != nil { + instanceError = mi.errorChan() + recreateTimer = emptyTimer() + } else { + instanceError = make(chan error) + recreateTimer = time.NewTimer(recreatePause) + } + var activityCheckTimer *time.Timer if m.remoteAddr != "" { activityCheckTimer = time.NewTimer(closeCheckPeriod) @@ -181,10 +179,15 @@ func (m *muxer) runInner() error { activityCheckTimer = emptyTimer() } + cumulatedOutboundFramesDiscarded := uint64(0) + for { select { case req := <-m.chGetInstance: - req.res <- mi + req.res <- muxerGetInstanceRes{ + instance: mi, + cumulatedOutboundFramesDiscarded: cumulatedOutboundFramesDiscarded, + } case err = <-instanceError: if m.remoteAddr != "" { @@ -193,24 +196,13 @@ func (m *muxer) runInner() error { m.Log(logger.Error, err.Error()) mi.close() + cumulatedOutboundFramesDiscarded += mi.reader.OutboundFramesDiscarded() mi = nil instanceError = make(chan error) recreateTimer = time.NewTimer(recreatePause) case <-recreateTimer.C: - mi = &muxerInstance{ - variant: m.variant, - segmentCount: m.segmentCount, - segmentDuration: m.segmentDuration, - partDuration: m.partDuration, - segmentMaxSize: m.segmentMaxSize, - directory: m.directory, - pathName: m.pathName, - stream: res.Stream, - bytesSent: m.bytesSent, - parent: m, - } - err = mi.initialize() + mi, err = m.createInstance(res.Stream) if err != nil { m.Log(logger.Error, err.Error()) mi = nil @@ -232,17 +224,31 @@ func (m *muxer) runInner() error { } } -func (m *muxer) getInstance() *muxerInstance { - atomic.StoreInt64(m.lastRequestTime, time.Now().UnixNano()) +func (m *muxer) createInstance(strm *stream.Stream) (*muxerInstance, error) { + mi := &muxerInstance{ + variant: m.variant, + segmentCount: m.segmentCount, + segmentDuration: m.segmentDuration, + partDuration: m.partDuration, + segmentMaxSize: m.segmentMaxSize, + directory: m.directory, + pathName: m.pathName, + stream: strm, + parent: m, + } + err := mi.initialize() + return mi, err +} - req := muxerGetInstanceReq{res: make(chan *muxerInstance)} +func (m *muxer) getInstance() muxerGetInstanceRes { + req := muxerGetInstanceReq{res: make(chan muxerGetInstanceRes)} select { case m.chGetInstance <- req: return <-req.res case <-m.ctx.Done(): - return nil + return muxerGetInstanceRes{} } } @@ -254,11 +260,37 @@ func (m *muxer) APIReaderDescribe() *defs.APIPathReader { } } +func (m *muxer) handleRequest(ctx *gin.Context) { + atomic.StoreInt64(m.lastRequestTime, time.Now().UnixNano()) + + res := m.getInstance() + if res.instance == nil { + ctx.Writer.WriteHeader(http.StatusNotFound) + return + } + + w := &responseWriterWithCounter{ + ResponseWriter: ctx.Writer, + bytesSent: m.bytesSent, + } + + res.instance.handleRequest(w, ctx.Request) +} + func (m *muxer) apiItem() *defs.APIHLSMuxer { + res := m.getInstance() + + outboundFramesDiscarded := res.cumulatedOutboundFramesDiscarded + if res.instance != nil { + outboundFramesDiscarded += res.instance.reader.OutboundFramesDiscarded() + } + return &defs.APIHLSMuxer{ - Path: m.pathName, - Created: m.created, - LastRequest: time.Unix(0, atomic.LoadInt64(m.lastRequestTime)), - BytesSent: atomic.LoadUint64(m.bytesSent), + Path: m.pathName, + Created: m.created, + LastRequest: time.Unix(0, atomic.LoadInt64(m.lastRequestTime)), + OutboundBytes: atomic.LoadUint64(m.bytesSent), + OutboundFramesDiscarded: outboundFramesDiscarded, + BytesSent: atomic.LoadUint64(m.bytesSent), } } diff --git a/internal/servers/hls/muxer_instance.go b/internal/servers/hls/muxer_instance.go index 2e7a0e76..9af7d5c1 100644 --- a/internal/servers/hls/muxer_instance.go +++ b/internal/servers/hls/muxer_instance.go @@ -1,6 +1,7 @@ package hls import ( + "net/http" "os" "path/filepath" "time" @@ -11,7 +12,6 @@ import ( "github.com/bluenviron/mediamtx/internal/logger" "github.com/bluenviron/mediamtx/internal/protocols/hls" "github.com/bluenviron/mediamtx/internal/stream" - "github.com/gin-gonic/gin" ) type muxerInstance struct { @@ -23,7 +23,6 @@ type muxerInstance struct { directory string pathName string stream *stream.Stream - bytesSent *uint64 parent logger.Writer hmuxer *gohlslib.Muxer @@ -89,11 +88,6 @@ func (mi *muxerInstance) errorChan() chan error { return mi.reader.Error() } -func (mi *muxerInstance) handleRequest(ctx *gin.Context) { - w := &responseWriterWithCounter{ - ResponseWriter: ctx.Writer, - bytesSent: mi.bytesSent, - } - - mi.hmuxer.Handle(w, ctx.Request) +func (mi *muxerInstance) handleRequest(w http.ResponseWriter, r *http.Request) { + mi.hmuxer.Handle(w, r) } diff --git a/internal/servers/rtmp/conn.go b/internal/servers/rtmp/conn.go index d6276671..6fcd4e2a 100644 --- a/internal/servers/rtmp/conn.go +++ b/internal/servers/rtmp/conn.go @@ -48,6 +48,7 @@ type conn struct { pathName string query string user string + reader *stream.Reader } func (c *conn) initialize() { @@ -210,6 +211,10 @@ func (c *conn) runRead() error { res.Stream.AddReader(r) defer res.Stream.RemoveReader(r) + c.mutex.Lock() + c.reader = r + c.mutex.Unlock() + select { case <-c.ctx.Done(): return fmt.Errorf("terminated") @@ -320,21 +325,29 @@ func (c *conn) apiItem() *defs.APIRTMPConn { bytesReceived := uint64(0) bytesSent := uint64(0) + outboundFramesDiscarded := uint64(0) if c.rconn != nil { bytesReceived = c.rconn.BytesReceived() bytesSent = c.rconn.BytesSent() } + if c.reader != nil { + outboundFramesDiscarded = c.reader.OutboundFramesDiscarded() + } + return &defs.APIRTMPConn{ - ID: c.uuid, - Created: c.created, - RemoteAddr: c.remoteAddr().String(), - State: c.state, - Path: c.pathName, - Query: c.query, - User: c.user, - BytesReceived: bytesReceived, - BytesSent: bytesSent, + ID: c.uuid, + Created: c.created, + RemoteAddr: c.remoteAddr().String(), + State: c.state, + Path: c.pathName, + Query: c.query, + User: c.user, + InboundBytes: bytesReceived, + OutboundBytes: bytesSent, + BytesReceived: bytesReceived, + BytesSent: bytesSent, + OutboundFramesDiscarded: outboundFramesDiscarded, } } diff --git a/internal/servers/rtmp/server_test.go b/internal/servers/rtmp/server_test.go index 9d0fa847..d3f6a0fa 100644 --- a/internal/servers/rtmp/server_test.go +++ b/internal/servers/rtmp/server_test.go @@ -196,15 +196,18 @@ func TestServerPublish(t *testing.T) { require.Equal(t, &defs.APIRTMPConnList{ Items: []defs.APIRTMPConn{ { - ID: list.Items[0].ID, - Created: list.Items[0].Created, - RemoteAddr: list.Items[0].RemoteAddr, - State: "publish", - Path: "teststream", - Query: "user=myuser&pass=mypass¶m=value", - User: "myuser", - BytesReceived: list.Items[0].BytesReceived, - BytesSent: list.Items[0].BytesSent, + ID: list.Items[0].ID, + Created: list.Items[0].Created, + RemoteAddr: list.Items[0].RemoteAddr, + State: "publish", + Path: "teststream", + Query: "user=myuser&pass=mypass¶m=value", + User: "myuser", + InboundBytes: list.Items[0].InboundBytes, + OutboundBytes: list.Items[0].OutboundBytes, + OutboundFramesDiscarded: list.Items[0].OutboundFramesDiscarded, + BytesReceived: list.Items[0].BytesReceived, + BytesSent: list.Items[0].BytesSent, }, }, }, list) @@ -354,15 +357,18 @@ func TestServerRead(t *testing.T) { require.Equal(t, &defs.APIRTMPConnList{ Items: []defs.APIRTMPConn{ { - ID: list.Items[0].ID, - Created: list.Items[0].Created, - RemoteAddr: list.Items[0].RemoteAddr, - State: "read", - Path: "teststream", - Query: "user=myuser&pass=mypass¶m=value", - User: "myuser", - BytesReceived: list.Items[0].BytesReceived, - BytesSent: list.Items[0].BytesSent, + ID: list.Items[0].ID, + Created: list.Items[0].Created, + RemoteAddr: list.Items[0].RemoteAddr, + State: "read", + Path: "teststream", + Query: "user=myuser&pass=mypass¶m=value", + User: "myuser", + InboundBytes: list.Items[0].InboundBytes, + OutboundBytes: list.Items[0].OutboundBytes, + OutboundFramesDiscarded: list.Items[0].OutboundFramesDiscarded, + BytesReceived: list.Items[0].BytesReceived, + BytesSent: list.Items[0].BytesSent, }, }, }, list) diff --git a/internal/servers/rtsp/session.go b/internal/servers/rtsp/session.go index cc83071a..d799f005 100644 --- a/internal/servers/rtsp/session.go +++ b/internal/servers/rtsp/session.go @@ -489,6 +489,7 @@ func (s *session) apiItem() *defs.APIRTSPSession { OutboundBytes: stats.OutboundBytes, OutboundRTPPackets: stats.OutboundRTPPackets, OutboundRTPPacketsReportedLost: stats.OutboundRTPPacketsReportedLost, + OutboundRTPPacketsDiscarded: s.outboundRTPPacketsDiscarded.Get(), OutboundRTCPPackets: stats.OutboundRTCPPackets, BytesReceived: stats.InboundBytes, BytesSent: stats.OutboundBytes, diff --git a/internal/servers/srt/conn.go b/internal/servers/srt/conn.go index a3658348..14b9110e 100644 --- a/internal/servers/srt/conn.go +++ b/internal/servers/srt/conn.go @@ -66,6 +66,7 @@ type conn struct { query string user string sconn srt.Conn + reader *stream.Reader } func (c *conn) initialize() { @@ -335,6 +336,10 @@ func (c *conn) runRead(streamID *streamID) error { res.Stream.AddReader(r) defer res.Stream.RemoveReader(r) + c.mutex.Lock() + c.reader = r + c.mutex.Unlock() + select { case <-c.ctx.Done(): return fmt.Errorf("terminated") @@ -433,5 +438,9 @@ func (c *conn) apiItem() *defs.APISRTConn { item.PacketsReceivedLossRate = s.Instantaneous.PktRecvLossRate } + if c.reader != nil { + item.OutboundFramesDiscarded = c.reader.OutboundFramesDiscarded() + } + return item } diff --git a/internal/servers/srt/server_test.go b/internal/servers/srt/server_test.go index 11389dfa..ef3d5d7a 100644 --- a/internal/servers/srt/server_test.go +++ b/internal/servers/srt/server_test.go @@ -216,6 +216,7 @@ func TestServerPublish(t *testing.T) { BytesSendDrop: list.Items[0].BytesSendDrop, BytesReceivedDrop: list.Items[0].BytesReceivedDrop, BytesReceivedUndecrypt: list.Items[0].BytesReceivedUndecrypt, + OutboundFramesDiscarded: list.Items[0].OutboundFramesDiscarded, UsPacketsSendPeriod: list.Items[0].UsPacketsSendPeriod, PacketsFlowWindow: list.Items[0].PacketsFlowWindow, PacketsFlightSize: list.Items[0].PacketsFlightSize, diff --git a/internal/servers/webrtc/server_test.go b/internal/servers/webrtc/server_test.go index 67ed9d64..da8a0fa1 100644 --- a/internal/servers/webrtc/server_test.go +++ b/internal/servers/webrtc/server_test.go @@ -342,6 +342,7 @@ func TestServerPublish(t *testing.T) { OutboundBytes: list.Items[0].OutboundBytes, OutboundRTPPackets: list.Items[0].OutboundRTPPackets, OutboundRTCPPackets: list.Items[0].OutboundRTCPPackets, + OutboundFramesDiscarded: list.Items[0].OutboundFramesDiscarded, BytesReceived: list.Items[0].BytesReceived, BytesSent: list.Items[0].BytesSent, RTPPacketsReceived: list.Items[0].RTPPacketsReceived, @@ -635,6 +636,7 @@ func TestServerRead(t *testing.T) { OutboundBytes: list.Items[0].OutboundBytes, OutboundRTPPackets: list.Items[0].OutboundRTPPackets, OutboundRTCPPackets: list.Items[0].OutboundRTCPPackets, + OutboundFramesDiscarded: list.Items[0].OutboundFramesDiscarded, 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 6454b7ba..157d5f02 100644 --- a/internal/servers/webrtc/session.go +++ b/internal/servers/webrtc/session.go @@ -63,6 +63,7 @@ type session struct { uuid uuid.UUID secret uuid.UUID mutex sync.RWMutex + reader *stream.Reader pc *webrtc.PeerConnection user string @@ -387,6 +388,10 @@ func (s *session) runRead() (int, error) { res.Stream.AddReader(r) defer res.Stream.RemoveReader(r) + s.mutex.Lock() + s.reader = r + s.mutex.Unlock() + select { case <-pc.Failed(): return 0, fmt.Errorf("peer connection closed") @@ -479,6 +484,7 @@ func (s *session) apiItem() *defs.APIWebRTCSession { rtpPacketsJitter := float64(0) rtcpPacketsReceived := uint64(0) rtcpPacketsSent := uint64(0) + outboundFramesDiscarded := uint64(0) if s.pc != nil { peerConnectionEstablished = true @@ -495,6 +501,10 @@ func (s *session) apiItem() *defs.APIWebRTCSession { rtcpPacketsSent = stats.RTCPPacketsSent } + if s.reader != nil { + outboundFramesDiscarded = s.reader.OutboundFramesDiscarded() + } + return &defs.APIWebRTCSession{ ID: s.uuid, Created: s.created, @@ -519,6 +529,7 @@ func (s *session) apiItem() *defs.APIWebRTCSession { OutboundBytes: bytesSent, OutboundRTPPackets: rtpPacketsSent, OutboundRTCPPackets: rtcpPacketsSent, + OutboundFramesDiscarded: outboundFramesDiscarded, BytesReceived: bytesReceived, BytesSent: bytesSent, RTPPacketsReceived: rtpPacketsReceived, diff --git a/internal/stream/reader.go b/internal/stream/reader.go index 5ad92e75..fe9d647c 100644 --- a/internal/stream/reader.go +++ b/internal/stream/reader.go @@ -65,6 +65,11 @@ func (r *Reader) Formats() []format.Format { return out } +// OutboundFramesDiscarded returns the number of frames discarded because the reader is too slow. +func (r *Reader) OutboundFramesDiscarded() uint64 { + return r.outboundFramesDiscarded.Get() +} + // error returns whenever there's an error. // It can be called only after stream.AddReader(). func (r *Reader) Error() chan error { diff --git a/internal/stream/stream.go b/internal/stream/stream.go index 5c2f5d69..7aeddd74 100644 --- a/internal/stream/stream.go +++ b/internal/stream/stream.go @@ -483,6 +483,11 @@ func (s *Stream) OutboundBytes() uint64 { return outboundBytes } +// InboundFramesInError returns the number of frames received with processing errors. +func (s *Stream) InboundFramesInError() uint64 { + return s.inboundFramesInError.Get() +} + // RTSPStream returns the RTSP stream. func (s *Stream) RTSPStream(server *gortsplib.Server) *gortsplib.ServerStream { s.mutex.Lock()