From a69a759e17a5d0f9e4dff32ee9c5f788e7be1dc8 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Thu, 9 Oct 2025 22:32:03 +0800 Subject: [PATCH] tunnel: add multiple entrypoints --- handler/http2/handler.go | 98 +++++++++++++-------------- handler/tunnel/handler.go | 132 +++++++++++++++++++++---------------- handler/tunnel/metadata.go | 52 +++++++++++++-- 3 files changed, 173 insertions(+), 109 deletions(-) diff --git a/handler/http2/handler.go b/handler/http2/handler.go index a594c445..3865222e 100644 --- a/handler/http2/handler.go +++ b/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 { - 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) - } - + if req.Method != http.MethodConnect { start := time.Now() log.Infof("%s <-> %s", req.RemoteAddr, host) - // xnet.Transport(rw, cc) - xnet.Pipe(ctx, xio.NewReadWriteCloser(rw, rw, req.Body), cc) + if err := h.forwardRequest(w, req, cc); err != nil { + 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 { - log.Info("%s - %s: %s", req.RemoteAddr, host, err) - } + // xnet.Transport(rw, cc) + 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 } diff --git a/handler/tunnel/handler.go b/handler/tunnel/handler.go index 1289ee4d..589654b5 100644 --- a/handler/tunnel/handler.go +++ b/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 - options handler.Options - pool *ConnectorPool - epSvc service.Service - ep *entrypoint - md metadata - log logger.Logger - stats *stats_util.HandlerStats - limiter traffic.TrafficLimiter - cancel context.CancelFunc + id string + options handler.Options + pool *ConnectorPool + entrypoints []service.Service + md metadata + log logger.Logger + stats *stats_util.HandlerStats + limiter traffic.TrafficLimiter + 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{ - 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 { + if err = h.initEntrypoints(); err != nil { return } @@ -134,20 +107,73 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) { return nil } -func (h *tunnelHandler) initEntrypoint() (err error) { - if h.md.entryPoint == "" { - return +func (h *tunnelHandler) initEntrypoints() (err error) { + if h.md.entryPoint != "" { + 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), - ) - go h.epSvc.Serve() - log.Infof("entrypoint: %s", h.epSvc.Addr()) - - return + ), nil } 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 { - h.epSvc.Close() + for _, ep := range h.entrypoints { + ep.Close() } h.pool.Close() diff --git a/handler/tunnel/metadata.go b/handler/tunnel/metadata.go index 023aed77..c6310eda 100644 --- a/handler/tunnel/metadata.go +++ b/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 - tunnelTTL time.Duration - ingress ingress.Ingress - sd sd.SD - muxCfg *mux.Config + directTunnel bool + tunnelTTL time.Duration + ingress ingress.Ingress + sd sd.SD + 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 +}