feat(ratelimier): Support disabling rate limiting
This commit is contained in:
9 files changed
+896
-60
No files matched your search
@@ -126,6 +126,57 @@ listener "tcp" {
|
||||
tls_disable = true
|
||||
}
|
||||
|
||||
listener "tcp" {
|
||||
address = "127.0.0.1:9501"
|
||||
purpose = "cluster"
|
||||
}
|
||||
`
|
||||
|
||||
ratelimitConfigDisabledReload = `
|
||||
disable_mlock = true
|
||||
|
||||
telemetry {
|
||||
prometheus_retention_time = "24h"
|
||||
disable_hostname = true
|
||||
}
|
||||
|
||||
controller {
|
||||
name = "test-controller"
|
||||
description = "A default controller created for tests"
|
||||
database {
|
||||
url = "%s"
|
||||
}
|
||||
|
||||
api_rate_limit_disable = true
|
||||
}
|
||||
|
||||
kms "aead" {
|
||||
purpose = "root"
|
||||
aead_type = "aes-gcm"
|
||||
key = "%s"
|
||||
key_id = "global_root"
|
||||
}
|
||||
|
||||
kms "aead" {
|
||||
purpose = "worker-auth"
|
||||
aead_type = "aes-gcm"
|
||||
key = "%s"
|
||||
key_id = "global_worker-auth"
|
||||
}
|
||||
|
||||
kms "aead" {
|
||||
purpose = "recovery"
|
||||
aead_type = "aes-gcm"
|
||||
key = "%s"
|
||||
key_id = "global_recovery"
|
||||
}
|
||||
|
||||
listener "tcp" {
|
||||
purpose = "api"
|
||||
address = "127.0.0.1:9500"
|
||||
tls_disable = true
|
||||
}
|
||||
|
||||
listener "tcp" {
|
||||
address = "127.0.0.1:9501"
|
||||
purpose = "cluster"
|
||||
@@ -169,7 +220,7 @@ func TestRealodControllerRateLimits(t *testing.T) {
|
||||
t.Fatal("timeout")
|
||||
}
|
||||
|
||||
// Change config so it is read for reloading
|
||||
// Change config so it is ready for reloading
|
||||
cfgHcl = fmt.Sprintf(ratelimitConfigReload, url, controllerKey, workerAuthKey, recoveryKey)
|
||||
require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644))
|
||||
|
||||
@@ -325,3 +376,200 @@ func TestRealodControllerRateLimitsSameConfig(t *testing.T) {
|
||||
cmd.ShutdownCh <- struct{}{}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func TestRealodControllerRateLimitsDisable(t *testing.T) {
|
||||
td := t.TempDir()
|
||||
|
||||
controllerKey := config.DevKeyGeneration()
|
||||
|
||||
closeDB, url, _, err := getInitDatabase(t, controllerKey)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, closeDB()) })
|
||||
|
||||
cmd := testServerCommand(t, testServerCommandOpts{})
|
||||
|
||||
workerAuthKey := config.DevKeyGeneration()
|
||||
recoveryKey := config.DevKeyGeneration()
|
||||
cfgHcl := fmt.Sprintf(ratelimitConfig, url, controllerKey, workerAuthKey, recoveryKey)
|
||||
require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644))
|
||||
|
||||
wg := &sync.WaitGroup{}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
args := []string{"-config", td + "/config.hcl"}
|
||||
exitCode := cmd.Run(args)
|
||||
if exitCode != 0 {
|
||||
output := cmd.UI.(*cli.MockUi).ErrorWriter.String() + cmd.UI.(*cli.MockUi).OutputWriter.String()
|
||||
t.Errorf("got a non-zero exit status: %s", output)
|
||||
}
|
||||
}()
|
||||
|
||||
// Wait until things are up and running (or timeout).
|
||||
select {
|
||||
case <-cmd.startedCh:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timeout")
|
||||
}
|
||||
|
||||
// Change config so it is ready for reloading
|
||||
cfgHcl = fmt.Sprintf(ratelimitConfigDisabledReload, url, controllerKey, workerAuthKey, recoveryKey)
|
||||
require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644))
|
||||
|
||||
c := http.Client{}
|
||||
r, err := c.Do(func() *http.Request {
|
||||
r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil)
|
||||
require.NoError(t, err)
|
||||
return r
|
||||
}())
|
||||
require.NoError(t, err)
|
||||
// unauthed request, so we expect a 400
|
||||
assert.Equal(t, http.StatusBadRequest, r.StatusCode)
|
||||
assert.Equal(t, `limit=2, remaining=1, reset=59`, r.Header.Get("Ratelimit"))
|
||||
assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy"))
|
||||
|
||||
r, err = c.Do(func() *http.Request {
|
||||
r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil)
|
||||
require.NoError(t, err)
|
||||
return r
|
||||
}())
|
||||
require.NoError(t, err)
|
||||
// unauthed request, so we expect a 400
|
||||
assert.Equal(t, http.StatusBadRequest, r.StatusCode)
|
||||
assert.Equal(t, `limit=2, remaining=0, reset=59`, r.Header.Get("Ratelimit"))
|
||||
assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy"))
|
||||
|
||||
r, err = c.Do(func() *http.Request {
|
||||
r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil)
|
||||
require.NoError(t, err)
|
||||
return r
|
||||
}())
|
||||
require.NoError(t, err)
|
||||
// out of quota, so we expect a 429
|
||||
assert.Equal(t, http.StatusTooManyRequests, r.StatusCode)
|
||||
assert.Equal(t, `limit=2, remaining=0, reset=59`, r.Header.Get("Ratelimit"))
|
||||
assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy"))
|
||||
|
||||
cmd.SighupCh <- struct{}{}
|
||||
select {
|
||||
case <-cmd.reloadedCh:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timeout")
|
||||
}
|
||||
|
||||
// Make another request, the limit should be disabled
|
||||
// should get reported via the headers.
|
||||
r, err = c.Do(func() *http.Request {
|
||||
r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil)
|
||||
require.NoError(t, err)
|
||||
return r
|
||||
}())
|
||||
require.NoError(t, err)
|
||||
// unauthed request, so we expect a 400
|
||||
assert.Equal(t, http.StatusBadRequest, r.StatusCode)
|
||||
assert.Equal(t, ``, r.Header.Get("Ratelimit"))
|
||||
assert.Equal(t, ``, r.Header.Get("Ratelimit-Policy"))
|
||||
|
||||
cmd.ShutdownCh <- struct{}{}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func TestRealodControllerRateLimitsEnable(t *testing.T) {
|
||||
td := t.TempDir()
|
||||
|
||||
controllerKey := config.DevKeyGeneration()
|
||||
|
||||
closeDB, url, _, err := getInitDatabase(t, controllerKey)
|
||||
require.NoError(t, err)
|
||||
t.Cleanup(func() { require.NoError(t, closeDB()) })
|
||||
|
||||
cmd := testServerCommand(t, testServerCommandOpts{})
|
||||
|
||||
workerAuthKey := config.DevKeyGeneration()
|
||||
recoveryKey := config.DevKeyGeneration()
|
||||
// Start with rate limiting diasabled
|
||||
cfgHcl := fmt.Sprintf(ratelimitConfigDisabledReload, url, controllerKey, workerAuthKey, recoveryKey)
|
||||
require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644))
|
||||
|
||||
wg := &sync.WaitGroup{}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
args := []string{"-config", td + "/config.hcl"}
|
||||
exitCode := cmd.Run(args)
|
||||
if exitCode != 0 {
|
||||
output := cmd.UI.(*cli.MockUi).ErrorWriter.String() + cmd.UI.(*cli.MockUi).OutputWriter.String()
|
||||
t.Errorf("got a non-zero exit status: %s", output)
|
||||
}
|
||||
}()
|
||||
|
||||
// Wait until things are up and running (or timeout).
|
||||
select {
|
||||
case <-cmd.startedCh:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timeout")
|
||||
}
|
||||
|
||||
// Change config so it is ready for reloading
|
||||
cfgHcl = fmt.Sprintf(ratelimitConfig, url, controllerKey, workerAuthKey, recoveryKey)
|
||||
require.NoError(t, os.WriteFile(td+"/config.hcl", []byte(cfgHcl), 0o644))
|
||||
|
||||
c := http.Client{}
|
||||
r, err := c.Do(func() *http.Request {
|
||||
r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil)
|
||||
require.NoError(t, err)
|
||||
return r
|
||||
}())
|
||||
require.NoError(t, err)
|
||||
// unauthed request, so we expect a 400
|
||||
assert.Equal(t, http.StatusBadRequest, r.StatusCode)
|
||||
assert.Equal(t, ``, r.Header.Get("Ratelimit"))
|
||||
assert.Equal(t, ``, r.Header.Get("Ratelimit-Policy"))
|
||||
|
||||
cmd.SighupCh <- struct{}{}
|
||||
select {
|
||||
case <-cmd.reloadedCh:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timeout")
|
||||
}
|
||||
|
||||
// Make another request, the limit should be enabled
|
||||
// should get reported via the headers.
|
||||
r, err = c.Do(func() *http.Request {
|
||||
r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil)
|
||||
require.NoError(t, err)
|
||||
return r
|
||||
}())
|
||||
require.NoError(t, err)
|
||||
// unauthed request, so we expect a 400
|
||||
assert.Equal(t, http.StatusBadRequest, r.StatusCode)
|
||||
assert.Equal(t, `limit=2, remaining=1, reset=59`, r.Header.Get("Ratelimit"))
|
||||
assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy"))
|
||||
|
||||
r, err = c.Do(func() *http.Request {
|
||||
r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil)
|
||||
require.NoError(t, err)
|
||||
return r
|
||||
}())
|
||||
require.NoError(t, err)
|
||||
// unauthed request, so we expect a 400
|
||||
assert.Equal(t, http.StatusBadRequest, r.StatusCode)
|
||||
assert.Equal(t, `limit=2, remaining=0, reset=59`, r.Header.Get("Ratelimit"))
|
||||
assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy"))
|
||||
|
||||
r, err = c.Do(func() *http.Request {
|
||||
r, err := http.NewRequest(http.MethodGet, `http://127.0.0.1:9500/v1/targets`, nil)
|
||||
require.NoError(t, err)
|
||||
return r
|
||||
}())
|
||||
require.NoError(t, err)
|
||||
// out of quota, so we expect a 429
|
||||
assert.Equal(t, http.StatusTooManyRequests, r.StatusCode)
|
||||
assert.Equal(t, `limit=2, remaining=0, reset=59`, r.Header.Get("Ratelimit"))
|
||||
assert.Equal(t, `2;w=60;comment="total", 60;w=30;comment="ip-address", 60;w=30;comment="auth-token"`, r.Header.Get("Ratelimit-Policy"))
|
||||
|
||||
cmd.ShutdownCh <- struct{}{}
|
||||
wg.Wait()
|
||||
}
|
||||
@@ -622,8 +622,6 @@ func (c *Command) StartController(ctx context.Context) error {
|
||||
RawConfig: c.Config,
|
||||
Server: c.Server,
|
||||
TestOverrideWorkerAuthCaCertificateLifetime: c.flagWorkerAuthCaCertificateLifetime,
|
||||
ApiRateLimits: c.Config.Controller.ApiRateLimits,
|
||||
ApiRateLimiterMaxEntries: c.Config.Controller.ApiRateLimiterMaxEntries,
|
||||
}
|
||||
|
||||
var err error
|
||||
@@ -962,7 +960,7 @@ func (c *Command) reloadControllerRateLimits(newConfig *config.Config) error {
|
||||
if c.controller == nil || newConfig == nil || newConfig.Controller == nil {
|
||||
return nil
|
||||
}
|
||||
return c.controller.ReloadRateLimiter(newConfig.Controller.ApiRateLimits, newConfig.Controller.ApiRateLimiterMaxEntries)
|
||||
return c.controller.ReloadRateLimiter(newConfig)
|
||||
}
|
||||
|
||||
// acquireSchemaManager returns a schema manager and generally acquires a shared lock on
|
||||
|
||||
@@ -217,6 +217,7 @@ type Controller struct {
|
||||
|
||||
ApiRateLimits ratelimit.Configs `hcl:"-"`
|
||||
ApiRateLimiterMaxEntries int `hcl:"api_rate_limit_max_entries"`
|
||||
ApiRateLimitDisable bool `hcl:"api_rate_limit_disable"`
|
||||
|
||||
// License is the license used by HCP builds
|
||||
License string `hcl:"license"`
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
|
||||
"github.com/hashicorp/boundary/internal/cmd/base"
|
||||
"github.com/hashicorp/boundary/internal/cmd/config"
|
||||
"github.com/hashicorp/boundary/internal/ratelimit"
|
||||
)
|
||||
|
||||
type Config struct {
|
||||
@@ -23,8 +22,8 @@ type Config struct {
|
||||
// Override worker auth CA certificate lifetime for testing
|
||||
TestOverrideWorkerAuthCaCertificateLifetime time.Duration
|
||||
|
||||
// These are duplicated from the config.Config to track the state if these
|
||||
// change via SIGHUP reloads.
|
||||
ApiRateLimits ratelimit.Configs
|
||||
ApiRateLimiterMaxEntries int
|
||||
// This is derived from the config.Config. It tracks the state of the
|
||||
// rate limiter's configuration, and is updated if the config changes via a
|
||||
// SIGHUP.
|
||||
rateLimiterConfig *rateLimiterConfig
|
||||
}
|
||||
@@ -51,7 +51,6 @@ import (
|
||||
external_plugins "github.com/hashicorp/boundary/sdk/plugins"
|
||||
"github.com/hashicorp/boundary/version"
|
||||
"github.com/hashicorp/go-hclog"
|
||||
"github.com/hashicorp/go-rate"
|
||||
"github.com/hashicorp/go-secure-stdlib/mlock"
|
||||
"github.com/hashicorp/go-secure-stdlib/pluginutil/v2"
|
||||
"github.com/hashicorp/nodeenrollment"
|
||||
@@ -121,7 +120,8 @@ type Controller struct {
|
||||
apiGrpcServerListener grpcServerListener
|
||||
apiGrpcGatewayTicket string
|
||||
|
||||
rateLimiter *atomic.Pointer[rate.Limiter]
|
||||
rateLimiter ratelimit.Limiter
|
||||
rateLimiterMu sync.RWMutex
|
||||
|
||||
// Repo factory methods
|
||||
AuthTokenRepoFn common.AuthTokenRepoFactory
|
||||
@@ -174,7 +174,6 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
|
||||
pkiConnManager: cluster.NewDownstreamManager(),
|
||||
workerStatusGracePeriod: new(atomic.Int64),
|
||||
livenessTimeToStale: new(atomic.Int64),
|
||||
rateLimiter: new(atomic.Pointer[rate.Limiter]),
|
||||
}
|
||||
|
||||
if downstreamReceiverFactory != nil {
|
||||
@@ -251,19 +250,9 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
|
||||
}
|
||||
c.clusterListener = clusterListeners[0]
|
||||
|
||||
rateLimits, err := conf.RawConfig.Controller.ApiRateLimits.Limits(c.baseContext)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error parsing rate limit configuration: %w", err)
|
||||
}
|
||||
|
||||
rateLimiter, err := ratelimit.NewLimiter(
|
||||
rateLimits,
|
||||
conf.RawConfig.Controller.ApiRateLimiterMaxEntries,
|
||||
)
|
||||
if err != nil {
|
||||
if err := c.initializeRateLimiter(conf.RawConfig); err != nil {
|
||||
return nil, fmt.Errorf("error initializing rate limiter: %w", err)
|
||||
}
|
||||
c.rateLimiter.Store(rateLimiter)
|
||||
|
||||
var pluginLogger hclog.Logger
|
||||
for _, enabledPlugin := range c.enabledPlugins {
|
||||
@@ -638,37 +627,3 @@ func (c *Controller) Shutdown() error {
|
||||
func (c *Controller) WorkerStatusUpdateTimes() *sync.Map {
|
||||
return c.workerStatusUpdateTimes
|
||||
}
|
||||
|
||||
func (c *Controller) getRateLimiter() *rate.Limiter {
|
||||
return c.rateLimiter.Load()
|
||||
}
|
||||
|
||||
// ReloadRateLimiter replaces the Controller's rate.Limiter with a new rate.Limiter
|
||||
// using the supplied ratelimit.Configs and max entries. If the configs and max
|
||||
// entries match the current values, the rate.Limiter is not replaced and no
|
||||
// error is returned. Otherwise the rate.Limiter is replaced, and the old rate.Limiter
|
||||
// is shutdown. This means that a client's quota is effectively reset if the
|
||||
// configuration changes.
|
||||
func (c *Controller) ReloadRateLimiter(newLimitConfigs ratelimit.Configs, newMaxEntries int) error {
|
||||
// Config has not changed, no need to reload.
|
||||
if c.conf.ApiRateLimits.Equal(newLimitConfigs) && c.conf.ApiRateLimiterMaxEntries == newMaxEntries {
|
||||
return nil
|
||||
}
|
||||
|
||||
ratelimits, err := newLimitConfigs.Limits(c.baseContext)
|
||||
if err != nil {
|
||||
return fmt.Errorf("error parsing rate limit configuration: %w", err)
|
||||
}
|
||||
|
||||
limiter, err := ratelimit.NewLimiter(ratelimits, newMaxEntries)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
old := c.rateLimiter.Swap(limiter)
|
||||
|
||||
c.conf.ApiRateLimits = newLimitConfigs
|
||||
c.conf.ApiRateLimiterMaxEntries = newMaxEntries
|
||||
|
||||
return old.Shutdown()
|
||||
}
|
||||
@@ -0,0 +1,141 @@
|
||||
// Copyright (c) HashiCorp, Inc.
|
||||
// SPDX-License-Identifier: BUSL-1.1
|
||||
|
||||
package controller
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/hashicorp/boundary/internal/cmd/config"
|
||||
"github.com/hashicorp/boundary/internal/errors"
|
||||
"github.com/hashicorp/boundary/internal/ratelimit"
|
||||
"github.com/hashicorp/go-rate"
|
||||
)
|
||||
|
||||
type rateLimiterConfig struct {
|
||||
maxSize int
|
||||
disabled bool
|
||||
configs ratelimit.Configs
|
||||
|
||||
limits []rate.Limit
|
||||
}
|
||||
|
||||
func newRateLimiterConfig(ctx context.Context, configs ratelimit.Configs, maxSize int, disabled bool) (*rateLimiterConfig, error) {
|
||||
const op = "controller.newRateLimiterConfig"
|
||||
|
||||
switch {
|
||||
case disabled && len(configs) != 0:
|
||||
return nil, errors.New(ctx, errors.InvalidConfiguration, op, "disabled rate limiter with rate limit configs")
|
||||
}
|
||||
|
||||
var limits []rate.Limit
|
||||
var err error
|
||||
if !disabled {
|
||||
if limits, err = configs.Limits(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return &rateLimiterConfig{
|
||||
maxSize: maxSize,
|
||||
disabled: disabled,
|
||||
configs: configs,
|
||||
limits: limits,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Controller) initializeRateLimiter(conf *config.Config) error {
|
||||
const op = "controller.(Controller).initializeRateLimiter"
|
||||
switch {
|
||||
case conf == nil:
|
||||
return errors.New(c.baseContext, errors.InvalidParameter, op, "nil config")
|
||||
case conf.Controller == nil:
|
||||
return errors.New(c.baseContext, errors.InvalidParameter, op, "nil config.Controller")
|
||||
}
|
||||
|
||||
c.rateLimiterMu.Lock()
|
||||
defer c.rateLimiterMu.Unlock()
|
||||
|
||||
rlConfig, err := newRateLimiterConfig(
|
||||
c.baseContext,
|
||||
conf.Controller.ApiRateLimits,
|
||||
conf.Controller.ApiRateLimiterMaxEntries,
|
||||
conf.Controller.ApiRateLimitDisable,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch {
|
||||
case rlConfig.disabled:
|
||||
c.rateLimiter = rate.NopLimiter
|
||||
default:
|
||||
c.rateLimiter, err = ratelimit.NewLimiter(rlConfig.limits, rlConfig.maxSize)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
c.conf.rateLimiterConfig = rlConfig
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Controller) getRateLimiter() ratelimit.Limiter {
|
||||
c.rateLimiterMu.RLock()
|
||||
defer c.rateLimiterMu.RUnlock()
|
||||
return c.rateLimiter
|
||||
}
|
||||
|
||||
// ReloadRateLimiter replaces the Controller's rate.Limiter with a new rate.Limiter
|
||||
// using the supplied ratelimit.Configs and max entries. If the configs and max
|
||||
// entries match the current values, the rate.Limiter is not replaced and no
|
||||
// error is returned. Otherwise the rate.Limiter is replaced, and the old rate.Limiter
|
||||
// is shutdown. This means that a client's quota is effectively reset if the
|
||||
// configuration changes.
|
||||
func (c *Controller) ReloadRateLimiter(newConfig *config.Config) error {
|
||||
const op = "controller.(Controller).ReloadRateLimiter"
|
||||
|
||||
switch {
|
||||
case newConfig == nil:
|
||||
return errors.New(c.baseContext, errors.InvalidParameter, op, "nil config")
|
||||
case newConfig.Controller == nil:
|
||||
return errors.New(c.baseContext, errors.InvalidParameter, op, "nil config.Controller")
|
||||
}
|
||||
|
||||
rlConfig, err := newRateLimiterConfig(
|
||||
c.baseContext,
|
||||
newConfig.Controller.ApiRateLimits,
|
||||
newConfig.Controller.ApiRateLimiterMaxEntries,
|
||||
newConfig.Controller.ApiRateLimitDisable,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Config has not changed, no need to reload.
|
||||
if c.conf.rateLimiterConfig.maxSize == rlConfig.maxSize &&
|
||||
c.conf.rateLimiterConfig.disabled == rlConfig.disabled &&
|
||||
c.conf.rateLimiterConfig.configs.Equal(rlConfig.configs) {
|
||||
return nil
|
||||
}
|
||||
|
||||
var limiter ratelimit.Limiter
|
||||
switch {
|
||||
case rlConfig.disabled:
|
||||
limiter = rate.NopLimiter
|
||||
default:
|
||||
limiter, err = ratelimit.NewLimiter(rlConfig.limits, rlConfig.maxSize)
|
||||
if err != nil {
|
||||
return errors.Wrap(c.baseContext, err, op)
|
||||
}
|
||||
|
||||
}
|
||||
c.rateLimiterMu.Lock()
|
||||
old := c.rateLimiter
|
||||
c.rateLimiter = limiter
|
||||
c.conf.rateLimiterConfig = rlConfig
|
||||
c.rateLimiterMu.Unlock()
|
||||
|
||||
return old.Shutdown()
|
||||
}
|
||||
@@ -0,0 +1,481 @@
|
||||
// Copyright (c) HashiCorp, Inc.
|
||||
// SPDX-License-Identifier: BUSL-1.1
|
||||
|
||||
package controller
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/go-cmp/cmp"
|
||||
"github.com/google/go-cmp/cmp/cmpopts"
|
||||
"github.com/hashicorp/boundary/internal/cmd/config"
|
||||
"github.com/hashicorp/boundary/internal/event"
|
||||
"github.com/hashicorp/boundary/internal/ratelimit"
|
||||
"github.com/hashicorp/go-rate"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func Test_newRateLimiterConfig(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
var configs ratelimit.Configs
|
||||
defaultLimits, err := configs.Limits(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
configs ratelimit.Configs
|
||||
maxSize int
|
||||
disabled bool
|
||||
want *rateLimiterConfig
|
||||
wantErr error
|
||||
}{
|
||||
{
|
||||
"disabled",
|
||||
nil,
|
||||
0,
|
||||
true,
|
||||
&rateLimiterConfig{disabled: true},
|
||||
nil,
|
||||
},
|
||||
{
|
||||
"defaults",
|
||||
nil,
|
||||
ratelimit.DefaultLimiterMaxEntries(),
|
||||
false,
|
||||
&rateLimiterConfig{
|
||||
maxSize: 296148,
|
||||
configs: nil,
|
||||
disabled: false,
|
||||
limits: defaultLimits,
|
||||
},
|
||||
nil,
|
||||
},
|
||||
{
|
||||
"disabledWithConfigs",
|
||||
ratelimit.Configs{
|
||||
{
|
||||
Resources: []string{"*"},
|
||||
Actions: []string{"*"},
|
||||
Per: "total",
|
||||
Limit: 100,
|
||||
Period: time.Minute,
|
||||
Unlimited: false,
|
||||
},
|
||||
},
|
||||
ratelimit.DefaultLimiterMaxEntries(),
|
||||
true,
|
||||
nil,
|
||||
fmt.Errorf("controller.newRateLimiterConfig: disabled rate limiter with rate limit configs: configuration issue: error #5000"),
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
got, err := newRateLimiterConfig(ctx, tc.configs, tc.maxSize, tc.disabled)
|
||||
if tc.wantErr != nil {
|
||||
assert.EqualError(t, err, tc.wantErr.Error())
|
||||
return
|
||||
}
|
||||
require.NoError(t, err)
|
||||
assert.Empty(t, cmp.Diff(
|
||||
tc.want,
|
||||
got,
|
||||
cmp.AllowUnexported(rateLimiterConfig{}),
|
||||
cmpopts.IgnoreFields(rateLimiterConfig{}, "limits"),
|
||||
))
|
||||
assert.ElementsMatch(t, got.limits, tc.want.limits)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestController_initializeRateLimiter(t *testing.T) {
|
||||
// Disabling eventing so reduce noise.
|
||||
event.TestWithoutEventing(t)
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
conf *config.Config
|
||||
wantNopLimiter bool
|
||||
wantErr error
|
||||
}{
|
||||
{
|
||||
"disabled",
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimitDisable: true,
|
||||
},
|
||||
},
|
||||
true,
|
||||
nil,
|
||||
},
|
||||
{
|
||||
"defaults",
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
},
|
||||
false,
|
||||
nil,
|
||||
},
|
||||
{
|
||||
"invalid",
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimits: ratelimit.Configs{
|
||||
{
|
||||
Resources: []string{"*"},
|
||||
Actions: []string{"*"},
|
||||
Per: "total",
|
||||
Limit: 100,
|
||||
Period: time.Minute,
|
||||
Unlimited: false,
|
||||
},
|
||||
},
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
ApiRateLimitDisable: true,
|
||||
},
|
||||
},
|
||||
false,
|
||||
fmt.Errorf("controller.newRateLimiterConfig: disabled rate limiter with rate limit configs: configuration issue: error #5000"),
|
||||
},
|
||||
{
|
||||
"nilConfig",
|
||||
nil,
|
||||
false,
|
||||
fmt.Errorf("controller.(Controller).initializeRateLimiter: nil config: parameter violation: error #100"),
|
||||
},
|
||||
{
|
||||
"nilConfigController",
|
||||
&config.Config{
|
||||
Controller: nil,
|
||||
},
|
||||
false,
|
||||
fmt.Errorf("controller.(Controller).initializeRateLimiter: nil config.Controller: parameter violation: error #100"),
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: tc.conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(tc.conf)
|
||||
if tc.wantErr != nil {
|
||||
assert.EqualError(t, err, tc.wantErr.Error())
|
||||
return
|
||||
}
|
||||
require.NoError(t, err)
|
||||
|
||||
if tc.wantNopLimiter {
|
||||
assert.Equal(t, rate.NopLimiter, c.rateLimiter)
|
||||
return
|
||||
}
|
||||
|
||||
_, ok := c.rateLimiter.(*rate.Limiter)
|
||||
assert.True(t, ok, "expected rate.Limiter")
|
||||
assert.NotNil(t, c.rateLimiter)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestControllerReloadRateLimiter(t *testing.T) {
|
||||
// Disabling eventing so reduce noise.
|
||||
event.TestWithoutEventing(t)
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
c *Controller
|
||||
conf *config.Config
|
||||
wantNewLimiter bool
|
||||
wantErr error
|
||||
}{
|
||||
{
|
||||
"newConfigs",
|
||||
func() *Controller {
|
||||
conf := &config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
}
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(conf)
|
||||
require.NoError(t, err)
|
||||
return c
|
||||
}(),
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimits: ratelimit.Configs{
|
||||
{
|
||||
Resources: []string{"*"},
|
||||
Actions: []string{"*"},
|
||||
Per: "total",
|
||||
Limit: 100,
|
||||
Period: time.Minute,
|
||||
Unlimited: false,
|
||||
},
|
||||
},
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
nil,
|
||||
},
|
||||
{
|
||||
"newMaxSize",
|
||||
func() *Controller {
|
||||
conf := &config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
}
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(conf)
|
||||
require.NoError(t, err)
|
||||
return c
|
||||
}(),
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: 3000,
|
||||
},
|
||||
},
|
||||
true,
|
||||
nil,
|
||||
},
|
||||
{
|
||||
"newDisabled",
|
||||
func() *Controller {
|
||||
conf := &config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
}
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(conf)
|
||||
require.NoError(t, err)
|
||||
return c
|
||||
}(),
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimitDisable: true,
|
||||
},
|
||||
},
|
||||
true,
|
||||
nil,
|
||||
},
|
||||
{
|
||||
"newEnabled",
|
||||
func() *Controller {
|
||||
conf := &config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimitDisable: true,
|
||||
},
|
||||
}
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(conf)
|
||||
require.NoError(t, err)
|
||||
return c
|
||||
}(),
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
},
|
||||
true,
|
||||
nil,
|
||||
},
|
||||
{
|
||||
"newInvalid",
|
||||
func() *Controller {
|
||||
conf := &config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
}
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(conf)
|
||||
require.NoError(t, err)
|
||||
return c
|
||||
}(),
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
ApiRateLimits: ratelimit.Configs{
|
||||
{
|
||||
Resources: []string{"*"},
|
||||
Actions: []string{"*"},
|
||||
Per: "total",
|
||||
Limit: 100,
|
||||
Period: time.Minute,
|
||||
Unlimited: false,
|
||||
},
|
||||
},
|
||||
ApiRateLimitDisable: true,
|
||||
},
|
||||
},
|
||||
false,
|
||||
fmt.Errorf("controller.newRateLimiterConfig: disabled rate limiter with rate limit configs: configuration issue: error #5000"),
|
||||
},
|
||||
{
|
||||
"newInvalidMaxSize",
|
||||
func() *Controller {
|
||||
conf := &config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
}
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(conf)
|
||||
require.NoError(t, err)
|
||||
return c
|
||||
}(),
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: 0,
|
||||
ApiRateLimits: ratelimit.Configs{
|
||||
{
|
||||
Resources: []string{"*"},
|
||||
Actions: []string{"*"},
|
||||
Per: "total",
|
||||
Limit: 100,
|
||||
Period: time.Minute,
|
||||
Unlimited: false,
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
false,
|
||||
fmt.Errorf("controller.(Controller).ReloadRateLimiter: unknown, unknown: error #0: rate.NewLimiter: rate.newExpirableStore: max size must be greater than zero: invalid max size"),
|
||||
},
|
||||
{
|
||||
"newConfigsNoChange",
|
||||
func() *Controller {
|
||||
conf := &config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
}
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(conf)
|
||||
require.NoError(t, err)
|
||||
return c
|
||||
}(),
|
||||
&config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
},
|
||||
false,
|
||||
nil,
|
||||
},
|
||||
{
|
||||
"nilNewConfigController",
|
||||
func() *Controller {
|
||||
conf := &config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
}
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(conf)
|
||||
require.NoError(t, err)
|
||||
return c
|
||||
}(),
|
||||
&config.Config{
|
||||
Controller: nil,
|
||||
},
|
||||
false,
|
||||
fmt.Errorf("controller.(Controller).ReloadRateLimiter: nil config.Controller: parameter violation: error #100"),
|
||||
},
|
||||
{
|
||||
"nilNewConfig",
|
||||
func() *Controller {
|
||||
conf := &config.Config{
|
||||
Controller: &config.Controller{
|
||||
ApiRateLimiterMaxEntries: ratelimit.DefaultLimiterMaxEntries(),
|
||||
},
|
||||
}
|
||||
c := &Controller{
|
||||
baseContext: context.Background(),
|
||||
conf: &Config{
|
||||
RawConfig: conf,
|
||||
},
|
||||
}
|
||||
err := c.initializeRateLimiter(conf)
|
||||
require.NoError(t, err)
|
||||
return c
|
||||
}(),
|
||||
nil,
|
||||
false,
|
||||
fmt.Errorf("controller.(Controller).ReloadRateLimiter: nil config: parameter violation: error #100"),
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
prevLimiter := tc.c.getRateLimiter()
|
||||
err := tc.c.ReloadRateLimiter(tc.conf)
|
||||
if tc.wantErr != nil {
|
||||
assert.EqualError(t, err, tc.wantErr.Error())
|
||||
} else {
|
||||
require.NoError(t, err)
|
||||
}
|
||||
if tc.wantNewLimiter {
|
||||
assert.NotSame(t, prevLimiter, tc.c.getRateLimiter())
|
||||
return
|
||||
}
|
||||
assert.Same(t, prevLimiter, tc.c.getRateLimiter())
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -699,7 +699,7 @@ func TestHandler(t *testing.T) {
|
||||
|
||||
next.ServeHTTP(rw, req)
|
||||
})
|
||||
}(ratelimit.Handler(ctx, func() *rate.Limiter { return tc.limiter }, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
}(ratelimit.Handler(ctx, func() ratelimit.Limiter { return tc.limiter }, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
return
|
||||
}))),
|
||||
@@ -799,7 +799,7 @@ func TestHandlerErrors(t *testing.T) {
|
||||
|
||||
next.ServeHTTP(rw, req)
|
||||
})
|
||||
}(ratelimit.Handler(ctx, func() *rate.Limiter { return r }, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
}(ratelimit.Handler(ctx, func() ratelimit.Limiter { return r }, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusOK)
|
||||
return
|
||||
}))),
|
||||
|
||||
@@ -3,7 +3,20 @@
|
||||
|
||||
package ratelimit
|
||||
|
||||
import "github.com/hashicorp/go-rate"
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
"github.com/hashicorp/go-rate"
|
||||
)
|
||||
|
||||
// Limiter is used to provide rate limiting. This is the interface implemented
|
||||
// by both rate.Limiter and rate.NopLimiter.
|
||||
type Limiter interface {
|
||||
SetPolicyHeader(string, string, http.Header) error
|
||||
SetUsageHeader(*rate.Quota, http.Header)
|
||||
Allow(string, string, string, string) (bool, *rate.Quota, error)
|
||||
Shutdown() error
|
||||
}
|
||||
|
||||
// NewLimiter creates a rate.Limiter.
|
||||
func NewLimiter(limits []rate.Limit, maxEntries int) (*rate.Limiter, error) {
|
||||
|
||||
Reference in new issue
Block a user