go-gost/x
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

router handler: add cache for sd service

Browse Source
This commit is contained in:
ginuerzh
2025-02-09 16:16:44 +08:00
parent 08ad260606
commit fc12e33786
13 changed files with 98 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/parsing/router/parse.go
+3
View File
@@ -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 != "" {
config/parsing/service/parse.go
+5 -5
View File
@@ -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)),
handler/http/handler.go
+5 -5
View File
@@ -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),
)
}
handler/http2/handler.go
+5 -5
View File
@@ -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),
)
}
handler/relay/handler.go
+5 -5
View File
@@ -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),
)
}
handler/router/associate.go
+28 -19
View File
@@ -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
}
handler/router/handler.go
+8 -5
View File
@@ -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),
)
}
handler/router/metadata.go
+14 -6
View File
@@ -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")
handler/socks/v4/handler.go
+5 -5
View File
@@ -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),
)
}
handler/socks/v5/handler.go
+5 -5
View File
@@ -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),
)
}
handler/tunnel/handler.go
+5 -5
View File
@@ -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),
)
}
internal/util/limiter/cache.go → internal/util/cache/cache.go
Vendored
+1 -1
View File
@@ -1,4 +1,4 @@
package limiter
package cache
import (
"sync"
internal/util/limiter/limiter.go → limiter/traffic/cache/limiter.go
Vendored
+9 -8
View File
@@ -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
}