add RTMPS metrics (#2783)

README.md

@@ -1485,6 +1485,11 @@ rtmp_conns{id="[id]",state="[state]"} 1
 rtmp_conns_bytes_received{id="[id]",state="[state]"} 1234
 rtmp_conns_bytes_sent{id="[id]",state="[state]"} 187
 
+# metrics of every RTMPS connection
+rtmps_conns{id="[id]",state="[state]"} 1
+rtmps_conns_bytes_received{id="[id]",state="[state]"} 1234
+rtmps_conns_bytes_sent{id="[id]",state="[state]"} 187
+
 # metrics of every SRT connection
 srt_conns{id="[id]",state="[state]"} 1
 srt_conns_bytes_received{id="[id]",state="[state]"} 1234

internal/core/core.go

@@ -270,11 +270,12 @@ func (p *Core) createResources(initial bool) error {
 
 	if p.conf.Metrics &&
 		p.metrics == nil {
-		p.metrics, err = newMetrics(
+		p.metrics = &metrics{
-			p.conf.MetricsAddress,
+			Address:     p.conf.MetricsAddress,
-			p.conf.ReadTimeout,
+			ReadTimeout: p.conf.ReadTimeout,
-			p,
+			Parent:      p,
-		)
+		}
+		err = p.metrics.initialize()
 		if err != nil {
 			return err
 		}
@@ -452,6 +453,10 @@ func (p *Core) createResources(initial bool) error {
 		if err != nil {
 			return err
 		}
+
+		if p.metrics != nil {
+			p.metrics.setRTMPSServer(p.rtmpsServer)
+		}
 	}
 
 	if p.conf.HLS &&
@@ -798,6 +803,10 @@ func (p *Core) closeResources(newConf *conf.Conf, calledByAPI bool) {
 	}
 
 	if closeRTMPSServer && p.rtmpsServer != nil {
+		if p.metrics != nil {
+			p.metrics.setRTMPSServer(nil)
+		}
+
 		p.rtmpsServer.close()
 		p.rtmpsServer = nil
 	}

internal/core/metrics.go

@@ -24,7 +24,9 @@ type metricsParent interface {
 }
 
 type metrics struct {
-	parent metricsParent
+	Address     string
+	ReadTimeout conf.StringDuration
+	Parent      metricsParent
 
 	httpServer    *httpserv.WrappedServer
 	mutex         sync.Mutex
@@ -32,44 +34,37 @@ type metrics struct {
 	rtspServer    apiRTSPServer
 	rtspsServer   apiRTSPServer
 	rtmpServer    apiRTMPServer
+	rtmpsServer   apiRTMPServer
 	srtServer     apiSRTServer
 	hlsManager    apiHLSManager
 	webRTCManager apiWebRTCManager
 }
 
-func newMetrics(
+func (m *metrics) initialize() error {
-	address string,
-	readTimeout conf.StringDuration,
-	parent metricsParent,
-) (*metrics, error) {
-	m := &metrics{
-		parent: parent,
-	}
-
 	router := gin.New()
 	router.SetTrustedProxies(nil) //nolint:errcheck
 
 	router.GET("/metrics", m.onMetrics)
 
-	network, address := restrictnetwork.Restrict("tcp", address)
+	network, address := restrictnetwork.Restrict("tcp", m.Address)
 
 	var err error
 	m.httpServer, err = httpserv.NewWrappedServer(
 		network,
 		address,
-		time.Duration(readTimeout),
+		time.Duration(m.ReadTimeout),
 		"",
 		"",
 		router,
 		m,
 	)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	m.Log(logger.Info, "listener opened on "+address)
 
-	return m, nil
+	return nil
 }
 
 func (m *metrics) close() {
@@ -78,7 +73,7 @@ func (m *metrics) close() {
 }
 
 func (m *metrics) Log(level logger.Level, format string, args ...interface{}) {
-	m.parent.Log(level, "[metrics] "+format, args...)
+	m.Parent.Log(level, "[metrics] "+format, args...)
 }
 
 func (m *metrics) onMetrics(ctx *gin.Context) {
@@ -201,6 +196,22 @@ func (m *metrics) onMetrics(ctx *gin.Context) {
 		}
 	}
 
+	if !interfaceIsEmpty(m.rtmpsServer) {
+		data, err := m.rtmpsServer.apiConnsList()
+		if err == nil && len(data.Items) != 0 {
+			for _, i := range data.Items {
+				tags := "{id=\"" + i.ID.String() + "\",state=\"" + string(i.State) + "\"}"
+				out += metric("rtmps_conns", tags, 1)
+				out += metric("rtmps_conns_bytes_received", tags, int64(i.BytesReceived))
+				out += metric("rtmps_conns_bytes_sent", tags, int64(i.BytesSent))
+			}
+		} else {
+			out += metric("rtmps_conns", "", 0)
+			out += metric("rtmps_conns_bytes_received", "", 0)
+			out += metric("rtmps_conns_bytes_sent", "", 0)
+		}
+	}
+
 	if !interfaceIsEmpty(m.srtServer) {
 		data, err := m.srtServer.apiConnsList()
 		if err == nil && len(data.Items) != 0 {
@@ -272,6 +283,13 @@ func (m *metrics) setRTMPServer(s apiRTMPServer) {
 	m.rtmpServer = s
 }
 
+// setRTMPSServer is called by core.
+func (m *metrics) setRTMPSServer(s apiRTMPServer) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	m.rtmpsServer = s
+}
+
 // setSRTServer is called by core.
 func (m *metrics) setSRTServer(s apiSRTServer) {
 	m.mutex.Lock()

internal/core/metrics_test.go

@@ -41,6 +41,9 @@ func TestMetrics(t *testing.T) {
 		"encryption: optional\n" +
 		"serverCert: " + serverCertFpath + "\n" +
 		"serverKey: " + serverKeyFpath + "\n" +
+		"rtmpEncryption: optional\n" +
+		"rtmpServerCert: " + serverCertFpath + "\n" +
+		"rtmpServerKey: " + serverKeyFpath + "\n" +
 		"paths:\n" +
 		"  all_others:\n")
 	require.Equal(t, true, ok)
@@ -69,6 +72,9 @@ rtsps_sessions_bytes_sent 0
 rtmp_conns 0
 rtmp_conns_bytes_received 0
 rtmp_conns_bytes_sent 0
+rtmps_conns 0
+rtmps_conns_bytes_received 0
+rtmps_conns_bytes_sent 0
 srt_conns 0
 srt_conns_bytes_received 0
 srt_conns_bytes_sent 0
@@ -81,7 +87,7 @@ webrtc_sessions_bytes_sent 0
 	t.Run("with data", func(t *testing.T) {
 		terminate := make(chan struct{})
 		var wg sync.WaitGroup
-		wg.Add(5)
+		wg.Add(6)
 
 		go func() {
 			defer wg.Done()
@@ -126,6 +132,23 @@ webrtc_sessions_bytes_sent 0
 			<-terminate
 		}()
 
+		go func() {
+			defer wg.Done()
+			u, err := url.Parse("rtmp://localhost:1936/rtmps_path")
+			require.NoError(t, err)
+
+			nconn, err := tls.Dial("tcp", u.Host, &tls.Config{InsecureSkipVerify: true})
+			require.NoError(t, err)
+			defer nconn.Close() //nolint:errcheck
+
+			conn, err := rtmp.NewClientConn(nconn, u, true)
+			require.NoError(t, err)
+
+			_, err = rtmp.NewWriter(conn, testFormatH264, nil)
+			require.NoError(t, err)
+			<-terminate
+		}()
+
 		go func() {
 			defer wg.Done()
 
@@ -219,6 +242,9 @@ webrtc_sessions_bytes_sent 0
 				`paths\{name=".*?",state="ready"\} 1`+"\n"+
 				`paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+
 				`paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+
+				`paths\{name=".*?",state="ready"\} 1`+"\n"+
+				`paths_bytes_received\{name=".*?",state="ready"\} [0-9]+`+"\n"+
+				`paths_bytes_sent\{name=".*?",state="ready"\} 0`+"\n"+
 				`hls_muxers\{name=".*?"\} 1`+"\n"+
 				`hls_muxers_bytes_sent\{name=".*?"\} [0-9]+`+"\n"+
 				`hls_muxers\{name=".*?"\} 1`+"\n"+
@@ -229,6 +255,8 @@ webrtc_sessions_bytes_sent 0
 				`hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+
 				`hls_muxers\{name=".*?"\} 1`+"\n"+
 				`hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+
+				`hls_muxers\{name=".*?"\} 1`+"\n"+
+				`hls_muxers_bytes_sent\{name=".*?"\} 0`+"\n"+
 				`rtsp_conns\{id=".*?"\} 1`+"\n"+
 				`rtsp_conns_bytes_received\{id=".*?"\} [0-9]+`+"\n"+
 				`rtsp_conns_bytes_sent\{id=".*?"\} [0-9]+`+"\n"+
@@ -244,6 +272,9 @@ webrtc_sessions_bytes_sent 0
 				`rtmp_conns\{id=".*?",state="publish"\} 1`+"\n"+
 				`rtmp_conns_bytes_received\{id=".*?",state="publish"\} [0-9]+`+"\n"+
 				`rtmp_conns_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+
+				`rtmps_conns\{id=".*?",state="publish"\} 1`+"\n"+
+				`rtmps_conns_bytes_received\{id=".*?",state="publish"\} [0-9]+`+"\n"+
+				`rtmps_conns_bytes_sent\{id=".*?",state="publish"\} [0-9]+`+"\n"+
 				`srt_conns\{id=".*?",state="publish"\} 1`+"\n"+
 				`srt_conns_bytes_received\{id=".*?",state="publish"\} [0-9]+`+"\n"+
 				`srt_conns_bytes_sent\{id=".*?",state="publish"\} 0`+"\n"+