From 379d9a73ad609f0a61c704c1f6d4d8eb3e8eda02 Mon Sep 17 00:00:00 2001 From: ginuerzh Date: Mon, 20 Jan 2025 23:22:58 +0800 Subject: [PATCH] service: add limiter.scope option --- config/cmd/cmd.go | 2 +- config/parsing/parse.go | 12 ++++ config/parsing/service/parse.go | 30 +++++++-- handler/http/handler.go | 8 ++- handler/http/metadata.go | 33 +++++----- handler/http2/handler.go | 8 ++- handler/http2/metadata.go | 21 +++---- handler/relay/handler.go | 9 ++- handler/relay/metadata.go | 15 ++--- handler/socks/v4/handler.go | 8 ++- handler/socks/v4/metadata.go | 15 ++--- handler/socks/v5/handler.go | 9 ++- handler/socks/v5/metadata.go | 31 +++++----- handler/tunnel/handler.go | 9 ++- handler/tunnel/metadata.go | 11 ++-- internal/util/limiter/limiter.go | 94 +++++++++++++++++++++++++---- limiter/traffic/traffic.go | 28 ++++----- limiter/traffic/wrapper/listener.go | 3 +- listener/dns/listener.go | 27 +++------ listener/dns/metadata.go | 21 ++----- listener/dtls/listener.go | 10 +-- listener/dtls/metadata.go | 15 +---- listener/ftcp/listener.go | 10 +-- listener/ftcp/metadata.go | 11 +--- listener/grpc/listener.go | 11 ++-- listener/grpc/metadata.go | 9 --- listener/http2/h2/listener.go | 9 +-- listener/http2/h2/metadata.go | 17 +----- listener/http2/listener.go | 10 +-- listener/http2/metadata.go | 15 +---- listener/http3/h3/listener.go | 4 +- listener/http3/h3/metadata.go | 10 --- listener/http3/wt/listener.go | 9 ++- listener/http3/wt/metadata.go | 10 --- listener/icmp/listener.go | 9 ++- listener/icmp/metadata.go | 9 --- listener/kcp/listener.go | 8 +-- listener/kcp/metadata.go | 14 +---- listener/mtcp/listener.go | 9 +-- listener/mtcp/metadata.go | 17 +----- listener/mtls/listener.go | 10 +-- listener/mtls/metadata.go | 17 +----- listener/mws/listener.go | 10 +-- listener/mws/metadata.go | 11 +--- listener/obfs/http/listener.go | 9 +-- listener/obfs/http/metadata.go | 14 +---- listener/obfs/tls/listener.go | 9 +-- listener/obfs/tls/metadata.go | 13 +--- listener/pht/listener.go | 4 +- listener/pht/metadata.go | 20 ++---- listener/quic/listener.go | 9 ++- listener/quic/metadata.go | 13 +--- listener/redirect/tcp/listener.go | 9 +-- listener/redirect/tcp/metadata.go | 15 +---- listener/redirect/udp/listener.go | 4 +- listener/redirect/udp/metadata.go | 13 +--- listener/rtcp/listener.go | 10 +-- listener/rtcp/metadata.go | 14 +---- listener/rudp/listener.go | 12 +--- listener/rudp/metadata.go | 17 ++---- listener/serial/listener.go | 6 +- listener/serial/metadata.go | 11 +--- listener/ssh/listener.go | 9 +-- listener/ssh/metadata.go | 18 ++---- listener/sshd/listener.go | 10 +-- listener/sshd/metadata.go | 18 ++---- listener/tap/listener.go | 6 +- listener/tap/metadata.go | 12 +--- listener/tcp/listener.go | 9 +-- listener/tcp/metadata.go | 13 +--- listener/tls/listener.go | 9 +-- listener/tls/metadata.go | 13 +--- listener/tun/listener.go | 6 +- listener/tun/metadata.go | 14 +---- listener/udp/listener.go | 7 +-- listener/udp/metadata.go | 9 --- listener/unix/listener.go | 10 +-- listener/unix/metadata.go | 15 +---- listener/ws/listener.go | 10 +-- listener/ws/metadata.go | 11 +--- 80 files changed, 374 insertions(+), 685 deletions(-) diff --git a/config/cmd/cmd.go b/config/cmd/cmd.go index 058562ca..bc306f39 100644 --- a/config/cmd/cmd.go +++ b/config/cmd/cmd.go @@ -279,7 +279,7 @@ func BuildConfigFromCmd(serviceList, nodeList []string) (*config.Config, error) } if in != "" || out != "" { limiter.Limits = append(limiter.Limits, - fmt.Sprintf("%s %s %s", traffic.GlobalLimitKey, in, out)) + fmt.Sprintf("%s %s %s", traffic.ServiceLimitKey, in, out)) } if cin != "" || cout != "" { limiter.Limits = append(limiter.Limits, diff --git a/config/parsing/parse.go b/config/parsing/parse.go index fc28dc81..87b7eb49 100644 --- a/config/parsing/parse.go +++ b/config/parsing/parse.go @@ -17,4 +17,16 @@ const ( MDKeyRecorderHexdump = "hexdump" MDKeyRecorderHTTPBody = "http.body" MDKeyRecorderHTTPMaxBodySize = "http.maxBodySize" + + MDKeyLimiterRefreshInterval = "limiter.refreshInterval" + MDKeyLimiterCleanupInterval = "limiter.cleanupInterval" + MDKeyLimiterScope = "limiter.scope" + + MDKeyObserverResetTraffic = "observer.resetTraffic" + MDKeyObserverPeriod = "observer.period" + + MDKeyNetns = "netns" + MDKeyNetnsOut = "netns.out" + + MDKeyDialTimeout = "dialTimeout" ) diff --git a/config/parsing/service/parse.go b/config/parsing/service/parse.go index c160ab02..86e35033 100644 --- a/config/parsing/service/parse.go +++ b/config/parsing/service/parse.go @@ -28,6 +28,7 @@ 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" "github.com/go-gost/x/metadata" mdutil "github.com/go-gost/x/metadata/util" @@ -106,6 +107,11 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { var observerPeriod time.Duration var netnsIn, netnsOut string var dialTimeout time.Duration + + var limiterRefreshInterval time.Duration + var limiterCleanupInterval time.Duration + var limiterScope string + if cfg.Metadata != nil { md := metadata.NewMetadata(cfg.Metadata) ppv = mdutil.GetInt(md, parsing.MDKeyProxyProtocol) @@ -124,13 +130,18 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { ignoreChain = mdutil.GetBool(md, parsing.MDKeyIgnoreChain) if mdutil.GetBool(md, parsing.MDKeyEnableStats) { - pStats = xstats.NewStats(mdutil.GetBool(md, "observer.resetTraffic")) + pStats = xstats.NewStats(mdutil.GetBool(md, parsing.MDKeyObserverResetTraffic)) } - observerPeriod = mdutil.GetDuration(md, "observePeriod", "observer.period") + observerPeriod = mdutil.GetDuration(md, parsing.MDKeyObserverPeriod, "observePeriod") - netnsIn = mdutil.GetString(md, "netns") - netnsOut = mdutil.GetString(md, "netns.out") - dialTimeout = mdutil.GetDuration(md, "dialTimeout") + netnsIn = mdutil.GetString(md, parsing.MDKeyNetns) + netnsOut = mdutil.GetString(md, parsing.MDKeyNetnsOut) + + dialTimeout = mdutil.GetDuration(md, parsing.MDKeyDialTimeout) + + limiterRefreshInterval = mdutil.GetDuration(md, parsing.MDKeyLimiterRefreshInterval) + limiterCleanupInterval = mdutil.GetDuration(md, parsing.MDKeyLimiterCleanupInterval) + limiterScope = mdutil.GetString(md, parsing.MDKeyLimiterScope) } listenerLogger := serviceLogger.WithFields(map[string]any{ @@ -159,7 +170,14 @@ func ParseService(cfg *config.ServiceConfig) (service.Service, error) { listener.AuthOption(auth_parser.Info(cfg.Listener.Auth)), listener.TLSConfigOption(tlsConfig), listener.AdmissionOption(xadmission.AdmissionGroup(admissions...)), - listener.TrafficLimiterOption(registry.TrafficLimiterRegistry().Get(cfg.Limiter)), + listener.TrafficLimiterOption( + limiter_util.NewCachedTrafficLimiter( + registry.TrafficLimiterRegistry().Get(cfg.Limiter), + limiter_util.RefreshIntervalOption(limiterRefreshInterval), + limiter_util.CleanupIntervalOption(limiterCleanupInterval), + limiter_util.ScopeOption(limiterScope), + ), + ), listener.ConnLimiterOption(registry.ConnLimiterRegistry().Get(cfg.CLimiter)), listener.ServiceOption(cfg.Name), listener.ProxyProtocolOption(ppv), diff --git a/handler/http/handler.go b/handler/http/handler.go index db522b51..0491d9af 100644 --- a/handler/http/handler.go +++ b/handler/http/handler.go @@ -89,8 +89,12 @@ func (h *httpHandler) Init(md md.Metadata) error { go h.observeStats(ctx) } - if limiter := h.options.Limiter; limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, h.md.limiterRefreshInterval, 60*time.Second) + 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), + ) } for _, ro := range h.options.Recorders { diff --git a/handler/http/metadata.go b/handler/http/metadata.go index e0cf247c..d1ed5654 100644 --- a/handler/http/metadata.go +++ b/handler/http/metadata.go @@ -20,16 +20,15 @@ const ( ) type metadata struct { - readTimeout time.Duration - keepalive bool - compression bool - probeResistance *probeResistance - enableUDP bool - header http.Header - hash string - authBasicRealm string - proxyAgent string - limiterRefreshInterval time.Duration + readTimeout time.Duration + keepalive bool + compression bool + probeResistance *probeResistance + enableUDP bool + header http.Header + hash string + authBasicRealm string + proxyAgent string observerPeriod time.Duration observerResetTraffic bool @@ -43,6 +42,9 @@ type metadata struct { privateKey crypto.PrivateKey alpn string mitmBypass bypass.Bypass + + limiterRefreshInterval time.Duration + limiterCleanupInterval time.Duration } func (h *httpHandler) parseMetadata(md mdata.Metadata) error { @@ -85,14 +87,6 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error { h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic") - h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if h.md.limiterRefreshInterval == 0 { - h.md.limiterRefreshInterval = 30 * time.Second - } - if h.md.limiterRefreshInterval < time.Second { - h.md.limiterRefreshInterval = time.Second - } - h.md.proxyAgent = mdutil.GetString(md, "http.proxyAgent", "proxyAgent") if h.md.proxyAgent == "" { h.md.proxyAgent = defaultProxyAgent @@ -119,6 +113,9 @@ func (h *httpHandler) parseMetadata(md mdata.Metadata) error { h.md.alpn = mdutil.GetString(md, "mitm.alpn") h.md.mitmBypass = registry.BypassRegistry().Get(mdutil.GetString(md, "mitm.bypass")) + h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") + h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval") + return nil } diff --git a/handler/http2/handler.go b/handler/http2/handler.go index 958abd27..3bf105d8 100644 --- a/handler/http2/handler.go +++ b/handler/http2/handler.go @@ -77,8 +77,12 @@ func (h *http2Handler) Init(md md.Metadata) error { go h.observeStats(ctx) } - if limiter := h.options.Limiter; limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, h.md.limiterRefreshInterval, 60*time.Second) + 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), + ) } for _, ro := range h.options.Recorders { diff --git a/handler/http2/metadata.go b/handler/http2/metadata.go index 539390b4..f7aeaf79 100644 --- a/handler/http2/metadata.go +++ b/handler/http2/metadata.go @@ -14,13 +14,15 @@ const ( ) type metadata struct { - probeResistance *probeResistance - header http.Header - hash string - authBasicRealm string - observerPeriod time.Duration - observerResetTraffic bool + probeResistance *probeResistance + header http.Header + hash string + authBasicRealm string + observerPeriod time.Duration + observerResetTraffic bool + limiterRefreshInterval time.Duration + limiterCleanupInterval time.Duration } func (h *http2Handler) parseMetadata(md mdata.Metadata) error { @@ -54,12 +56,7 @@ func (h *http2Handler) parseMetadata(md mdata.Metadata) error { h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic") h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if h.md.limiterRefreshInterval == 0 { - h.md.limiterRefreshInterval = 30 * time.Second - } - if h.md.limiterRefreshInterval < time.Second { - h.md.limiterRefreshInterval = time.Second - } + h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval") return nil } diff --git a/handler/relay/handler.go b/handler/relay/handler.go index f062c342..0444d0aa 100644 --- a/handler/relay/handler.go +++ b/handler/relay/handler.go @@ -9,6 +9,7 @@ import ( "github.com/go-gost/core/handler" "github.com/go-gost/core/hop" + "github.com/go-gost/core/limiter" "github.com/go-gost/core/limiter/traffic" md "github.com/go-gost/core/metadata" "github.com/go-gost/core/observer" @@ -71,8 +72,12 @@ func (h *relayHandler) Init(md md.Metadata) (err error) { go h.observeStats(ctx) } - if limiter := h.options.Limiter; limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, h.md.limiterRefreshInterval, 60*time.Second) + 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), + ) } for _, ro := range h.options.Recorders { diff --git a/handler/relay/metadata.go b/handler/relay/metadata.go index da714168..0cab5a3b 100644 --- a/handler/relay/metadata.go +++ b/handler/relay/metadata.go @@ -21,7 +21,6 @@ type metadata struct { noDelay bool hash string muxCfg *mux.Config - limiterRefreshInterval time.Duration observerPeriod time.Duration observerResetTraffic bool @@ -35,6 +34,9 @@ type metadata struct { privateKey crypto.PrivateKey alpn string mitmBypass bypass.Bypass + + limiterRefreshInterval time.Duration + limiterCleanupInterval time.Duration } func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) { @@ -74,14 +76,6 @@ func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic") - h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if h.md.limiterRefreshInterval == 0 { - h.md.limiterRefreshInterval = 30 * time.Second - } - if h.md.limiterRefreshInterval < time.Second { - h.md.limiterRefreshInterval = time.Second - } - h.md.sniffing = mdutil.GetBool(md, "sniffing") h.md.sniffingTimeout = mdutil.GetDuration(md, "sniffing.timeout") h.md.sniffingWebsocket = mdutil.GetBool(md, "sniffing.websocket") @@ -103,5 +97,8 @@ func (h *relayHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.alpn = mdutil.GetString(md, "mitm.alpn") h.md.mitmBypass = registry.BypassRegistry().Get(mdutil.GetString(md, "mitm.bypass")) + h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") + h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval") + return } diff --git a/handler/socks/v4/handler.go b/handler/socks/v4/handler.go index b14b08c0..86d7e3e7 100644 --- a/handler/socks/v4/handler.go +++ b/handler/socks/v4/handler.go @@ -76,8 +76,12 @@ func (h *socks4Handler) Init(md md.Metadata) (err error) { go h.observeStats(ctx) } - if limiter := h.options.Limiter; limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, h.md.limiterRefreshInterval, 60*time.Second) + 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), + ) } for _, ro := range h.options.Recorders { diff --git a/handler/socks/v4/metadata.go b/handler/socks/v4/metadata.go index 6c3fabbe..b3e7550c 100644 --- a/handler/socks/v4/metadata.go +++ b/handler/socks/v4/metadata.go @@ -15,7 +15,6 @@ import ( type metadata struct { readTimeout time.Duration hash string - limiterRefreshInterval time.Duration observerPeriod time.Duration observerResetTraffic bool @@ -29,6 +28,9 @@ type metadata struct { privateKey crypto.PrivateKey alpn string mitmBypass bypass.Bypass + + limiterRefreshInterval time.Duration + limiterCleanupInterval time.Duration } func (h *socks4Handler) parseMetadata(md mdata.Metadata) (err error) { @@ -49,14 +51,6 @@ func (h *socks4Handler) parseMetadata(md mdata.Metadata) (err error) { h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic") - h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if h.md.limiterRefreshInterval == 0 { - h.md.limiterRefreshInterval = 30 * time.Second - } - if h.md.limiterRefreshInterval < time.Second { - h.md.limiterRefreshInterval = time.Second - } - h.md.sniffing = mdutil.GetBool(md, "sniffing") h.md.sniffingTimeout = mdutil.GetDuration(md, "sniffing.timeout") h.md.sniffingWebsocket = mdutil.GetBool(md, "sniffing.websocket") @@ -78,5 +72,8 @@ func (h *socks4Handler) parseMetadata(md mdata.Metadata) (err error) { h.md.alpn = mdutil.GetString(md, "mitm.alpn") h.md.mitmBypass = registry.BypassRegistry().Get(mdutil.GetString(md, "mitm.bypass")) + h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") + h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval") + return } diff --git a/handler/socks/v5/handler.go b/handler/socks/v5/handler.go index affa27fd..6a6c4302 100644 --- a/handler/socks/v5/handler.go +++ b/handler/socks/v5/handler.go @@ -7,6 +7,7 @@ import ( "time" "github.com/go-gost/core/handler" + "github.com/go-gost/core/limiter" "github.com/go-gost/core/limiter/traffic" md "github.com/go-gost/core/metadata" "github.com/go-gost/core/observer" @@ -76,8 +77,12 @@ func (h *socks5Handler) Init(md md.Metadata) (err error) { go h.observeStats(ctx) } - if limiter := h.options.Limiter; limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, h.md.limiterRefreshInterval, 60*time.Second) + 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), + ) } for _, ro := range h.options.Recorders { diff --git a/handler/socks/v5/metadata.go b/handler/socks/v5/metadata.go index a218f560..2fb17111 100644 --- a/handler/socks/v5/metadata.go +++ b/handler/socks/v5/metadata.go @@ -15,15 +15,14 @@ import ( ) type metadata struct { - readTimeout time.Duration - noTLS bool - enableBind bool - enableUDP bool - udpBufferSize int - compatibilityMode bool - hash string - muxCfg *mux.Config - limiterRefreshInterval time.Duration + readTimeout time.Duration + noTLS bool + enableBind bool + enableUDP bool + udpBufferSize int + compatibilityMode bool + hash string + muxCfg *mux.Config observerPeriod time.Duration observerResetTraffic bool @@ -37,6 +36,9 @@ type metadata struct { privateKey crypto.PrivateKey alpn string mitmBypass bypass.Bypass + + limiterRefreshInterval time.Duration + limiterCleanupInterval time.Duration } func (h *socks5Handler) parseMetadata(md mdata.Metadata) (err error) { @@ -77,14 +79,6 @@ func (h *socks5Handler) parseMetadata(md mdata.Metadata) (err error) { } h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic") - h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if h.md.limiterRefreshInterval == 0 { - h.md.limiterRefreshInterval = 30 * time.Second - } - if h.md.limiterRefreshInterval < time.Second { - h.md.limiterRefreshInterval = time.Second - } - h.md.sniffing = mdutil.GetBool(md, "sniffing") h.md.sniffingTimeout = mdutil.GetDuration(md, "sniffing.timeout") h.md.sniffingWebsocket = mdutil.GetBool(md, "sniffing.websocket") @@ -106,5 +100,8 @@ func (h *socks5Handler) parseMetadata(md mdata.Metadata) (err error) { h.md.alpn = mdutil.GetString(md, "mitm.alpn") h.md.mitmBypass = registry.BypassRegistry().Get(mdutil.GetString(md, "mitm.bypass")) + h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") + h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval") + return nil } diff --git a/handler/tunnel/handler.go b/handler/tunnel/handler.go index 13bbcad8..7cbc675c 100644 --- a/handler/tunnel/handler.go +++ b/handler/tunnel/handler.go @@ -10,6 +10,7 @@ import ( "time" "github.com/go-gost/core/handler" + "github.com/go-gost/core/limiter" "github.com/go-gost/core/limiter/traffic" "github.com/go-gost/core/listener" "github.com/go-gost/core/logger" @@ -121,8 +122,12 @@ func (h *tunnelHandler) Init(md md.Metadata) (err error) { go h.observeStats(ctx) } - if limiter := h.options.Limiter; limiter != nil { - h.limiter = limiter_util.NewCachedTrafficLimiter(limiter, h.md.limiterRefreshInterval, 60*time.Second) + 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), + ) } return nil diff --git a/handler/tunnel/metadata.go b/handler/tunnel/metadata.go index 97818149..023aed77 100644 --- a/handler/tunnel/metadata.go +++ b/handler/tunnel/metadata.go @@ -36,10 +36,12 @@ type metadata struct { ingress ingress.Ingress sd sd.SD muxCfg *mux.Config - limiterRefreshInterval time.Duration observerPeriod time.Duration observerResetTraffic bool + + limiterRefreshInterval time.Duration + limiterCleanupInterval time.Duration } func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) { @@ -117,12 +119,7 @@ func (h *tunnelHandler) parseMetadata(md mdata.Metadata) (err error) { h.md.observerResetTraffic = mdutil.GetBool(md, "observer.resetTraffic") h.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if h.md.limiterRefreshInterval == 0 { - h.md.limiterRefreshInterval = 30 * time.Second - } - if h.md.limiterRefreshInterval < time.Second { - h.md.limiterRefreshInterval = time.Second - } + h.md.limiterCleanupInterval = mdutil.GetDuration(md, "limiter.cleanupInterval") return } diff --git a/internal/util/limiter/limiter.go b/internal/util/limiter/limiter.go index b488898a..b25a3df0 100644 --- a/internal/util/limiter/limiter.go +++ b/internal/util/limiter/limiter.go @@ -8,23 +8,73 @@ import ( "github.com/go-gost/core/limiter/traffic" ) -type cachedTrafficLimiter struct { - inLimits *Cache - outLimits *Cache - limiter traffic.TrafficLimiter +const ( + defaultRefreshInterval = 30 * time.Second + defaultCleanupInterval = 60 * time.Second +) + +type options struct { refreshInterval time.Duration + cleanupInterval time.Duration + scope string } -func NewCachedTrafficLimiter(limiter traffic.TrafficLimiter, refreshInterval time.Duration, cleanupInterval time.Duration) traffic.TrafficLimiter { +type Option func(*options) + +func RefreshIntervalOption(interval time.Duration) Option { + return func(o *options) { + o.refreshInterval = interval + } +} + +func CleanupIntervalOption(interval time.Duration) Option { + return func(o *options) { + o.cleanupInterval = interval + } +} + +func ScopeOption(scope string) Option { + return func(o *options) { + o.scope = scope + } +} + +type cachedTrafficLimiter struct { + inLimits *Cache + outLimits *Cache + limiter traffic.TrafficLimiter + options options +} + +func NewCachedTrafficLimiter(limiter traffic.TrafficLimiter, opts ...Option) traffic.TrafficLimiter { if limiter == nil { return nil } + var options options + for _, opt := range opts { + opt(&options) + } + + if options.refreshInterval == 0 { + options.refreshInterval = defaultRefreshInterval + } + if options.refreshInterval < time.Second { + options.refreshInterval = time.Second + } + + if options.cleanupInterval == 0 { + options.cleanupInterval = defaultCleanupInterval + } + if options.cleanupInterval < time.Second { + options.cleanupInterval = time.Second + } + lim := &cachedTrafficLimiter{ - inLimits: NewCache(cleanupInterval), - outLimits: NewCache(cleanupInterval), - limiter: limiter, - refreshInterval: refreshInterval, + inLimits: NewCache(options.cleanupInterval), + outLimits: NewCache(options.cleanupInterval), + limiter: limiter, + options: options, } return lim } @@ -34,6 +84,15 @@ func (p *cachedTrafficLimiter) In(ctx context.Context, key string, opts ...limit return nil } + var options limiter.Options + for _, opt := range opts { + opt(&options) + } + + if p.options.scope != "" && p.options.scope != options.Scope { + return nil + } + item := p.inLimits.Get(key) lim, _ := item.Value().(traffic.Limiter) if !item.Expired() { @@ -45,11 +104,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.refreshInterval)) + p.inLimits.Set(key, NewItem(limNew, p.options.refreshInterval)) return limNew } - p.inLimits.Set(key, NewItem(lim, p.refreshInterval)) + p.inLimits.Set(key, NewItem(lim, p.options.refreshInterval)) return lim } @@ -59,6 +118,15 @@ func (p *cachedTrafficLimiter) Out(ctx context.Context, key string, opts ...limi return nil } + var options limiter.Options + for _, opt := range opts { + opt(&options) + } + + if p.options.scope != "" && p.options.scope != options.Scope { + return nil + } + item := p.outLimits.Get(key) lim, _ := item.Value().(traffic.Limiter) if !item.Expired() { @@ -70,11 +138,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.refreshInterval)) + p.outLimits.Set(key, NewItem(limNew, p.options.refreshInterval)) return limNew } - p.outLimits.Set(key, NewItem(lim, p.refreshInterval)) + p.outLimits.Set(key, NewItem(lim, p.options.refreshInterval)) return lim } diff --git a/limiter/traffic/traffic.go b/limiter/traffic/traffic.go index 2c68cfe0..6f9c2b25 100644 --- a/limiter/traffic/traffic.go +++ b/limiter/traffic/traffic.go @@ -19,8 +19,8 @@ import ( ) const ( - GlobalLimitKey = "$" - ConnLimitKey = "$$" + ServiceLimitKey = "$" + ConnLimitKey = "$$" ) const ( @@ -130,7 +130,7 @@ func (l *trafficLimiter) In(ctx context.Context, key string, opts ...limiter.Opt switch options.Scope { case limiter.ScopeService: - if lim, ok := l.inLimits.Get(GlobalLimitKey); ok && lim != nil { + if lim, ok := l.inLimits.Get(ServiceLimitKey); ok && lim != nil { return lim.(traffic.Limiter) } return nil @@ -209,7 +209,7 @@ func (l *trafficLimiter) Out(ctx context.Context, key string, opts ...limiter.Op switch options.Scope { case limiter.ScopeService: - if lim, ok := l.outLimits.Get(GlobalLimitKey); ok && lim != nil { + if lim, ok := l.outLimits.Get(ServiceLimitKey); ok && lim != nil { return lim.(traffic.Limiter) } return nil @@ -307,33 +307,33 @@ func (l *trafficLimiter) reload(ctx context.Context) error { // service level limiter, never expired { - value := values[GlobalLimitKey] - if v, _ := l.inLimits.Get(GlobalLimitKey); v != nil { + value := values[ServiceLimitKey] + if v, _ := l.inLimits.Get(ServiceLimitKey); v != nil { lim := v.(traffic.Limiter) if value.in <= 0 { - l.inLimits.Delete(GlobalLimitKey) + l.inLimits.Delete(ServiceLimitKey) } else { lim.Set(value.in) } } else { if value.in > 0 { - l.inLimits.Set(GlobalLimitKey, NewLimiter(value.in), cache.NoExpiration) + l.inLimits.Set(ServiceLimitKey, NewLimiter(value.in), cache.NoExpiration) } } - if v, _ := l.outLimits.Get(GlobalLimitKey); v != nil { + if v, _ := l.outLimits.Get(ServiceLimitKey); v != nil { lim := v.(traffic.Limiter) if value.out <= 0 { - l.outLimits.Delete(GlobalLimitKey) + l.outLimits.Delete(ServiceLimitKey) } else { lim.Set(value.out) } } else { if value.out > 0 { - l.outLimits.Set(GlobalLimitKey, NewLimiter(value.out), cache.NoExpiration) + l.outLimits.Set(ServiceLimitKey, NewLimiter(value.out), cache.NoExpiration) } } - delete(values, GlobalLimitKey) + delete(values, ServiceLimitKey) } // connection level limiters @@ -379,8 +379,8 @@ func (l *trafficLimiter) reload(ctx context.Context) error { inLimits := l.inLimits.Items() outLimits := l.outLimits.Items() - delete(inLimits, GlobalLimitKey) - delete(outLimits, GlobalLimitKey) + delete(inLimits, ServiceLimitKey) + delete(outLimits, ServiceLimitKey) for key, value := range values { if _, ipNet, _ := net.ParseCIDR(key); ipNet != nil { diff --git a/limiter/traffic/wrapper/listener.go b/limiter/traffic/wrapper/listener.go index 9b579ce7..8af59e6c 100644 --- a/limiter/traffic/wrapper/listener.go +++ b/limiter/traffic/wrapper/listener.go @@ -5,6 +5,7 @@ import ( "github.com/go-gost/core/limiter" "github.com/go-gost/core/limiter/traffic" + traffic_limiter "github.com/go-gost/x/limiter/traffic" ) type listener struct { @@ -31,7 +32,7 @@ func (ln *listener) Accept() (net.Conn, error) { return nil, err } - return WrapConn(c, ln.limiter, "", + return WrapConn(c, ln.limiter, traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(ln.service), limiter.NetworkOption(ln.Addr().Network()), diff --git a/listener/dns/listener.go b/listener/dns/listener.go index aadcb4c2..5ac48857 100644 --- a/listener/dns/listener.go +++ b/listener/dns/listener.go @@ -10,7 +10,6 @@ import ( "net" "net/http" "strings" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" @@ -18,7 +17,7 @@ import ( md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" @@ -80,11 +79,7 @@ func (l *dnsListener) Init(md md.Metadata) (err error) { return } - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) l.server = &dnsServer{ server: &dns.Server{ @@ -121,11 +116,7 @@ func (l *dnsListener) Init(md md.Metadata) (err error) { } ln = tls.NewListener(ln, l.options.TLSConfig) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) l.server = &dnsServer{ server: &dns.Server{ @@ -163,11 +154,7 @@ func (l *dnsListener) Init(md md.Metadata) (err error) { } ln = tls.NewListener(ln, l.options.TLSConfig) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) l.server = &dohServer{ addr: l.options.Addr, @@ -205,8 +192,8 @@ func (l *dnsListener) Init(md md.Metadata) (err error) { pc = limiter_wrapper.WrapPacketConn( pc, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - "", + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(network), @@ -251,7 +238,7 @@ func (l *dnsListener) Accept() (conn net.Conn, err error) { conn = admission.WrapConn(l.options.Admission, conn) conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/dns/metadata.go b/listener/dns/metadata.go index 53c277ba..47d9051f 100644 --- a/listener/dns/metadata.go +++ b/listener/dns/metadata.go @@ -12,13 +12,12 @@ const ( ) type metadata struct { - mode string - readBufferSize int - readTimeout time.Duration - writeTimeout time.Duration - backlog int - mptcp bool - limiterRefreshInterval time.Duration + mode string + readBufferSize int + readTimeout time.Duration + writeTimeout time.Duration + backlog int + mptcp bool } func (l *dnsListener) parseMetadata(md mdata.Metadata) (err error) { @@ -41,13 +40,5 @@ func (l *dnsListener) parseMetadata(md mdata.Metadata) (err error) { } l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/dtls/listener.go b/listener/dtls/listener.go index c573e9b8..4a3c08e0 100644 --- a/listener/dtls/listener.go +++ b/listener/dtls/listener.go @@ -14,7 +14,6 @@ import ( xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" xdtls "github.com/go-gost/x/internal/util/dtls" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -84,11 +83,7 @@ func (l *dtlsListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.ln = ln @@ -104,13 +99,14 @@ func (l *dtlsListener) Accept() (conn net.Conn, err error) { conn = xdtls.Conn(c, l.md.bufferSize) conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), limiter.SrcOption(conn.RemoteAddr().String()), ) + return } diff --git a/listener/dtls/metadata.go b/listener/dtls/metadata.go index 4bf7a84d..d10a52e7 100644 --- a/listener/dtls/metadata.go +++ b/listener/dtls/metadata.go @@ -12,10 +12,9 @@ const ( ) type metadata struct { - mtu int - bufferSize int - flightInterval time.Duration - limiterRefreshInterval time.Duration + mtu int + bufferSize int + flightInterval time.Duration } func (l *dtlsListener) parseMetadata(md mdata.Metadata) (err error) { @@ -27,13 +26,5 @@ func (l *dtlsListener) parseMetadata(md mdata.Metadata) (err error) { l.md.flightInterval = mdutil.GetDuration(md, "dtls.flightInterval", "flightInterval") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return nil } diff --git a/listener/ftcp/listener.go b/listener/ftcp/listener.go index d1d7cae7..79f5950b 100644 --- a/listener/ftcp/listener.go +++ b/listener/ftcp/listener.go @@ -2,7 +2,6 @@ package ftcp import ( "net" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" @@ -11,7 +10,7 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/udp" - limiter_util "github.com/go-gost/x/internal/util/limiter" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" @@ -55,13 +54,14 @@ func (l *ftcpListener) Init(md md.Metadata) (err error) { if err != nil { return } + conn = metrics.WrapPacketConn(l.options.Service, conn) conn = stats.WrapPacketConn(conn, l.options.Stats) conn = admission.WrapPacketConn(l.options.Admission, conn) conn = limiter_wrapper.WrapPacketConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - "", + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), @@ -88,7 +88,7 @@ func (l *ftcpListener) Accept() (conn net.Conn, err error) { conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/ftcp/metadata.go b/listener/ftcp/metadata.go index 1edb397f..e2715013 100644 --- a/listener/ftcp/metadata.go +++ b/listener/ftcp/metadata.go @@ -15,8 +15,7 @@ const ( ) type metadata struct { - ttl time.Duration - limiterRefreshInterval time.Duration + ttl time.Duration readBufferSize int readQueueSize int @@ -50,13 +49,5 @@ func (l *ftcpListener) parseMetadata(md mdata.Metadata) (err error) { l.md.backlog = defaultBacklog } - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/grpc/listener.go b/listener/grpc/listener.go index 112bd08c..65a46fcf 100644 --- a/listener/grpc/listener.go +++ b/listener/grpc/listener.go @@ -13,7 +13,6 @@ import ( xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" pb "github.com/go-gost/x/internal/util/grpc/proto" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -67,15 +66,12 @@ func (l *grpcListener) Init(md md.Metadata) (err error) { if err != nil { return } + ln = proxyproto.WrapListener(l.options.ProxyProtocol, ln, 10*time.Second) ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) var opts []grpc.ServerOption @@ -124,13 +120,14 @@ func (l *grpcListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), limiter.SrcOption(conn.RemoteAddr().String()), ) + case err, ok = <-l.errChan: if !ok { err = listener.ErrClosed diff --git a/listener/grpc/metadata.go b/listener/grpc/metadata.go index 2248fb28..3383a357 100644 --- a/listener/grpc/metadata.go +++ b/listener/grpc/metadata.go @@ -22,7 +22,6 @@ type metadata struct { keepalivePermitWithoutStream bool keepaliveMaxConnectionIdle time.Duration mptcp bool - limiterRefreshInterval time.Duration } func (l *grpcListener) parseMetadata(md mdata.Metadata) (err error) { @@ -54,13 +53,5 @@ func (l *grpcListener) parseMetadata(md mdata.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") } - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/http2/h2/listener.go b/listener/http2/h2/listener.go index b8d6641b..1aa32405 100644 --- a/listener/http2/h2/listener.go +++ b/listener/http2/h2/listener.go @@ -16,7 +16,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -92,11 +91,7 @@ func (l *h2Listener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) if l.h2c { @@ -130,7 +125,7 @@ func (l *h2Listener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/http2/h2/metadata.go b/listener/http2/h2/metadata.go index 061d7904..42672fd8 100644 --- a/listener/http2/h2/metadata.go +++ b/listener/http2/h2/metadata.go @@ -1,8 +1,6 @@ package h2 import ( - "time" - mdata "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/x/metadata/util" ) @@ -12,10 +10,9 @@ const ( ) type metadata struct { - path string - backlog int - mptcp bool - limiterRefreshInterval time.Duration + path string + backlog int + mptcp bool } func (l *h2Listener) parseMetadata(md mdata.Metadata) (err error) { @@ -32,13 +29,5 @@ func (l *h2Listener) parseMetadata(md mdata.Metadata) (err error) { l.md.path = mdutil.GetString(md, path) l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/http2/listener.go b/listener/http2/listener.go index 20aeb7ad..31973b3a 100644 --- a/listener/http2/listener.go +++ b/listener/http2/listener.go @@ -14,7 +14,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" mdx "github.com/go-gost/x/metadata" @@ -81,11 +80,7 @@ func (l *http2Listener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) ln = tls.NewListener( @@ -111,13 +106,14 @@ func (l *http2Listener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), limiter.SrcOption(conn.RemoteAddr().String()), ) + case err, ok = <-l.errChan: if !ok { err = listener.ErrClosed diff --git a/listener/http2/metadata.go b/listener/http2/metadata.go index d03aa0cc..5e300153 100644 --- a/listener/http2/metadata.go +++ b/listener/http2/metadata.go @@ -1,8 +1,6 @@ package http2 import ( - "time" - mdata "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/x/metadata/util" ) @@ -12,9 +10,8 @@ const ( ) type metadata struct { - backlog int - mptcp bool - limiterRefreshInterval time.Duration + backlog int + mptcp bool } func (l *http2Listener) parseMetadata(md mdata.Metadata) (err error) { @@ -28,13 +25,5 @@ func (l *http2Listener) parseMetadata(md mdata.Metadata) (err error) { } l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/http3/h3/listener.go b/listener/http3/h3/listener.go index 0ed9ea92..cfb770c7 100644 --- a/listener/http3/h3/listener.go +++ b/listener/http3/h3/listener.go @@ -2,7 +2,6 @@ package h3 import ( "net" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" @@ -10,7 +9,6 @@ import ( md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" pht_util "github.com/go-gost/x/internal/util/pht" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -93,7 +91,7 @@ func (l *http3Listener) Accept() (conn net.Conn, err error) { conn = admission.WrapConn(l.options.Admission, conn) conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/http3/h3/metadata.go b/listener/http3/h3/metadata.go index 28930340..c743e318 100644 --- a/listener/http3/h3/metadata.go +++ b/listener/http3/h3/metadata.go @@ -26,8 +26,6 @@ type metadata struct { maxIdleTimeout time.Duration handshakeTimeout time.Duration maxStreams int - - limiterRefreshInterval time.Duration } func (l *http3Listener) parseMetadata(md mdata.Metadata) (err error) { @@ -69,13 +67,5 @@ func (l *http3Listener) parseMetadata(md mdata.Metadata) (err error) { l.md.maxIdleTimeout = mdutil.GetDuration(md, maxIdleTimeout) l.md.maxStreams = mdutil.GetInt(md, maxStreams) - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/http3/wt/listener.go b/listener/http3/wt/listener.go index e92aeffe..75e21183 100644 --- a/listener/http3/wt/listener.go +++ b/listener/http3/wt/listener.go @@ -4,7 +4,6 @@ import ( "net" "net/http" "net/http/httputil" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" @@ -12,8 +11,8 @@ import ( md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" wt_util "github.com/go-gost/x/internal/util/wt" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" @@ -79,8 +78,8 @@ func (l *wtListener) Init(md md.Metadata) (err error) { pc = admission.WrapPacketConn(l.options.Admission, pc) pc = limiter_wrapper.WrapPacketConn( pc, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - "", + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(network), @@ -130,7 +129,7 @@ func (l *wtListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/http3/wt/metadata.go b/listener/http3/wt/metadata.go index a88085e9..ecaa7449 100644 --- a/listener/http3/wt/metadata.go +++ b/listener/http3/wt/metadata.go @@ -21,8 +21,6 @@ type metadata struct { maxIdleTimeout time.Duration handshakeTimeout time.Duration maxStreams int - - limiterRefreshInterval time.Duration } func (l *wtListener) parseMetadata(md mdata.Metadata) (err error) { @@ -56,13 +54,5 @@ func (l *wtListener) parseMetadata(md mdata.Metadata) (err error) { l.md.maxIdleTimeout = mdutil.GetDuration(md, maxIdleTimeout) l.md.maxStreams = mdutil.GetInt(md, maxStreams) - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/icmp/listener.go b/listener/icmp/listener.go index 748338f5..08a3e46f 100644 --- a/listener/icmp/listener.go +++ b/listener/icmp/listener.go @@ -3,7 +3,6 @@ package quic import ( "context" "net" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" @@ -11,7 +10,7 @@ import ( md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" icmp_pkg "github.com/go-gost/x/internal/util/icmp" - limiter_util "github.com/go-gost/x/internal/util/limiter" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" @@ -83,8 +82,8 @@ func (l *icmpListener) Init(md md.Metadata) (err error) { conn = admission.WrapPacketConn(l.options.Admission, conn) conn = limiter_wrapper.WrapPacketConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - "", + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), @@ -123,7 +122,7 @@ func (l *icmpListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/icmp/metadata.go b/listener/icmp/metadata.go index 956d8393..8ed1ee8c 100644 --- a/listener/icmp/metadata.go +++ b/listener/icmp/metadata.go @@ -17,7 +17,6 @@ type metadata struct { maxIdleTimeout time.Duration backlog int - limiterRefreshInterval time.Duration } func (l *icmpListener) parseMetadata(md mdata.Metadata) (err error) { @@ -35,13 +34,5 @@ func (l *icmpListener) parseMetadata(md mdata.Metadata) (err error) { l.md.handshakeTimeout = mdutil.GetDuration(md, "handshakeTimeout") l.md.maxIdleTimeout = mdutil.GetDuration(md, "maxIdleTimeout") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/kcp/listener.go b/listener/kcp/listener.go index 89e61932..71dd0531 100644 --- a/listener/kcp/listener.go +++ b/listener/kcp/listener.go @@ -11,7 +11,7 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" kcp_util "github.com/go-gost/x/internal/util/kcp" - limiter_util "github.com/go-gost/x/internal/util/limiter" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" @@ -82,8 +82,8 @@ func (l *kcpListener) Init(md md.Metadata) (err error) { conn = admission.WrapUDPConn(l.options.Admission, conn) conn = limiter_wrapper.WrapUDPConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - "", + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), @@ -124,7 +124,7 @@ func (l *kcpListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/kcp/metadata.go b/listener/kcp/metadata.go index bff4bfff..539d2adf 100644 --- a/listener/kcp/metadata.go +++ b/listener/kcp/metadata.go @@ -2,7 +2,6 @@ package kcp import ( "encoding/json" - "time" mdata "github.com/go-gost/core/metadata" kcp_util "github.com/go-gost/x/internal/util/kcp" @@ -14,9 +13,8 @@ const ( ) type metadata struct { - config *kcp_util.Config - backlog int - limiterRefreshInterval time.Duration + config *kcp_util.Config + backlog int } func (l *kcpListener) parseMetadata(md mdata.Metadata) (err error) { @@ -66,13 +64,5 @@ func (l *kcpListener) parseMetadata(md mdata.Metadata) (err error) { l.md.backlog = defaultBacklog } - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/mtcp/listener.go b/listener/mtcp/listener.go index 7e0dcf4e..c90b00ee 100644 --- a/listener/mtcp/listener.go +++ b/listener/mtcp/listener.go @@ -12,7 +12,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" "github.com/go-gost/x/internal/util/mux" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" @@ -71,11 +70,7 @@ func (l *mtcpListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.ln = ln @@ -101,7 +96,7 @@ func (l *mtcpListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/mtcp/metadata.go b/listener/mtcp/metadata.go index 3102f844..8b8db627 100644 --- a/listener/mtcp/metadata.go +++ b/listener/mtcp/metadata.go @@ -1,8 +1,6 @@ package mtcp import ( - "time" - md "github.com/go-gost/core/metadata" "github.com/go-gost/x/internal/util/mux" mdutil "github.com/go-gost/x/metadata/util" @@ -13,10 +11,9 @@ const ( ) type metadata struct { - mptcp bool - muxCfg *mux.Config - backlog int - limiterRefreshInterval time.Duration + mptcp bool + muxCfg *mux.Config + backlog int } func (l *mtcpListener) parseMetadata(md md.Metadata) (err error) { @@ -40,13 +37,5 @@ func (l *mtcpListener) parseMetadata(md md.Metadata) (err error) { l.md.backlog = defaultBacklog } - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/mtls/listener.go b/listener/mtls/listener.go index 76bd45d3..8aa78ead 100644 --- a/listener/mtls/listener.go +++ b/listener/mtls/listener.go @@ -13,7 +13,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" "github.com/go-gost/x/internal/util/mux" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" @@ -70,11 +69,7 @@ func (l *mtlsListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.Listener = tls.NewListener(ln, l.options.TLSConfig) @@ -92,13 +87,14 @@ func (l *mtlsListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), limiter.SrcOption(conn.RemoteAddr().String()), ) + case err, ok = <-l.errChan: if !ok { err = listener.ErrClosed diff --git a/listener/mtls/metadata.go b/listener/mtls/metadata.go index 4365538f..6f6d8432 100644 --- a/listener/mtls/metadata.go +++ b/listener/mtls/metadata.go @@ -1,8 +1,6 @@ package mtls import ( - "time" - mdata "github.com/go-gost/core/metadata" "github.com/go-gost/x/internal/util/mux" mdutil "github.com/go-gost/x/metadata/util" @@ -13,10 +11,9 @@ const ( ) type metadata struct { - muxCfg *mux.Config - backlog int - mptcp bool - limiterRefreshInterval time.Duration + muxCfg *mux.Config + backlog int + mptcp bool } func (l *mtlsListener) parseMetadata(md mdata.Metadata) (err error) { @@ -36,13 +33,5 @@ func (l *mtlsListener) parseMetadata(md mdata.Metadata) (err error) { } l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/mws/listener.go b/listener/mws/listener.go index cafbcb84..6626f069 100644 --- a/listener/mws/listener.go +++ b/listener/mws/listener.go @@ -15,7 +15,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" "github.com/go-gost/x/internal/util/mux" ws_util "github.com/go-gost/x/internal/util/ws" climiter "github.com/go-gost/x/limiter/conn/wrapper" @@ -112,11 +111,7 @@ func (l *mwsListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) if l.tlsEnabled { @@ -142,13 +137,14 @@ func (l *mwsListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), limiter.SrcOption(conn.RemoteAddr().String()), ) + case err, ok = <-l.errChan: if !ok { err = listener.ErrClosed diff --git a/listener/mws/metadata.go b/listener/mws/metadata.go index dd35bf16..b2fafab0 100644 --- a/listener/mws/metadata.go +++ b/listener/mws/metadata.go @@ -27,8 +27,7 @@ type metadata struct { muxCfg *mux.Config - mptcp bool - limiterRefreshInterval time.Duration + mptcp bool } func (l *mwsListener) parseMetadata(md mdata.Metadata) (err error) { @@ -68,13 +67,5 @@ func (l *mwsListener) parseMetadata(md mdata.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/obfs/http/listener.go b/listener/obfs/http/listener.go index d9996532..a7e87dce 100644 --- a/listener/obfs/http/listener.go +++ b/listener/obfs/http/listener.go @@ -12,7 +12,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -65,11 +64,7 @@ func (l *obfsListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.Listener = ln @@ -84,7 +79,7 @@ func (l *obfsListener) Accept() (net.Conn, error) { conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/obfs/http/metadata.go b/listener/obfs/http/metadata.go index 9ca6baf6..e620089e 100644 --- a/listener/obfs/http/metadata.go +++ b/listener/obfs/http/metadata.go @@ -2,16 +2,14 @@ package http import ( "net/http" - "time" mdata "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/x/metadata/util" ) type metadata struct { - header http.Header - mptcp bool - limiterRefreshInterval time.Duration + header http.Header + mptcp bool } func (l *obfsListener) parseMetadata(md mdata.Metadata) (err error) { @@ -29,13 +27,5 @@ func (l *obfsListener) parseMetadata(md mdata.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/obfs/tls/listener.go b/listener/obfs/tls/listener.go index acf77495..d8025ae1 100644 --- a/listener/obfs/tls/listener.go +++ b/listener/obfs/tls/listener.go @@ -12,7 +12,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -65,11 +64,7 @@ func (l *obfsListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.Listener = ln @@ -84,7 +79,7 @@ func (l *obfsListener) Accept() (net.Conn, error) { conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/obfs/tls/metadata.go b/listener/obfs/tls/metadata.go index f70dc22c..b4e7cae9 100644 --- a/listener/obfs/tls/metadata.go +++ b/listener/obfs/tls/metadata.go @@ -1,27 +1,16 @@ package tls import ( - "time" - md "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/x/metadata/util" ) type metadata struct { - mptcp bool - limiterRefreshInterval time.Duration + mptcp bool } func (l *obfsListener) parseMetadata(md md.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/pht/listener.go b/listener/pht/listener.go index 2292e600..3aa9db50 100644 --- a/listener/pht/listener.go +++ b/listener/pht/listener.go @@ -4,7 +4,6 @@ package pht import ( "net" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" @@ -12,7 +11,6 @@ import ( md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" pht_util "github.com/go-gost/x/internal/util/pht" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -100,7 +98,7 @@ func (l *phtListener) Accept() (conn net.Conn, err error) { conn = admission.WrapConn(l.options.Admission, conn) conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/pht/metadata.go b/listener/pht/metadata.go index 2f4df9fb..d65a5dcc 100644 --- a/listener/pht/metadata.go +++ b/listener/pht/metadata.go @@ -2,7 +2,6 @@ package pht import ( "strings" - "time" mdata "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/x/metadata/util" @@ -16,12 +15,11 @@ const ( ) type metadata struct { - authorizePath string - pushPath string - pullPath string - backlog int - mptcp bool - limiterRefreshInterval time.Duration + authorizePath string + pushPath string + pullPath string + backlog int + mptcp bool } func (l *phtListener) parseMetadata(md mdata.Metadata) (err error) { @@ -53,13 +51,5 @@ func (l *phtListener) parseMetadata(md mdata.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/quic/listener.go b/listener/quic/listener.go index d458a1f0..d286bc87 100644 --- a/listener/quic/listener.go +++ b/listener/quic/listener.go @@ -4,7 +4,6 @@ import ( "context" "net" "strings" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" @@ -12,8 +11,8 @@ import ( md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" quic_util "github.com/go-gost/x/internal/util/quic" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" @@ -78,8 +77,8 @@ func (l *quicListener) Init(md md.Metadata) (err error) { conn = admission.WrapPacketConn(l.options.Admission, conn) conn = limiter_wrapper.WrapPacketConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - "", + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), @@ -119,7 +118,7 @@ func (l *quicListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/quic/metadata.go b/listener/quic/metadata.go index 6606d4c2..cec46285 100644 --- a/listener/quic/metadata.go +++ b/listener/quic/metadata.go @@ -18,9 +18,8 @@ type metadata struct { maxIdleTimeout time.Duration maxStreams int - cipherKey []byte - backlog int - limiterRefreshInterval time.Duration + cipherKey []byte + backlog int } func (l *quicListener) parseMetadata(md mdata.Metadata) (err error) { @@ -54,13 +53,5 @@ func (l *quicListener) parseMetadata(md mdata.Metadata) (err error) { l.md.maxIdleTimeout = mdutil.GetDuration(md, maxIdleTimeout) l.md.maxStreams = mdutil.GetInt(md, maxStreams) - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/redirect/tcp/listener.go b/listener/redirect/tcp/listener.go index 0dc05e39..d948af72 100644 --- a/listener/redirect/tcp/listener.go +++ b/listener/redirect/tcp/listener.go @@ -12,7 +12,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -70,11 +69,7 @@ func (l *redirectListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.ln = ln return @@ -88,7 +83,7 @@ func (l *redirectListener) Accept() (conn net.Conn, err error) { conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/redirect/tcp/metadata.go b/listener/redirect/tcp/metadata.go index 46f07aa1..b10d6971 100644 --- a/listener/redirect/tcp/metadata.go +++ b/listener/redirect/tcp/metadata.go @@ -1,29 +1,18 @@ package tcp import ( - "time" - mdata "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/x/metadata/util" ) type metadata struct { - tproxy bool - mptcp bool - limiterRefreshInterval time.Duration + tproxy bool + mptcp bool } func (l *redirectListener) parseMetadata(md mdata.Metadata) (err error) { l.md.tproxy = mdutil.GetBool(md, "tproxy") l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/redirect/udp/listener.go b/listener/redirect/udp/listener.go index 99bd0344..0529f377 100644 --- a/listener/redirect/udp/listener.go +++ b/listener/redirect/udp/listener.go @@ -2,14 +2,12 @@ package udp import ( "net" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" - limiter_util "github.com/go-gost/x/internal/util/limiter" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" @@ -62,7 +60,7 @@ func (l *redirectListener) Accept() (conn net.Conn, err error) { conn = admission.WrapConn(l.options.Admission, conn) conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/redirect/udp/metadata.go b/listener/redirect/udp/metadata.go index 8c7423ee..60603622 100644 --- a/listener/redirect/udp/metadata.go +++ b/listener/redirect/udp/metadata.go @@ -13,9 +13,8 @@ const ( ) type metadata struct { - ttl time.Duration - readBufferSize int - limiterRefreshInterval time.Duration + ttl time.Duration + readBufferSize int } func (l *redirectListener) parseMetadata(md mdata.Metadata) (err error) { @@ -34,13 +33,5 @@ func (l *redirectListener) parseMetadata(md mdata.Metadata) (err error) { l.md.readBufferSize = defaultReadBufferSize } - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/rtcp/listener.go b/listener/rtcp/listener.go index 71b4051f..42b405e8 100644 --- a/listener/rtcp/listener.go +++ b/listener/rtcp/listener.go @@ -4,7 +4,6 @@ import ( "context" "net" "sync" - "time" "github.com/go-gost/core/chain" "github.com/go-gost/core/limiter" @@ -13,7 +12,6 @@ import ( md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -85,11 +83,7 @@ func (l *rtcpListener) Accept() (conn net.Conn, err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.setListener(ln) } @@ -110,7 +104,7 @@ func (l *rtcpListener) Accept() (conn net.Conn, err error) { conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/rtcp/metadata.go b/listener/rtcp/metadata.go index 19910cd3..42d52d59 100644 --- a/listener/rtcp/metadata.go +++ b/listener/rtcp/metadata.go @@ -1,23 +1,11 @@ package rtcp import ( - "time" - mdata "github.com/go-gost/core/metadata" - mdutil "github.com/go-gost/x/metadata/util" ) -type metadata struct { - limiterRefreshInterval time.Duration -} +type metadata struct{} func (l *rtcpListener) parseMetadata(md mdata.Metadata) (err error) { - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } return } diff --git a/listener/rudp/listener.go b/listener/rudp/listener.go index 16f41b2c..ae1847aa 100644 --- a/listener/rudp/listener.go +++ b/listener/rudp/listener.go @@ -4,7 +4,6 @@ import ( "context" "net" "sync" - "time" "github.com/go-gost/core/chain" "github.com/go-gost/core/limiter" @@ -13,7 +12,6 @@ import ( md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -86,11 +84,7 @@ func (l *rudpListener) Accept() (conn net.Conn, err error) { return nil, listener.NewAcceptError(err) } - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.setListener(ln) @@ -116,8 +110,8 @@ func (l *rudpListener) Accept() (conn net.Conn, err error) { uc = admission.WrapUDPConn(l.options.Admission, uc) conn = limiter_wrapper.WrapUDPConn( uc, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - "", + l.options.TrafficLimiter, + conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), diff --git a/listener/rudp/metadata.go b/listener/rudp/metadata.go index 834fa7cf..94da2522 100644 --- a/listener/rudp/metadata.go +++ b/listener/rudp/metadata.go @@ -15,11 +15,10 @@ const ( ) type metadata struct { - ttl time.Duration - readBufferSize int - readQueueSize int - backlog int - limiterRefreshInterval time.Duration + ttl time.Duration + readBufferSize int + readQueueSize int + backlog int } func (l *rudpListener) parseMetadata(md mdata.Metadata) (err error) { @@ -49,13 +48,5 @@ func (l *rudpListener) parseMetadata(md mdata.Metadata) (err error) { l.md.backlog = defaultBacklog } - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/serial/listener.go b/listener/serial/listener.go index 8163bc9f..c8840d95 100644 --- a/listener/serial/listener.go +++ b/listener/serial/listener.go @@ -9,8 +9,8 @@ import ( "github.com/go-gost/core/listener" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" - limiter_util "github.com/go-gost/x/internal/util/limiter" serial "github.com/go-gost/x/internal/util/serial" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" @@ -101,8 +101,8 @@ func (l *serialListener) listenLoop() { conn = stats.WrapConn(conn, l.options.Stats) conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - "", + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), diff --git a/listener/serial/metadata.go b/listener/serial/metadata.go index 6763f0c3..0f71a674 100644 --- a/listener/serial/metadata.go +++ b/listener/serial/metadata.go @@ -8,20 +8,11 @@ import ( ) type metadata struct { - timeout time.Duration - limiterRefreshInterval time.Duration + timeout time.Duration } func (l *serialListener) parseMetadata(md md.Metadata) (err error) { l.md.timeout = mdutil.GetDuration(md, "timeout", "serial.timeout", "listener.serial.timeout") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/ssh/listener.go b/listener/ssh/listener.go index ff2efd10..791b449a 100644 --- a/listener/ssh/listener.go +++ b/listener/ssh/listener.go @@ -13,7 +13,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" ssh_util "github.com/go-gost/x/internal/util/ssh" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" @@ -72,11 +71,7 @@ func (l *sshListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.Listener = ln @@ -104,7 +99,7 @@ func (l *sshListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/ssh/metadata.go b/listener/ssh/metadata.go index ecd937ff..ef680b33 100644 --- a/listener/ssh/metadata.go +++ b/listener/ssh/metadata.go @@ -2,7 +2,6 @@ package ssh import ( "os" - "time" mdata "github.com/go-gost/core/metadata" ssh_util "github.com/go-gost/x/internal/util/ssh" @@ -16,11 +15,10 @@ const ( ) type metadata struct { - signer ssh.Signer - authorizedKeys map[string]bool - backlog int - mptcp bool - limiterRefreshInterval time.Duration + signer ssh.Signer + authorizedKeys map[string]bool + backlog int + mptcp bool } func (l *sshListener) parseMetadata(md mdata.Metadata) (err error) { @@ -74,13 +72,5 @@ func (l *sshListener) parseMetadata(md mdata.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/sshd/listener.go b/listener/sshd/listener.go index c6910247..b8e9864c 100644 --- a/listener/sshd/listener.go +++ b/listener/sshd/listener.go @@ -14,7 +14,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" ssh_util "github.com/go-gost/x/internal/util/ssh" sshd_util "github.com/go-gost/x/internal/util/sshd" climiter "github.com/go-gost/x/limiter/conn/wrapper" @@ -80,11 +79,7 @@ func (l *sshdListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.Listener = ln @@ -112,13 +107,14 @@ func (l *sshdListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), limiter.SrcOption(conn.RemoteAddr().String()), ) + case err, ok = <-l.errChan: if !ok { err = listener.ErrClosed diff --git a/listener/sshd/metadata.go b/listener/sshd/metadata.go index 5de6eda0..15df0173 100644 --- a/listener/sshd/metadata.go +++ b/listener/sshd/metadata.go @@ -3,7 +3,6 @@ package ssh import ( "fmt" "os" - "time" mdata "github.com/go-gost/core/metadata" ssh_util "github.com/go-gost/x/internal/util/ssh" @@ -18,11 +17,10 @@ const ( ) type metadata struct { - signer ssh.Signer - authorizedKeys map[string]bool - backlog int - mptcp bool - limiterRefreshInterval time.Duration + signer ssh.Signer + authorizedKeys map[string]bool + backlog int + mptcp bool } func (l *sshdListener) parseMetadata(md mdata.Metadata) (err error) { @@ -84,13 +82,5 @@ func (l *sshdListener) parseMetadata(md mdata.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/tap/listener.go b/listener/tap/listener.go index 7461a4fa..157560cc 100644 --- a/listener/tap/listener.go +++ b/listener/tap/listener.go @@ -10,7 +10,7 @@ import ( "github.com/go-gost/core/logger" mdata "github.com/go-gost/core/metadata" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" mdx "github.com/go-gost/x/metadata" metrics "github.com/go-gost/x/metrics/wrapper" @@ -96,8 +96,8 @@ func (l *tapListener) listenLoop() { c = stats.WrapConn(c, l.options.Stats) c = limiter_wrapper.WrapConn( c, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - c.RemoteAddr().String(), + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(c.LocalAddr().Network()), diff --git a/listener/tap/metadata.go b/listener/tap/metadata.go index 6d762023..dd2ecfc8 100644 --- a/listener/tap/metadata.go +++ b/listener/tap/metadata.go @@ -3,7 +3,6 @@ package tap import ( "net" "strings" - "time" mdata "github.com/go-gost/core/metadata" tap_util "github.com/go-gost/x/internal/util/tap" @@ -15,8 +14,7 @@ const ( ) type metadata struct { - config *tap_util.Config - limiterRefreshInterval time.Duration + config *tap_util.Config } func (l *tapListener) parseMetadata(md mdata.Metadata) (err error) { @@ -73,13 +71,5 @@ func (l *tapListener) parseMetadata(md mdata.Metadata) (err error) { l.md.config = config - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/tcp/listener.go b/listener/tcp/listener.go index 972e3683..1fff684f 100644 --- a/listener/tcp/listener.go +++ b/listener/tcp/listener.go @@ -12,7 +12,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -68,11 +67,7 @@ func (l *tcpListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.ln = ln @@ -87,7 +82,7 @@ func (l *tcpListener) Accept() (conn net.Conn, err error) { conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/tcp/metadata.go b/listener/tcp/metadata.go index c7ac1ae9..28c91886 100644 --- a/listener/tcp/metadata.go +++ b/listener/tcp/metadata.go @@ -1,27 +1,16 @@ package tcp import ( - "time" - md "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/x/metadata/util" ) type metadata struct { - mptcp bool - limiterRefreshInterval time.Duration + mptcp bool } func (l *tcpListener) parseMetadata(md md.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/tls/listener.go b/listener/tls/listener.go index 7aecdbf5..a15aaeb0 100644 --- a/listener/tls/listener.go +++ b/listener/tls/listener.go @@ -13,7 +13,6 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -66,11 +65,7 @@ func (l *tlsListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.ln = tls.NewListener(ln, l.options.TLSConfig) @@ -86,7 +81,7 @@ func (l *tlsListener) Accept() (conn net.Conn, err error) { conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/tls/metadata.go b/listener/tls/metadata.go index 0fd467de..9e96e713 100644 --- a/listener/tls/metadata.go +++ b/listener/tls/metadata.go @@ -1,27 +1,16 @@ package tls import ( - "time" - mdata "github.com/go-gost/core/metadata" mdutil "github.com/go-gost/x/metadata/util" ) type metadata struct { - mptcp bool - limiterRefreshInterval time.Duration + mptcp bool } func (l *tlsListener) parseMetadata(md mdata.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/tun/listener.go b/listener/tun/listener.go index 8e629dd4..5524d14f 100644 --- a/listener/tun/listener.go +++ b/listener/tun/listener.go @@ -11,7 +11,7 @@ import ( mdata "github.com/go-gost/core/metadata" "github.com/go-gost/core/router" xnet "github.com/go-gost/x/internal/net" - limiter_util "github.com/go-gost/x/internal/util/limiter" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" mdx "github.com/go-gost/x/metadata" metrics "github.com/go-gost/x/metrics/wrapper" @@ -97,8 +97,8 @@ func (l *tunListener) listenLoop() { c = stats.WrapConn(c, l.options.Stats) c = limiter_wrapper.WrapConn( c, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - c.RemoteAddr().String(), + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(c.LocalAddr().Network()), diff --git a/listener/tun/metadata.go b/listener/tun/metadata.go index d1641640..b52245c5 100644 --- a/listener/tun/metadata.go +++ b/listener/tun/metadata.go @@ -3,7 +3,6 @@ package tun import ( "net" "strings" - "time" "github.com/go-gost/core/logger" mdata "github.com/go-gost/core/metadata" @@ -20,9 +19,8 @@ const ( ) type metadata struct { - config *tun_util.Config - readBufferSize int - limiterRefreshInterval time.Duration + config *tun_util.Config + readBufferSize int } func (l *tunListener) parseMetadata(md mdata.Metadata) (err error) { @@ -113,13 +111,5 @@ func (l *tunListener) parseMetadata(md mdata.Metadata) (err error) { l.md.config = config - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/udp/listener.go b/listener/udp/listener.go index 2c21d4f5..6f21f7c5 100644 --- a/listener/udp/listener.go +++ b/listener/udp/listener.go @@ -2,7 +2,6 @@ package udp import ( "net" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" @@ -11,7 +10,7 @@ import ( admission "github.com/go-gost/x/admission/wrapper" xnet "github.com/go-gost/x/internal/net" "github.com/go-gost/x/internal/net/udp" - limiter_util "github.com/go-gost/x/internal/util/limiter" + traffic_limiter "github.com/go-gost/x/limiter/traffic" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" stats "github.com/go-gost/x/observer/stats/wrapper" @@ -64,8 +63,8 @@ func (l *udpListener) Init(md md.Metadata) (err error) { conn = admission.WrapPacketConn(l.options.Admission, conn) conn = limiter_wrapper.WrapPacketConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - "", + l.options.TrafficLimiter, + traffic_limiter.ServiceLimitKey, limiter.ScopeOption(limiter.ScopeService), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), diff --git a/listener/udp/metadata.go b/listener/udp/metadata.go index dbeae83b..db8a5e13 100644 --- a/listener/udp/metadata.go +++ b/listener/udp/metadata.go @@ -20,7 +20,6 @@ type metadata struct { backlog int keepalive bool ttl time.Duration - limiterRefreshInterval time.Duration } func (l *udpListener) parseMetadata(md mdata.Metadata) (err error) { @@ -52,13 +51,5 @@ func (l *udpListener) parseMetadata(md mdata.Metadata) (err error) { } l.md.keepalive = mdutil.GetBool(md, keepalive) - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/unix/listener.go b/listener/unix/listener.go index cea3a850..57168640 100644 --- a/listener/unix/listener.go +++ b/listener/unix/listener.go @@ -2,14 +2,12 @@ package unix import ( "net" - "time" "github.com/go-gost/core/limiter" "github.com/go-gost/core/listener" "github.com/go-gost/core/logger" md "github.com/go-gost/core/metadata" admission "github.com/go-gost/x/admission/wrapper" - limiter_util "github.com/go-gost/x/internal/util/limiter" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" metrics "github.com/go-gost/x/metrics/wrapper" @@ -52,11 +50,7 @@ func (l *unixListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) l.ln = ln @@ -71,7 +65,7 @@ func (l *unixListener) Accept() (conn net.Conn, err error) { conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), diff --git a/listener/unix/metadata.go b/listener/unix/metadata.go index e8b596bd..767bce40 100644 --- a/listener/unix/metadata.go +++ b/listener/unix/metadata.go @@ -1,24 +1,11 @@ package unix import ( - "time" - md "github.com/go-gost/core/metadata" - mdutil "github.com/go-gost/x/metadata/util" ) -type metadata struct { - limiterRefreshInterval time.Duration -} +type metadata struct{} func (l *unixListener) parseMetadata(md md.Metadata) (err error) { - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return } diff --git a/listener/ws/listener.go b/listener/ws/listener.go index 096134e9..2e501181 100644 --- a/listener/ws/listener.go +++ b/listener/ws/listener.go @@ -16,7 +16,6 @@ import ( xnet "github.com/go-gost/x/internal/net" xhttp "github.com/go-gost/x/internal/net/http" "github.com/go-gost/x/internal/net/proxyproto" - limiter_util "github.com/go-gost/x/internal/util/limiter" ws_util "github.com/go-gost/x/internal/util/ws" climiter "github.com/go-gost/x/limiter/conn/wrapper" limiter_wrapper "github.com/go-gost/x/limiter/traffic/wrapper" @@ -108,11 +107,7 @@ func (l *wsListener) Init(md md.Metadata) (err error) { ln = metrics.WrapListener(l.options.Service, ln) ln = stats.WrapListener(ln, l.options.Stats) ln = admission.WrapListener(l.options.Admission, ln) - ln = limiter_wrapper.WrapListener( - l.options.Service, - ln, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), - ) + ln = limiter_wrapper.WrapListener(l.options.Service, ln, l.options.TrafficLimiter) ln = climiter.WrapListener(l.options.ConnLimiter, ln) if l.tlsEnabled { @@ -138,13 +133,14 @@ func (l *wsListener) Accept() (conn net.Conn, err error) { case conn = <-l.cqueue: conn = limiter_wrapper.WrapConn( conn, - limiter_util.NewCachedTrafficLimiter(l.options.TrafficLimiter, l.md.limiterRefreshInterval, 60*time.Second), + l.options.TrafficLimiter, conn.RemoteAddr().String(), limiter.ScopeOption(limiter.ScopeConn), limiter.ServiceOption(l.options.Service), limiter.NetworkOption(conn.LocalAddr().Network()), limiter.SrcOption(conn.RemoteAddr().String()), ) + case err, ok = <-l.errChan: if !ok { err = listener.ErrClosed diff --git a/listener/ws/metadata.go b/listener/ws/metadata.go index 5d2bf20b..f6fa1eec 100644 --- a/listener/ws/metadata.go +++ b/listener/ws/metadata.go @@ -24,8 +24,7 @@ type metadata struct { enableCompression bool header http.Header - mptcp bool - limiterRefreshInterval time.Duration + mptcp bool } func (l *wsListener) parseMetadata(md mdata.Metadata) (err error) { @@ -55,13 +54,5 @@ func (l *wsListener) parseMetadata(md mdata.Metadata) (err error) { l.md.mptcp = mdutil.GetBool(md, "mptcp") - l.md.limiterRefreshInterval = mdutil.GetDuration(md, "limiter.refreshInterval") - if l.md.limiterRefreshInterval == 0 { - l.md.limiterRefreshInterval = 30 * time.Second - } - if l.md.limiterRefreshInterval < time.Second { - l.md.limiterRefreshInterval = time.Second - } - return }