pimhwang
/
mediamtx
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

fix crash when requesting metrics and RTMP, SRT or WebRTC servers are not present anymore (#2782)

pull/2783/head
Alessandro Ros 2 years ago committed by GitHub
parent
20123fd5b9
commit
ed72fa7db1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 274 additions and 269 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 68
      internal/core/core.go
  2. 15
      internal/core/hls_manager.go
  3. 22
      internal/core/metrics.go
  4. 383
      internal/core/metrics_test.go
  5. 11
      internal/core/path_manager.go
  6. 11
      internal/core/rtmp_server.go
  7. 19
      internal/core/rtsp_server.go
  8. 9
      internal/core/srt_server.go
  9. 5
      internal/core/webrtc_manager.go

68
internal/core/core.go
Unescape View File

@ -313,9 +313,12 @@ func (p *Core) createResources(initial bool) error {
p.conf.UDPMaxPayloadSize, p.conf.UDPMaxPayloadSize,
p.conf.Paths, p.conf.Paths,
p.externalCmdPool, p.externalCmdPool,
p.metrics,
p, p,
) )
if p.metrics != nil {
p.metrics.setPathManager(p.pathManager)
}
} }
if p.conf.RTSP && if p.conf.RTSP &&
@ -347,13 +350,16 @@ func (p *Core) createResources(initial bool) error {
p.conf.RunOnConnectRestart, p.conf.RunOnConnectRestart,
p.conf.RunOnDisconnect, p.conf.RunOnDisconnect,
p.externalCmdPool, p.externalCmdPool,
p.metrics,
p.pathManager, p.pathManager,
p, p,
) )
if err != nil { if err != nil {
return err return err
} }
if p.metrics != nil {
p.metrics.setRTSPServer(p.rtspServer)
}
} }
if p.conf.RTSP && if p.conf.RTSP &&
@ -382,13 +388,16 @@ func (p *Core) createResources(initial bool) error {
p.conf.RunOnConnectRestart, p.conf.RunOnConnectRestart,
p.conf.RunOnDisconnect, p.conf.RunOnDisconnect,
p.externalCmdPool, p.externalCmdPool,
p.metrics,
p.pathManager, p.pathManager,
p, p,
) )
if err != nil { if err != nil {
return err return err
} }
if p.metrics != nil {
p.metrics.setRTSPSServer(p.rtspsServer)
}
} }
if p.conf.RTMP && if p.conf.RTMP &&
@ -408,13 +417,16 @@ func (p *Core) createResources(initial bool) error {
p.conf.RunOnConnectRestart, p.conf.RunOnConnectRestart,
p.conf.RunOnDisconnect, p.conf.RunOnDisconnect,
p.externalCmdPool, p.externalCmdPool,
p.metrics,
p.pathManager, p.pathManager,
p, p,
) )
if err != nil { if err != nil {
return err return err
} }
if p.metrics != nil {
p.metrics.setRTMPServer(p.rtmpServer)
}
} }
if p.conf.RTMP && if p.conf.RTMP &&
@ -434,7 +446,6 @@ func (p *Core) createResources(initial bool) error {
p.conf.RunOnConnectRestart, p.conf.RunOnConnectRestart,
p.conf.RunOnDisconnect, p.conf.RunOnDisconnect,
p.externalCmdPool, p.externalCmdPool,
p.metrics,
p.pathManager, p.pathManager,
p, p,
) )
@ -463,12 +474,17 @@ func (p *Core) createResources(initial bool) error {
p.conf.ReadTimeout, p.conf.ReadTimeout,
p.conf.WriteQueueSize, p.conf.WriteQueueSize,
p.pathManager, p.pathManager,
p.metrics,
p, p,
) )
if err != nil { if err != nil {
return err return err
} }
p.pathManager.setHLSManager(p.hlsManager)
if p.metrics != nil {
p.metrics.setHLSManager(p.hlsManager)
}
} }
if p.conf.WebRTC && if p.conf.WebRTC &&
@ -490,7 +506,6 @@ func (p *Core) createResources(initial bool) error {
ICEServers: p.conf.WebRTCICEServers2, ICEServers: p.conf.WebRTCICEServers2,
ExternalCmdPool: p.externalCmdPool, ExternalCmdPool: p.externalCmdPool,
PathManager: p.pathManager, PathManager: p.pathManager,
Metrics: p.metrics,
Parent: p, Parent: p,
} }
err = p.webRTCManager.initialize() err = p.webRTCManager.initialize()
@ -498,6 +513,10 @@ func (p *Core) createResources(initial bool) error {
p.webRTCManager = nil p.webRTCManager = nil
return err return err
} }
if p.metrics != nil {
p.metrics.setWebRTCManager(p.webRTCManager)
}
} }
if p.conf.SRT && if p.conf.SRT &&
@ -513,13 +532,16 @@ func (p *Core) createResources(initial bool) error {
p.conf.RunOnConnectRestart, p.conf.RunOnConnectRestart,
p.conf.RunOnDisconnect, p.conf.RunOnDisconnect,
p.externalCmdPool, p.externalCmdPool,
p.metrics,
p.pathManager, p.pathManager,
p, p,
) )
if err != nil { if err != nil {
return err return err
} }
if p.metrics != nil {
p.metrics.setSRTServer(p.srtServer)
}
} }
if p.conf.API && if p.conf.API &&
@ -747,16 +769,30 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
} }
if closeSRTServer && p.srtServer != nil { if closeSRTServer && p.srtServer != nil {
if p.metrics != nil {
p.metrics.setSRTServer(nil)
}
p.srtServer.close() p.srtServer.close()
p.srtServer = nil p.srtServer = nil
} }
if closeWebRTCManager && p.webRTCManager != nil { if closeWebRTCManager && p.webRTCManager != nil {
if p.metrics != nil {
p.metrics.setWebRTCManager(nil)
}
p.webRTCManager.close() p.webRTCManager.close()
p.webRTCManager = nil p.webRTCManager = nil
} }
if closeHLSManager && p.hlsManager != nil { if closeHLSManager && p.hlsManager != nil {
if p.metrics != nil {
p.metrics.setHLSManager(nil)
}
p.pathManager.setHLSManager(nil)
p.hlsManager.close() p.hlsManager.close()
p.hlsManager = nil p.hlsManager = nil
} }
@ -767,21 +803,37 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
} }
if closeRTMPServer && p.rtmpServer != nil { if closeRTMPServer && p.rtmpServer != nil {
if p.metrics != nil {
p.metrics.setRTMPServer(nil)
}
p.rtmpServer.close() p.rtmpServer.close()
p.rtmpServer = nil p.rtmpServer = nil
} }
if closeRTSPSServer && p.rtspsServer != nil { if closeRTSPSServer && p.rtspsServer != nil {
if p.metrics != nil {
p.metrics.setRTSPSServer(nil)
}
p.rtspsServer.close() p.rtspsServer.close()
p.rtspsServer = nil p.rtspsServer = nil
} }
if closeRTSPServer && p.rtspServer != nil { if closeRTSPServer && p.rtspServer != nil {
if p.metrics != nil {
p.metrics.setRTSPServer(nil)
}
p.rtspServer.close() p.rtspServer.close()
p.rtspServer = nil p.rtspServer = nil
} }
if closePathManager && p.pathManager != nil { if closePathManager && p.pathManager != nil {
if p.metrics != nil {
p.metrics.setPathManager(nil)
}
p.pathManager.close() p.pathManager.close()
p.pathManager = nil p.pathManager = nil
} }

15
internal/core/hls_manager.go
Unescape View File

@ -45,7 +45,6 @@ type hlsManager struct {
directory string directory string
writeQueueSize int writeQueueSize int
pathManager *pathManager pathManager *pathManager
metrics *metrics
parent hlsManagerParent parent hlsManagerParent
ctx context.Context ctx context.Context
@ -81,7 +80,6 @@ func newHLSManager(
readTimeout conf.StringDuration, readTimeout conf.StringDuration,
writeQueueSize int, writeQueueSize int,
pathManager *pathManager, pathManager *pathManager,
metrics *metrics,
parent hlsManagerParent, parent hlsManagerParent,
) (*hlsManager, error) { ) (*hlsManager, error) {
ctx, ctxCancel := context.WithCancel(context.Background()) ctx, ctxCancel := context.WithCancel(context.Background())
@ -98,7 +96,6 @@ func newHLSManager(
writeQueueSize: writeQueueSize, writeQueueSize: writeQueueSize,
pathManager: pathManager, pathManager: pathManager,
parent: parent, parent: parent,
metrics: metrics,
ctx: ctx, ctx: ctx,
ctxCancel: ctxCancel, ctxCancel: ctxCancel,
muxers: make(map[string]*hlsMuxer), muxers: make(map[string]*hlsMuxer),
@ -129,12 +126,6 @@ func newHLSManager(
m.Log(logger.Info, "listener opened on "+address) m.Log(logger.Info, "listener opened on "+address)
m.pathManager.setHLSManager(m)
if m.metrics != nil {
m.metrics.setHLSManager(m)
}
m.wg.Add(1) m.wg.Add(1)
go m.run() go m.run()
@ -223,12 +214,6 @@ outer:
m.ctxCancel() m.ctxCancel()
m.httpServer.close() m.httpServer.close()
m.pathManager.setHLSManager(nil)
if m.metrics != nil {
m.metrics.setHLSManager(nil)
}
} }
func (m *hlsManager) createMuxer(pathName string, remoteAddr string) *hlsMuxer { func (m *hlsManager) createMuxer(pathName string, remoteAddr string) *hlsMuxer {

22
internal/core/metrics.go
Unescape View File

@ -237,50 +237,50 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
io.WriteString(ctx.Writer, out) //nolint:errcheck io.WriteString(ctx.Writer, out) //nolint:errcheck
} }
// pathManagerSet is called by pathManager. // setPathManager is called by core.
func (m *metrics) pathManagerSet(s apiPathManager) { func (m *metrics) setPathManager(s apiPathManager) {
m.mutex.Lock() m.mutex.Lock()
defer m.mutex.Unlock() defer m.mutex.Unlock()
m.pathManager = s m.pathManager = s
} }
// setHLSManager is called by hlsManager. // setHLSManager is called by core.
func (m *metrics) setHLSManager(s apiHLSManager) { func (m *metrics) setHLSManager(s apiHLSManager) {
m.mutex.Lock() m.mutex.Lock()
defer m.mutex.Unlock() defer m.mutex.Unlock()
m.hlsManager = s m.hlsManager = s
} }
// setRTSPServer is called by rtspServer (plain). // setRTSPServer is called by core.
func (m *metrics) setRTSPServer(s apiRTSPServer) { func (m *metrics) setRTSPServer(s apiRTSPServer) {
m.mutex.Lock() m.mutex.Lock()
defer m.mutex.Unlock() defer m.mutex.Unlock()
m.rtspServer = s m.rtspServer = s
} }
// setRTSPSServer is called by rtspServer (tls). // setRTSPSServer is called by core.
func (m *metrics) setRTSPSServer(s apiRTSPServer) { func (m *metrics) setRTSPSServer(s apiRTSPServer) {
m.mutex.Lock() m.mutex.Lock()
defer m.mutex.Unlock() defer m.mutex.Unlock()
m.rtspsServer = s m.rtspsServer = s
} }
// rtmpServerSet is called by rtmpServer. // setRTMPServer is called by core.
func (m *metrics) rtmpServerSet(s apiRTMPServer) { func (m *metrics) setRTMPServer(s apiRTMPServer) {
m.mutex.Lock() m.mutex.Lock()
defer m.mutex.Unlock() defer m.mutex.Unlock()
m.rtmpServer = s m.rtmpServer = s
} }
// srtServerSet is called by srtServer. // setSRTServer is called by core.
func (m *metrics) srtServerSet(s apiSRTServer) { func (m *metrics) setSRTServer(s apiSRTServer) {
m.mutex.Lock() m.mutex.Lock()
defer m.mutex.Unlock() defer m.mutex.Unlock()
m.srtServer = s m.srtServer = s
} }
// webRTCManagerSet is called by webRTCManager. // setWebRTCManager is called by core.
func (m *metrics) webRTCManagerSet(s apiWebRTCManager) { func (m *metrics) setWebRTCManager(s apiWebRTCManager) {
m.mutex.Lock() m.mutex.Lock()
defer m.mutex.Unlock() defer m.mutex.Unlock()
m.webRTCManager = s m.webRTCManager = s

383
internal/core/metrics_test.go
Unescape View File

@ -16,7 +16,7 @@ import (
"github.com/bluenviron/gortsplib/v4/pkg/description" "github.com/bluenviron/gortsplib/v4/pkg/description"
"github.com/bluenviron/gortsplib/v4/pkg/format" "github.com/bluenviron/gortsplib/v4/pkg/format"
"github.com/bluenviron/mediacommon/pkg/formats/mpegts" "github.com/bluenviron/mediacommon/pkg/formats/mpegts"
"github.com/datarhei/gosrt" srt "github.com/datarhei/gosrt"
"github.com/pion/rtp" "github.com/pion/rtp"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
@ -33,7 +33,8 @@ func TestMetrics(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
defer os.Remove(serverKeyFpath) defer os.Remove(serverKeyFpath)
p, ok := newInstance("hlsAlwaysRemux: yes\n" + p, ok := newInstance("api: yes\n" +
"hlsAlwaysRemux: yes\n" +
"metrics: yes\n" + "metrics: yes\n" +
"webrtcServerCert: " + serverCertFpath + "\n" + "webrtcServerCert: " + serverCertFpath + "\n" +
"webrtcServerKey: " + serverKeyFpath + "\n" + "webrtcServerKey: " + serverKeyFpath + "\n" +
@ -47,9 +48,10 @@ func TestMetrics(t *testing.T) {
hc := &http.Client{Transport: &http.Transport{}} hc := &http.Client{Transport: &http.Transport{}}
bo := httpPullFile(t, hc, "http://localhost:9998/metrics") t.Run("initial", func(t *testing.T) {
bo := httpPullFile(t, hc, "http://localhost:9998/metrics")
require.Equal(t, `paths 0 require.Equal(t, `paths 0
hls_muxers 0 hls_muxers 0
hls_muxers_bytes_sent 0 hls_muxers_bytes_sent 0
rtsp_conns 0 rtsp_conns 0
@ -74,181 +76,200 @@ webrtc_sessions 0
webrtc_sessions_bytes_received 0 webrtc_sessions_bytes_received 0
webrtc_sessions_bytes_sent 0 webrtc_sessions_bytes_sent 0
`, string(bo)) `, string(bo))
})
terminate := make(chan struct{})
var wg sync.WaitGroup t.Run("with data", func(t *testing.T) {
wg.Add(5) terminate := make(chan struct{})
var wg sync.WaitGroup
go func() { wg.Add(5)
defer wg.Done()
source := gortsplib.Client{} go func() {
err := source.StartRecording("rtsp://localhost:8554/rtsp_path", defer wg.Done()
&description.Session{Medias: []*description.Media{{ source := gortsplib.Client{}
Type: description.MediaTypeVideo, err := source.StartRecording("rtsp://localhost:8554/rtsp_path",
Formats: []format.Format{testFormatH264}, &description.Session{Medias: []*description.Media{{
}}}) Type: description.MediaTypeVideo,
require.NoError(t, err) Formats: []format.Format{testFormatH264},
defer source.Close() }}})
<-terminate require.NoError(t, err)
}() defer source.Close()
<-terminate
go func() { }()
defer wg.Done()
source2 := gortsplib.Client{TLSConfig: &tls.Config{InsecureSkipVerify: true}} go func() {
err := source2.StartRecording("rtsps://localhost:8322/rtsps_path", defer wg.Done()
&description.Session{Medias: []*description.Media{{ source2 := gortsplib.Client{TLSConfig: &tls.Config{InsecureSkipVerify: true}}
Type: description.MediaTypeVideo, err := source2.StartRecording("rtsps://localhost:8322/rtsps_path",
Formats: []format.Format{testFormatH264}, &description.Session{Medias: []*description.Media{{
}}}) Type: description.MediaTypeVideo,
require.NoError(t, err) Formats: []format.Format{testFormatH264},
defer source2.Close() }}})
<-terminate require.NoError(t, err)
}() defer source2.Close()
<-terminate
go func() { }()
defer wg.Done()
u, err := url.Parse("rtmp://localhost:1935/rtmp_path") go func() {
require.NoError(t, err) defer wg.Done()
u, err := url.Parse("rtmp://localhost:1935/rtmp_path")
nconn, err := net.Dial("tcp", u.Host) require.NoError(t, err)
require.NoError(t, err)
defer nconn.Close() nconn, err := net.Dial("tcp", u.Host)
require.NoError(t, err)
conn, err := rtmp.NewClientConn(nconn, u, true) defer nconn.Close()
require.NoError(t, err)
conn, err := rtmp.NewClientConn(nconn, u, true)
_, err = rtmp.NewWriter(conn, testFormatH264, nil) require.NoError(t, err)
require.NoError(t, err)
<-terminate _, err = rtmp.NewWriter(conn, testFormatH264, nil)
}() require.NoError(t, err)
<-terminate
go func() { }()
defer wg.Done()
go func() {
su, err := url.Parse("http://localhost:8889/webrtc_path/whip") defer wg.Done()
require.NoError(t, err)
su, err := url.Parse("http://localhost:8889/webrtc_path/whip")
s := &webrtc.WHIPClient{ require.NoError(t, err)
HTTPClient: &http.Client{Transport: &http.Transport{}},
URL: su, s := &webrtc.WHIPClient{
} HTTPClient: &http.Client{Transport: &http.Transport{}},
URL: su,
tracks, err := s.Publish(context.Background(), testMediaH264.Formats[0], nil) }
require.NoError(t, err)
defer checkClose(t, s.Close) tracks, err := s.Publish(context.Background(), testMediaH264.Formats[0], nil)
require.NoError(t, err)
err = tracks[0].WriteRTP(&rtp.Packet{ defer checkClose(t, s.Close)
Header: rtp.Header{
Version: 2, err = tracks[0].WriteRTP(&rtp.Packet{
Marker: true, Header: rtp.Header{
PayloadType: 96, Version: 2,
SequenceNumber: 123, Marker: true,
Timestamp: 45343, PayloadType: 96,
SSRC: 563423, SequenceNumber: 123,
}, Timestamp: 45343,
Payload: []byte{1}, SSRC: 563423,
}) },
require.NoError(t, err) Payload: []byte{1},
<-terminate })
}() require.NoError(t, err)
<-terminate
go func() { }()
defer wg.Done()
go func() {
srtConf := srt.DefaultConfig() defer wg.Done()
address, err := srtConf.UnmarshalURL("srt://localhost:8890?streamid=publish:srt_path")
require.NoError(t, err) srtConf := srt.DefaultConfig()
address, err := srtConf.UnmarshalURL("srt://localhost:8890?streamid=publish:srt_path")
err = srtConf.Validate() require.NoError(t, err)
require.NoError(t, err)
err = srtConf.Validate()
publisher, err := srt.Dial("srt", address, srtConf) require.NoError(t, err)
require.NoError(t, err)
defer publisher.Close() publisher, err := srt.Dial("srt", address, srtConf)
require.NoError(t, err)
track := &mpegts.Track{ defer publisher.Close()
Codec: &mpegts.CodecH264{},
} track := &mpegts.Track{
Codec: &mpegts.CodecH264{},
bw := bufio.NewWriter(publisher) }
w := mpegts.NewWriter(bw, []*mpegts.Track{track})
require.NoError(t, err) bw := bufio.NewWriter(publisher)
w := mpegts.NewWriter(bw, []*mpegts.Track{track})
err = w.WriteH26x(track, 0, 0, true, [][]byte{ require.NoError(t, err)
{ // SPS
0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02, err = w.WriteH26x(track, 0, 0, true, [][]byte{
0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04, { // SPS
0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9, 0x67, 0x42, 0xc0, 0x28, 0xd9, 0x00, 0x78, 0x02,
0x20, 0x27, 0xe5, 0x84, 0x00, 0x00, 0x03, 0x00, 0x04,
}, 0x00, 0x00, 0x03, 0x00, 0xf0, 0x3c, 0x60, 0xc9,
{ // PPS 0x20,
0x08, 0x06, 0x07, 0x08, },
}, { // PPS
{ // IDR 0x08, 0x06, 0x07, 0x08,
0x05, 1, },
}, { // IDR
}) 0x05, 1,
require.NoError(t, err) },
})
err = bw.Flush() require.NoError(t, err)
require.NoError(t, err)
<-terminate err = bw.Flush()
}() require.NoError(t, err)
<-terminate
time.Sleep(500 * time.Millisecond) }()
bo = httpPullFile(t, hc, "http://localhost:9998/metrics") time.Sleep(500 * time.Millisecond)
require.Regexp(t, bo := httpPullFile(t, hc, "http://localhost:9998/metrics")
`^paths\{name=".*?",state="ready"\} 1`+"\n"+
`paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ require.Regexp(t,
`paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+ `^paths\{name=".*?",state="ready"\} 1`+"\n"+
`paths\{name=".*?",state="ready"\} 1`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+
`paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+
`paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+ `paths\{name=".*?",state="ready"\} 1`+"\n"+
`paths\{name=".*?",state="ready"\} 1`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+
`paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+
`paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+ `paths\{name=".*?",state="ready"\} 1`+"\n"+
`paths\{name=".*?",state="ready"\} 1`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+
`paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+
`paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+ `paths\{name=".*?",state="ready"\} 1`+"\n"+
`paths\{name=".*?",state="ready"\} 1`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+
`paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+
`paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+ `paths\{name=".*?",state="ready"\} 1`+"\n"+
`hls_muxers\{name=".*?"\} 1`+"\n"+ `paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+
`hls_muxers_bytes_sent\{name=".*?"\} [0-9]+`+"\n"+ `paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+
`hls_muxers\{name=".*?"\} 1`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+
`hls_muxers_bytes_sent\{name=".*?"\} [0-9]+`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} [0-9]+`+"\n"+
`hls_muxers\{name=".*?"\} 1`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+
`hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} [0-9]+`+"\n"+
`hls_muxers\{name=".*?"\} 1`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+
`hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+
`hls_muxers\{name=".*?"\} 1`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+
`hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+
`rtsp_conns\{id=".*?"\} 1`+"\n"+ `hls_muxers\{name=".*?"\} 1`+"\n"+
`rtsp_conns_bytes_received\{id=".*?"\} [0-9]+`+"\n"+ `hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+
`rtsp_conns_bytes_sent\{id=".*?"\} [0-9]+`+"\n"+ `rtsp_conns\{id=".*?"\} 1`+"\n"+
`rtsp_sessions\{id=".*?",state="publish"\} 1`+"\n"+ `rtsp_conns_bytes_received\{id=".*?"\} [0-9]+`+"\n"+
`rtsp_sessions_bytes_received\{id=".*?",state="publish"\} 0`+"\n"+ `rtsp_conns_bytes_sent\{id=".*?"\} [0-9]+`+"\n"+
`rtsp_sessions_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsp_sessions\{id=".*?",state="publish"\} 1`+"\n"+
`rtsps_conns\{id=".*?"\} 1`+"\n"+ `rtsp_sessions_bytes_received\{id=".*?",state="publish"\} 0`+"\n"+
`rtsps_conns_bytes_received\{id=".*?"\} [0-9]+`+"\n"+ `rtsp_sessions_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+
`rtsps_conns_bytes_sent\{id=".*?"\} [0-9]+`+"\n"+ `rtsps_conns\{id=".*?"\} 1`+"\n"+
`rtsps_sessions\{id=".*?",state="publish"\} 1`+"\n"+ `rtsps_conns_bytes_received\{id=".*?"\} [0-9]+`+"\n"+
`rtsps_sessions_bytes_received\{id=".*?",state="publish"\} 0`+"\n"+ `rtsps_conns_bytes_sent\{id=".*?"\} [0-9]+`+"\n"+
`rtsps_sessions_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsps_sessions\{id=".*?",state="publish"\} 1`+"\n"+
`rtmp_conns\{id=".*?",state="publish"\} 1`+"\n"+ `rtsps_sessions_bytes_received\{id=".*?",state="publish"\} 0`+"\n"+
`rtmp_conns_bytes_received\{id=".*?",state="publish"\} [0-9]+`+"\n"+ `rtsps_sessions_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+
`rtmp_conns_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+ `rtmp_conns\{id=".*?",state="publish"\} 1`+"\n"+
`srt_conns\{id=".*?",state="publish"\} 1`+"\n"+ `rtmp_conns_bytes_received\{id=".*?",state="publish"\} [0-9]+`+"\n"+
`srt_conns_bytes_received\{id=".*?",state="publish"\} [0-9]+`+"\n"+ `rtmp_conns_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+
`srt_conns_bytes_sent\{id=".*?",state="publish"\} 0`+"\n"+ `srt_conns\{id=".*?",state="publish"\} 1`+"\n"+
`webrtc_sessions\{id=".*?",state="publish"\} 1`+"\n"+ `srt_conns_bytes_received\{id=".*?",state="publish"\} [0-9]+`+"\n"+
`webrtc_sessions_bytes_received\{id=".*?",state="publish"\} [0-9]+`+"\n"+ `srt_conns_bytes_sent\{id=".*?",state="publish"\} 0`+"\n"+
`webrtc_sessions_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+ `webrtc_sessions\{id=".*?",state="publish"\} 1`+"\n"+
"$", `webrtc_sessions_bytes_received\{id=".*?",state="publish"\} [0-9]+`+"\n"+
string(bo)) `webrtc_sessions_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+
"$",
close(terminate) string(bo))
wg.Wait()
close(terminate)
wg.Wait()
})
t.Run("servers deleted", func(t *testing.T) {
httpRequest(t, hc, http.MethodPatch, "http://localhost:9997/v3/config/global/patch", map[string]interface{}{
"rtsp": false,
"rtmp": false,
"srt": false,
"hls": false,
"webrtc": false,
}, nil)
time.Sleep(500 * time.Millisecond)
bo := httpPullFile(t, hc, "http://localhost:9998/metrics")
require.Equal(t, "paths 0\n", string(bo))
})
} }

11
internal/core/path_manager.go
Unescape View File

@ -78,7 +78,6 @@ type pathManager struct {
udpMaxPayloadSize int udpMaxPayloadSize int
pathConfs map[string]*conf.Path pathConfs map[string]*conf.Path
externalCmdPool *externalcmd.Pool externalCmdPool *externalcmd.Pool
metrics *metrics
parent pathManagerParent parent pathManagerParent
ctx context.Context ctx context.Context
@ -112,7 +111,6 @@ func newPathManager(
udpMaxPayloadSize int, udpMaxPayloadSize int,
pathConfs map[string]*conf.Path, pathConfs map[string]*conf.Path,
externalCmdPool *externalcmd.Pool, externalCmdPool *externalcmd.Pool,
metrics *metrics,
parent pathManagerParent, parent pathManagerParent,
) *pathManager { ) *pathManager {
ctx, ctxCancel := context.WithCancel(context.Background()) ctx, ctxCancel := context.WithCancel(context.Background())
@ -127,7 +125,6 @@ func newPathManager(
udpMaxPayloadSize: udpMaxPayloadSize, udpMaxPayloadSize: udpMaxPayloadSize,
pathConfs: pathConfs, pathConfs: pathConfs,
externalCmdPool: externalCmdPool, externalCmdPool: externalCmdPool,
metrics: metrics,
parent: parent, parent: parent,
ctx: ctx, ctx: ctx,
ctxCancel: ctxCancel, ctxCancel: ctxCancel,
@ -152,10 +149,6 @@ func newPathManager(
} }
} }
if pm.metrics != nil {
pm.metrics.pathManagerSet(pm)
}
pm.Log(logger.Debug, "path manager created") pm.Log(logger.Debug, "path manager created")
pm.wg.Add(1) pm.wg.Add(1)
@ -220,10 +213,6 @@ outer:
} }
pm.ctxCancel() pm.ctxCancel()
if pm.metrics != nil {
pm.metrics.pathManagerSet(nil)
}
} }
func (pm *pathManager) doReloadConf(newPaths map[string]*conf.Path) { func (pm *pathManager) doReloadConf(newPaths map[string]*conf.Path) {

11
internal/core/rtmp_server.go
Unescape View File

@ -59,7 +59,6 @@ type rtmpServer struct {
runOnConnectRestart bool runOnConnectRestart bool
runOnDisconnect string runOnDisconnect string
externalCmdPool *externalcmd.Pool externalCmdPool *externalcmd.Pool
metrics *metrics
pathManager *pathManager pathManager *pathManager
parent rtmpServerParent parent rtmpServerParent
@ -91,7 +90,6 @@ func newRTMPServer(
runOnConnectRestart bool, runOnConnectRestart bool,
runOnDisconnect string, runOnDisconnect string,
externalCmdPool *externalcmd.Pool, externalCmdPool *externalcmd.Pool,
metrics *metrics,
pathManager *pathManager, pathManager *pathManager,
parent rtmpServerParent, parent rtmpServerParent,
) (*rtmpServer, error) { ) (*rtmpServer, error) {
@ -124,7 +122,6 @@ func newRTMPServer(
runOnDisconnect: runOnDisconnect, runOnDisconnect: runOnDisconnect,
isTLS: isTLS, isTLS: isTLS,
externalCmdPool: externalCmdPool, externalCmdPool: externalCmdPool,
metrics: metrics,
pathManager: pathManager, pathManager: pathManager,
parent: parent, parent: parent,
ctx: ctx, ctx: ctx,
@ -141,10 +138,6 @@ func newRTMPServer(
s.Log(logger.Info, "listener opened on %s", address) s.Log(logger.Info, "listener opened on %s", address)
if s.metrics != nil {
s.metrics.rtmpServerSet(s)
}
newRTMPListener( newRTMPListener(
s.ln, s.ln,
&s.wg, &s.wg,
@ -247,10 +240,6 @@ outer:
s.ctxCancel() s.ctxCancel()
s.ln.Close() s.ln.Close()
if s.metrics != nil {
s.metrics.rtmpServerSet(s)
}
} }
func (s *rtmpServer) findConnByUUID(uuid uuid.UUID) *rtmpConn { func (s *rtmpServer) findConnByUUID(uuid uuid.UUID) *rtmpConn {

19
internal/core/rtsp_server.go
Unescape View File

@ -51,7 +51,6 @@ type rtspServer struct {
runOnConnectRestart bool runOnConnectRestart bool
runOnDisconnect string runOnDisconnect string
externalCmdPool *externalcmd.Pool externalCmdPool *externalcmd.Pool
metrics *metrics
pathManager *pathManager pathManager *pathManager
parent rtspServerParent parent rtspServerParent
@ -86,7 +85,6 @@ func newRTSPServer(
runOnConnectRestart bool, runOnConnectRestart bool,
runOnDisconnect string, runOnDisconnect string,
externalCmdPool *externalcmd.Pool, externalCmdPool *externalcmd.Pool,
metrics *metrics,
pathManager *pathManager, pathManager *pathManager,
parent rtspServerParent, parent rtspServerParent,
) (*rtspServer, error) { ) (*rtspServer, error) {
@ -102,7 +100,6 @@ func newRTSPServer(
runOnConnectRestart: runOnConnectRestart, runOnConnectRestart: runOnConnectRestart,
runOnDisconnect: runOnDisconnect, runOnDisconnect: runOnDisconnect,
externalCmdPool: externalCmdPool, externalCmdPool: externalCmdPool,
metrics: metrics,
pathManager: pathManager, pathManager: pathManager,
parent: parent, parent: parent,
ctx: ctx, ctx: ctx,
@ -146,14 +143,6 @@ func newRTSPServer(
s.Log(logger.Info, "listener opened on %s", printAddresses(s.srv)) s.Log(logger.Info, "listener opened on %s", printAddresses(s.srv))
if metrics != nil {
if !isTLS {
metrics.setRTSPServer(s)
} else {
metrics.setRTSPSServer(s)
}
}
s.wg.Add(1) s.wg.Add(1)
go s.run() go s.run()
@ -205,14 +194,6 @@ outer:
} }
s.ctxCancel() s.ctxCancel()
if s.metrics != nil {
if !s.isTLS {
s.metrics.setRTSPServer(nil)
} else {
s.metrics.setRTSPSServer(nil)
}
}
} }
// OnConnOpen implements gortsplib.ServerHandlerOnConnOpen. // OnConnOpen implements gortsplib.ServerHandlerOnConnOpen.

9
internal/core/srt_server.go
Unescape View File

@ -7,7 +7,7 @@ import (
"sync" "sync"
"time" "time"
"github.com/datarhei/gosrt" srt "github.com/datarhei/gosrt"
"github.com/google/uuid" "github.com/google/uuid"
"github.com/bluenviron/mediamtx/internal/conf" "github.com/bluenviron/mediamtx/internal/conf"
@ -67,7 +67,6 @@ type srtServer struct {
runOnConnectRestart bool runOnConnectRestart bool
runOnDisconnect string runOnDisconnect string
externalCmdPool *externalcmd.Pool externalCmdPool *externalcmd.Pool
metrics *metrics
pathManager *pathManager pathManager *pathManager
parent srtServerParent parent srtServerParent
@ -97,7 +96,6 @@ func newSRTServer(
runOnConnectRestart bool, runOnConnectRestart bool,
runOnDisconnect string, runOnDisconnect string,
externalCmdPool *externalcmd.Pool, externalCmdPool *externalcmd.Pool,
metrics *metrics,
pathManager *pathManager, pathManager *pathManager,
parent srtServerParent, parent srtServerParent,
) (*srtServer, error) { ) (*srtServer, error) {
@ -122,7 +120,6 @@ func newSRTServer(
runOnConnectRestart: runOnConnectRestart, runOnConnectRestart: runOnConnectRestart,
runOnDisconnect: runOnDisconnect, runOnDisconnect: runOnDisconnect,
externalCmdPool: externalCmdPool, externalCmdPool: externalCmdPool,
metrics: metrics,
pathManager: pathManager, pathManager: pathManager,
parent: parent, parent: parent,
ctx: ctx, ctx: ctx,
@ -139,10 +136,6 @@ func newSRTServer(
s.Log(logger.Info, "listener opened on "+address+" (UDP)") s.Log(logger.Info, "listener opened on "+address+" (UDP)")
if s.metrics != nil {
s.metrics.srtServerSet(s)
}
newSRTListener( newSRTListener(
s.ln, s.ln,
&s.wg, &s.wg,

5
internal/core/webrtc_manager.go
Unescape View File

@ -180,7 +180,6 @@ type webRTCManager struct {
ICEServers []conf.WebRTCICEServer ICEServers []conf.WebRTCICEServer
ExternalCmdPool *externalcmd.Pool ExternalCmdPool *externalcmd.Pool
PathManager *pathManager PathManager *pathManager
Metrics *metrics
Parent webRTCManagerParent Parent webRTCManagerParent
ctx context.Context ctx context.Context
@ -284,10 +283,6 @@ func (m *webRTCManager) initialize() error {
} }
m.Log(logger.Info, str) m.Log(logger.Info, str)
if m.Metrics != nil {
m.Metrics.webRTCManagerSet(m)
}
go m.run() go m.run()
return nil return nil

Write Preview
Loading…
Cancel
Save