From fc12e33786843977bb162860665bcc19a357eef6 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Sun, 9 Feb 2025 16:16:44 +0800 Subject: [PATCH] router handler: add cache for sd service --- config/parsing/router/parse.go | 3 ++ config/parsing/service/parse.go | 10 ++-- handler/http/handler.go | 10 ++-- handler/http2/handler.go | 10 ++-- handler/relay/handler.go | 10 ++-- handler/router/associate.go | 47 +++++++++++-------- handler/router/handler.go | 13 +++-- handler/router/metadata.go | 20 +++++--- handler/socks/v4/handler.go | 10 ++-- handler/socks/v5/handler.go | 10 ++-- handler/tunnel/handler.go | 10 ++-- internal/util/{limiter => cache}/cache.go | 2 +- .../traffic/cache}/limiter.go | 17 +++---- 13 files changed, 98 insertions(+), 74 deletions(-) rename internal/util/{limiter => cache}/cache.go (98%) rename {internal/util/limiter => limiter/traffic/cache}/limiter.go (85%) diff --git a/config/parsing/router/parse.go b/config/parsing/router/parse.go index 3c355c02..da0f6983 100644 --- a/config/parsing/router/parse.go +++ b/config/parsing/router/parse.go @@ -45,6 +45,9 @@ func ParseRouter(cfg *config.RouterConfig) router.Router { var routes []*router.Route for _, route := range cfg.Routes { + if route == nil { + continue + } _, ipNet, _ := net.ParseCIDR(route.Net) dst := route.Dst if dst != "" { diff --git a/config/parsing/service/parse.go b/config/parsing/service/parse.go index 86e35033..4ff061f6 100644 --- a/config/parsing/service/parse.go +++ b/config/parsing/service/parse.go @@ -28,8 +28,8 @@ import ( hop_parser "github.com/go-gost/x/config/parsing/hop" logger_parser "github.com/go-gost/x/config/parsing/logger" selector_parser "github.com/go-gost/x/config/parsing/selector" - limiter_util "github.com/go-gost/x/internal/util/limiter" tls_util "github.com/go-gost/x/internal/util/tls" + cache_limiter "github.com/go-gost/x/limiter/traffic/cache" "github.com/go-gost/x/metadata" mdutil "github.com/go-gost/x/metadata/util" xstats "github.com/go-gost/x/observer/stats" @@ -171,11 +171,11 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { listener.TLSConfigOption(tlsConfig), listener.AdmissionOption(xadmission.AdmissionGroup(admissions...)), listener.TrafficLimiterOption( - limiter_util.NewCachedTrafficLimiter( + cache_limiter.NewCachedTrafficLimiter( registry.TrafficLimiterRegistry().Get(cfg.Limiter), - limiter_util.RefreshIntervalOption(limiterRefreshInterval), - limiter_util.CleanupIntervalOption(limiterCleanupInterval), - limiter_util.ScopeOption(limiterScope), + cache_limiter.RefreshIntervalOption(limiterRefreshInterval), + cache_limiter.CleanupIntervalOption(limiterCleanupInterval), + cache_limiter.ScopeOption(limiterScope), ), ), listener.ConnLimiterOption(registry.ConnLimiterRegistry().Get(cfg.CLimiter)), diff --git a/handler/http/handler.go b/handler/http/handler.go index 0491d9af..e0581e76 100644 --- a/handler/http/handler.go +++ b/handler/http/handler.go @@ -35,12 +35,12 @@ import ( xio "github.com/go-gost/x/internal/io" xnet "github.com/go-gost/x/internal/net" xhttp "github.com/go-gost/x/internal/net/http" - limiter_util "github.com/go-gost/x/internal/util/limiter" "github.com/go-gost/x/internal/util/sniffing" stats_util "github.com/go-gost/x/internal/util/stats" tls_util "github.com/go-gost/x/internal/util/tls" ws_util "github.com/go-gost/x/internal/util/ws" rate_limiter "github.com/go-gost/x/limiter/rate" + cache_limiter "github.com/go-gost/x/limiter/traffic/cache" traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" xstats "github.com/go-gost/x/observer/stats" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" @@ -90,10 +90,10 @@ func (h *httpHandler) Init(md md.Metadata) error { } if h.options.Limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(h.options.Limiter, - limiter_util.RefreshIntervalOption(h.md.limiterRefreshInterval), - limiter_util.CleanupIntervalOption(h.md.limiterCleanupInterval), - limiter_util.ScopeOption(limiter.ScopeClient), + h.limiter = cache_limiter.NewCachedTrafficLimiter(h.options.Limiter, + cache_limiter.RefreshIntervalOption(h.md.limiterRefreshInterval), + cache_limiter.CleanupIntervalOption(h.md.limiterCleanupInterval), + cache_limiter.ScopeOption(limiter.ScopeClient), ) } diff --git a/handler/http2/handler.go b/handler/http2/handler.go index 3bf105d8..e3f7e4fd 100644 --- a/handler/http2/handler.go +++ b/handler/http2/handler.go @@ -31,9 +31,9 @@ import ( xio "github.com/go-gost/x/internal/io" xnet "github.com/go-gost/x/internal/net" xhttp "github.com/go-gost/x/internal/net/http" - limiter_util "github.com/go-gost/x/internal/util/limiter" stats_util "github.com/go-gost/x/internal/util/stats" rate_limiter "github.com/go-gost/x/limiter/rate" + cache_limiter "github.com/go-gost/x/limiter/traffic/cache" traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" @@ -78,10 +78,10 @@ func (h *http2Handler) Init(md md.Metadata) error { } if h.options.Limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(h.options.Limiter, - limiter_util.RefreshIntervalOption(h.md.limiterRefreshInterval), - limiter_util.CleanupIntervalOption(h.md.limiterCleanupInterval), - limiter_util.ScopeOption(limiter.ScopeClient), + h.limiter = cache_limiter.NewCachedTrafficLimiter(h.options.Limiter, + cache_limiter.RefreshIntervalOption(h.md.limiterRefreshInterval), + cache_limiter.CleanupIntervalOption(h.md.limiterCleanupInterval), + cache_limiter.ScopeOption(limiter.ScopeClient), ) } diff --git a/handler/relay/handler.go b/handler/relay/handler.go index 0444d0aa..035c873d 100644 --- a/handler/relay/handler.go +++ b/handler/relay/handler.go @@ -17,10 +17,10 @@ import ( "github.com/go-gost/core/recorder" "github.com/go-gost/relay" ctxvalue "github.com/go-gost/x/ctx" - limiter_util "github.com/go-gost/x/internal/util/limiter" stats_util "github.com/go-gost/x/internal/util/stats" tls_util "github.com/go-gost/x/internal/util/tls" rate_limiter "github.com/go-gost/x/limiter/rate" + cache_limiter "github.com/go-gost/x/limiter/traffic/cache" xstats "github.com/go-gost/x/observer/stats" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" @@ -73,10 +73,10 @@ func (h *relayHandler) Init(md md.Metadata) (err error) { } if h.options.Limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(h.options.Limiter, - limiter_util.RefreshIntervalOption(h.md.limiterRefreshInterval), - limiter_util.CleanupIntervalOption(h.md.limiterCleanupInterval), - limiter_util.ScopeOption(limiter.ScopeClient), + h.limiter = cache_limiter.NewCachedTrafficLimiter(h.options.Limiter, + cache_limiter.RefreshIntervalOption(h.md.limiterRefreshInterval), + cache_limiter.CleanupIntervalOption(h.md.limiterCleanupInterval), + cache_limiter.ScopeOption(limiter.ScopeClient), ) } diff --git a/handler/router/associate.go b/handler/router/associate.go index ddc7b357..4a7caf6c 100644 --- a/handler/router/associate.go +++ b/handler/router/associate.go @@ -19,6 +19,7 @@ import ( "github.com/go-gost/core/sd" "github.com/go-gost/relay" xip "github.com/go-gost/x/internal/net/ip" + "github.com/go-gost/x/internal/util/cache" traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" "github.com/go-gost/x/registry" @@ -189,25 +190,7 @@ func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID return nil } - if h.md.sd == nil { - return nil - } - - clientID := fmt.Sprintf("%s@%s", route.Gateway, routerID) - ss, _ := h.md.sd.Get(ctx, clientID) - - var service *sd.Service - for _, s := range ss { - if s.Node != h.id { - service = s - break - } - } - if service == nil { - return nil - } - - raddr, _ := net.ResolveUDPAddr("udp", service.Address) + raddr := h.getAddrforRoute(ctx, rid, route.Gateway) if raddr == nil { return nil } @@ -235,6 +218,32 @@ func (h *routerHandler) handlePacket(ctx context.Context, data []byte, routerID return nil } +func (h *routerHandler) getAddrforRoute(ctx context.Context, routerID, gateway string) net.Addr { + if h.md.sd == nil { + return nil + } + clientID := fmt.Sprintf("%s@%s", gateway, routerID) + + if item := h.sdCache.Get(clientID); item != nil && !item.Expired() { + addr, _ := item.Value().(net.Addr) + return addr + } + + ss, _ := h.md.sd.Get(ctx, clientID) + + service := &sd.Service{} + for _, s := range ss { + if s.Node != h.id { + service = s + break + } + } + raddr, _ := net.ResolveUDPAddr("udp", service.Address) + h.sdCache.Set(clientID, cache.NewItem(raddr, h.md.sdCacheExpiration)) + + return raddr +} + type packetConn struct { net.Conn } diff --git a/handler/router/handler.go b/handler/router/handler.go index 91e0d2cd..21bc2280 100644 --- a/handler/router/handler.go +++ b/handler/router/handler.go @@ -17,9 +17,10 @@ import ( "github.com/go-gost/relay" ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" + "github.com/go-gost/x/internal/util/cache" stats_util "github.com/go-gost/x/internal/util/stats" rate_limiter "github.com/go-gost/x/limiter/rate" + cache_limiter "github.com/go-gost/x/limiter/traffic/cache" "github.com/go-gost/x/registry" "github.com/google/uuid" ) @@ -45,6 +46,7 @@ type routerHandler struct { stats *stats_util.HandlerStats limiter traffic.TrafficLimiter cancel context.CancelFunc + sdCache *cache.Cache } func NewHandler(opts ...handler.Option) handler.Handler { @@ -55,6 +57,7 @@ func NewHandler(opts ...handler.Option) handler.Handler { return &routerHandler{ options: options, + sdCache: cache.NewCache(time.Minute), } } @@ -87,10 +90,10 @@ func (h *routerHandler) Init(md md.Metadata) (err error) { } if h.options.Limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(h.options.Limiter, - limiter_util.RefreshIntervalOption(h.md.limiterRefreshInterval), - limiter_util.CleanupIntervalOption(h.md.limiterCleanupInterval), - limiter_util.ScopeOption(limiter.ScopeClient), + h.limiter = cache_limiter.NewCachedTrafficLimiter(h.options.Limiter, + cache_limiter.RefreshIntervalOption(h.md.limiterRefreshInterval), + cache_limiter.CleanupIntervalOption(h.md.limiterCleanupInterval), + cache_limiter.ScopeOption(limiter.ScopeClient), ) } diff --git a/handler/router/metadata.go b/handler/router/metadata.go index d9b198eb..35d8b5b8 100644 --- a/handler/router/metadata.go +++ b/handler/router/metadata.go @@ -12,18 +12,20 @@ import ( ) const ( - defaultTTL = 15 * time.Second - defaultBufferSize = 4096 + defaultTTL = 15 * time.Second + defaultBufferSize = 4096 + defaultCacheExpiration = time.Second ) type metadata struct { readTimeout time.Duration bufferSize int - entryPoint string - ingress ingress.Ingress - sd sd.SD - router router.Router + entryPoint string + ingress ingress.Ingress + sd sd.SD + sdCacheExpiration time.Duration + router router.Router observerPeriod time.Duration observerResetTraffic bool @@ -41,7 +43,13 @@ func (h *routerHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.entryPoint = mdutil.GetString(md, "entrypoint") 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") + if h.md.sdCacheExpiration <= 0 { + h.md.sdCacheExpiration = defaultCacheExpiration + } + h.md.router = registry.RouterRegistry().Get(mdutil.GetString(md, "router")) h.md.observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period", "observer.observePeriod") diff --git a/handler/socks/v4/handler.go b/handler/socks/v4/handler.go index 86d7e3e7..8cc9e8c4 100644 --- a/handler/socks/v4/handler.go +++ b/handler/socks/v4/handler.go @@ -20,11 +20,11 @@ import ( "github.com/go-gost/gosocks4" ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" "github.com/go-gost/x/internal/util/sniffing" stats_util "github.com/go-gost/x/internal/util/stats" tls_util "github.com/go-gost/x/internal/util/tls" rate_limiter "github.com/go-gost/x/limiter/rate" + cache_limiter "github.com/go-gost/x/limiter/traffic/cache" traffic_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" xstats "github.com/go-gost/x/observer/stats" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" @@ -77,10 +77,10 @@ func (h *socks4Handler) Init(md md.Metadata) (err error) { } if h.options.Limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(h.options.Limiter, - limiter_util.RefreshIntervalOption(h.md.limiterRefreshInterval), - limiter_util.CleanupIntervalOption(h.md.limiterCleanupInterval), - limiter_util.ScopeOption(limiter.ScopeClient), + h.limiter = cache_limiter.NewCachedTrafficLimiter(h.options.Limiter, + cache_limiter.RefreshIntervalOption(h.md.limiterRefreshInterval), + cache_limiter.CleanupIntervalOption(h.md.limiterCleanupInterval), + cache_limiter.ScopeOption(limiter.ScopeClient), ) } diff --git a/handler/socks/v5/handler.go b/handler/socks/v5/handler.go index 6a6c4302..8d7f6955 100644 --- a/handler/socks/v5/handler.go +++ b/handler/socks/v5/handler.go @@ -15,11 +15,11 @@ import ( "github.com/go-gost/core/recorder" "github.com/go-gost/gosocks5" ctxvalue "github.com/go-gost/x/ctx" - limiter_util "github.com/go-gost/x/internal/util/limiter" "github.com/go-gost/x/internal/util/socks" stats_util "github.com/go-gost/x/internal/util/stats" tls_util "github.com/go-gost/x/internal/util/tls" rate_limiter "github.com/go-gost/x/limiter/rate" + cache_limiter "github.com/go-gost/x/limiter/traffic/cache" xstats "github.com/go-gost/x/observer/stats" stats_wrapper "github.com/go-gost/x/observer/stats/wrapper" xrecorder "github.com/go-gost/x/recorder" @@ -78,10 +78,10 @@ func (h *socks5Handler) Init(md md.Metadata) (err error) { } if h.options.Limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(h.options.Limiter, - limiter_util.RefreshIntervalOption(h.md.limiterRefreshInterval), - limiter_util.CleanupIntervalOption(h.md.limiterCleanupInterval), - limiter_util.ScopeOption(limiter.ScopeClient), + h.limiter = cache_limiter.NewCachedTrafficLimiter(h.options.Limiter, + cache_limiter.RefreshIntervalOption(h.md.limiterRefreshInterval), + cache_limiter.CleanupIntervalOption(h.md.limiterCleanupInterval), + cache_limiter.ScopeOption(limiter.ScopeClient), ) } diff --git a/handler/tunnel/handler.go b/handler/tunnel/handler.go index 7cbc675c..45e34b31 100644 --- a/handler/tunnel/handler.go +++ b/handler/tunnel/handler.go @@ -20,9 +20,9 @@ import ( "github.com/go-gost/relay" ctxvalue "github.com/go-gost/x/ctx" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" stats_util "github.com/go-gost/x/internal/util/stats" rate_limiter "github.com/go-gost/x/limiter/rate" + cache_limiter "github.com/go-gost/x/limiter/traffic/cache" xrecorder "github.com/go-gost/x/recorder" "github.com/go-gost/x/registry" xservice "github.com/go-gost/x/service" @@ -123,10 +123,10 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) { } if h.options.Limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(h.options.Limiter, - limiter_util.RefreshIntervalOption(h.md.limiterRefreshInterval), - limiter_util.CleanupIntervalOption(h.md.limiterCleanupInterval), - limiter_util.ScopeOption(limiter.ScopeClient), + h.limiter = cache_limiter.NewCachedTrafficLimiter(h.options.Limiter, + cache_limiter.RefreshIntervalOption(h.md.limiterRefreshInterval), + cache_limiter.CleanupIntervalOption(h.md.limiterCleanupInterval), + cache_limiter.ScopeOption(limiter.ScopeClient), ) } diff --git a/internal/util/limiter/cache.go b/internal/util/cache/cache.go similarity index 98% rename from internal/util/limiter/cache.go rename to internal/util/cache/cache.go index b2174e58..7506c3d7 100644 --- a/internal/util/limiter/cache.go +++ b/internal/util/cache/cache.go @@ -1,4 +1,4 @@ -package limiter +package cache import ( "sync" diff --git a/internal/util/limiter/limiter.go b/limiter/traffic/cache/limiter.go similarity index 85% rename from internal/util/limiter/limiter.go rename to limiter/traffic/cache/limiter.go index b25a3df0..c52d685a 100644 --- a/internal/util/limiter/limiter.go +++ b/limiter/traffic/cache/limiter.go @@ -6,6 +6,7 @@ import ( "github.com/go-gost/core/limiter" "github.com/go-gost/core/limiter/traffic" + "github.com/go-gost/x/internal/util/cache" ) const ( @@ -40,8 +41,8 @@ func ScopeOption(scope string) Option { } type cachedTrafficLimiter struct { - inLimits *Cache - outLimits *Cache + inLimits *cache.Cache + outLimits *cache.Cache limiter traffic.TrafficLimiter options options } @@ -71,8 +72,8 @@ func NewCachedTrafficLimiter(limiter traffic.TrafficLimiter, opts ...Option) tra } lim := &cachedTrafficLimiter{ - inLimits: NewCache(options.cleanupInterval), - outLimits: NewCache(options.cleanupInterval), + inLimits: cache.NewCache(options.cleanupInterval), + outLimits: cache.NewCache(options.cleanupInterval), limiter: limiter, options: options, } @@ -104,11 +105,11 @@ func (p *cachedTrafficLimiter) In(ctx context.Context, key string, opts ...limit limNew = lim } if item == nil || !p.equal(lim, limNew) { - p.inLimits.Set(key, NewItem(limNew, p.options.refreshInterval)) + p.inLimits.Set(key, cache.NewItem(limNew, p.options.refreshInterval)) return limNew } - p.inLimits.Set(key, NewItem(lim, p.options.refreshInterval)) + p.inLimits.Set(key, cache.NewItem(lim, p.options.refreshInterval)) return lim } @@ -138,11 +139,11 @@ func (p *cachedTrafficLimiter) Out(ctx context.Context, key string, opts ...limi limNew = lim } if item == nil || !p.equal(lim, limNew) { - p.outLimits.Set(key, NewItem(limNew, p.options.refreshInterval)) + p.outLimits.Set(key, cache.NewItem(limNew, p.options.refreshInterval)) return limNew } - p.outLimits.Set(key, NewItem(lim, p.options.refreshInterval)) + p.outLimits.Set(key, cache.NewItem(lim, p.options.refreshInterval)) return lim }