Remove the redis-reporting path (eng#3813 phase 3)

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

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"
 )

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.

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,177 +69,21 @@ 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:
-//
-// <used>/<allowed>/<asof>
-//
-// <used> is the string representation of a 64-bit unsigned integer
-// <allowed> is the string representation of a 64-bit unsigned integer
-// <asof> 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
 	instrument    instrument.Instrument
 }
 
 // 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,

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

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=

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,