tunnel: add multiple entrypoints

handler/http2/handler.go

@@ -276,68 +276,68 @@ func (h *http2Handler) roundTrip(ctx context.Context, w http.ResponseWriter, req
 	ro.SrcAddr = cc.LocalAddr().String()
 	ro.DstAddr = cc.RemoteAddr().String()
 
-	if req.Method == http.MethodConnect {
+	if req.Method != http.MethodConnect {
-		resp.StatusCode = http.StatusOK
-		w.WriteHeader(http.StatusOK)
-		if fw, ok := w.(http.Flusher); ok {
-			fw.Flush()
-		}
-
-		rw := xio.NewReadWriter(req.Body, flushWriter{w})
-
-		// compatible with HTTP1.x
-		if hj, ok := w.(http.Hijacker); ok && req.ProtoMajor == 1 {
-			// we take over the underly connection
-			conn, _, err := hj.Hijack()
-			if err != nil {
-				log.Error(err)
-				resp.StatusCode = http.StatusInternalServerError
-				w.WriteHeader(http.StatusInternalServerError)
-				return err
-			}
-			defer conn.Close()
-
-			rw = conn
-		}
-
-		rw = traffic_wrapper.WrapReadWriter(
-			h.limiter,
-			rw,
-			clientID,
-			limiter.ScopeOption(limiter.ScopeClient),
-			limiter.ServiceOption(h.options.Service),
-			limiter.NetworkOption("tcp"),
-			limiter.AddrOption(host),
-			limiter.ClientOption(clientID),
-			limiter.SrcOption(req.RemoteAddr),
-		)
-		if h.options.Observer != nil {
-			pstats := h.stats.Stats(clientID)
-			pstats.Add(stats.KindTotalConns, 1)
-			pstats.Add(stats.KindCurrentConns, 1)
-			defer pstats.Add(stats.KindCurrentConns, -1)
-			rw = stats_wrapper.WrapReadWriter(rw, pstats)
-		}
-
 		start := time.Now()
 		log.Infof("%s <-> %s", req.RemoteAddr, host)
-		// xnet.Transport(rw, cc)
+		if err := h.forwardRequest(w, req, cc); err != nil {
-		xnet.Pipe(ctx, xio.NewReadWriteCloser(rw, rw, req.Body), cc)
+			log.Info("%s - %s: %s", req.RemoteAddr, host, err)
+		}
 		log.WithFields(map[string]any{
 			"duration": time.Since(start),
 		}).Infof("%s >-< %s", req.RemoteAddr, host)
+
 		return nil
 	}
 
+	resp.StatusCode = http.StatusOK
+	w.WriteHeader(http.StatusOK)
+	if fw, ok := w.(http.Flusher); ok {
+		fw.Flush()
+	}
+
+	rw := xio.NewReadWriter(req.Body, flushWriter{w})
+
+	// compatible with HTTP1.x
+	if hj, ok := w.(http.Hijacker); ok && req.ProtoMajor == 1 {
+		// we take over the underly connection
+		conn, _, err := hj.Hijack()
+		if err != nil {
+			log.Error(err)
+			resp.StatusCode = http.StatusInternalServerError
+			w.WriteHeader(http.StatusInternalServerError)
+			return err
+		}
+		defer conn.Close()
+
+		rw = conn
+	}
+
+	rw = traffic_wrapper.WrapReadWriter(
+		h.limiter,
+		rw,
+		clientID,
+		limiter.ScopeOption(limiter.ScopeClient),
+		limiter.ServiceOption(h.options.Service),
+		limiter.NetworkOption("tcp"),
+		limiter.AddrOption(host),
+		limiter.ClientOption(clientID),
+		limiter.SrcOption(req.RemoteAddr),
+	)
+	if h.options.Observer != nil {
+		pstats := h.stats.Stats(clientID)
+		pstats.Add(stats.KindTotalConns, 1)
+		pstats.Add(stats.KindCurrentConns, 1)
+		defer pstats.Add(stats.KindCurrentConns, -1)
+		rw = stats_wrapper.WrapReadWriter(rw, pstats)
+	}
+
 	start := time.Now()
 	log.Infof("%s <-> %s", req.RemoteAddr, host)
-	if err := h.forwardRequest(w, req, cc); err != nil {
+	// xnet.Transport(rw, cc)
-		log.Info("%s - %s: %s", req.RemoteAddr, host, err)
+	xnet.Pipe(ctx, xio.NewReadWriteCloser(rw, rw, req.Body), cc)
-	}
 	log.WithFields(map[string]any{
 		"duration": time.Since(start),
 	}).Infof("%s >-< %s", req.RemoteAddr, host)
-
 	return nil
 }
 

handler/tunnel/handler.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/go-gost/core/auth"
 	"github.com/go-gost/core/handler"
+	"github.com/go-gost/core/ingress"
 	"github.com/go-gost/core/limiter"
 	"github.com/go-gost/core/limiter/traffic"
 	"github.com/go-gost/core/listener"
@@ -45,16 +46,15 @@ func init() {
 }
 
 type tunnelHandler struct {
-	id      string
+	id          string
-	options handler.Options
+	options     handler.Options
-	pool    *ConnectorPool
+	pool        *ConnectorPool
-	epSvc   service.Service
+	entrypoints []service.Service
-	ep      *entrypoint
+	md          metadata
-	md      metadata
+	log         logger.Logger
-	log     logger.Logger
+	stats       *stats_util.HandlerStats
-	stats   *stats_util.HandlerStats
+	limiter     traffic.TrafficLimiter
-	limiter traffic.TrafficLimiter
+	cancel      context.CancelFunc
-	cancel  context.CancelFunc
 }
 
 func NewHandler(opts ...handler.Option) handler.Handler {
@@ -84,34 +84,7 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) {
 	})
 	h.pool = NewConnectorPool(h.id)
 
-	h.ep = &entrypoint{
+	if err = h.initEntrypoints(); err != nil {
-		node:    h.id,
-		service: h.options.Service,
-		pool:    h.pool,
-		ingress: h.md.ingress,
-		sd:      h.md.sd,
-		log: h.log.WithFields(map[string]any{
-			"kind": "entrypoint",
-		}),
-		sniffingWebsocket:   h.md.sniffingWebsocket,
-		websocketSampleRate: h.md.sniffingWebsocketSampleRate,
-		readTimeout:         h.md.entryPointReadTimeout,
-	}
-	h.ep.transport = &http.Transport{
-		DialContext:           h.ep.dial,
-		IdleConnTimeout:       30 * time.Second,
-		ResponseHeaderTimeout: h.md.entryPointReadTimeout,
-		DisableKeepAlives:     !h.md.entryPointKeepalive,
-		DisableCompression:    !h.md.entryPointCompression,
-	}
-
-	for _, ro := range h.options.Recorders {
-		if ro.Record == xrecorder.RecorderServiceHandler {
-			h.ep.recorder = ro
-			break
-		}
-	}
-	if err = h.initEntrypoint(); err != nil {
 		return
 	}
 
@@ -134,20 +107,73 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) {
 	return nil
 }
 
-func (h *tunnelHandler) initEntrypoint() (err error) {
+func (h *tunnelHandler) initEntrypoints() (err error) {
-	if h.md.entryPoint == "" {
+	if h.md.entryPoint != "" {
-		return
+		svc, err := h.createEntrypointService(h.md.entryPoint, h.md.ingress)
+		if err != nil {
+			return err
+		}
+		go svc.Serve()
+
+		h.entrypoints = append(h.entrypoints, svc)
+		h.log.Infof("entrypoint: %s", svc.Addr())
+	}
+
+	for _, ep := range h.md.entrypoints {
+		if ep.Addr == "" {
+			continue
+		}
+		svc, err := h.createEntrypointService(ep.Addr, ep.Ingress)
+		if err != nil {
+			return err
+		}
+		go svc.Serve()
+
+		h.entrypoints = append(h.entrypoints, svc)
+		h.log.Infof("entrypoint: %s %s", ep.Name, svc.Addr())
+	}
+
+	return
+}
+
+func (h *tunnelHandler) createEntrypointService(addr string, ingress ingress.Ingress) (service.Service, error) {
+	ep := &entrypoint{
+		node:    h.id,
+		service: h.options.Service,
+		pool:    h.pool,
+		ingress: ingress,
+		sd:      h.md.sd,
+		log: h.log.WithFields(map[string]any{
+			"kind": "entrypoint",
+		}),
+		sniffingWebsocket:   h.md.sniffingWebsocket,
+		websocketSampleRate: h.md.sniffingWebsocketSampleRate,
+		readTimeout:         h.md.entryPointReadTimeout,
+	}
+	ep.transport = &http.Transport{
+		DialContext:           ep.dial,
+		IdleConnTimeout:       30 * time.Second,
+		ResponseHeaderTimeout: h.md.entryPointReadTimeout,
+		DisableKeepAlives:     !h.md.entryPointKeepalive,
+		DisableCompression:    !h.md.entryPointCompression,
+	}
+
+	for _, ro := range h.options.Recorders {
+		if ro.Record == xrecorder.RecorderServiceHandler {
+			ep.recorder = ro
+			break
+		}
 	}
 
 	network := "tcp"
-	if xnet.IsIPv4(h.md.entryPoint) {
+	if xnet.IsIPv4(addr) {
 		network = "tcp4"
 	}
 
-	ln, err := net.Listen(network, h.md.entryPoint)
+	ln, err := net.Listen(network, addr)
 	if err != nil {
 		h.log.Error(err)
-		return
+		return nil, err
 	}
 
 	serviceName := fmt.Sprintf("%s-ep-%s", h.options.Service, ln.Addr())
@@ -166,23 +192,19 @@ func (h *tunnelHandler) initEntrypoint() (err error) {
 		})),
 	)
 	if err = epListener.Init(nil); err != nil {
-		return
+		return nil, err
 	}
 	epHandler := &entrypointHandler{
-		ep: h.ep,
+		ep: ep,
 	}
 	if err = epHandler.Init(nil); err != nil {
-		return
+		return nil, err
 	}
 
-	h.epSvc = xservice.NewService(
+	return xservice.NewService(
 		serviceName, epListener, epHandler,
 		xservice.LoggerOption(log),
-	)
+	), nil
-	go h.epSvc.Serve()
-	log.Infof("entrypoint: %s", h.epSvc.Addr())
-
-	return
 }
 
 func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handler.HandleOption) (err error) {
@@ -306,8 +328,8 @@ func (h *tunnelHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl
 
 // Close implements io.Closer interface.
 func (h *tunnelHandler) Close() error {
-	if h.epSvc != nil {
+	for _, ep := range h.entrypoints {
-		h.epSvc.Close()
+		ep.Close()
 	}
 	h.pool.Close()
 

handler/tunnel/metadata.go

@@ -1,6 +1,7 @@
 package tunnel
 
 import (
+	"encoding/json"
 	"strings"
 	"time"
 
@@ -22,6 +23,8 @@ const (
 type metadata struct {
 	readTimeout time.Duration
 
+	entrypoints []entrypointConfig
+
 	entryPoint                  string
 	entryPointID                relay.TunnelID
 	entryPointProxyProtocol     int
@@ -31,11 +34,11 @@ type metadata struct {
 	sniffingWebsocket           bool
 	sniffingWebsocketSampleRate float64
 
-	directTunnel           bool
+	directTunnel bool
-	tunnelTTL              time.Duration
+	tunnelTTL    time.Duration
-	ingress                ingress.Ingress
+	ingress      ingress.Ingress
-	sd                     sd.SD
+	sd           sd.SD
-	muxCfg                 *mux.Config
+	muxCfg       *mux.Config
 
 	observerPeriod       time.Duration
 	observerResetTraffic bool
@@ -91,6 +94,39 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
 			)
 		}
 	}
+
+	if md != nil {
+		if v := md.Get("entrypoints"); v != nil {
+			b, err := json.Marshal(v)
+			if err != nil {
+				return err
+			}
+
+			var entrypoints []struct {
+				Addr    string
+				Ingress string
+			}
+
+			if err := json.Unmarshal(b, &entrypoints); err != nil {
+				return err
+			}
+
+			for _, ep := range entrypoints {
+				if ep.Addr == "" {
+					continue
+				}
+				ingress := registry.IngressRegistry().Get(ep.Ingress)
+				if ingress == nil {
+					ingress = h.md.ingress
+				}
+				h.md.entrypoints = append(h.md.entrypoints, entrypointConfig{
+					Addr:    ep.Addr,
+					Ingress: ingress,
+				})
+			}
+		}
+	}
+
 	h.md.sd = registry.SDRegistry().Get(mdutil.GetString(md, "sd"))
 
 	h.md.muxCfg = &mux.Config{
@@ -123,3 +159,9 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) {
 
 	return
 }
+
+type entrypointConfig struct {
+	Name    string
+	Addr    string
+	Ingress ingress.Ingress
+}