From c92d6d2bb464ac434b3db7cb964bc5f9742f1941 Mon Sep 17 00:00:00 2001 From: Ilya Yakelzon Date: Fri, 21 Aug 2026 15:49:31 +0200 Subject: [PATCH] Remove the redis-reporting path (eng#3813 phase 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The fleet fully cut over to the datacap sidecar on 2026-08-21 — the reporting redis has zero fleet connections (verified via CLIENT LIST + MONITOR). Delete the legacy accounting/enforcement path: - redis/ (Lua submit reporter, device fetcher, client), usage/ cache, and the throttle/ package (redis-backed _throttle config; the forced config had no remaining callers either) - devicefilter.NewPre (redis usage path); NewDatacapPre stays as the single accounting filter - reportingredis/throttlerefresh flags, the redis client wiring, and the ReportingRedisClient/ThrottleRefreshInterval proxy fields - bitnami test-redis fixture in test.bash/Makefile and the redis integration tests Part of getlantern/engineering#3813. --- Makefile | 3 +- common/headers.go | 1 - config.ini.default | 2 - devicefilter/devicefilter.go | 194 ++------------------ go.mod | 4 +- go.sum | 2 - http-proxy/main.go | 27 +-- http_proxy.go | 32 +--- redis/devices.go | 130 -------------- redis/measured_reporter.go | 251 -------------------------- redis/redis.go | 26 --- redis/redis_test.go | 118 ------------ reporting.go | 31 +--- test.bash | 31 +--- test/test-redis-data/redis-cert.pem | 22 --- test/test-redis-data/redis-key.pem | 28 --- throttle/throttle.go | 268 ---------------------------- throttle/throttle_test.go | 137 -------------- throttle_integration_test.go | 240 ------------------------- usage/usage.go | 33 ---- 20 files changed, 36 insertions(+), 1544 deletions(-) delete mode 100644 redis/devices.go delete mode 100644 redis/measured_reporter.go delete mode 100644 redis/redis.go delete mode 100644 redis/redis_test.go delete mode 100644 test/test-redis-data/redis-cert.pem delete mode 100644 test/test-redis-data/redis-key.pem delete mode 100644 throttle/throttle.go delete mode 100644 throttle/throttle_test.go delete mode 100644 throttle_integration_test.go delete mode 100644 usage/usage.go diff --git a/Makefile b/Makefile index c117e1ee..9c8235d4 100644 --- a/Makefile +++ b/Makefile @@ -15,7 +15,6 @@ get-command = $(shell which="$$(which $(1) 2> /dev/null)" && if [[ ! -z "$$which GO := $(call get-command,go) # Controls whether logs from Redis are included in test output. -REDIS_LOGS ?= false BUILD_TYPE = stable ifeq ($(BUILD_CANARY),true) @@ -96,4 +95,4 @@ system-checks: @if [[ -z "$(GO)" ]]; then echo 'Missing "go" command.'; exit 1; fi test: - ./test.bash $(REDIS_LOGS) + ./test.bash diff --git a/common/headers.go b/common/headers.go index 9ef0f00e..4d6d24c7 100644 --- a/common/headers.go +++ b/common/headers.go @@ -41,7 +41,6 @@ const ( OriginPort = "origin_port" ProbingError = "probing_error" ClientIP = "client_ip" - ThrottleSettings = "throttle_settings" TimeZone = "time_zone" SupportedDataCaps = "supported_data_caps" ) diff --git a/config.ini.default b/config.ini.default index 5578736a..b827bd25 100644 --- a/config.ini.default +++ b/config.ini.default @@ -69,7 +69,6 @@ psmux-max-stream-buffer = 0 # psmux max stream buffer psmux-version = 0 # psmux protocol version quic-bbr = false # Should quic-go use BBR instead of CUBIC quic-ietf-addr = # Address at which to listen for IETF QUIC connections. -reportingredis = # The address of the reporting Redis instance in "redis[s]://host:port" format require-session-tickets = true # Specifies whether or not to require TLS session tickets in ClientHellos sessionticketkey = # File name for storing rotating session ticket keys shadowsocks-addr = # Address at which to listen for shadowsocks connections. @@ -84,7 +83,6 @@ smux-version = 0 # smux protocol version stackdriver-creds = /home/lantern/lantern-stackdriver.json # Optional full json file path containing stackdriver credentials stackdriver-project-id = lantern-http-proxy # Optional project ID for stackdriver error reporting as in http-proxy-lantern stackdriver-sample-percentage = 0.003 # The percentage of devices to report to Stackdriver (0.01 = 1%) -throttlerefresh = 5m0s # Specifies how frequently to refresh throttling configuration from redis. Defaults to 5 minutes. tlslistener-allow-tls13 = false # Allow tlslistener to offer tls13. Because of session ticket issues, this is likely experimental until they can be worked out tlsmasq-addr = # Address at which to listen for tlsmasq connections. tlsmasq-origin-addr = # Address of tlsmasq origin with port. diff --git a/devicefilter/devicefilter.go b/devicefilter/devicefilter.go index 265bc302..6531c862 100644 --- a/devicefilter/devicefilter.go +++ b/devicefilter/devicefilter.go @@ -5,11 +5,8 @@ import ( "net" "net/http" "net/http/httputil" - "sync" "time" - "github.com/dustin/go-humanize" - "github.com/getlantern/golog" "github.com/getlantern/proxy/v3/filters" @@ -20,9 +17,6 @@ import ( "github.com/getlantern/http-proxy-lantern/v2/datacap" "github.com/getlantern/http-proxy-lantern/v2/domains" "github.com/getlantern/http-proxy-lantern/v2/instrument" - "github.com/getlantern/http-proxy-lantern/v2/redis" - "github.com/getlantern/http-proxy-lantern/v2/throttle" - "github.com/getlantern/http-proxy-lantern/v2/usage" ) var ( @@ -75,166 +69,10 @@ func setXBQHeaders(resp *http.Response, usedBytes, capBytes int64, asOf time.Tim resp.Header.Set(common.XBQHeaderv2, fmt.Sprintf("%s/%d", xbq, ttlSeconds)) } -// deviceFilterPre does the device-based filtering, with usage fetched from the -// reporting Redis. Its datacap-sidecar counterpart is datacapFilterPre. -type deviceFilterPre struct { - deviceFetcher *redis.DeviceFetcher - throttleConfig throttle.Config - sendXBQHeader bool - instrument instrument.Instrument - limitersByDevice map[string]*listeners.RateLimiter - limitersByDeviceMx sync.Mutex -} - -// deviceFilterPost cleans up -type deviceFilterPost struct { - bl *blacklist.Blacklist -} - -// NewPre creates a filter which throttling all connections from a device if its data usage threshold is reached. -// * df is used to fetch device data usage across all proxies from a central Redis. -// * throttleConfig is to determine the threshold and throttle rate. They can -// be fixed values or fetched from Redis periodically. -// * If sendXBQHeader is true, it attaches a common.XBQHeader to inform the -// clients the usage information before this request is made. The header is -// expected to follow this format: -// -// // -// -// is the string representation of a 64-bit unsigned integer -// is the string representation of a 64-bit unsigned integer -// is the 64-bit signed integer representing seconds since a custom -// epoch (00:00:00 01/01/2016 UTC). -func NewPre(df *redis.DeviceFetcher, throttleConfig throttle.Config, sendXBQHeader bool, instrument instrument.Instrument) filters.Filter { - if throttleConfig != nil { - log.Debug("Throttling enabled") - } - - return &deviceFilterPre{ - deviceFetcher: df, - throttleConfig: throttleConfig, - sendXBQHeader: sendXBQHeader, - instrument: instrument, - limitersByDevice: make(map[string]*listeners.RateLimiter, 0), - } -} - -func (f *deviceFilterPre) Apply(cs *filters.ConnectionState, req *http.Request, next filters.Next) (*http.Response, *filters.ConnectionState, error) { - if log.IsTraceEnabled() { - reqStr, _ := httputil.DumpRequest(req, true) - log.Tracef("DeviceFilter Middleware received request:\n%s", reqStr) - } - - // Attached the uid to connection to report stats to redis correctly - // "conn" in context is previously attached in server.go - wc := cs.Downstream().(listeners.WrapConn) - lanternDeviceID := req.Header.Get(common.DeviceIdHeader) - - // Even if a device hasn't hit its data cap, we always throttle to a default throttle rate to - // keep bandwidth hogs from using too much bandwidth. Note - this does not apply to pro proxies - // which don't use the devicefilter at all. - throttleDefault := func(message string) { - if DefaultThrottleRate <= 0 { - f.instrument.Throttle(req.Context(), false, message) - } - limiter := f.rateLimiterForDevice(lanternDeviceID, DefaultThrottleRate, DefaultThrottleRate) - if log.IsTraceEnabled() { - log.Tracef("Throttling connection to %v per second by default", - humanize.Bytes(uint64(DefaultThrottleRate))) - } - f.instrument.Throttle(req.Context(), true, "default") - wc.ControlMessage("throttle", limiter) - } - - // Some domains are excluded from being throttled and don't count towards the - // bandwidth cap. - if domains.ConfigForRequest(req).Unthrottled { - throttleDefault("domain-excluded") - return next(cs, req) - } - - if throttleSentinelDevice(f.instrument, wc, req, lanternDeviceID) { - return next(cs, req) - } - - if f.throttleConfig == nil { - f.instrument.Throttle(req.Context(), false, "no-config") - return next(cs, req) - } - - // Throttling enabled - u := usage.Get(lanternDeviceID) - if u == nil { - // Eagerly request device ID data from Redis and store it in usage - f.deviceFetcher.RequestNewDeviceUsage(lanternDeviceID) - throttleDefault("no-usage-data") - return next(cs, req) - } - - settings, capOn := f.throttleConfig.SettingsFor(lanternDeviceID, u.CountryCode, req.Header.Get(common.PlatformHeader), req.Header.Get(common.AppHeader), req.Header[common.SupportedDataCapsHeader]) - - measuredCtx := map[string]interface{}{ - "throttled": false, - } - - // To turn the data cap off in Redis we simply set the threshold to 0 or - // below. This will also turn off the cap in the UI on desktop and in newer - // versions on mobile. - if capOn { - log.Tracef("Got throttle settings: %v", settings) - capOn = settings.Threshold > 0 - - // Send throttle settings to measured as well - measuredCtx["throttle_settings"] = settings - } - - if capOn && u.Bytes > settings.Threshold { - // per connection limiter - // Note - when people hit the data cap, we only throttle writes back to the client, not reads. - // This way, they can continue to upload videos or other bandwidth intensive content for sharing. - limiter := f.rateLimiterForDevice(lanternDeviceID, DefaultThrottleRate, settings.Rate) - if log.IsTraceEnabled() { - log.Tracef("Throttling connection from device %s to %v per second", lanternDeviceID, - humanize.Bytes(uint64(settings.Rate))) - } - f.instrument.Throttle(req.Context(), true, "datacap") - wc.ControlMessage("throttle", limiter) - measuredCtx["throttled"] = true - } else { - // default case is not throttling - throttleDefault("") - } - wc.ControlMessage("measured", measuredCtx) - - resp, nextCtx, err := next(cs, req) - if resp == nil || err != nil { - return resp, nextCtx, err - } - if !capOn || !f.sendXBQHeader { - return resp, nextCtx, err - } - setXBQHeaders(resp, u.Bytes, settings.Threshold, u.AsOf, u.TTLSeconds) - f.instrument.XBQHeaderSent(req.Context()) - return resp, nextCtx, err -} - -func (f *deviceFilterPre) rateLimiterForDevice(deviceID string, rateLimitRead, rateLimitWrite int64) *listeners.RateLimiter { - f.limitersByDeviceMx.Lock() - defer f.limitersByDeviceMx.Unlock() - - limiter := f.limitersByDevice[deviceID] - if limiter == nil || limiter.GetRateRead() != rateLimitRead || limiter.GetRateWrite() != rateLimitWrite { - limiter = listeners.NewRateLimiter(rateLimitRead, rateLimitWrite) - f.limitersByDevice[deviceID] = limiter - } - return limiter -} - -// datacapFilterPre is deviceFilterPre's counterpart for proxies whose byte -// accounting runs through the local datacap sidecar. The throttle decision -// itself is made asynchronously by the tracker; the filter attaches the -// device's shared limiter and surfaces the tracker's latest view of the device -// to the client via the XBQ headers. +// datacapFilterPre throttles devices whose byte accounting runs through the +// local datacap sidecar. The throttle decision itself is made asynchronously +// by the tracker; the filter attaches the device's shared limiter and surfaces +// the tracker's latest view of the device to the client via the XBQ headers. type datacapFilterPre struct { tracker *datacap.Tracker sendXBQHeader bool @@ -242,10 +80,10 @@ type datacapFilterPre struct { } // NewDatacapPre creates the filter for proxies whose byte accounting runs -// through the local datacap sidecar. Unlike the Redis path, the limiter it -// attaches is shared across all of a device's connections and is re-rated by -// the tracker as reports come back, so crossing the cap slows down transfers -// that are already in flight rather than only the next one. +// through the local datacap sidecar. The limiter it attaches is shared across +// all of a device's connections and is re-rated by the tracker as reports +// come back, so crossing the cap slows down transfers that are already in +// flight rather than only the next one. func NewDatacapPre(tracker *datacap.Tracker, sendXBQHeader bool, instrument instrument.Instrument) filters.Filter { return &datacapFilterPre{ tracker: tracker, @@ -268,12 +106,11 @@ func (f *datacapFilterPre) Apply(cs *filters.ConnectionState, req *http.Request, // just never held to the capped rate — hence a separate limiter that the // tracker never re-rates. // - // This check deliberately precedes the sentinel-device guards, matching the - // Redis path: a request to an excluded domain gets the default rate even - // with a missing device ID. Moving the guards first would newly subject - // old clients to alwaysThrottle on domains we have decided not to throttle. - // Such requests share one limiter under the empty device ID, exactly as - // rateLimiterForDevice("") does on the Redis path. + // This check deliberately precedes the sentinel-device guards: a request + // to an excluded domain gets the default rate even with a missing device + // ID. Moving the guards first would newly subject old clients to + // alwaysThrottle on domains we have decided not to throttle. Such requests + // share one limiter under the empty device ID. if domains.ConfigForRequest(req).Unthrottled { f.instrument.Throttle(req.Context(), true, "default") wc.ControlMessage("throttle", f.tracker.Limiter(deviceID, true)) @@ -325,6 +162,11 @@ func (f *datacapFilterPre) Apply(cs *filters.ConnectionState, req *http.Request, return resp, nextCtx, err } +// deviceFilterPost cleans up +type deviceFilterPost struct { + bl *blacklist.Blacklist +} + func NewPost(bl *blacklist.Blacklist) filters.Filter { return &deviceFilterPost{ bl: bl, diff --git a/go.mod b/go.mod index d73afed4..22bbf5c4 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,6 @@ require ( github.com/Jigsaw-Code/outline-ss-server v1.5.0 github.com/OperatorFoundation/Replicant-go/Replicant/v3 v3.0.23 github.com/OperatorFoundation/Starbridge-go/Starbridge/v3 v3.0.17 - github.com/dustin/go-humanize v1.0.1 github.com/getlantern/broflake v0.0.0-20260405220006-86ebdd4b757a github.com/getlantern/cmux/v2 v2.0.0-20230301223233-dac79088a4c0 github.com/getlantern/cmuxprivate v0.0.0-20211216020409-d29d0d38be54 @@ -43,7 +42,6 @@ require ( github.com/getlantern/tlsdefaults v0.0.0-20171004213447-cf35cfd0b1b4 github.com/getlantern/tlsmasq v0.4.7-0.20230302000139-6e479a593298 github.com/getlantern/tlsutil v0.5.3 - github.com/getlantern/waitforserver v1.0.1 github.com/getlantern/withtimeout v0.0.0-20160829163843-511f017cd913 github.com/go-redis/redis/v8 v8.11.5 github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da @@ -53,7 +51,6 @@ require ( github.com/refraction-networking/utls v1.6.7 github.com/sagernet/sing v0.6.0-alpha.18 github.com/siddontang/go v0.0.0-20180604090527-bdc77568d726 - github.com/spaolacci/murmur3 v1.1.0 github.com/stretchr/testify v1.11.1 github.com/vharitonsky/iniflags v0.0.0-20180513140207-a33cd0b5f3de github.com/xtaci/smux v1.5.35-0.20250217141229-e6b0586a4539 @@ -117,6 +114,7 @@ require ( github.com/dchest/siphash v1.2.3 // indirect github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dsnet/compress v0.0.2-0.20230904184137-39efe44ab707 // indirect + github.com/dustin/go-humanize v1.0.1 // indirect github.com/dvyukov/go-fuzz v0.0.0-20210429054444-fca39067bc72 // indirect github.com/edsrzf/mmap-go v1.1.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect diff --git a/go.sum b/go.sum index b20256f3..253f1e1b 100644 --- a/go.sum +++ b/go.sum @@ -710,8 +710,6 @@ github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8 h1:TG/diQgUe0pntT/2D github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8/go.mod h1:P5HUIBuIWKbyjl083/loAegFkfbFNx5i2qEP4CNbm7E= github.com/sorairolake/lzip-go v0.3.8 h1:j5Q2313INdTA80ureWYRhX+1K78mUXfMoPZCw/ivWik= github.com/sorairolake/lzip-go v0.3.8/go.mod h1:JcBqGMV0frlxwrsE9sMWXDjqn3EeVf0/54YPsw66qkU= -github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI= -github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/spf13/afero v1.15.0 h1:b/YBCLWAJdFWJTN9cLhiXXcD7mzKn9Dm86dNnfyQw1I= github.com/spf13/afero v1.15.0/go.mod h1:NC2ByUVxtQs4b3sIUphxK0NioZnmxgyCrfzeuq8lxMg= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= diff --git a/http-proxy/main.go b/http-proxy/main.go index 1980763a..7f290a8e 100644 --- a/http-proxy/main.go +++ b/http-proxy/main.go @@ -17,7 +17,6 @@ import ( "syscall" "time" - "github.com/go-redis/redis/v8" "github.com/mitchellh/panicwrap" "github.com/vharitonsky/iniflags" @@ -30,10 +29,8 @@ import ( "github.com/getlantern/http-proxy-lantern/v2/datacap" "github.com/getlantern/http-proxy-lantern/v2/googlefilter" "github.com/getlantern/http-proxy-lantern/v2/obfs4listener" - lanternredis "github.com/getlantern/http-proxy-lantern/v2/redis" "github.com/getlantern/http-proxy-lantern/v2/shadowsocks" "github.com/getlantern/http-proxy-lantern/v2/stackdrivererror" - "github.com/getlantern/http-proxy-lantern/v2/throttle" "github.com/getlantern/http-proxy-lantern/v2/tlslistener" ) @@ -102,8 +99,6 @@ var ( // built into the proxy package. See eng#3695. legacyAPIHosts = flag.String("legacyapihosts", "", "Comma-separated additional hostnames exempted from the BlockLocal filter") - throttleRefreshInterval = flag.Duration("throttlerefresh", throttle.DefaultRefreshInterval, "Specifies how frequently to refresh throttling configuration from redis. Defaults to 5 minutes.") - enableMultipath = flag.Bool("enablemultipath", false, "Enable multipath. Only clients support multipath can communicate with it.") externalIP = flag.String("externalip", "", "The external IP of this proxy, used for reporting") @@ -120,12 +115,7 @@ var ( proxiedSitesSamplePercentage = flag.Float64("proxied-sites-sample-percentage", 0, "The percentage of requests to sample (0.01 = 1%)") proxiedSitesTrackingId = flag.String("proxied-sites-tracking-id", "UA-21815217-16", "The Google Analytics property id for tracking proxied sites") - reportingRedisAddr = flag.String("reportingredis", "", "The address of the reporting Redis instance in \"redis[s]://host:port\" format") - - // Successor to reportingredis. When both are set datacapurl wins: a proxy - // must account and enforce from exactly one source, and the sidecar is the - // one every other proxy flavor already reports to. - datacapURL = flag.String("datacapurl", "", "Base URL of the local datacap sidecar, e.g. \"http://127.0.0.1:8078\". Enables byte accounting and data-cap throttling through the sidecar, superseding reportingredis.") + datacapURL = flag.String("datacapurl", "", "Base URL of the local datacap sidecar, e.g. \"http://127.0.0.1:8078\". Enables byte accounting and data-cap throttling through the sidecar.") datacapReportInterval = flag.Duration("datacapreportinterval", datacap.DefaultReportInterval, "How frequently to flush accumulated per-device usage to the datacap sidecar.") // default value of tunnelPorts matches ports in flashlight/client/client.go @@ -397,17 +387,10 @@ func main() { }) go periodicallyForceGC() - var reportingRedisClient *redis.Client - switch { - case *datacapURL != "": + if *datacapURL != "" { log.Debugf("reporting bandwidth to the datacap sidecar at %v", *datacapURL) - case *reportingRedisAddr != "": - reportingRedisClient, err = lanternredis.NewClient(*reportingRedisAddr) - if err != nil { - log.Errorf("failed to initialize redis client, will not be able to perform bandwidth limiting: %v", err) - } - default: - log.Debug("neither a datacap sidecar nor a redis address configured for bandwidth reporting") + } else { + log.Debug("no datacap sidecar configured for bandwidth reporting") } p := &proxy.Proxy{ @@ -422,7 +405,6 @@ func main() { BanditCallbackTTL: *banditCallbackTTL, LegacyAPIHosts: *legacyAPIHosts, EnableMultipath: *enableMultipath, - ThrottleRefreshInterval: *throttleRefreshInterval, TracesSampleRate: *tracesSampleRate, TeleportSampleRate: *teleportSampleRate, ExternalIP: *externalIP, @@ -435,7 +417,6 @@ func main() { Pro: *pro, ProxiedSitesSamplePercentage: *proxiedSitesSamplePercentage, ProxiedSitesTrackingID: *proxiedSitesTrackingId, - ReportingRedisClient: reportingRedisClient, DatacapURL: *datacapURL, DatacapReportInterval: *datacapReportInterval, Token: *token, diff --git a/http_proxy.go b/http_proxy.go index 8ab4b440..b69782e5 100644 --- a/http_proxy.go +++ b/http_proxy.go @@ -17,8 +17,6 @@ import ( "strings" "time" - rclient "github.com/go-redis/redis/v8" - "github.com/getlantern/cmux/v2" "github.com/getlantern/cmuxprivate" "github.com/getlantern/enhttp" @@ -66,8 +64,6 @@ import ( "github.com/getlantern/http-proxy-lantern/v2/mimic" "github.com/getlantern/http-proxy-lantern/v2/obfs4listener" "github.com/getlantern/http-proxy-lantern/v2/ping" - "github.com/getlantern/http-proxy-lantern/v2/redis" - "github.com/getlantern/http-proxy-lantern/v2/throttle" "github.com/getlantern/http-proxy-lantern/v2/tlslistener" "github.com/getlantern/http-proxy-lantern/v2/tlsmasq" "github.com/getlantern/http-proxy-lantern/v2/tokenfilter" @@ -116,8 +112,6 @@ type Proxy struct { Pro bool ProxiedSitesSamplePercentage float64 ProxiedSitesTrackingID string - ReportingRedisClient *rclient.Client - ThrottleRefreshInterval time.Duration Token string TunnelPorts string Obfs4Addr string @@ -220,12 +214,10 @@ type Proxy struct { VMessUUIDs []string // DatacapURL is the base URL of the local datacap sidecar. When set, byte - // accounting and data-cap throttling run through the sidecar and the - // reporting-Redis path is left unused. + // accounting and data-cap throttling run through the sidecar. DatacapURL string DatacapReportInterval time.Duration - throttleConfig throttle.Config datacapTracker *datacap.Tracker instrument instrument.Instrument } @@ -267,7 +259,6 @@ func (p *Proxy) ListenAndServe(ctx context.Context) error { log.Errorf("Unable to set up packet forwarding, will continue to start up: %v", err) } p.setBenchmarkMode() - p.loadThrottleConfig() p.loadDatacapTracker() if p.ENHTTPAddr != "" { @@ -569,17 +560,11 @@ func (p *Proxy) createFilterChain(bl *blacklist.Blacklist) (filters.Chain, proxy ) } - switch { - case p.datacapTracker != nil: + if p.datacapTracker != nil { filterChain = filterChain.Append( proxy.OnFirstOnly(devicefilter.NewDatacapPre(p.datacapTracker, !p.Pro, p.instrument)), ) - case p.ReportingRedisClient != nil: - filterChain = filterChain.Append( - proxy.OnFirstOnly(devicefilter.NewPre( - redis.NewDeviceFetcher(p.ReportingRedisClient), p.throttleConfig, !p.Pro, p.instrument)), - ) - default: + } else { log.Debug("Not enabling bandwidth limiting") } @@ -741,16 +726,7 @@ func (p *Proxy) buildOTELOpts(includeProxyName bool) *otel.Opts { } func (p *Proxy) configureBandwidthReporting() *reportingConfig { - return newReportingConfig(p.CountryLookup, p.ReportingRedisClient, p.instrument, p.throttleConfig, p.datacapTracker) -} - -func (p *Proxy) loadThrottleConfig() { - if !p.Pro && p.ThrottleRefreshInterval > 0 && p.ReportingRedisClient != nil { - p.throttleConfig = throttle.NewRedisConfig(p.ReportingRedisClient, p.ThrottleRefreshInterval) - } else { - log.Debug("Not loading throttle config") - return - } + return newReportingConfig(p.instrument, p.datacapTracker) } // loadDatacapTracker starts the sidecar-backed accounting pipeline. Pro tracks diff --git a/redis/devices.go b/redis/devices.go deleted file mode 100644 index dc36a58f..00000000 --- a/redis/devices.go +++ /dev/null @@ -1,130 +0,0 @@ -package redis - -import ( - "context" - "strconv" - "sync" - "time" - - "github.com/getlantern/http-proxy-lantern/v2/usage" - "github.com/go-redis/redis/v8" -) - -const getUsageScript = ` - local clientKey = KEYS[1] - - local usage = redis.call("hmget", clientKey, "bytesIn", "bytesOut", "countryCode") - local ttl = redis.call("ttl", clientKey) - - return {usage[1], usage[2], usage[3], ttl} -` - -type ongoingSet struct { - set map[string]bool - sync.RWMutex -} - -func (s *ongoingSet) add(dev string) { - s.Lock() - s.set[dev] = true - s.Unlock() -} - -func (s *ongoingSet) del(dev string) { - s.Lock() - delete(s.set, dev) - s.Unlock() -} - -func (s *ongoingSet) isMember(dev string) bool { - s.RLock() - _, ok := s.set[dev] - s.RUnlock() - return ok -} - -// DeviceFetcher retrieves device information from Redis -type DeviceFetcher struct { - rc *redis.Client - ongoing *ongoingSet - queue chan string - ctx context.Context -} - -// NewDeviceFetcher creates a new DeviceFetcher -func NewDeviceFetcher(rc *redis.Client) *DeviceFetcher { - df := &DeviceFetcher{ - rc: rc, - ongoing: &ongoingSet{set: make(map[string]bool, 512)}, - queue: make(chan string, 512), - ctx: context.Background(), - } - - go df.processDeviceUsageRequests() - - return df -} - -// RequestNewDeviceUsage adds a new request for device usage to the queue -func (df *DeviceFetcher) RequestNewDeviceUsage(deviceID string) { - if df.ongoing.isMember(deviceID) { - return - } - select { - case df.queue <- deviceID: - df.ongoing.add(deviceID) - // ok - default: - // queue full, ignore - } -} - -func (df *DeviceFetcher) processDeviceUsageRequests() { - var scriptSHA string - for deviceID := range df.queue { - if scriptSHA == "" { - var err error - scriptSHA, err = df.rc.ScriptLoad(df.ctx, getUsageScript).Result() - if err != nil { - log.Errorf("Unable to load script, skip fetching usage: %v", err) - continue - } - } - - if err := df.retrieveDeviceUsage(scriptSHA, deviceID); err != nil { - log.Errorf("Error retrieving device usage: %v", err) - } - } -} - -func (df *DeviceFetcher) retrieveDeviceUsage(scriptSHA string, deviceID string) error { - clientKey := "_client:" + deviceID - _vals, err := df.rc.EvalSha(df.ctx, scriptSHA, []string{clientKey}).Result() - if err != nil { - return err - } - vals := _vals.([]interface{}) - if vals[0] == nil || vals[1] == nil || vals[2] == nil || vals[3] == nil { - // No entry found or partially stored, means no usage data so far. - usage.Set(deviceID, "", 0, time.Now(), 0) - return nil - } - - _bytesIn := vals[0].(string) - bytesIn, err := strconv.ParseInt(_bytesIn, 10, 64) - if err != nil { - log.Debugf("Error parsing bytesIn: %v", err) - return nil - } - _bytesOut := vals[1].(string) - bytesOut, err := strconv.ParseInt(_bytesOut, 10, 64) - if err != nil { - log.Debugf("Error parsing bytesOut: %v", err) - return nil - } - countryCode := vals[2].(string) - ttl := vals[3].(int64) - usage.Set(deviceID, countryCode, bytesIn+bytesOut, time.Now(), ttl) - df.ongoing.del(deviceID) - return nil -} diff --git a/redis/measured_reporter.go b/redis/measured_reporter.go deleted file mode 100644 index 5c3d96bf..00000000 --- a/redis/measured_reporter.go +++ /dev/null @@ -1,251 +0,0 @@ -package redis - -import ( - "context" - "math/rand" - "net" - "strconv" - "strings" - "time" - - "github.com/go-redis/redis/v8" - - "github.com/getlantern/geo" - "github.com/getlantern/golog" - "github.com/getlantern/http-proxy-lantern/v2/common" - "github.com/getlantern/http-proxy-lantern/v2/listeners" - "github.com/getlantern/http-proxy-lantern/v2/throttle" - "github.com/getlantern/http-proxy-lantern/v2/usage" - "github.com/getlantern/measured" -) - -const ( - updateUsageScript = ` - local clientKey = KEYS[1] - - local bytesIn = redis.call("hincrby", clientKey, "bytesIn", ARGV[1]) - local bytesOut = redis.call("hincrby", clientKey, "bytesOut", ARGV[2]) - local countryCode = redis.call("hget", clientKey, "countryCode") - if not countryCode or countryCode == "" then - countryCode = ARGV[3] - redis.call("hset", clientKey, "countryCode", countryCode) - -- record the IP on which we based the countryCode for auditing - redis.call("hset", clientKey, "clientIP", ARGV[4]) - redis.call("expireat", clientKey, ARGV[5]) - end - - local ttl = redis.call("ttl", clientKey) - return {bytesIn, bytesOut, countryCode, ttl} -` - - sixtyDays = 60 * 24 * time.Hour -) - -var ( - log = golog.LoggerFor("redis") -) - -type statsAndContext struct { - ctx map[string]interface{} - stats *measured.Stats -} - -func (sac *statsAndContext) add(other *statsAndContext) *statsAndContext { - newStats := *other.stats - if sac != nil { - newStats.SentTotal += sac.stats.SentTotal - newStats.RecvTotal += sac.stats.RecvTotal - } - return &statsAndContext{other.ctx, &newStats} -} - -func NewMeasuredReporter(countryLookup geo.CountryLookup, rc *redis.Client, reportInterval time.Duration, throttleConfig throttle.Config) listeners.MeasuredReportFN { - // Provide some buffering so that we don't lose data while submitting to Redis - statsCh := make(chan *statsAndContext, 10000) - go reportPeriodically(countryLookup, rc, reportInterval, throttleConfig, statsCh) - return func(ctx map[string]interface{}, stats *measured.Stats, deltaStats *measured.Stats, final bool) { - select { - case statsCh <- &statsAndContext{ctx, deltaStats}: - // submitted successfully - default: - // data lost, probably because Redis submission is taking longer than expected - } - } -} - -func reportPeriodically(countryLookup geo.CountryLookup, rc *redis.Client, reportInterval time.Duration, throttleConfig throttle.Config, statsCh chan *statsAndContext) { - // randomize the interval to evenly distribute traffic to reporting Redis. - randomized := time.Duration(reportInterval.Nanoseconds()/2 + rand.Int63n(reportInterval.Nanoseconds())) - log.Debugf("Will report data usage to Redis every %v", randomized) - ticker := time.NewTicker(randomized) - statsByDeviceID := make(map[string]*statsAndContext) - var scriptSHA string - for { - select { - case sac := <-statsCh: - _deviceID := sac.ctx[common.DeviceID] - if _deviceID == nil { - // ignore - continue - } - deviceID := _deviceID.(string) - statsByDeviceID[deviceID] = statsByDeviceID[deviceID].add(sac) - case <-ticker.C: - if log.IsTraceEnabled() { - log.Tracef("Submitting %d stats", len(statsByDeviceID)) - } - if scriptSHA == "" { - var err error - scriptSHA, err = rc.ScriptLoad(context.Background(), updateUsageScript).Result() - if err != nil { - log.Errorf("Unable to load script, skip submitting stats: %v", err) - continue - } - } - - err := submit(countryLookup, rc, scriptSHA, statsByDeviceID, throttleConfig) - if err != nil { - log.Errorf("Unable to submit stats: %v", err) - } - // Reset stats - statsByDeviceID = make(map[string]*statsAndContext) - } - } -} - -func submit(countryLookup geo.CountryLookup, rc *redis.Client, scriptSHA string, statsByDeviceID map[string]*statsAndContext, throttleConfig throttle.Config) error { - for deviceID, sac := range statsByDeviceID { - now := time.Now() - - _clientIP := sac.ctx[common.ClientIP] - if _clientIP == nil { - log.Error("Missing client_ip in context, this shouldn't happen. Ignoring.") - continue - } - clientIP := _clientIP.(string) - countryCode := countryLookup.CountryCode(net.ParseIP(clientIP)) - - var platform string - _platform, ok := sac.ctx[common.Platform] - if ok { - platform = _platform.(string) - } - - var appName string - _appName, ok := sac.ctx[common.App] - if ok { - appName = _appName.(string) - } - - var supportedDataCaps []string - _supportedDataCaps, ok := sac.ctx[common.SupportedDataCaps] - if ok { - supportedDataCaps = _supportedDataCaps.([]string) - } - throttleSettings, hasThrottleSettings := throttleConfig.SettingsFor(deviceID, countryCode, platform, appName, supportedDataCaps) - - pl := rc.Pipeline() - throttleCohort := "" - var updateUsage *redis.Cmd - if !hasThrottleSettings { - throttleCohort = "uncapped" - } else { - stats := sac.stats - throttleCohort = throttleSettings.Label - - timeZone := "" - _timeZone, hasTimeZone := sac.ctx[common.TimeZone] - if hasTimeZone { - timeZone = _timeZone.(string) - } else { - // default timeZone to now - timeZone = now.Location().String() - } - - clientKey := "_client:" + deviceID - updateUsage = pl.EvalSha(context.Background(), scriptSHA, []string{clientKey}, - strconv.Itoa(stats.RecvTotal), - strconv.Itoa(stats.SentTotal), - strings.ToLower(countryCode), - clientIP, - expirationFor(now, throttleSettings.CapResets, timeZone)) - } - log.Tracef("device %v on platform %v in country %v with supported data caps %v is in throttle cohort %v", deviceID, platform, countryCode, supportedDataCaps, throttleCohort) - countryCodeLower := strings.ToLower(countryCode) - - nowUTC := now.In(time.UTC) - today := nowUTC.Format("2006-01-02") - uniqueDevicesKey := "_devices:" + countryCodeLower + ":" + today + ":" + throttleCohort - pl.SAdd(context.Background(), uniqueDevicesKey, deviceID) - // we don't keep these around forever to save space, however we do need to keep them around for longer than the purchase data from pro-server, - // to make sure that we can identify the device cohort for all purchases - pl.ExpireAt(context.Background(), uniqueDevicesKey, daysFrom(nowUTC.In(time.UTC), 4)) - - deviceLastSeenKey := "_deviceLastSeen:" + countryCodeLower + ":" + throttleCohort + ":" + deviceID - pl.Set(context.Background(), deviceLastSeenKey, now.Unix(), 0) - pl.Expire(context.Background(), deviceLastSeenKey, sixtyDays) // nb: test fails if we try to set expiration in the above Set call - - throttled := sac.ctx["throttled"] == true - if throttled { - deviceFirstThrottledKey := "_deviceFirstThrottled:" + deviceID - pl.Set(context.Background(), deviceFirstThrottledKey, now.Unix(), 0) - pl.Expire(context.Background(), deviceFirstThrottledKey, sixtyDays) // nb: test fails if we try to set expiration in the above Set call - } - - _, err := pl.Exec(context.Background()) - if err != nil { - return err - } - - if hasThrottleSettings { - _result, err := updateUsage.Result() - if err != nil { - return err - } - result := _result.([]interface{}) - bytesIn, _ := result[0].(int64) - bytesOut, _ := result[1].(int64) - _countryCode := result[2] - // In production it should never be nil but LedisDB (for unit testing) - // has a bug which treats empty string as nil when `EvalSha`. - if _countryCode == nil { - countryCode = "" - } else { - countryCode = _countryCode.(string) - } - ttlSeconds := result[3].(int64) - usage.Set(deviceID, countryCode, bytesIn+bytesOut, now, ttlSeconds) - } - } - return nil -} - -func expirationFor(now time.Time, ttl throttle.CapInterval, timeZoneName string) int64 { - tz, err := time.LoadLocation(timeZoneName) - if err == nil { - // adjust to given timeZone - now = now.In(tz) - } - switch ttl { - case throttle.Daily: - return daysFrom(now, 1).Unix() - case throttle.Weekly: - daysFromSunday := int(now.Weekday()) - daysToNextMonday := 8 - daysFromSunday - if daysToNextMonday > 7 { - // today's Sunday, so next Monday is in just 1 day - daysToNextMonday = 1 - } - nextMonday := now.AddDate(0, 0, daysToNextMonday) - return time.Date(nextMonday.Year(), nextMonday.Month(), nextMonday.Day(), 0, 0, 0, 0, now.Location()).Add(-1 * time.Nanosecond).Unix() - case throttle.Monthly, throttle.Legacy: - nextMonth := now.AddDate(0, 1, 0) - return time.Date(nextMonth.Year(), nextMonth.Month(), 1, 0, 0, 0, 0, now.Location()).Add(-1 * time.Nanosecond).Unix() - } - return 0 -} - -func daysFrom(start time.Time, days int) time.Time { - next := start.AddDate(0, 0, days) - return time.Date(next.Year(), next.Month(), next.Day(), 0, 0, 0, 0, start.Location()).Add(-1 * time.Nanosecond) -} diff --git a/redis/redis.go b/redis/redis.go deleted file mode 100644 index aaf5f1ed..00000000 --- a/redis/redis.go +++ /dev/null @@ -1,26 +0,0 @@ -package redis - -import ( - "crypto/tls" - "fmt" - - "github.com/go-redis/redis/v8" -) - -// Creates a new redis client with the specified redis URL to use, in the form: -// rediss://:password@host -func NewClient(redisURL string) (*redis.Client, error) { - opt, err := redis.ParseURL(redisURL) - if err != nil { - return nil, fmt.Errorf("failed to parse URL") - } - - return redis.NewClient(&redis.Options{ - Addr: opt.Addr, - Password: opt.Password, - PoolSize: 2, - TLSConfig: &tls.Config{ - ClientSessionCache: tls.NewLRUClientSessionCache(20), - }, - }), nil -} diff --git a/redis/redis_test.go b/redis/redis_test.go deleted file mode 100644 index 3ee085ca..00000000 --- a/redis/redis_test.go +++ /dev/null @@ -1,118 +0,0 @@ -package redis - -import ( - "context" - "net" - "strconv" - "testing" - "time" - - "github.com/getlantern/measured" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - - "github.com/getlantern/http-proxy-lantern/v2/common" - "github.com/getlantern/http-proxy-lantern/v2/internal/testutil" - "github.com/getlantern/http-proxy-lantern/v2/throttle" - "github.com/getlantern/http-proxy-lantern/v2/usage" -) - -func TestRedisUrl(t *testing.T) { - cl, err := NewClient("rediss://:password@host:6379") - assert.NoError(t, err) - assert.NotNil(t, cl) - - cl, err = NewClient("rediss://127.0.0.1:5252") - assert.NoError(t, err) - assert.NotNil(t, cl) -} - -func TestReportPeriodically(t *testing.T) { - redisClient := testutil.TestRedis(t) - - deviceID := "device12" - clientIP := "1.1.1.1" - fetcher := NewDeviceFetcher(redisClient) - statsCh := make(chan *statsAndContext, 10000) - newStats := func() { - statsCh <- &statsAndContext{map[string]interface{}{common.DeviceID: deviceID, "client_ip": clientIP, "app_platform": "windows", "throttled": true}, &measured.Stats{RecvTotal: 2, SentTotal: 1}} - } - lookup := &fakeLookup{} - go reportPeriodically(lookup, redisClient, time.Millisecond, throttle.NewForcedConfig(5000, 500, throttle.Monthly), statsCh) - - fetcher.RequestNewDeviceUsage(deviceID) - time.Sleep(100 * time.Millisecond) - localCopy := usage.Get(deviceID) - assert.Equal(t, "", localCopy.CountryCode) - assert.EqualValues(t, 0, localCopy.Bytes) - newStats() - time.Sleep(300 * time.Millisecond) - result := redisClient.HGetAll(context.Background(), "_client:"+deviceID).Val() - assert.Equal(t, "2", result["bytesIn"]) - assert.Equal(t, "1", result["bytesOut"]) - assert.Equal(t, "", result["countryCode"]) - assert.True(t, redisClient.TTL(context.Background(), "_client:"+deviceID).Val() > 0, "should have set TTL to the key") - localCopy = usage.Get(deviceID) - assert.Equal(t, "", localCopy.CountryCode) - assert.EqualValues(t, 3, localCopy.Bytes) - - lookup.countryCode = "ir" - newStats() - time.Sleep(10 * time.Millisecond) - result = redisClient.HGetAll(context.Background(), "_client:"+deviceID).Val() - assert.Equal(t, "4", result["bytesIn"]) - assert.Equal(t, "2", result["bytesOut"]) - assert.Equal(t, "ir", result["countryCode"]) - localCopy = usage.Get(deviceID) - assert.Equal(t, "ir", localCopy.CountryCode) - assert.EqualValues(t, 6, localCopy.Bytes) - - lookup.countryCode = "" - newStats() - time.Sleep(10 * time.Millisecond) - result = redisClient.HGetAll(context.Background(), "_client:"+deviceID).Val() - assert.Equal(t, "ir", result["countryCode"], "country code should have been remembered once set") - - uniqueDevicesForToday := redisClient.SMembers(context.Background(), "_devices:ir:"+time.Now().In(time.UTC).Format("2006-01-02")+":forced").Val() - assert.Equal(t, []string{deviceID}, uniqueDevicesForToday) - - _deviceLastSeen := redisClient.Get(context.Background(), "_deviceLastSeen:ir:forced:"+deviceID).Val() - deviceLastSeen, err := strconv.Atoi(_deviceLastSeen) - require.NoError(t, err) - _deviceFirstThrottled := redisClient.Get(context.Background(), "_deviceFirstThrottled:"+deviceID).Val() - deviceFirstThrottled, _ := strconv.Atoi(_deviceFirstThrottled) - - nowUnix := int(time.Now().Unix()) - assert.Greater(t, deviceLastSeen, nowUnix-10) - assert.Less(t, deviceLastSeen, nowUnix+10) - assert.Greater(t, deviceFirstThrottled, nowUnix-10) - assert.Less(t, deviceFirstThrottled, nowUnix+10) -} - -type fakeLookup struct{ countryCode string } - -func (l *fakeLookup) CountryCode(ip net.IP) string { - return l.countryCode -} - -func TestExpirationFor(t *testing.T) { - timeZone := "Asia/Shanghai" - tz, err := time.LoadLocation(timeZone) - require.NoError(t, err) - - thursday := time.Date(2020, 12, 31, 23, 0, 0, 0, tz).In(time.UTC) - friday := time.Date(2021, 1, 1, 0, 0, 0, 0, tz).Add(-1 * time.Nanosecond) - sunday := time.Date(2021, 1, 3, 0, 0, 0, 0, tz).Add(-1 * time.Nanosecond) - nextMonday := time.Date(2021, 1, 4, 0, 0, 0, 0, tz).Add(-1 * time.Nanosecond) - - require.Equal(t, friday.Unix(), expirationFor(thursday, throttle.Daily, timeZone), 0) - require.Equal(t, friday.Unix(), expirationFor(thursday.Add(5*time.Minute), throttle.Daily, timeZone), 0) - require.Equal(t, friday.Unix(), expirationFor(thursday, throttle.Monthly, timeZone), 0) - require.Equal(t, friday.Unix(), expirationFor(thursday, throttle.Legacy, timeZone), 0) - require.Equal(t, friday.Unix(), expirationFor(thursday.Add(5*time.Minute), throttle.Monthly, timeZone), 0) - require.Equal(t, friday.Unix(), expirationFor(thursday.Add(5*time.Minute), throttle.Legacy, timeZone), 0) - - require.Equal(t, nextMonday.Unix(), expirationFor(thursday, throttle.Weekly, timeZone), 0) - require.Equal(t, nextMonday.Unix(), expirationFor(thursday.Add(5*time.Minute), throttle.Weekly, timeZone), 0) - require.Equal(t, nextMonday.Unix(), expirationFor(sunday, throttle.Weekly, timeZone), 0) -} diff --git a/reporting.go b/reporting.go index 53f98c53..91859382 100644 --- a/reporting.go +++ b/reporting.go @@ -6,17 +6,12 @@ import ( "strings" "time" - rclient "github.com/go-redis/redis/v8" - - "github.com/getlantern/geo" "github.com/getlantern/http-proxy-lantern/v2/common" "github.com/getlantern/http-proxy-lantern/v2/datacap" "github.com/getlantern/http-proxy-lantern/v2/listeners" "github.com/getlantern/measured" "github.com/getlantern/http-proxy-lantern/v2/instrument" - "github.com/getlantern/http-proxy-lantern/v2/redis" - "github.com/getlantern/http-proxy-lantern/v2/throttle" ) var ( @@ -28,7 +23,7 @@ type reportingConfig struct { wrapper func(ls net.Listener) net.Listener } -func newReportingConfig(countryLookup geo.CountryLookup, rc *rclient.Client, instrument instrument.Instrument, throttleConfig throttle.Config, datacapTracker *datacap.Tracker) *reportingConfig { +func newReportingConfig(instrument instrument.Instrument, datacapTracker *datacap.Tracker) *reportingConfig { proxiedBytesReporter := func(ctx map[string]interface{}, stats *measured.Stats, deltaStats *measured.Stats, final bool) { noDelta := deltaStats.SentTotal == 0 && deltaStats.RecvTotal == 0 if noDelta && !final { @@ -66,27 +61,15 @@ func newReportingConfig(countryLookup geo.CountryLookup, rc *rclient.Client, ins probingError := fromContext(ctx, common.ProbingError) arch := fromContext(ctx, common.KernelArch) - dataCapCohort := "" - throttleSettings, hasThrottleSettings := ctx[common.ThrottleSettings] - if hasThrottleSettings { - dataCapCohort = throttleSettings.(*throttle.Settings).Label - } - - instrument.ProxiedBytes(context.Background(), deltaStats.SentTotal, deltaStats.RecvTotal, platform, platformVersion, libraryVersion, appVersion, app, locale, dataCapCohort, probingError, client_ip, deviceID, originHost, arch) + instrument.ProxiedBytes(context.Background(), deltaStats.SentTotal, deltaStats.RecvTotal, platform, platformVersion, libraryVersion, appVersion, app, locale, "", probingError, client_ip, deviceID, originHost, arch) } - var reporter listeners.MeasuredReportFN - switch { - case datacapTracker != nil: + reporter := func(ctx map[string]interface{}, stats *measured.Stats, deltaStats *measured.Stats, + final bool) { + // noop: no accounting source configured (pro tracks, local testing) + } + if datacapTracker != nil { reporter = datacapTracker.Reporter() - case throttleConfig == nil: - log.Debug("No throttling configured, don't bother reporting bandwidth usage to Redis") - reporter = func(ctx map[string]interface{}, stats *measured.Stats, deltaStats *measured.Stats, - final bool) { - // noop - } - case rc != nil: - reporter = redis.NewMeasuredReporter(countryLookup, rc, measuredReportingInterval, throttleConfig) } reporter = combineReporter(reporter, proxiedBytesReporter) wrapper := func(ls net.Listener) net.Listener { diff --git a/test.bash b/test.bash index 3f2637e1..d48a4be8 100755 --- a/test.bash +++ b/test.bash @@ -1,32 +1,3 @@ #! /bin/bash -TEST_REDIS_CONTAINER=http-proxy-lantern-test-redis - -function fail() { - echo "$1" - exit 1 -} - -function tearDown() { - echo "Shutting down local test Redis:" - docker stop $TEST_REDIS_CONTAINER -} - -function printRedisLogs() { - if [ "$1" == "true" ]; then - echo "Test Redis logs:" - docker logs $TEST_REDIS_CONTAINER - fi -} - -trap tearDown EXIT - -echo "Starting local test Redis. Container ID:" -docker run \ - --name $TEST_REDIS_CONTAINER \ - -p 6379:6379 \ - -v "$PWD"/test/test-redis-data:/opt/getlantern/ \ - -e ALLOW_EMPTY_PASSWORD=yes \ - --rm -d bitnami/redis:latest || fail "Failed to start local Redis server" - -go test ./... || (printRedisLogs "$1"; exit 1) \ No newline at end of file +go test ./... diff --git a/test/test-redis-data/redis-cert.pem b/test/test-redis-data/redis-cert.pem deleted file mode 100644 index 0c4566fb..00000000 --- a/test/test-redis-data/redis-cert.pem +++ /dev/null @@ -1,22 +0,0 @@ ------BEGIN CERTIFICATE----- -MIIDozCCAougAwIBAgIUMUmJiunMUmiyEPofRVMXlyTa5IIwDQYJKoZIhvcNAQEL -BQAwaDELMAkGA1UEBhMCQ1oxEDAOBgNVBAgMB0JvaGVtaWExDzANBgNVBAcMBlBy -YWd1ZTEUMBIGA1UECgwLV29ybGRzIEZhaXIxDDAKBgNVBAsMA1ImRDESMBAGA1UE -AwwJbG9jYWxob3N0MB4XDTIxMDYxODAwMTcwN1oXDTMxMDYxNjAwMTcwN1owaDEL -MAkGA1UEBhMCQ1oxEDAOBgNVBAgMB0JvaGVtaWExDzANBgNVBAcMBlByYWd1ZTEU -MBIGA1UECgwLV29ybGRzIEZhaXIxDDAKBgNVBAsMA1ImRDESMBAGA1UEAwwJbG9j -YWxob3N0MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAvSE9p6nYAR0r -bXH28C7VbCz/jUEHzJqioj4r26rLrlyuPgIhiX0VsObw6j5SSA2L54U93jq/5Wiy -lhufU4vJjMhJzb7pltuG1kdehLoDFg3paIEj9PKHLo4mKchhQ6jZLIphJYeWVQ9R -GTSp2rbUP4EWU6dTTa9aYb0YNY3e3K/fqD2tLnTDMliuZJPViBp4j2XgoH18w7Qk -7uUAfro2KHDPa6j5sj8PD9mTk+bYixeRRBm9OstP9rgNesQuT6QpAGiJRkqHHeIQ -6j8+ZHXLTUBb8lnKiibGFg9F2q6K2Tt4M1LxmmqVTa/XDA+ASeKTm+jZIzzkqY7o -NYR3NjyCZQIDAQABo0UwQzALBgNVHQ8EBAMCBDAwEwYDVR0lBAwwCgYIKwYBBQUH -AwEwHwYDVR0RBBgwFoIJbG9jYWxob3N0ggkxMjcuMC4wLjEwDQYJKoZIhvcNAQEL -BQADggEBAEfzv/Cs+dUvcIMka3OckoHCycS2Le5cbN+cg7R+J2VHpPTee1XEuJAw -V98KYbw27xYko7bdDuXZPkAKNTIBGmQ63DZ86wtO9pqSHZr15c0r4Ybs5K1aj5Ck -MXp170sVsU0FP36vHO0iW64KX/t6o/YiD+xqoPup9LO3OpmSO1/zTHaZqD8V+AdV -YnsQylPFrdLhwP9Eozkzq03FSsCjUPAJqa/s+x7XFTZI87GLiDcCuq7wwUpNvzTH -rzc0PT6W5lfQC2sZh/1z/Q7gjrlOW02NMBqDTkpkvUt7REPP3cLWR29xqTwFAzT9 -eQu4ArP9p4D2/HKXGIAveo3SS7OWOKQ= ------END CERTIFICATE----- diff --git a/test/test-redis-data/redis-key.pem b/test/test-redis-data/redis-key.pem deleted file mode 100644 index 0febd105..00000000 --- a/test/test-redis-data/redis-key.pem +++ /dev/null @@ -1,28 +0,0 @@ ------BEGIN PRIVATE KEY----- -MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC9IT2nqdgBHStt -cfbwLtVsLP+NQQfMmqKiPivbqsuuXK4+AiGJfRWw5vDqPlJIDYvnhT3eOr/laLKW -G59Ti8mMyEnNvumW24bWR16EugMWDelogSP08ocujiYpyGFDqNksimElh5ZVD1EZ -NKnattQ/gRZTp1NNr1phvRg1jd7cr9+oPa0udMMyWK5kk9WIGniPZeCgfXzDtCTu -5QB+ujYocM9rqPmyPw8P2ZOT5tiLF5FEGb06y0/2uA16xC5PpCkAaIlGSocd4hDq -Pz5kdctNQFvyWcqKJsYWD0XarorZO3gzUvGaapVNr9cMD4BJ4pOb6NkjPOSpjug1 -hHc2PIJlAgMBAAECggEAIwUPrIb3dyVWkDpk9g5F+HTQkPA0qH8NlCsc5dzXQB7r -KJIucY3WwV3RyB3oEG1ptWHnlcmdFLZxMvhEZSziEg5YFB/Aku1tJyop5zumLKRA -ztfyt31c6JLroBr5X5TZObUZzeAwRunCI3+r39KwtF2Lq8u00hKhEq332Jq/ZG9e -a062opeNmWQKbezpaj2BFBh7domie2sx7bzfDm6iLNhqeztnsSlyqtXPcKDk3yHP -3YTXzbXTYr70mEijXX0l3L9Y0q1QZdgsQ1uf0MVorLJRPM3jRZMFfsJ+YNQiHBzl -JpnnQ2Rvu+Q/py4Z51D2HpGS+mxXAUVbP/ngucMmoQKBgQD1UMMh+APB4N+Q8CNR -yANPPZuikZvzM4E37ieZeSpZ/cLB4lw5POvsvYc0/eKog4t9z28erZwDtP5TvGRZ -mTcwWKiglYeF7PyFBx1+EhrzMqQEfvNFonvwKBi/Ox0fg80waivx+jo43/dUW9zX -qGUMUc04lvh92QoC+WhhMcELqQKBgQDFXgQBrm54N21Y53Kd+C+7RSREs/SF+c/1 -BVJoaFxvrlPcTkEmjQ4G9D6eKPwz6RzAoDb8P48tXDGbYacJji9M32rmr5PVGNXO -eeLhqogLIgljY3hoLvyaMLYC6gaxeQUG4kwxkyhtITS2V+Ur3Dl2rP3JLcrmlIOz -/V4UofzWXQKBgHbHJ/Qmlgslypnu6+sJITEY72aDgjL7WTGvilTxKeRwzf/6jhTY -vpLeIb0ywLA/ISdFbNQ50zSoSgxZR21qliI+i0dVn0zlNk9i5pDIS2w9tNTIcxng -Vootn+3XvX4o3wz2G1kKg25OYv7hi8iPrH2OjgM8dgzZLdPK1xNYd9QhAoGAf2yb -mZpMqEDg0C1dyq+Z4YZaTCEDFhybLs4ozk/Snigm7G15krIvBD5czzAW5Ez0y0il -N2axwc9sF62Mcpbh4ve7xQRGpaZLI+t9l8TJ4KYw+YIXEuyDGempGWiOubYwkKk6 -GCeySgLOTYuh4hdk4kH9+A0gmo34GhTS0O9kleECgYEAvJnVlzI7VimR4kCGyfzj -9yVmS2GWDYgIt1zHtzCY//6YJSj2iafVafoTeGfNOUvl4+XqqMEQQRu/tyuy6YF6 -gile2U20UbdOjFQ4Tcz94q291mhPKRmhRXryEOoLSAxNfxuL4YN+tp0sfD0vQOse -OnmyXxZkUXitoKKMYclX2ts= ------END PRIVATE KEY----- diff --git a/throttle/throttle.go b/throttle/throttle.go deleted file mode 100644 index 96e655ce..00000000 --- a/throttle/throttle.go +++ /dev/null @@ -1,268 +0,0 @@ -// Package throttle provides the ability to read throttling configurations from -// redis. Configurations are stored in redis as maps under the keys -// "_throttle:desktop" and "_throttle:mobile". The key/value pairs in each map -// are the 2-digit lowercase ISO-3166 country code plus a pipe-delimited -// threshold and rate, for example: -// -// _throttle:mobile -// "__" "524288000|10240" -// "cn" "104857600|10240" -// -package throttle - -import ( - "context" - "encoding/json" - "strings" - "sync" - "time" - - "github.com/go-redis/redis/v8" - "github.com/spaolacci/murmur3" - - "github.com/getlantern/errors" - "github.com/getlantern/golog" -) - -const ( - DefaultRefreshInterval = 5 * time.Minute -) - -var ( - log = golog.LoggerFor("flashlight.throttle") -) - -type CapInterval string - -const ( - Daily = "daily" - Weekly = "weekly" - Monthly = "monthly" - Legacy = "legacy" // like Monthly for old clients -) - -type Settings struct { - // Label uniquely identifies this set of settings for reporting purposes - Label string - - // AppName constrains this setting to a particular application name. Leave blank to apply to all applications. - AppName string - - // DeviceFloor is an optional number between 0 and 1 that sets the floor (inclusive) of devices included in the cohort that gets these settings - DeviceFloor float64 - - // DeviceCeil is an optional number between 0 and 1 that sets the floor (exclusive) of devices included in the cohort that gets these settings. - // If DeviceCeil is 1, the 1 is treated as inclusive. - DeviceCeil float64 - - // Threshold at which we start throttling (in bytes) - Threshold int64 - - // Rate to which to throttle (in bytes per second) - Rate int64 - - // How frequently the usage cap resets, one of "daily", "weekly" or "monthly" - CapResets CapInterval -} - -func (settings *Settings) Validate() error { - if settings.Label == "" { - return errors.New("Missing label") - } - - if settings.CapResets != Daily && settings.CapResets != Weekly && settings.CapResets != Monthly { - return errors.New("Unknown CapResets interval %v: ", settings.CapResets) - } - - if settings.Threshold > 0 && settings.Rate <= 0 { - return errors.New("Throttling threshold specified without a rate") - } - - return nil -} - -// Config is a per-country throttling config -type Config interface { - // SettingsFor returns the throttling settings for the given deviceID in the given - // countryCode on the given platform (windows, darwin, linux, android or ios). At the each level - // (country and platform) this should fall back to default values if a specific value isn't provided. - // supportedDataCaps identifies which cap intervals the client supports ("daily", "weekly" or "monthly"). - // If this list is empty, the client is assumed to support "monthly" (legacy clients). - SettingsFor(deviceID, countryCode, platform, appName string, supportedDataCaps []string) (settings *Settings, ok bool) -} - -// NewForcedConfig returns a new Config that uses the forced threshold, rate and TTL -func NewForcedConfig(threshold int64, rate int64, capResets CapInterval) Config { - return &forcedConfig{ - Settings: Settings{ - Label: "forced", - Threshold: threshold, - Rate: rate, - CapResets: capResets, - }, - } -} - -type forcedConfig struct { - Settings -} - -func (cfg *forcedConfig) SettingsFor(deviceID, countryCode, platform, appName string, supportedDataCaps []string) (settings *Settings, ok bool) { - return &cfg.Settings, true -} - -// SettingsByCountryAndPlatform organizes slices of SettingsWithConstraints by -// country -> platform -type SettingsByCountryAndPlatform map[string]map[string][]*Settings - -func (sbcap SettingsByCountryAndPlatform) Validate() error { - for _, platforms := range sbcap { - for _, cohorts := range platforms { - for _, settings := range cohorts { - err := settings.Validate() - if err != nil { - return err - } - } - } - } - return nil -} - -func decodeSettingsByCountryAndPlatform(encoded []byte) (settings SettingsByCountryAndPlatform, err error) { - settings = make(SettingsByCountryAndPlatform) - err = json.Unmarshal(encoded, &settings) - return -} - -type redisConfig struct { - rc *redis.Client - refreshInterval time.Duration - settings SettingsByCountryAndPlatform - mx sync.RWMutex - ctx context.Context -} - -// NewRedisConfig returns a new Config that uses the given redis client to load -// its configuration information and reload that information every -// refreshInterval. -func NewRedisConfig(rc *redis.Client, refreshInterval time.Duration) Config { - cfg := &redisConfig{ - rc: rc, - refreshInterval: refreshInterval, - ctx: context.Background(), - } - cfg.refreshSettings() - go cfg.keepCurrent() - return cfg -} - -func (cfg *redisConfig) keepCurrent() { - if cfg.refreshInterval <= 0 { - log.Debugf("Defaulting refresh interval to %v", DefaultRefreshInterval) - cfg.refreshInterval = DefaultRefreshInterval - } - - log.Debugf("Refreshing every %v", cfg.refreshInterval) - for { - time.Sleep(cfg.refreshInterval) - cfg.refreshSettings() - } -} - -func (cfg *redisConfig) refreshSettings() { - encoded, err := cfg.rc.Get(cfg.ctx, "_throttle").Bytes() - if err != nil { - log.Errorf("Unable to load throttle settings from redis: %v", err) - return - } - settings, err := decodeSettingsByCountryAndPlatform(encoded) - if err != nil { - log.Errorf("Unable to decode throttle settings: %v", err) - return - } - - // too much info in this log. disabled - // log.Debugf("Loaded throttle config: %v", string(encoded)) - - cfg.mx.Lock() - cfg.settings = settings - cfg.mx.Unlock() -} - -func (cfg *redisConfig) SettingsFor(deviceID, countryCode, platform, appName string, supportedDataCaps []string) (*Settings, bool) { - cfg.mx.RLock() - settings := cfg.settings - cfg.mx.RUnlock() - - platformSettings := settings[strings.ToLower(countryCode)] - if platformSettings == nil { - log.Tracef("No settings found for country %v, use default", countryCode) - platformSettings = settings["default"] - if platformSettings == nil { - log.Trace("No settings for default country, not throttling") - return nil, false - } - } - - constrainedSettings := platformSettings[strings.ToLower(platform)] - if len(constrainedSettings) == 0 { - log.Tracef("No settings found for platform %v, use default", platform) - constrainedSettings = platformSettings["default"] - if len(constrainedSettings) == 0 { - log.Trace("No settings for default platform, not throttling") - return nil, false - } - } - - clientSupportsInterval := func(requested CapInterval) bool { - if requested == Legacy && len(supportedDataCaps) == 0 { - // legacy client - return true - } - for _, supported := range supportedDataCaps { - if requested == CapInterval(supported) { - return true - } - } - return false - } - - hash := murmur3.New64() - hash.Write([]byte(deviceID)) - hashOfDeviceID := hash.Sum64() - const scale = 1000000 // do not change this, as it will result in users being segmented differently than they were before - segment := float64((hashOfDeviceID % scale)) / float64(scale) - - settingsForAppName := func(checkAppName string) *Settings { - for _, candidateSettings := range constrainedSettings { - if clientSupportsInterval(candidateSettings.CapResets) { - appMatches := candidateSettings.AppName == checkAppName - deviceMatches := candidateSettings.DeviceFloor <= segment && (candidateSettings.DeviceCeil > segment || (candidateSettings.DeviceCeil == 1 && segment == 1)) - if appMatches && deviceMatches { - return candidateSettings - } - } - } - - log.Tracef("No setting for segment %v, using first supported in list", segment) - for _, candidateSettings := range constrainedSettings { - if clientSupportsInterval(candidateSettings.CapResets) { - appMatches := candidateSettings.AppName == checkAppName - if appMatches { - return candidateSettings - } - } - } - - return nil - } - - result := settingsForAppName(appName) - if result == nil && appName != "" { - log.Tracef("No applicable settings found for app name %v, trying with no app name", appName) - result = settingsForAppName("") - } - - return result, result != nil -} diff --git a/throttle/throttle_test.go b/throttle/throttle_test.go deleted file mode 100644 index a0044816..00000000 --- a/throttle/throttle_test.go +++ /dev/null @@ -1,137 +0,0 @@ -package throttle - -import ( - "context" - "crypto/tls" - "strings" - "testing" - "time" - - "github.com/getlantern/golog/testlog" - "github.com/getlantern/http-proxy-lantern/v2/internal/testutil" - "github.com/go-redis/redis/v8" - "github.com/stretchr/testify/require" -) - -const ( - refreshInterval = 10 * time.Millisecond - - deviceIDInSegment1 = "74" // this falls in segment 0.300786 - deviceIDInSegment2 = "78" // this falls in segment 0.914739 - deviceIDWithNoSegment = "55" // this falls in segment 0.016255 - - goodSettings = ` -{ - "default": { - "default": [ - {"label": "cohort 1", "deviceFloor": 0.1, "deviceCeil": 0.5, "threshold": 1000, "rate": 100, "capResets": "weekly"}, - {"label": "cohort 2", "deviceFloor": 0.5, "deviceCeil": 1.0, "threshold": 1100, "rate": 110, "capResets": "monthly"} - ], - "windows": [ - {"label": "cohort 3", "deviceFloor": 0.1, "deviceCeil": 0.5, "threshold": 2000, "rate": 200, "capResets": "weekly"}, - {"label": "cohort 4", "deviceFloor": 0.5, "deviceCeil": 1.0, "threshold": 2100, "rate": 210, "capResets": "monthly"} - ] - }, - "cn": { - "default": [ - {"label": "cohort 5", "deviceFloor": 0.1, "deviceCeil": 0.5, "threshold": 3000, "rate": 300, "capResets": "weekly"}, - {"label": "cohort 6", "deviceFloor": 0.5, "deviceCeil": 1.0, "threshold": 3100, "rate": 310, "capResets": "monthly"}, - {"label": "cohort 6", "deviceFloor": 0.5, "deviceCeil": 1.0, "threshold": 3200, "rate": 320, "capResets": "legacy"} - ], - "windows": [ - {"label": "cohort 7", "deviceFloor": 0.1, "deviceCeil": 0.5, "threshold": 4000, "rate": 400, "capResets": "weekly"}, - {"label": "cohort 8", "deviceFloor": 0.5, "deviceCeil": 1.0, "threshold": 4100, "rate": 410, "capResets": "monthly"}, - {"label": "cohort 8", "deviceFloor": 0.5, "deviceCeil": 1.0, "threshold": 4200, "rate": 420, "capResets": "legacy"} - ] - }, - "ir": { - "default": [ - {"label": "capped", "threshold": 1000, "rate": 100, "capResets": "monthly"}, - {"label": "notcapped", "threshold": 1000000, "rate": 100, "capResets": "monthly", "appName": "specialapp"} - ] - } -}` -) - -func doTest(t *testing.T, cfg Config, deviceID, countryCode, platform, appName string, supportedDataCaps []string, expectedThreshold int64, expectedRate int64, expectedCapResets CapInterval, testCase string) { - settings, ok := cfg.SettingsFor(deviceID, countryCode, platform, appName, supportedDataCaps) - require.True(t, ok, "valid config for "+testCase) - require.NotNil(t, settings, "non-nil settings for "+testCase) - require.Equal(t, expectedThreshold, settings.Threshold, "correct threshold for "+testCase) - require.Equal(t, expectedRate, settings.Rate, testCase, "correct rate for "+testCase) - require.Equal(t, expectedCapResets, settings.CapResets, testCase, "correct ttl for "+testCase) -} - -func TestThrottleConfig(t *testing.T) { - stopCapture := testlog.Capture(t) - defer stopCapture() - - rc := testutil.TestRedis(t) - - // try a bad config first - require.NoError(t, rc.Set(context.Background(), "_throttle", "blah I'm bad settings blah", 0).Err()) - cfg := NewRedisConfig(rc, refreshInterval) - _, ok := cfg.SettingsFor(deviceIDInSegment1, "cn", "windows", "lantern", []string{"monthly", "weekly"}) - require.False(t, ok, "Loading throttle settings from bad config should fail") - - // now do a good config - require.NoError(t, rc.Set(context.Background(), "_throttle", goodSettings, 0).Err()) - cfg = NewRedisConfig(rc, refreshInterval) - - doTest(t, cfg, deviceIDInSegment1, "cn", "windows", "lantern", []string{"monthly", "weekly"}, 4000, 400, "weekly", "known country, known platform, segment 1") - doTest(t, cfg, deviceIDInSegment2, "cn", "windows", "lantern", []string{"monthly", "weekly"}, 4100, 410, "monthly", "known country, known platform, segment 2") - doTest(t, cfg, deviceIDInSegment1, "cn", "windows", "lantern", []string{"monthly", "weekly"}, 4000, 400, "weekly", "known country, known platform, unknown segment") - doTest(t, cfg, deviceIDInSegment1, "cn", "windows", "lantern", nil, 4200, 420, "legacy", "known country, known platform, segment 1, legacy client") - - doTest(t, cfg, deviceIDInSegment1, "cn", "", "lantern", []string{"monthly", "weekly"}, 3000, 300, "weekly", "known country, unknown platform, segment 1") - doTest(t, cfg, deviceIDInSegment2, "cn", "", "lantern", []string{"monthly", "weekly"}, 3100, 310, "monthly", "known country, unknown platform, segment 2") - doTest(t, cfg, deviceIDInSegment1, "cn", "", "lantern", []string{"monthly", "weekly"}, 3000, 300, "weekly", "known country, unknown platform, unknown segment") - - doTest(t, cfg, deviceIDInSegment1, "de", "windows", "lantern", []string{"monthly", "weekly"}, 2000, 200, "weekly", "unknown country, known platform, segment 1") - doTest(t, cfg, deviceIDInSegment2, "de", "windows", "lantern", []string{"monthly", "weekly"}, 2100, 210, "monthly", "unknown country, known platform, segment 2") - doTest(t, cfg, deviceIDInSegment1, "de", "windows", "lantern", []string{"monthly", "weekly"}, 2000, 200, "weekly", "unknown country, known platform, unknown segment") - - doTest(t, cfg, deviceIDInSegment1, "de", "", "lantern", []string{"monthly", "weekly"}, 1000, 100, "weekly", "unknown country, unknown platform, segment 1") - doTest(t, cfg, deviceIDInSegment2, "de", "", "lantern", []string{"monthly", "weekly"}, 1100, 110, "monthly", "unknown country, unknown platform, segment 2") - doTest(t, cfg, deviceIDInSegment1, "de", "", "lantern", []string{"monthly", "weekly"}, 1000, 100, "weekly", "unknown country, unknown platform, unknown segment") - - doTest(t, cfg, deviceIDInSegment1, "ir", "", "lantern", []string{"monthly", "weekly"}, 1000, 100, "monthly", "capped named app") - doTest(t, cfg, deviceIDInSegment1, "ir", "", "", []string{"monthly", "weekly"}, 1000, 100, "monthly", "capped unnamed app") - doTest(t, cfg, deviceIDInSegment1, "ir", "", "specialapp", []string{"monthly", "weekly"}, 1000000, 100, "monthly", "uncapped app") - - // update settings - require.NoError(t, rc.Set(context.Background(), "_throttle", strings.ReplaceAll(goodSettings, "4", "5"), 0).Err()) - time.Sleep(refreshInterval * 2) - - doTest(t, cfg, deviceIDInSegment1, "cn", "windows", "lantern", []string{"monthly", "weekly"}, 5000, 500, "weekly", "known country, known platform, segment 1, after update") -} - -func TestForcedConfig(t *testing.T) { - stopCapture := testlog.Capture(t) - defer stopCapture() - - cfg := NewForcedConfig(1024, 512, "weekly") - doTest(t, cfg, deviceIDInSegment1, "", "", "lantern", []string{"monthly", "weekly"}, 1024, 512, "weekly", "forced config") -} - -func TestFailToConnectRedis(t *testing.T) { - stopCapture := testlog.Capture(t) - defer stopCapture() - - bogusClient := redis.NewClient(&redis.Options{ - Addr: "localhost:80", - TLSConfig: &tls.Config{InsecureSkipVerify: true}, - }) - - cfg := NewRedisConfig(bogusClient, refreshInterval) - _, ok := cfg.SettingsFor(deviceIDInSegment1, "cn", "windows", "lantern", []string{"monthly", "weekly"}) - require.False(t, ok, "Loading throttle settings when unable to contact redis should fail") - - redisClient := testutil.TestRedis(t) - cfg = NewRedisConfig(redisClient, refreshInterval) - require.NoError(t, redisClient.Set(context.Background(), "_throttle", goodSettings, 0).Err()) - - time.Sleep(refreshInterval * 2) - // Should load the config when Redis is back up online - doTest(t, cfg, deviceIDInSegment1, "cn", "windows", "lantern", []string{"monthly", "weekly"}, 4000, 400, "weekly", "known country, known platform, segment 1, redis back online") -} diff --git a/throttle_integration_test.go b/throttle_integration_test.go deleted file mode 100644 index a49d5821..00000000 --- a/throttle_integration_test.go +++ /dev/null @@ -1,240 +0,0 @@ -package proxy - -import ( - "context" - "fmt" - "io" - "io/ioutil" - "math/rand" - "net" - "net/http" - "net/http/httptest" - "net/url" - "strconv" - "strings" - "sync" - "sync/atomic" - "testing" - "time" - - "github.com/getlantern/golog/testlog" - . "github.com/getlantern/waitforserver" - "github.com/stretchr/testify/assert" - "github.com/stretchr/testify/require" - - "github.com/getlantern/http-proxy-lantern/v2/common" - "github.com/getlantern/http-proxy-lantern/v2/internal/testutil" - "github.com/getlantern/http-proxy-lantern/v2/throttle" -) - -const ( - timezone = "Asia/Shanghai" -) - -// Requires a Redis setup created in `make test` -func TestThrottling(t *testing.T) { - stopCapture := testlog.Capture(t) - defer stopCapture() - - origMeasuredReportingInterval := measuredReportingInterval - measuredReportingInterval = 10 * time.Millisecond - defer func() { - measuredReportingInterval = origMeasuredReportingInterval - }() - - throttleThreshold := 10485760 - throttleRate := 10240 - t.Run("free_config_when_redis_is_down", func(t *testing.T) { - doTestThrottling(t, false, "127.0.0.1:18707", false, throttleThreshold, throttleRate) - }) - t.Run("disabling_throttling_via_redis", func(t *testing.T) { - doTestThrottling(t, true, "127.0.0.1:18709", true, 0, throttleRate) - }) - t.Run("free_config_from_redis", func(t *testing.T) { - doTestThrottling(t, false, "127.0.0.1:18711", true, throttleThreshold, throttleRate) - }) -} - -func doTestThrottling(t *testing.T, pro bool, serverAddr string, redisIsUp bool, throttleThreshold, throttleRate int) { - deviceId := fmt.Sprintf("dev-%d", rand.Int()) - sizeHeader := "X-Test-Size" - originSite := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) { - n, _ := strconv.Atoi(req.Header.Get(sizeHeader)) - io.CopyN(rw, rand.New(rand.NewSource(time.Now().UnixNano())), int64(n)) - })) - originAddr := originSite.Listener.Addr().String() - log.Debugf("Waiting for origin server at %s...", originAddr) - require.NoError(t, WaitForServer("tcp", originAddr, 10*time.Second)) - - redisClient := testutil.TestRedis(t) - if redisIsUp { - settings := fmt.Sprintf(`{"default": { "default": [{"capResets": "daily", "threshold": %d, "rate": %d}] } }`, throttleThreshold, throttleRate) - require.NoError(t, redisClient.Set(context.Background(), "_throttle", settings, 0).Err()) - } - - durationForBytes := func(bytes int) time.Duration { - // the buckets will be full initially, so these will be available immediately. - lbytes := bytes - throttleRate - if lbytes <= 0 { - return 0 - } - - return time.Duration(1000*float64(lbytes)/float64(throttleRate)) * time.Millisecond - } - - proxy := &Proxy{ - HTTPAddr: serverAddr, - ReportingRedisClient: redisClient, - Token: validToken, - IdleTimeout: 1 * time.Minute, - Pro: pro, - ThrottleRefreshInterval: throttle.DefaultRefreshInterval, - TestingLocal: true, - GoogleSearchRegex: "bequiet", - GoogleCaptchaRegex: "bequiet", - } - go func() { - assert.NoError(t, proxy.ListenAndServe(context.Background())) - }() - - require.NoError(t, WaitForServer("tcp", serverAddr, 10*time.Second)) - - makeRequest := func(u string, testSize int) (*http.Response, int, error) { - var conn *ReadSizeConn - client := &http.Client{ - Transport: &http.Transport{ - DisableKeepAlives: true, - Proxy: func(req *http.Request) (*url.URL, error) { - return url.Parse("http://" + serverAddr) - }, - DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { - var d net.Dialer - c, err := d.DialContext(ctx, network, addr) - if err != nil { - return nil, err - } - wrapped := &ReadSizeConn{Conn: c} - conn = wrapped - return wrapped, nil - }, - }, - } - - req, _ := http.NewRequest(http.MethodGet, u, nil) - req.Header.Set(common.TokenHeader, validToken) - req.Header.Set(common.DeviceIdHeader, deviceId) - req.Header.Add(common.SupportedDataCapsHeader, throttle.Daily) - req.Header.Set(common.TimeZoneHeader, timezone) - req.Header.Set(sizeHeader, strconv.Itoa(testSize)) - - resp, err := client.Do(req) - if err != nil { - return nil, 0, err - } - - _, err = io.Copy(ioutil.Discard, resp.Body) - - rs := 0 - if conn != nil { - rs = conn.readSize - } - return resp, rs, err - } - - resp, _, err := makeRequest(originSite.URL, 9*1024*1024) - require.NoError(t, err) - - resp, _, err = makeRequest(originSite.URL, 1024*1024) - require.NoError(t, err) - - time.Sleep(time.Second) - - start := time.Now() - resp, sz, err := makeRequest(originSite.URL, 3*throttleRate) - require.NoError(t, err) - xbq := resp.Header.Get(common.XBQHeader) - xbqv2 := resp.Header.Get(common.XBQHeaderv2) - if !redisIsUp || throttleThreshold <= 0 { - assert.Empty(t, xbq) - assert.Empty(t, xbqv2) - return - } - - if pro { - assert.Empty(t, xbq) - } else { - assert.InDelta(t, durationForBytes(sz), time.Since(start), float64(100*time.Millisecond), - fmt.Sprintf("per connection throttling should be in effect for Free proxy sz=%d", sz)) - - require.NotEmpty(t, xbq) - - parts := strings.Split(xbqv2, "/") - require.Len(t, parts, 4) - require.Len(t, strings.Split(xbq, "/"), 3) - - log.Debugf("XBQ is: %v", xbq) - assert.NotEqual(t, "0", parts[0], "Should show some usage") - assert.Equal(t, "10", parts[1], "Should show correct bandwidth limit") - - time.Sleep(time.Second) - // Now test throttling concurrent connections from a single device - readers := 16 - readSize := 3 * throttleRate - - errors := make(chan error, readers) - var sz int64 - var wg sync.WaitGroup - - start := time.Now() - for i := 0; i < readers; i++ { - wg.Add(1) - go func() { - _, ss, err := makeRequest(originSite.URL, readSize) - atomic.AddInt64(&sz, int64(ss)) - errors <- err - wg.Done() - }() - } - wg.Wait() - endTime := time.Since(start) - close(errors) - for err := range errors { - assert.NoError(t, err) - } - assert.InDelta(t, durationForBytes(int(sz)), endTime, float64(150*time.Millisecond), - fmt.Sprintf("throttling should be applied to each connection generated by the device sz=%d", sz)) - } - - result, err := redisClient.HMGet(context.Background(), "_client:"+deviceId, "bytesIn", "bytesOut", "countryCode", "clientIP").Result() - - bytesIn, err := strconv.Atoi(result[0].(string)) - require.NoError(t, err) - bytesOut, err := strconv.Atoi(result[1].(string)) - require.NoError(t, err) - assert.True(t, bytesIn > 0) - assert.True(t, bytesOut > 0) - assert.Equal(t, "", result[2]) - assert.Equal(t, "127.0.0.1", result[3]) - - // Can't run the below test condition because reading TTL doesn't work with the LedisDB that we use in testing - // tz, err := time.LoadLocation(timezone) - // require.NoError(t, err) - // now := time.Now().In(tz) - // tomorrow := now.AddDate(0, 0, 1) - // beginningOfTomorrow := time.Date(tomorrow.Year(), tomorrow.Month(), tomorrow.Day(), 0, 0, 0, 0, now.Location()).Add(-1 * time.Nanosecond) - // timeUntilBeginningOfTomorrow := beginningOfTomorrow.Sub(now) - // ttl := rc.TTL("_client:" + deviceId).Val() - // require.True(t, ttl-time.Minute < timeUntilBeginningOfTomorrow && ttl+time.Minute > timeUntilBeginningOfTomorrow, "ttl of %v should be in the right ballpark of %v until %v", ttl, timeUntilBeginningOfTomorrow, beginningOfTomorrow) -} - -// utility for observing bytes read -type ReadSizeConn struct { - net.Conn - readSize int -} - -func (c *ReadSizeConn) Read(b []byte) (n int, err error) { - n, err = c.Conn.Read(b) - c.readSize += n - return -} diff --git a/usage/usage.go b/usage/usage.go deleted file mode 100644 index 8783f5c6..00000000 --- a/usage/usage.go +++ /dev/null @@ -1,33 +0,0 @@ -package usage - -import ( - "sync" - "time" -) - -var ( - mutex sync.RWMutex - usageByDeviceID = make(map[string]*Usage) -) - -type Usage struct { - CountryCode string - Bytes int64 - AsOf time.Time - TTLSeconds int64 -} - -// Set sets the Usage in bytes for the given device as of the given time and known to be resetting within ttlSeconds -func Set(dev string, countryCode string, usage int64, asOf time.Time, ttlSeconds int64) { - mutex.Lock() - usageByDeviceID[dev] = &Usage{countryCode, usage, asOf, ttlSeconds} - mutex.Unlock() -} - -// Get gets the Usage for the given device. -func Get(dev string) *Usage { - mutex.RLock() - result := usageByDeviceID[dev] - mutex.RUnlock() - return result -}