From 1791465f8c1d614f3b1f777d45ba57111a1f3892 Mon Sep 17 00:00:00 2001 From: Matias <83959431+mativm02@users.noreply.github.com> Date: Wed, 6 Dec 2023 10:58:02 -0300 Subject: [PATCH 1/4] Adding RateLimiter --- .../internal/driver/redisv8/ratelimiter.go | 99 ++++++++++++ temporal/model/types.go | 8 + temporal/ratelimiter/ratelimiter.go | 20 +++ temporal/ratelimiter/ratelimiter_test.go | 142 ++++++++++++++++++ temporal/temperr/errors.go | 1 + 5 files changed, 270 insertions(+) create mode 100644 temporal/internal/driver/redisv8/ratelimiter.go create mode 100644 temporal/ratelimiter/ratelimiter.go create mode 100644 temporal/ratelimiter/ratelimiter_test.go diff --git a/temporal/internal/driver/redisv8/ratelimiter.go b/temporal/internal/driver/redisv8/ratelimiter.go new file mode 100644 index 00000000..b6c83685 --- /dev/null +++ b/temporal/internal/driver/redisv8/ratelimiter.go @@ -0,0 +1,99 @@ +package redisv8 + +import ( + "context" + "strconv" + "time" + + "github.com/TykTechnologies/storage/temporal/temperr" + "github.com/go-redis/redis/v8" +) + +// SetRollingWindow sets a rolling window of values in a Redis sorted set. +// It returns a slice of strings (the values in the rolling window) and an error if any occurs. +func (r *RedisV8) SetRollingWindow(ctx context.Context, keyName string, + per int64, value_override string, pipeline bool, +) ([]string, error) { + now := time.Now() + onePeriodAgo := now.Add(time.Duration(-1*per) * time.Second) + var zrange *redis.StringSliceCmd + var err error + + if keyName == "" { + return []string{}, temperr.KeyEmpty + } + + if per <= 0 { + return []string{}, temperr.InvalidPeriod + } + + pipeFn := func(pipe redis.Pipeliner) error { + pipe.ZRemRangeByScore(ctx, keyName, "-inf", strconv.Itoa(int(onePeriodAgo.UnixNano()))) + zrange = pipe.ZRange(ctx, keyName, 0, -1) + + element := redis.Z{ + Score: float64(now.UnixNano()), + } + + if value_override != "-1" { + element.Member = value_override + } else { + element.Member = strconv.Itoa(int(now.UnixNano())) + } + + pipe.ZAdd(ctx, keyName, &element) + pipe.Expire(ctx, keyName, time.Duration(per)*time.Second) + + return nil + } + + if pipeline { + _, err = r.client.Pipelined(ctx, pipeFn) + } else { + _, err = r.client.TxPipelined(ctx, pipeFn) + } + + if err != nil { + return []string{}, err + } + + return zrange.Result() +} + +// GetRollingWindow retrieves a rolling window of values from a Redis sorted set. +// It returns a slice of strings (the values in the rolling window) and an error if any occurs. +func (r *RedisV8) GetRollingWindow(ctx context.Context, keyName string, per int64, pipeline bool) ([]string, error) { + now := time.Now() + onePeriodAgo := now.Add(time.Duration(-1*per) * time.Second) + + var zrange *redis.StringSliceCmd + var err error + + pipeFn := func(pipe redis.Pipeliner) error { + pipe.ZRemRangeByScore(ctx, keyName, "-inf", strconv.FormatInt(onePeriodAgo.UnixNano(), 10)) + zrange = pipe.ZRange(ctx, keyName, 0, -1) + + return nil + } + + if pipeline { + _, err = r.client.Pipelined(ctx, pipeFn) + } else { + _, err = r.client.TxPipelined(ctx, pipeFn) + } + + if err != nil { + return nil, err + } + + values, err := zrange.Result() + if err != nil { + return nil, err + } + + if values == nil { + return []string{}, nil + } + + return values, nil +} diff --git a/temporal/model/types.go b/temporal/model/types.go index 451fb9de..11ec2e91 100644 --- a/temporal/model/types.go +++ b/temporal/model/types.go @@ -160,3 +160,11 @@ type Message interface { // - an empty string, returning an error Payload() (string, error) } + +type RateLimit interface { + // SetRollingWindow sets the rolling window for a key with a per second rate limit + SetRollingWindow(ctx context.Context, keyName string, per int64, + value_override string, pipeline bool) ([]string, error) + // GetRollingWindow gets the rolling window for a key with a per second rate limit + GetRollingWindow(ctx context.Context, keyName string, per int64, pipeline bool) ([]string, error) +} diff --git a/temporal/ratelimiter/ratelimiter.go b/temporal/ratelimiter/ratelimiter.go new file mode 100644 index 00000000..99a3d53c --- /dev/null +++ b/temporal/ratelimiter/ratelimiter.go @@ -0,0 +1,20 @@ +package ratelimiter + +import ( + "github.com/TykTechnologies/storage/temporal/internal/driver/redisv8" + "github.com/TykTechnologies/storage/temporal/model" + "github.com/TykTechnologies/storage/temporal/temperr" +) + +type RateLimit = model.RateLimit + +var _ RateLimit = (*redisv8.RedisV8)(nil) + +func NewRateLimit(conn model.Connector) (RateLimit, error) { + switch conn.Type() { + case model.RedisV8Type: + return redisv8.NewRedisV8WithConnection(conn) + default: + return nil, temperr.InvalidHandlerType + } +} diff --git a/temporal/ratelimiter/ratelimiter_test.go b/temporal/ratelimiter/ratelimiter_test.go new file mode 100644 index 00000000..33264ef1 --- /dev/null +++ b/temporal/ratelimiter/ratelimiter_test.go @@ -0,0 +1,142 @@ +package ratelimiter + +import ( + "context" + "testing" + + "github.com/TykTechnologies/storage/temporal/flusher" + "github.com/TykTechnologies/storage/temporal/internal/testutil" + "github.com/TykTechnologies/storage/temporal/model" + "github.com/TykTechnologies/storage/temporal/temperr" + "github.com/stretchr/testify/assert" +) + +func TestRedisCluster_SetRollingWindow(t *testing.T) { + connectors := testutil.TestConnectors(t) + defer testutil.CloseConnectors(t, connectors) + + tcs := []struct { + name string + keyName string + per int64 + valueOverride string + pipeline bool + expectedErr error + expectedLen int + }{ + { + name: "valid_rolling_window", + keyName: "key1", + per: 60, + valueOverride: "value1", + pipeline: false, + expectedErr: nil, + expectedLen: 0, + }, + { + name: "empty_key_name", + keyName: "", + per: 60, + valueOverride: "value2", + pipeline: false, + expectedErr: temperr.KeyEmpty, + expectedLen: 0, + }, + { + name: "negative_period", + keyName: "key2", + per: -10, + valueOverride: "value3", + pipeline: false, + expectedErr: temperr.InvalidPeriod, + expectedLen: 0, + }, + } + + for _, connector := range connectors { + for _, tc := range tcs { + t.Run(connector.Type()+"_"+tc.name, func(t *testing.T) { + ctx := context.Background() + + rateLimiter, err := NewRateLimit(connector) + assert.Nil(t, err) + + flusher, err := flusher.NewFlusher(connector) + assert.Nil(t, err) + defer assert.Nil(t, flusher.FlushAll(ctx)) + + result, err := rateLimiter.SetRollingWindow(ctx, tc.keyName, tc.per, tc.valueOverride, tc.pipeline) + + assert.Equal(t, tc.expectedErr, err) + + if err == nil { + assert.Equal(t, tc.expectedLen, len(result)) + // Executing SetRollingWindow again should return expectedLen + 1 if err == nil + result, err = rateLimiter.SetRollingWindow(ctx, tc.keyName, tc.per, tc.valueOverride, tc.pipeline) + assert.NoError(t, err) + assert.Equal(t, tc.expectedLen+1, len(result)) + } + }) + } + } +} + +func TestRedisCluster_GetRollingWindow(t *testing.T) { + connectors := testutil.TestConnectors(t) + defer testutil.CloseConnectors(t, connectors) + + tcs := []struct { + name string + keyName string + per int64 + pipeline bool + expectedErr error + expectedLen int + preTest func(ctx context.Context, rateLimiter model.RateLimit) + }{ + { + name: "empty_sorted_set", + keyName: "key_empty", + per: 60, + pipeline: false, + expectedErr: nil, + expectedLen: 0, + }, + { + name: "non_empty_sorted_set", + keyName: "key_non_empty", + per: 60, + pipeline: false, + expectedErr: nil, + expectedLen: 2, + preTest: func(ctx context.Context, rateLimiter model.RateLimit) { + _, err := rateLimiter.SetRollingWindow(ctx, "key_non_empty", 60, "value1", false) + assert.Nil(t, err) + _, err = rateLimiter.SetRollingWindow(ctx, "key_non_empty", 60, "value2", false) + assert.Nil(t, err) + }, + }, + } + + for _, connector := range connectors { + for _, tc := range tcs { + t.Run(connector.Type()+"_"+tc.name, func(t *testing.T) { + ctx := context.Background() + + rateLimiter, err := NewRateLimit(connector) + assert.Nil(t, err) + + if tc.preTest != nil { + tc.preTest(ctx, rateLimiter) + } + + result, err := rateLimiter.GetRollingWindow(ctx, tc.keyName, tc.per, tc.pipeline) + + assert.Equal(t, tc.expectedErr, err) + if err == nil { + assert.Equal(t, tc.expectedLen, len(result)) + } + }) + } + } +} diff --git a/temporal/temperr/errors.go b/temporal/temperr/errors.go index 91fb02e4..c685c3cb 100644 --- a/temporal/temperr/errors.go +++ b/temporal/temperr/errors.go @@ -17,6 +17,7 @@ var ( // Redis related errors InvalidRedisClient = errors.New("invalid redis client") + InvalidPeriod = errors.New("invalid period specified") // TLS related errors // TLS related errors From 982b61e89e7b6c91ee36d53f6b4ef3acc647ad5c Mon Sep 17 00:00:00 2001 From: Matias <83959431+mativm02@users.noreply.github.com> Date: Wed, 6 Dec 2023 11:05:58 -0300 Subject: [PATCH 2/4] increasing test coverage --- .../internal/driver/redisv8/ratelimiter.go | 8 ++++ temporal/ratelimiter/ratelimiter_test.go | 42 +++++++++++++++++++ 2 files changed, 50 insertions(+) diff --git a/temporal/internal/driver/redisv8/ratelimiter.go b/temporal/internal/driver/redisv8/ratelimiter.go index b6c83685..ec9c04b4 100644 --- a/temporal/internal/driver/redisv8/ratelimiter.go +++ b/temporal/internal/driver/redisv8/ratelimiter.go @@ -69,6 +69,14 @@ func (r *RedisV8) GetRollingWindow(ctx context.Context, keyName string, per int6 var zrange *redis.StringSliceCmd var err error + if keyName == "" { + return []string{}, temperr.KeyEmpty + } + + if per <= 0 { + return []string{}, temperr.InvalidPeriod + } + pipeFn := func(pipe redis.Pipeliner) error { pipe.ZRemRangeByScore(ctx, keyName, "-inf", strconv.FormatInt(onePeriodAgo.UnixNano(), 10)) zrange = pipe.ZRange(ctx, keyName, 0, -1) diff --git a/temporal/ratelimiter/ratelimiter_test.go b/temporal/ratelimiter/ratelimiter_test.go index 33264ef1..5e7a8d26 100644 --- a/temporal/ratelimiter/ratelimiter_test.go +++ b/temporal/ratelimiter/ratelimiter_test.go @@ -51,6 +51,24 @@ func TestRedisCluster_SetRollingWindow(t *testing.T) { expectedErr: temperr.InvalidPeriod, expectedLen: 0, }, + { + name: "pipeline_enabled", + keyName: "key_pipeline", + per: 60, + valueOverride: "pipeline_value", + pipeline: true, + expectedErr: nil, + expectedLen: 0, + }, + { + name: "value_override_minus_one", + keyName: "key_value_override", + per: 60, + valueOverride: "-1", // value_override set to "-1" + pipeline: false, + expectedErr: nil, + expectedLen: 0, + }, } for _, connector := range connectors { @@ -116,6 +134,30 @@ func TestRedisCluster_GetRollingWindow(t *testing.T) { assert.Nil(t, err) }, }, + { + name: "pipeline_enabled", + keyName: "key_pipeline", + per: 60, + pipeline: true, + expectedErr: nil, + expectedLen: 0, + }, + { + name: "negative_period", + keyName: "key_negative_period", + per: -10, + pipeline: false, + expectedErr: temperr.InvalidPeriod, + expectedLen: 0, + }, + { + name: "empty_key_name", + keyName: "", + per: 60, + pipeline: false, + expectedErr: temperr.KeyEmpty, + expectedLen: 0, + }, } for _, connector := range connectors { From 99a92e6601dfbc2111c6e3e986bb33c6346505de Mon Sep 17 00:00:00 2001 From: Matias <83959431+mativm02@users.noreply.github.com> Date: Wed, 6 Dec 2023 11:26:24 -0300 Subject: [PATCH 3/4] renaming parameter to make it camelCase --- temporal/internal/driver/redisv8/ratelimiter.go | 6 +++--- temporal/model/types.go | 2 +- temporal/ratelimiter/ratelimiter_test.go | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/temporal/internal/driver/redisv8/ratelimiter.go b/temporal/internal/driver/redisv8/ratelimiter.go index ec9c04b4..95b48c33 100644 --- a/temporal/internal/driver/redisv8/ratelimiter.go +++ b/temporal/internal/driver/redisv8/ratelimiter.go @@ -12,7 +12,7 @@ import ( // SetRollingWindow sets a rolling window of values in a Redis sorted set. // It returns a slice of strings (the values in the rolling window) and an error if any occurs. func (r *RedisV8) SetRollingWindow(ctx context.Context, keyName string, - per int64, value_override string, pipeline bool, + per int64, valueOverride string, pipeline bool, ) ([]string, error) { now := time.Now() onePeriodAgo := now.Add(time.Duration(-1*per) * time.Second) @@ -35,8 +35,8 @@ func (r *RedisV8) SetRollingWindow(ctx context.Context, keyName string, Score: float64(now.UnixNano()), } - if value_override != "-1" { - element.Member = value_override + if valueOverride != "-1" { + element.Member = valueOverride } else { element.Member = strconv.Itoa(int(now.UnixNano())) } diff --git a/temporal/model/types.go b/temporal/model/types.go index 11ec2e91..faffa981 100644 --- a/temporal/model/types.go +++ b/temporal/model/types.go @@ -164,7 +164,7 @@ type Message interface { type RateLimit interface { // SetRollingWindow sets the rolling window for a key with a per second rate limit SetRollingWindow(ctx context.Context, keyName string, per int64, - value_override string, pipeline bool) ([]string, error) + valueOverride string, pipeline bool) ([]string, error) // GetRollingWindow gets the rolling window for a key with a per second rate limit GetRollingWindow(ctx context.Context, keyName string, per int64, pipeline bool) ([]string, error) } diff --git a/temporal/ratelimiter/ratelimiter_test.go b/temporal/ratelimiter/ratelimiter_test.go index 5e7a8d26..7aa466c8 100644 --- a/temporal/ratelimiter/ratelimiter_test.go +++ b/temporal/ratelimiter/ratelimiter_test.go @@ -61,10 +61,10 @@ func TestRedisCluster_SetRollingWindow(t *testing.T) { expectedLen: 0, }, { - name: "value_override_minus_one", + name: "valueOverride", keyName: "key_value_override", per: 60, - valueOverride: "-1", // value_override set to "-1" + valueOverride: "-1", pipeline: false, expectedErr: nil, expectedLen: 0, From 5fb31f5e3c634cbcddeaac6b82686b5d8ca9e5d1 Mon Sep 17 00:00:00 2001 From: Matias <83959431+mativm02@users.noreply.github.com> Date: Tue, 26 Dec 2023 09:47:06 -0300 Subject: [PATCH 4/4] Refactor RedisV8 SetRollingWindow and GetRollingWindow methods --- .../internal/driver/redisv8/ratelimiter.go | 101 ++++++++---------- temporal/model/types.go | 4 +- temporal/ratelimiter/ratelimiter_test.go | 13 ++- 3 files changed, 54 insertions(+), 64 deletions(-) diff --git a/temporal/internal/driver/redisv8/ratelimiter.go b/temporal/internal/driver/redisv8/ratelimiter.go index 95b48c33..eac3873f 100644 --- a/temporal/internal/driver/redisv8/ratelimiter.go +++ b/temporal/internal/driver/redisv8/ratelimiter.go @@ -9,99 +9,86 @@ import ( "github.com/go-redis/redis/v8" ) -// SetRollingWindow sets a rolling window of values in a Redis sorted set. -// It returns a slice of strings (the values in the rolling window) and an error if any occurs. -func (r *RedisV8) SetRollingWindow(ctx context.Context, keyName string, - per int64, valueOverride string, pipeline bool, -) ([]string, error) { - now := time.Now() - onePeriodAgo := now.Add(time.Duration(-1*per) * time.Second) - var zrange *redis.StringSliceCmd - var err error - +// SetRollingWindow updates a sorted set in Redis to represent a rolling time window of values. +func (r *RedisV8) SetRollingWindow(ctx context.Context, now time.Time, keyName string, per int64, valueOverride string, pipeline bool) ([]string, error) { if keyName == "" { return []string{}, temperr.KeyEmpty } - if per <= 0 { return []string{}, temperr.InvalidPeriod } - pipeFn := func(pipe redis.Pipeliner) error { - pipe.ZRemRangeByScore(ctx, keyName, "-inf", strconv.Itoa(int(onePeriodAgo.UnixNano()))) - zrange = pipe.ZRange(ctx, keyName, 0, -1) - - element := redis.Z{ - Score: float64(now.UnixNano()), - } + onePeriodAgo := now.Add(time.Duration(-1*per) * time.Second) + expire := time.Duration(per) * time.Second - if valueOverride != "-1" { - element.Member = valueOverride - } else { - element.Member = strconv.Itoa(int(now.UnixNano())) - } + memberValue := valueOverride + if valueOverride == "-1" { + memberValue = strconv.Itoa(int(now.UnixNano())) + } + element := redis.Z{ + Score: float64(now.UnixNano()), + Member: memberValue, + } - pipe.ZAdd(ctx, keyName, &element) - pipe.Expire(ctx, keyName, time.Duration(per)*time.Second) + var zrange *redis.StringSliceCmd + var err error - return nil + exec := r.client.TxPipelined + if pipeline { + exec = r.client.Pipelined } - if pipeline { - _, err = r.client.Pipelined(ctx, pipeFn) - } else { - _, err = r.client.TxPipelined(ctx, pipeFn) + pipeFn := func(pipe redis.Pipeliner) error { + // removing elements outside the rolling window. + pipe.ZRemRangeByScore(ctx, keyName, "-inf", strconv.Itoa(int(onePeriodAgo.UnixNano()))) + // getting the current range of values within the window. + zrange = pipe.ZRange(ctx, keyName, 0, -1) + // adding the new element and set the expiration time. + pipe.ZAdd(ctx, keyName, &element) + pipe.Expire(ctx, keyName, expire) + return nil } + _, err = exec(ctx, pipeFn) if err != nil { - return []string{}, err + return nil, err } return zrange.Result() } -// GetRollingWindow retrieves a rolling window of values from a Redis sorted set. -// It returns a slice of strings (the values in the rolling window) and an error if any occurs. -func (r *RedisV8) GetRollingWindow(ctx context.Context, keyName string, per int64, pipeline bool) ([]string, error) { - now := time.Now() - onePeriodAgo := now.Add(time.Duration(-1*per) * time.Second) - - var zrange *redis.StringSliceCmd - var err error - +// GetRollingWindow removes a part of a sorted set in Redis and extracts a timed window of values. +func (r *RedisV8) GetRollingWindow(ctx context.Context, now time.Time, keyName string, per int64, pipeline bool) ([]string, error) { if keyName == "" { return []string{}, temperr.KeyEmpty } - if per <= 0 { return []string{}, temperr.InvalidPeriod } - pipeFn := func(pipe redis.Pipeliner) error { - pipe.ZRemRangeByScore(ctx, keyName, "-inf", strconv.FormatInt(onePeriodAgo.UnixNano(), 10)) - zrange = pipe.ZRange(ctx, keyName, 0, -1) + onePeriodAgo := now.Add(time.Duration(-1*per) * time.Second) + period := strconv.FormatInt(onePeriodAgo.UnixNano(), 10) - return nil - } + var zrange *redis.StringSliceCmd + var err error + exec := r.client.TxPipelined if pipeline { - _, err = r.client.Pipelined(ctx, pipeFn) - } else { - _, err = r.client.TxPipelined(ctx, pipeFn) + exec = r.client.Pipelined } - if err != nil { - return nil, err + pipeFn := func(pipe redis.Pipeliner) error { + // removing old elements outside the rolling window + pipe.ZRemRangeByScore(ctx, keyName, "-inf", period) + // retrieving the current range of values + zrange = pipe.ZRange(ctx, keyName, 0, -1) + return nil } - values, err := zrange.Result() + _, err = exec(ctx, pipeFn) if err != nil { return nil, err } - if values == nil { - return []string{}, nil - } - - return values, nil + return zrange.Result() } diff --git a/temporal/model/types.go b/temporal/model/types.go index faffa981..d755e9f4 100644 --- a/temporal/model/types.go +++ b/temporal/model/types.go @@ -163,8 +163,8 @@ type Message interface { type RateLimit interface { // SetRollingWindow sets the rolling window for a key with a per second rate limit - SetRollingWindow(ctx context.Context, keyName string, per int64, + SetRollingWindow(ctx context.Context, now time.Time, keyName string, per int64, valueOverride string, pipeline bool) ([]string, error) // GetRollingWindow gets the rolling window for a key with a per second rate limit - GetRollingWindow(ctx context.Context, keyName string, per int64, pipeline bool) ([]string, error) + GetRollingWindow(ctx context.Context, now time.Time, keyName string, per int64, pipeline bool) ([]string, error) } diff --git a/temporal/ratelimiter/ratelimiter_test.go b/temporal/ratelimiter/ratelimiter_test.go index 7aa466c8..88ff1e17 100644 --- a/temporal/ratelimiter/ratelimiter_test.go +++ b/temporal/ratelimiter/ratelimiter_test.go @@ -3,6 +3,7 @@ package ratelimiter import ( "context" "testing" + "time" "github.com/TykTechnologies/storage/temporal/flusher" "github.com/TykTechnologies/storage/temporal/internal/testutil" @@ -74,6 +75,7 @@ func TestRedisCluster_SetRollingWindow(t *testing.T) { for _, connector := range connectors { for _, tc := range tcs { t.Run(connector.Type()+"_"+tc.name, func(t *testing.T) { + now := time.Now() ctx := context.Background() rateLimiter, err := NewRateLimit(connector) @@ -83,14 +85,14 @@ func TestRedisCluster_SetRollingWindow(t *testing.T) { assert.Nil(t, err) defer assert.Nil(t, flusher.FlushAll(ctx)) - result, err := rateLimiter.SetRollingWindow(ctx, tc.keyName, tc.per, tc.valueOverride, tc.pipeline) + result, err := rateLimiter.SetRollingWindow(ctx, now, tc.keyName, tc.per, tc.valueOverride, tc.pipeline) assert.Equal(t, tc.expectedErr, err) if err == nil { assert.Equal(t, tc.expectedLen, len(result)) // Executing SetRollingWindow again should return expectedLen + 1 if err == nil - result, err = rateLimiter.SetRollingWindow(ctx, tc.keyName, tc.per, tc.valueOverride, tc.pipeline) + result, err = rateLimiter.SetRollingWindow(ctx, now, tc.keyName, tc.per, tc.valueOverride, tc.pipeline) assert.NoError(t, err) assert.Equal(t, tc.expectedLen+1, len(result)) } @@ -128,9 +130,10 @@ func TestRedisCluster_GetRollingWindow(t *testing.T) { expectedErr: nil, expectedLen: 2, preTest: func(ctx context.Context, rateLimiter model.RateLimit) { - _, err := rateLimiter.SetRollingWindow(ctx, "key_non_empty", 60, "value1", false) + now := time.Now() + _, err := rateLimiter.SetRollingWindow(ctx, now, "key_non_empty", 60, "value1", false) assert.Nil(t, err) - _, err = rateLimiter.SetRollingWindow(ctx, "key_non_empty", 60, "value2", false) + _, err = rateLimiter.SetRollingWindow(ctx, now, "key_non_empty", 60, "value2", false) assert.Nil(t, err) }, }, @@ -172,7 +175,7 @@ func TestRedisCluster_GetRollingWindow(t *testing.T) { tc.preTest(ctx, rateLimiter) } - result, err := rateLimiter.GetRollingWindow(ctx, tc.keyName, tc.per, tc.pipeline) + result, err := rateLimiter.GetRollingWindow(ctx, time.Now(), tc.keyName, tc.per, tc.pipeline) assert.Equal(t, tc.expectedErr, err) if err == nil {