From ef440e8b4a82d352b3bebdaf139314ac64528d06 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Tue, 11 Feb 2025 20:54:52 +0800 Subject: [PATCH] router: router.id is optional --- connector/router/connector.go | 8 +++-- connector/router/metadata.go | 8 ----- handler/router/associate.go | 65 ++++++++++++++++++++++++++++------- handler/router/handler.go | 32 ++++++++--------- handler/router/metadata.go | 17 +++++++-- 5 files changed, 86 insertions(+), 44 deletions(-) diff --git a/connector/router/connector.go b/connector/router/connector.go index c6fa9eda..e5b20523 100644 --- a/connector/router/connector.go +++ b/connector/router/connector.go @@ -90,9 +90,11 @@ func (c *routerConnector) Connect(ctx context.Context, conn net.Conn, network, a af.ParseFrom(address) req.Features = append(req.Features, af) // dst address - req.Features = append(req.Features, &relay.TunnelFeature{ - ID: c.md.routerID, - }) + if !c.md.routerID.IsZero() { + req.Features = append(req.Features, &relay.TunnelFeature{ + ID: c.md.routerID, + }) + } if _, err := req.WriteTo(conn); err != nil { return nil, err diff --git a/connector/router/metadata.go b/connector/router/metadata.go index 54d199e0..377d62e2 100644 --- a/connector/router/metadata.go +++ b/connector/router/metadata.go @@ -30,13 +30,5 @@ func (c *routerConnector) parseMetadata(md mdata.Metadata) (err error) { c.md.routerID = relay.NewTunnelID(uuid[:]) } - if c.md.routerID.IsZero() { - uuid, err := uuid.NewUUID() - if err != nil { - return err - } - c.md.routerID = relay.NewTunnelID(uuid[:]) - } - return } diff --git a/handler/router/associate.go b/handler/router/associate.go index 4a7caf6c..d6f20e7e 100644 --- a/handler/router/associate.go +++ b/handler/router/associate.go @@ -10,6 +10,7 @@ import ( "math" "net" "sync" + "time" "github.com/go-gost/core/common/bufpool" "github.com/go-gost/core/limiter" @@ -48,7 +49,7 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw rid = parseRouterID(rule.Endpoint) } - if rid.IsZero() || !rid.Equal(routerID) { + if !rid.Equal(routerID) { resp.Status = relay.StatusHostUnreachable resp.WriteTo(conn) err := fmt.Errorf("no route to host %s", host) @@ -111,13 +112,19 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw Name: clientID, Node: h.id, }) + + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + go h.sdRenew(ctx, clientID, connectorID.String()) } log.Debugf("%s/%s: router=%s, connector=%s, weight=%d established", host, network, routerID, connectorID, connectorID.Weight()) - for { - b := bufpool.Get(h.md.bufferSize) + b := bufpool.Get(h.md.bufferSize) + defer bufpool.Put(b) + for { n, err := conn.Read(b) if err != nil { if err == io.EOF { @@ -130,9 +137,25 @@ func (h *routerHandler) handleAssociate(ctx context.Context, conn net.Conn, netw } } -func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID relay.TunnelID, log logger.Logger) error { - defer bufpool.Put(data) +func (h *routerHandler) sdRenew(ctx context.Context, clientID string, connectorID string) { + tc := time.NewTicker(h.md.sdRenewInterval) + defer tc.Stop() + for { + select { + case <-tc.C: + h.md.sd.Renew(ctx, &sd.Service{ + ID: connectorID, + Name: clientID, + Node: h.id, + }) + case <-ctx.Done(): + return + } + } +} + +func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID relay.TunnelID, log logger.Logger) error { var dstIP net.IP if waterutil.IsIPv4(data) { header, err := ipv4.ParseHeader(data) @@ -167,16 +190,10 @@ func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID rid := routerID.String() - var route *router.Route - if r := registry.RouterRegistry().Get(rid); r != nil { - route = r.GetRoute(ctx, dstIP.String(), router.IDOption(rid)) - } - if route == nil && h.md.router != nil { - route = h.md.router.GetRoute(ctx, dstIP.String(), router.IDOption(rid)) - } + route := h.getRoute(ctx, rid, dstIP.String()) if route == nil || route.Gateway == "" { // no route to host, discard - return fmt.Errorf("route to host %s", dstIP) + return fmt.Errorf("no route to host %s", dstIP) } if log.IsLevelEnabled(logger.TraceLevel) { @@ -218,6 +235,28 @@ func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID return nil } +func (h *routerHandler) getRoute(ctx context.Context, rid string, dst string) *router.Route { + if h.md.routerCacheEnabled { + if item := h.routeCache.Get(dst); item != nil && !item.Expired() { + v, _ := item.Value().(*router.Route) + return v + } + } + + var route *router.Route + if r := registry.RouterRegistry().Get(rid); r != nil { + route = r.GetRoute(ctx, dst, router.IDOption(rid)) + } + if route == nil && h.md.router != nil { + route = h.md.router.GetRoute(ctx, dst, router.IDOption(rid)) + } + + if h.md.routerCacheEnabled { + h.routeCache.Set(dst, cache.NewItem(route, h.md.routerCacheExpiration)) + } + return route +} + func (h *routerHandler) getAddrforRoute(ctx context.Context, routerID, gateway string) net.Addr { if h.md.sd == nil { return nil diff --git a/handler/router/handler.go b/handler/router/handler.go index 21bc2280..f1c70669 100644 --- a/handler/router/handler.go +++ b/handler/router/handler.go @@ -37,16 +37,17 @@ func init() { } type routerHandler struct { - id string - options handler.Options - pool *ConnectorPool - epConn net.PacketConn - md metadata - log logger.Logger - stats *stats_util.HandlerStats - limiter traffic.TrafficLimiter - cancel context.CancelFunc - sdCache *cache.Cache + id string + options handler.Options + pool *ConnectorPool + epConn net.PacketConn + md metadata + log logger.Logger + stats *stats_util.HandlerStats + limiter traffic.TrafficLimiter + cancel context.CancelFunc + sdCache *cache.Cache + routeCache *cache.Cache } func NewHandler(opts ...handler.Option) handler.Handler { @@ -56,8 +57,9 @@ func NewHandler(opts ...handler.Option) handler.Handler { } return &routerHandler{ - options: options, - sdCache: cache.NewCache(time.Minute), + options: options, + sdCache: cache.NewCache(time.Minute), + routeCache: cache.NewCache(time.Minute), } } @@ -203,12 +205,6 @@ func (h *routerHandler) Handle(ctx context.Context, conn net.Conn, opts ...handl } } - if routerID.IsZero() { - resp.Status = relay.StatusBadRequest - resp.WriteTo(conn) - return ErrRouterID - } - if user != "" { log = log.WithFields(map[string]any{"user": user}) } diff --git a/handler/router/metadata.go b/handler/router/metadata.go index 35d8b5b8..e3c125b0 100644 --- a/handler/router/metadata.go +++ b/handler/router/metadata.go @@ -25,7 +25,11 @@ type metadata struct { ingress ingress.Ingress sd sd.SD sdCacheExpiration time.Duration - router router.Router + sdRenewInterval time.Duration + + router router.Router + routerCacheEnabled bool + routerCacheExpiration time.Duration observerPeriod time.Duration observerResetTraffic bool @@ -45,12 +49,21 @@ func (h *routerHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.ingress = registry.IngressRegistry().Get(mdutil.GetString(md, "ingress")) h.md.sd = registry.SDRegistry().Get(mdutil.GetString(md, "sd")) - h.md.sdCacheExpiration = mdutil.GetDuration(md, "sd.cacheExpiration") + h.md.sdCacheExpiration = mdutil.GetDuration(md, "sd.cache.expiration") if h.md.sdCacheExpiration <= 0 { h.md.sdCacheExpiration = defaultCacheExpiration } + h.md.sdRenewInterval = mdutil.GetDuration(md, "sd.renewInterval") + if h.md.sdRenewInterval < time.Second { + h.md.sdRenewInterval = defaultTTL + } h.md.router = registry.RouterRegistry().Get(mdutil.GetString(md, "router")) + h.md.routerCacheEnabled = mdutil.GetBool(md, "router.cache") + h.md.routerCacheExpiration = mdutil.GetDuration(md, "router.cache.expiration") + if h.md.routerCacheExpiration <= 0 { + h.md.routerCacheExpiration = defaultCacheExpiration + } h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod") if h.md.observerPeriod == 0 {