Rolling window limit support

README.md

@@ -181,6 +181,20 @@ The rate limit block specifies the actual rate limit that will be used when ther
 Currently the service supports per second, minute, hour, and day limits. More types of limits may be added in the
 future based on user demand.
 
+### Rate limit algorithm
+
+Ratelimit supports two algorithms:
+
+1. Fixed window
+For a limit of 60 requests per hour, there can only 60 requests in a single time window (e.g: 01:00 - 01:59).
+Fixed window algorithm does not care when did the request arrive, all 60 can arrive at 01:01 or 01:50 and the limit will still reset at 02:00.
+
+2. Rolling window
+For a limit of 60 requests per hour. Initially rate limiter can take a burst of 60 requests at once, then the limit is restored by 1 each minute. Requests are allowed as long as there's still some available limit.
+
+Configure rate limit algorithm with `RATE_LIMIT_ALGORITHM` environment variable.
+Use `FIXED_WINDOW` and `ROLLING_WINDOW` respectively.
+
 ### Examples
 
 #### Example 1

go.mod

@@ -9,7 +9,7 @@ require (
 	github.com/coocood/freecache v1.1.0
 	github.com/envoyproxy/go-control-plane v0.9.7
 	github.com/fsnotify/fsnotify v1.4.7 // indirect
-	github.com/golang/mock v1.4.1
+	github.com/golang/mock v1.4.4
 	github.com/golang/protobuf v1.4.2
 	github.com/gorilla/mux v1.7.4-0.20191121170500-49c01487a141
 	github.com/kavu/go_reuseport v1.2.0
@@ -26,4 +26,5 @@ require (
 	google.golang.org/grpc v1.27.0
 	google.golang.org/protobuf v1.25.0 // indirect
 	gopkg.in/yaml.v2 v2.3.0
+	rsc.io/quote/v3 v3.1.0 // indirect
 )

go.sum

@@ -35,6 +35,8 @@ github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfU
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/mock v1.4.1 h1:ocYkMQY5RrXTYgXl7ICpV0IXwlEQGwKIsery4gyXa1U=
 github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
+github.com/golang/mock v1.4.4 h1:l75CXGRSwbaYNpl/Z2X1XIIAMSCquvXgpVZDhwEIJsc=
+github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=

src/algorithm/base_window.go

@@ -0,0 +1,104 @@
+package algorithm
+
+import (
+	"github.com/coocood/freecache"
+	pb "github.com/envoyproxy/go-control-plane/envoy/service/ratelimit/v3"
+	"github.com/envoyproxy/ratelimit/src/config"
+	"github.com/envoyproxy/ratelimit/src/utils"
+	logger "github.com/sirupsen/logrus"
+)
+
+type WindowImpl struct {
+	algorithm         RatelimitAlgorithm
+	cacheKeyGenerator utils.CacheKeyGenerator
+	localCache        *freecache.Cache
+	timeSource        utils.TimeSource
+}
+
+func (w *WindowImpl) GetResponseDescriptorStatus(key string, limit *config.RateLimit, results int64,
+	isOverLimitWithLocalCache bool, hitsAddend int64) *pb.RateLimitResponse_DescriptorStatus {
+	if key == "" {
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:           pb.RateLimitResponse_OK,
+			CurrentLimit:   nil,
+			LimitRemaining: 0,
+		}
+	}
+
+	if isOverLimitWithLocalCache {
+		PopulateStats(limit, 0, uint64(hitsAddend), uint64(hitsAddend))
+
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:               pb.RateLimitResponse_OVER_LIMIT,
+			CurrentLimit:       limit.Limit,
+			LimitRemaining:     0,
+			DurationUntilReset: w.algorithm.CalculateSimpleReset(limit, w.timeSource),
+		}
+	}
+
+	isOverLimit, limitRemaining, durationUntilReset := w.algorithm.IsOverLimit(limit, int64(results), hitsAddend)
+
+	if !isOverLimit {
+		duration := w.algorithm.CalculateReset(isOverLimit, limit, w.timeSource)
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:               pb.RateLimitResponse_OK,
+			CurrentLimit:       limit.Limit,
+			LimitRemaining:     uint32(limitRemaining),
+			DurationUntilReset: duration,
+		}
+	} else {
+		if w.localCache != nil {
+			durationUntilReset = utils.MaxInt(1, durationUntilReset)
+
+			err := w.localCache.Set([]byte(key), []byte{}, durationUntilReset)
+			if err != nil {
+				logger.Errorf("Failing to set local cache key: %s", key)
+			}
+		}
+		duration := w.algorithm.CalculateReset(isOverLimit, limit, w.timeSource)
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:               pb.RateLimitResponse_OVER_LIMIT,
+			CurrentLimit:       limit.Limit,
+			LimitRemaining:     0,
+			DurationUntilReset: duration,
+		}
+	}
+}
+
+func (w *WindowImpl) IsOverLimitWithLocalCache(key string) bool {
+	if w.localCache != nil {
+		_, err := w.localCache.Get([]byte(key))
+		if err == nil {
+			return true
+		}
+	}
+	return false
+}
+
+func (w *WindowImpl) GenerateCacheKeys(request *pb.RateLimitRequest,
+	limits []*config.RateLimit, hitsAddend int64, timestamp int64) []utils.CacheKey {
+	return w.cacheKeyGenerator.GenerateCacheKeys(request, limits, uint32(hitsAddend), timestamp)
+}
+
+func (w *WindowImpl) GetExpirationSeconds() int64 {
+	return w.algorithm.GetExpirationSeconds()
+}
+
+func (w *WindowImpl) GetResultsAfterIncrease() int64 {
+	return w.algorithm.GetResultsAfterIncrease()
+}
+
+func PopulateStats(limit *config.RateLimit, nearLimit uint64, overLimit uint64, overLimitWithLocalCache uint64) {
+	limit.Stats.NearLimit.Add(nearLimit)
+	limit.Stats.OverLimit.Add(overLimit)
+	limit.Stats.OverLimitWithLocalCache.Add(overLimitWithLocalCache)
+}
+
+func NewWindow(algorithm RatelimitAlgorithm, cacheKeyPrefix string, localCache *freecache.Cache, timeSource utils.TimeSource) *WindowImpl {
+	return &WindowImpl{
+		algorithm:         algorithm,
+		cacheKeyGenerator: utils.NewCacheKeyGenerator(cacheKeyPrefix),
+		localCache:        localCache,
+		timeSource:        timeSource,
+	}
+}

src/algorithm/fixed_window.go

@@ -0,0 +1,72 @@
+package algorithm
+
+import (
+	"math"
+
+	"github.com/golang/protobuf/ptypes/duration"
+
+	"github.com/coocood/freecache"
+	"github.com/envoyproxy/ratelimit/src/config"
+	"github.com/envoyproxy/ratelimit/src/utils"
+)
+
+var _ RatelimitAlgorithm = (*FixedWindowImpl)(nil)
+
+type FixedWindowImpl struct {
+	timeSource        utils.TimeSource
+	cacheKeyGenerator utils.CacheKeyGenerator
+	localCache        *freecache.Cache
+	nearLimitRatio    float32
+}
+
+func (fw *FixedWindowImpl) IsOverLimit(limit *config.RateLimit, results int64, hitsAddend int64) (bool, int64, int) {
+	limitAfterIncrease := results
+	limitBeforeIncrease := limitAfterIncrease - int64(hitsAddend)
+	overLimitThreshold := int64(limit.Limit.RequestsPerUnit)
+	nearLimitThreshold := int64(math.Floor(float64(float32(overLimitThreshold) * fw.nearLimitRatio)))
+
+	if limitAfterIncrease > overLimitThreshold {
+		if limitBeforeIncrease >= overLimitThreshold {
+			PopulateStats(limit, 0, uint64(hitsAddend), 0)
+		} else {
+			PopulateStats(limit, uint64(overLimitThreshold-utils.MaxInt64(nearLimitThreshold, limitBeforeIncrease)), uint64(limitAfterIncrease-overLimitThreshold), 0)
+		}
+
+		return true, 0, int(utils.UnitToDivider(limit.Limit.Unit))
+	} else {
+		if limitAfterIncrease > nearLimitThreshold {
+			if limitBeforeIncrease >= nearLimitThreshold {
+				PopulateStats(limit, uint64(hitsAddend), 0, 0)
+			} else {
+				PopulateStats(limit, uint64(limitAfterIncrease-nearLimitThreshold), 0, 0)
+			}
+		}
+
+		return false, overLimitThreshold - limitAfterIncrease, int(utils.UnitToDivider(limit.Limit.Unit))
+	}
+}
+
+func (fw *FixedWindowImpl) GetExpirationSeconds() int64 {
+	return 0
+}
+
+func (fw *FixedWindowImpl) GetResultsAfterIncrease() int64 {
+	return 0
+}
+
+func (fw *FixedWindowImpl) CalculateSimpleReset(limit *config.RateLimit, timeSource utils.TimeSource) *duration.Duration {
+	return utils.CalculateFixedReset(limit.Limit, timeSource)
+}
+
+func (fw *FixedWindowImpl) CalculateReset(isOverLimit bool, limit *config.RateLimit, timeSource utils.TimeSource) *duration.Duration {
+	return fw.CalculateSimpleReset(limit, timeSource)
+}
+
+func NewFixedWindowAlgorithm(timeSource utils.TimeSource, localCache *freecache.Cache, nearLimitRatio float32, cacheKeyPrefix string) *FixedWindowImpl {
+	return &FixedWindowImpl{
+		timeSource:        timeSource,
+		cacheKeyGenerator: utils.NewCacheKeyGenerator(cacheKeyPrefix),
+		localCache:        localCache,
+		nearLimitRatio:    nearLimitRatio,
+	}
+}

src/algorithm/ratelimit_algorithm.go

@@ -0,0 +1,15 @@
+package algorithm
+
+import (
+	"github.com/envoyproxy/ratelimit/src/config"
+	"github.com/envoyproxy/ratelimit/src/utils"
+	"github.com/golang/protobuf/ptypes/duration"
+)
+
+type RatelimitAlgorithm interface {
+	CalculateSimpleReset(limit *config.RateLimit, timeSource utils.TimeSource) *duration.Duration
+	CalculateReset(isOverLimit bool, limit *config.RateLimit, timeSource utils.TimeSource) *duration.Duration
+	IsOverLimit(limit *config.RateLimit, results int64, hitsAddend int64) (bool, int64, int)
+	GetExpirationSeconds() int64
+	GetResultsAfterIncrease() int64
+}

src/algorithm/rolling_window.go

@@ -0,0 +1,100 @@
+package algorithm
+
+import (
+	"math"
+
+	"github.com/coocood/freecache"
+	"github.com/envoyproxy/ratelimit/src/config"
+	"github.com/envoyproxy/ratelimit/src/utils"
+	"github.com/golang/protobuf/ptypes/duration"
+)
+
+var _ RatelimitAlgorithm = (*RollingWindowImpl)(nil)
+
+type RollingWindowImpl struct {
+	timeSource        utils.TimeSource
+	cacheKeyGenerator utils.CacheKeyGenerator
+	localCache        *freecache.Cache
+	nearLimitRatio    float32
+	arrivedAt         int64
+	tat               int64
+	newTat            int64
+	diff              int64
+}
+
+func (rw *RollingWindowImpl) IsOverLimit(limit *config.RateLimit, results int64, hitsAddend int64) (bool, int64, int) {
+	now := rw.timeSource.UnixNanoNow()
+
+	// Time during computation should be in nanosecond
+	rw.arrivedAt = now
+	// Tat is set to current request timestamp if not set before
+	rw.tat = utils.MaxInt64(results, rw.arrivedAt)
+	totalLimit := int64(limit.Limit.RequestsPerUnit)
+	period := utils.SecondsToNanoseconds(utils.UnitToDivider(limit.Limit.Unit))
+	quantity := int64(hitsAddend)
+
+	// GCRA computation
+	// Emission interval is the cost of each request
+	emissionInterval := period / totalLimit
+	// New tat define the end of the window
+	rw.newTat = rw.tat + emissionInterval*quantity
+	// We allow the request if it's inside the window
+	allowAt := rw.newTat - period
+	rw.diff = rw.arrivedAt - allowAt
+
+	previousAllowAt := rw.tat - period
+	previousLimitRemaining := int64(math.Ceil(float64((rw.arrivedAt - previousAllowAt) / emissionInterval)))
+	previousLimitRemaining = utils.MaxInt64(previousLimitRemaining, 0)
+	nearLimitWindow := int64(math.Ceil(float64(float32(limit.Limit.RequestsPerUnit) * (1.0 - rw.nearLimitRatio))))
+	limitRemaining := int64(math.Ceil(float64(rw.diff / emissionInterval)))
+	hitNearLimit := quantity - (utils.MaxInt64(previousLimitRemaining, nearLimitWindow) - nearLimitWindow)
+
+	if rw.diff < 0 {
+		PopulateStats(limit, uint64(utils.MinInt64(previousLimitRemaining, nearLimitWindow)), uint64(quantity-previousLimitRemaining), 0)
+
+		return true, 0, int(utils.NanosecondsToSeconds(-rw.diff))
+	} else {
+		if hitNearLimit > 0 {
+			PopulateStats(limit, uint64(hitNearLimit), 0, 0)
+		}
+
+		return false, limitRemaining, 0
+	}
+}
+
+func (rw *RollingWindowImpl) GetExpirationSeconds() int64 {
+	if rw.diff < 0 {
+		return utils.NanosecondsToSeconds(rw.tat-rw.arrivedAt) + 1
+	}
+	return utils.NanosecondsToSeconds(rw.newTat-rw.arrivedAt) + 1
+}
+
+func (rw *RollingWindowImpl) GetResultsAfterIncrease() int64 {
+	if rw.diff < 0 {
+		return rw.tat
+	}
+	return rw.newTat
+}
+
+func (rw *RollingWindowImpl) CalculateSimpleReset(limit *config.RateLimit, timeSource utils.TimeSource) *duration.Duration {
+	secondsToReset := utils.UnitToDivider(limit.Limit.Unit)
+	secondsToReset -= utils.NanosecondsToSeconds(timeSource.UnixNanoNow()) % secondsToReset
+	return &duration.Duration{Seconds: secondsToReset}
+}
+
+func (rw *RollingWindowImpl) CalculateReset(isOverLimit bool, limit *config.RateLimit, timeSource utils.TimeSource) *duration.Duration {
+	if !isOverLimit {
+		return utils.NanosecondsToDuration(rw.newTat - rw.arrivedAt)
+	} else {
+		return utils.NanosecondsToDuration(int64(math.Ceil(float64(rw.tat - rw.arrivedAt))))
+	}
+}
+
+func NewRollingWindowAlgorithm(timeSource utils.TimeSource, localCache *freecache.Cache, nearLimitRatio float32, cacheKeyPrefix string) *RollingWindowImpl {
+	return &RollingWindowImpl{
+		timeSource:        timeSource,
+		cacheKeyGenerator: utils.NewCacheKeyGenerator(cacheKeyPrefix),
+		localCache:        localCache,
+		nearLimitRatio:    nearLimitRatio,
+	}
+}