metrics: add handshake duration metrics (#162)

Ehco1996 · Nov 12, 2022 · 1514443 · 1514443
1 parent 1b3f1e1
commit 1514443
Show file tree

Hide file tree

Showing 8 changed files with 1,781 additions and 264 deletions.
diff --git a/internal/constant/constant.go b/internal/constant/constant.go
@@ -37,7 +37,7 @@ const (
 
 	SmuxGCDuration       = 30 * time.Second
 	SmuxMaxAliveDuration = 1 * time.Hour
-	SmuxMaxStreamCnt     = 50
+	SmuxMaxStreamCnt     = 5
 
 	Listen_RAW  = "raw"
 	Listen_WS   = "ws"

diff --git a/internal/relay/relay.go b/internal/relay/relay.go
@@ -158,9 +158,11 @@ func (r *Relay) RunLocalTCPServer() error {
 			web.CurConnectionCount.WithLabelValues(remote.Label, web.METRIC_CONN_TCP).Inc()
 			defer web.CurConnectionCount.WithLabelValues(remote.Label, web.METRIC_CONN_TCP).Dec()
 			defer c.Close()
+			t1 := time.Now()
 			if err := r.TP.HandleTCPConn(c, remote); err != nil {
 				r.L.Errorf("HandleTCPConn meet error from:%s to:%s err:%s", c.RemoteAddr(), remote.Address, err)
 			}
+			web.HandShakeDuration.WithLabelValues(remote.Label).Observe(float64(time.Since(t1).Milliseconds()))
 		}(c)
 	}
 }
@@ -192,6 +194,35 @@ func (r *Relay) RunLocalUDPServer() error {
 	}
 }
 
+func (r *Relay) RunLocalMTCPServer() error {
+	mTCPServer := transporter.NewMTCPServer(r.L.Named("MTCPServer"), r.LocalTCPAddr)
+	r.closeTcpF = func() error {
+		return mTCPServer.Close()
+	}
+
+	go func() {
+		r.L.Infof("Start MTCP relay server %s", r.Name)
+		mTCPServer.ListenAndServe()
+	}()
+
+	tp := r.TP.(*transporter.Raw)
+	for {
+		conn, e := mTCPServer.Accept()
+		if e != nil {
+			return e
+		}
+		go func(c net.Conn) {
+			remote := tp.GetRemote()
+			web.CurConnectionCount.WithLabelValues(remote.Label, web.METRIC_CONN_TCP).Inc()
+			defer web.CurConnectionCount.WithLabelValues(remote.Label, web.METRIC_CONN_TCP).Dec()
+			defer c.Close()
+			if err := tp.HandleTCPConn(c, remote); err != nil {
+				r.L.Errorf("HandleTCPConn meet error from:%s to:%s err:%s", c.RemoteAddr(), remote.Address, err)
+			}
+		}(conn)
+	}
+}
+
 func (r *Relay) RunLocalWSServer() error {
 	tp := r.TP.(*transporter.Raw)
 	mux := mux.NewRouter()
@@ -234,7 +265,7 @@ func (r *Relay) RunLocalWSSServer() error {
 	r.closeTcpF = func() error {
 		return lis.Close()
 	}
-	r.L.Infof("Start WSS relay %s", r.Name)
+	r.L.Infof("Start WSS relay Server %s", r.Name)
 	return server.Serve(tls.NewListener(lis, server.TLSConfig))
 }
 
@@ -277,32 +308,3 @@ func (r *Relay) RunLocalMWSSServer() error {
 		go tp.HandleMWssRequest(conn)
 	}
 }
-
-func (r *Relay) RunLocalMTCPServer() error {
-	mTCPServer := transporter.NewMTCPServer(r.L.Named("MTCPServer"), r.LocalTCPAddr)
-	r.closeTcpF = func() error {
-		return mTCPServer.Close()
-	}
-
-	go func() {
-		r.L.Infof("Start MTCP relay server %s", r.Name)
-		mTCPServer.ListenAndServe()
-	}()
-
-	tp := r.TP.(*transporter.Raw)
-	for {
-		conn, e := mTCPServer.Accept()
-		if e != nil {
-			return e
-		}
-		go func(c net.Conn) {
-			remote := tp.GetRemote()
-			web.CurConnectionCount.WithLabelValues(remote.Label, web.METRIC_CONN_TCP).Inc()
-			defer web.CurConnectionCount.WithLabelValues(remote.Label, web.METRIC_CONN_TCP).Dec()
-			defer c.Close()
-			if err := tp.HandleTCPConn(c, remote); err != nil {
-				r.L.Errorf("HandleTCPConn meet error from:%s to:%s err:%s", c.RemoteAddr(), remote.Address, err)
-			}
-		}(conn)
-	}
-}
diff --git a/internal/transporter/mtcp.go b/internal/transporter/mtcp.go
@@ -121,9 +121,7 @@ type MTCPClient struct {
 }
 
 func NewMTCPClient(l *zap.SugaredLogger) *MTCPClient {
-	return &MTCPClient{
-		l: l.Named("MTCPClient"),
-	}
+	return &MTCPClient{l: l}
 }
 
 func (c *MTCPClient) InitNewSession(ctx context.Context, addr string) (*smux.Session, error) {
@@ -138,6 +136,6 @@ func (c *MTCPClient) InitNewSession(ctx context.Context, addr string) (*smux.Ses
 	if err != nil {
 		return nil, err
 	}
-	c.l.Infof("Init new session to: %s", rc.RemoteAddr())
+	c.l.Infof("init new session to: %s", rc.RemoteAddr())
 	return session, nil
 }
diff --git a/internal/transporter/mwss.go b/internal/transporter/mwss.go
@@ -135,6 +135,6 @@ func (c *MWSSClient) InitNewSession(ctx context.Context, addr string) (*smux.Ses
 	if err != nil {
 		return nil, err
 	}
-	c.l.Infof("Init new session to: %s", rc.RemoteAddr())
+	c.l.Infof("init new session to: %s", rc.RemoteAddr())
 	return session, nil
 }
diff --git a/internal/transporter/smux.go b/internal/transporter/smux.go
@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/Ehco1996/ehco/internal/constant"
+	"github.com/Ehco1996/ehco/internal/web"
 	"github.com/xtaci/smux"
 	"go.uber.org/zap"
 )
@@ -90,6 +91,7 @@ func (tr *smuxTransporter) Dial(ctx context.Context, addr string) (conn net.Conn
 	}
 
 	// create new one
+	t1 := time.Now()
 	if session == nil {
 		session, err = tr.initSessionF(ctx, addr)
 		if err != nil {
@@ -105,5 +107,6 @@ func (tr *smuxTransporter) Dial(ctx context.Context, addr string) (conn net.Conn
 		session.Close()
 		return nil, err
 	}
+	web.HandShakeDuration.WithLabelValues(addr).Observe(float64(time.Since(t1).Milliseconds()))
 	return stream, nil
 }
diff --git a/internal/web/server.go b/internal/web/server.go
@@ -52,6 +52,7 @@ func registerMetrics(cfg *config.Config) {
 	prometheus.MustRegister(EhcoAlive)
 	prometheus.MustRegister(CurConnectionCount)
 	prometheus.MustRegister(NetWorkTransmitBytes)
+	prometheus.MustRegister(HandShakeDuration)
 
 	EhcoAlive.Set(EhcoAliveStateInit)
 

diff --git a/internal/web/traffic_metrics.go b/internal/web/traffic_metrics.go
@@ -34,4 +34,12 @@ var (
 		Help:        "传输流量总量bytes",
 		ConstLabels: ConstLabels,
 	}, []string{METRIC_LABEL_REMOTE, METRIC_LABEL_CONN_TYPE})
+
+	HandShakeDuration = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Subsystem:   METRIC_SUBSYSTEM_TRAFFIC,
+		Namespace:   METRIC_NS,
+		Name:        "handshake_duration",
+		Help:        "握手时间ms",
+		ConstLabels: ConstLabels,
+	}, []string{METRIC_LABEL_REMOTE})
 )