diff --git a/pkg/core/server.go b/pkg/core/server.go index 1a76303..d760804 100644 --- 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 index c4f0f56..64f1e18 100644 --- 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)