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 it is able to 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/fixed_window.go

@@ -0,0 +1,131 @@
+package algorithm
+
+import (
+	"math"
+
+	"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"
+)
+
+var _ RatelimitAlgorithm = (*FixedWindowImpl)(nil)
+
+type FixedWindowImpl struct {
+	timeSource        utils.TimeSource
+	cacheKeyGenerator utils.CacheKeyGenerator
+	localCache        *freecache.Cache
+	nearLimitRatio    float32
+}
+
+func (fw *FixedWindowImpl) 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 {
+		fw.PopulateStats(limit, 0, uint64(hitsAddend), uint64(hitsAddend))
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:               pb.RateLimitResponse_OVER_LIMIT,
+			CurrentLimit:       limit.Limit,
+			LimitRemaining:     0,
+			DurationUntilReset: utils.CalculateFixedReset(limit.Limit, fw.timeSource),
+		}
+	}
+
+	isOverLimit, limitRemaining, durationUntilReset := fw.IsOverLimit(limit, int64(results), hitsAddend)
+
+	if !isOverLimit {
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:               pb.RateLimitResponse_OK,
+			CurrentLimit:       limit.Limit,
+			LimitRemaining:     uint32(limitRemaining),
+			DurationUntilReset: utils.CalculateFixedReset(limit.Limit, fw.timeSource),
+		}
+	} else {
+		if fw.localCache != nil {
+			durationUntilReset = utils.MaxInt(1, durationUntilReset)
+
+			err := fw.localCache.Set([]byte(key), []byte{}, durationUntilReset)
+			if err != nil {
+				logger.Errorf("Failing to set local cache key: %s", key)
+			}
+		}
+
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:               pb.RateLimitResponse_OVER_LIMIT,
+			CurrentLimit:       limit.Limit,
+			LimitRemaining:     uint32(limitRemaining),
+			DurationUntilReset: utils.CalculateFixedReset(limit.Limit, fw.timeSource),
+		}
+	}
+}
+
+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 {
+			fw.PopulateStats(limit, 0, uint64(hitsAddend), 0)
+		} else {
+			fw.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 {
+				fw.PopulateStats(limit, uint64(hitsAddend), 0, 0)
+			} else {
+				fw.PopulateStats(limit, uint64(limitAfterIncrease-nearLimitThreshold), 0, 0)
+			}
+		}
+
+		return false, overLimitThreshold - limitAfterIncrease, int(utils.UnitToDivider(limit.Limit.Unit))
+	}
+}
+
+func (fw *FixedWindowImpl) IsOverLimitWithLocalCache(key string) bool {
+	if fw.localCache != nil {
+		_, err := fw.localCache.Get([]byte(key))
+		if err == nil {
+			return true
+		}
+	}
+	return false
+}
+
+func (fw *FixedWindowImpl) GenerateCacheKeys(request *pb.RateLimitRequest,
+	limits []*config.RateLimit, hitsAddend int64) []utils.CacheKey {
+	return fw.cacheKeyGenerator.GenerateCacheKeys(request, limits, uint32(hitsAddend), fw.timeSource.UnixNow())
+}
+
+func (fw *FixedWindowImpl) 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 (fw *FixedWindowImpl) GetExpirationSeconds() int64 {
+	return 0
+}
+
+func (fw *FixedWindowImpl) GetResultsAfterIncrease() int64 {
+	return 0
+}
+
+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,20 @@
+package algorithm
+
+import (
+	pb "github.com/envoyproxy/go-control-plane/envoy/service/ratelimit/v3"
+	"github.com/envoyproxy/ratelimit/src/config"
+	"github.com/envoyproxy/ratelimit/src/utils"
+)
+
+type RatelimitAlgorithm interface {
+	IsOverLimit(limit *config.RateLimit, results int64, hitsAddend int64) (bool, int64, int)
+	IsOverLimitWithLocalCache(key string) bool
+
+	GetResponseDescriptorStatus(key string, limit *config.RateLimit, results int64, isOverLimitWithLocalCache bool, hitsAddend int64) *pb.RateLimitResponse_DescriptorStatus
+	GetExpirationSeconds() int64
+	GetResultsAfterIncrease() int64
+
+	GenerateCacheKeys(request *pb.RateLimitRequest,
+		limits []*config.RateLimit, hitsAddend int64) []utils.CacheKey
+	PopulateStats(limit *config.RateLimit, nearLimit uint64, overLimit uint64, overLimitWithLocalCache uint64)
+}

src/algorithm/rolling_window.go

@@ -0,0 +1,161 @@
+package algorithm
+
+import (
+	"math"
+
+	"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"
+	"github.com/golang/protobuf/ptypes/duration"
+	logger "github.com/sirupsen/logrus"
+)
+
+const DummyCacheKeyTime = 0
+
+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) 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 {
+		rw.PopulateStats(limit, 0, uint64(hitsAddend), uint64(hitsAddend))
+
+		secondsToReset := utils.UnitToDivider(limit.Limit.Unit)
+		secondsToReset -= utils.NanosecondsToSeconds(rw.timeSource.UnixNanoNow()) % secondsToReset
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:               pb.RateLimitResponse_OVER_LIMIT,
+			CurrentLimit:       limit.Limit,
+			LimitRemaining:     0,
+			DurationUntilReset: &duration.Duration{Seconds: secondsToReset},
+		}
+	}
+
+	isOverLimit, limitRemaining, durationUntilReset := rw.IsOverLimit(limit, int64(results), hitsAddend)
+
+	if !isOverLimit {
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:               pb.RateLimitResponse_OK,
+			CurrentLimit:       limit.Limit,
+			LimitRemaining:     uint32(limitRemaining),
+			DurationUntilReset: utils.NanosecondsToDuration(rw.newTat - rw.arrivedAt),
+		}
+	} else {
+		if rw.localCache != nil {
+			durationUntilReset = utils.MaxInt(1, durationUntilReset)
+
+			err := rw.localCache.Set([]byte(key), []byte{}, durationUntilReset)
+			if err != nil {
+				logger.Errorf("Failing to set local cache key: %s", key)
+			}
+		}
+
+		return &pb.RateLimitResponse_DescriptorStatus{
+			Code:               pb.RateLimitResponse_OVER_LIMIT,
+			CurrentLimit:       limit.Limit,
+			LimitRemaining:     0,
+			DurationUntilReset: utils.NanosecondsToDuration(int64(math.Ceil(float64(rw.tat - rw.arrivedAt)))),
+		}
+	}
+}
+
+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 {
+		rw.PopulateStats(limit, uint64(utils.MinInt64(previousLimitRemaining, nearLimitWindow)), uint64(quantity-previousLimitRemaining), 0)
+
+		return true, 0, int(utils.NanosecondsToSeconds(-rw.diff))
+	} else {
+		if hitNearLimit > 0 {
+			rw.PopulateStats(limit, uint64(hitNearLimit), 0, 0)
+		}
+
+		return false, limitRemaining, 0
+	}
+}
+
+func (rw *RollingWindowImpl) IsOverLimitWithLocalCache(key string) bool {
+	if rw.localCache != nil {
+		_, err := rw.localCache.Get([]byte(key))
+		if err == nil {
+			return true
+		}
+	}
+	return false
+}
+
+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) GenerateCacheKeys(request *pb.RateLimitRequest,
+	limits []*config.RateLimit, hitsAddend int64) []utils.CacheKey {
+	return rw.cacheKeyGenerator.GenerateCacheKeys(request, limits, uint32(hitsAddend), DummyCacheKeyTime)
+}
+
+func (rw *RollingWindowImpl) 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 NewRollingWindowAlgorithm(timeSource utils.TimeSource, localCache *freecache.Cache, nearLimitRatio float32, cacheKeyPrefix string) *RollingWindowImpl {
+	return &RollingWindowImpl{
+		timeSource:        timeSource,
+		cacheKeyGenerator: utils.NewCacheKeyGenerator(cacheKeyPrefix),
+		localCache:        localCache,
+		nearLimitRatio:    nearLimitRatio,
+	}
+}