Merge pull request apernet#60 from HyNetwork/wip-conn-gauge

Prometheus Active Conn Gauge
deerpine · Apr 28, 2021 · 39a33ae · 39a33ae
2 parents 56f35d9 + 7170158
commit 39a33ae
Show file tree

Hide file tree

Showing 4 changed files with 24 additions and 6 deletions.
diff --git a/README.md b/README.md
@@ -183,6 +183,9 @@ You can make Hysteria expose a Prometheus HTTP client endpoint for monitoring tr
 If configured on port 8080, the endpoint would be at `http://example.com:8080/metrics`.
 
 ```text
+hysteria_active_conn{auth="55m95auW5oCq"} 32
+hysteria_active_conn{auth="aGFja2VyISE="} 7
+
 hysteria_traffic_downlink_bytes_total{auth="55m95auW5oCq"} 122639
 hysteria_traffic_downlink_bytes_total{auth="aGFja2VyISE="} 3.225058e+06
 

diff --git a/README.zh.md b/README.zh.md
@@ -171,6 +171,9 @@ Hysteria 是专门针对恶劣网络环境进行优化的 TCP/UDP 转发和代
 例如如果配置在 8080 端口，则 API 地址是 `http://example.com:8080/metrics`
 
 ```text
+hysteria_active_conn{auth="55m95auW5oCq"} 32
+hysteria_active_conn{auth="aGFja2VyISE="} 7
+
 hysteria_traffic_downlink_bytes_total{auth="55m95auW5oCq"} 122639
 hysteria_traffic_downlink_bytes_total{auth="aGFja2VyISE="} 3.225058e+06
 

diff --git a/pkg/core/server.go b/pkg/core/server.go
@@ -34,6 +34,7 @@ type Server struct {
 	udpErrorFunc   UDPErrorFunc
 
 	upCounterVec, downCounterVec *prometheus.CounterVec
+	connGaugeVec                 *prometheus.GaugeVec
 
 	listener quic.Listener
 }
@@ -93,7 +94,10 @@ func NewServer(addr string, tlsConfig *tls.Config, quicConfig *quic.Config, tran
 		s.downCounterVec = prometheus.NewCounterVec(prometheus.CounterOpts{
 			Name: "hysteria_traffic_downlink_bytes_total",
 		}, []string{"auth"})
-		promRegistry.MustRegister(s.upCounterVec, s.downCounterVec)
+		s.connGaugeVec = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Name: "hysteria_active_conn",
+		}, []string{"auth"})
+		promRegistry.MustRegister(s.upCounterVec, s.downCounterVec, s.connGaugeVec)
 	}
 	return s, nil
 }
@@ -133,7 +137,8 @@ func (s *Server) handleClient(cs quic.Session) {
 	}
 	// Start accepting streams and messages
 	sc := newServerClient(cs, s.transport, auth, s.disableUDP, s.aclEngine,
-		s.tcpRequestFunc, s.tcpErrorFunc, s.udpRequestFunc, s.udpErrorFunc, s.upCounterVec, s.downCounterVec)
+		s.tcpRequestFunc, s.tcpErrorFunc, s.udpRequestFunc, s.udpErrorFunc,
+		s.upCounterVec, s.downCounterVec, s.connGaugeVec)
 	sc.Run()
 	_ = cs.CloseWithError(closeErrorCodeGeneric, "")
 }

diff --git a/pkg/core/server_client.go b/pkg/core/server_client.go
@@ -30,6 +30,7 @@ type serverClient struct {
 	CUDPErrorFunc   UDPErrorFunc
 
 	UpCounter, DownCounter prometheus.Counter
+	ConnGauge              prometheus.Gauge
 
 	udpSessionMutex  sync.RWMutex
 	udpSessionMap    map[uint32]*net.UDPConn
@@ -39,7 +40,8 @@ type serverClient struct {
 func newServerClient(cs quic.Session, transport transport.Transport, auth []byte, disableUDP bool, ACLEngine *acl.Engine,
 	CTCPRequestFunc TCPRequestFunc, CTCPErrorFunc TCPErrorFunc,
 	CUDPRequestFunc UDPRequestFunc, CUDPErrorFunc UDPErrorFunc,
-	UpCounterVec, DownCounterVec *prometheus.CounterVec) *serverClient {
+	UpCounterVec, DownCounterVec *prometheus.CounterVec,
+	ConnGaugeVec *prometheus.GaugeVec) *serverClient {
 	sc := &serverClient{
 		CS:              cs,
 		Transport:       transport,
@@ -53,10 +55,11 @@ func newServerClient(cs quic.Session, transport transport.Transport, auth []byte
 		CUDPErrorFunc:   CUDPErrorFunc,
 		udpSessionMap:   make(map[uint32]*net.UDPConn),
 	}
-	if UpCounterVec != nil && DownCounterVec != nil {
+	if UpCounterVec != nil && DownCounterVec != nil && ConnGaugeVec != nil {
 		authB64 := base64.StdEncoding.EncodeToString(auth)
 		sc.UpCounter = UpCounterVec.WithLabelValues(authB64)
 		sc.DownCounter = DownCounterVec.WithLabelValues(authB64)
+		sc.ConnGauge = ConnGaugeVec.WithLabelValues(authB64)
 	}
 	return sc
 }
@@ -78,12 +81,16 @@ func (c *serverClient) Run() {
 		if err != nil {
 			break
 		}
-		go c.handleStream(stream)
+		c.ConnGauge.Inc()
+		go func() {
+			c.handleStream(stream)
+			_ = stream.Close()
+			c.ConnGauge.Dec()
+		}()
 	}
 }
 
 func (c *serverClient) handleStream(stream quic.Stream) {
-	defer stream.Close()
 	// Read request
 	var req clientRequest
 	err := struc.Unpack(stream, &req)