feat(controller): Reload api rate limits on SIGHUP

internal/cmd/commands/server/controller_ratelimit_reload_test.go

@@ -0,0 +1,327 @@
+// Copyright (c) HashiCorp, Inc.
+// SPDX-License-Identifier: BUSL-1.1
+
+package server
+
+import (
+	"fmt"
+	"net/http"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hashicorp/boundary/internal/cmd/config"
+	"github.com/mitchellh/cli"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	ratelimitConfig = `
+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 {
+		resources = ["*"]
+		actions   = ["*"]
+		per       = "total"
+		limit     = 2
+		period    = "1m"
+	}
+}
+
+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"
+}
+`
+
+	ratelimitConfigReload = `
+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 {
+		resources = ["*"]
+		actions   = ["*"]
+		per       = "total"
+		limit     = 5
+		period    = "1m"
+	}
+}
+
+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"
+}
+`
+)
+
+func TestRealodControllerRateLimits(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 read for reloading
+	cfgHcl = fmt.Sprintf(ratelimitConfigReload, 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 have reset and the new limit
+	// 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=5, remaining=4, reset=59`, r.Header.Get("Ratelimit"))
+	assert.Equal(t, `5;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()
+}
+
+func TestRealodControllerRateLimitsSameConfig(t *testing.T) {
+	td := t.TempDir()
+
+	// Create and migrate database A and B.
+	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")
+	}
+
+	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 not have reset, and the current
+	// quota should still apply.
+	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)
+	// should still be rate limited, so 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()
+}

internal/cmd/commands/server/server.go

@@ -622,6 +622,8 @@ 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
@@ -841,6 +843,10 @@ func (c *Command) Reload(newConf *config.Config) error {
 		reloadErrors = stderrors.Join(reloadErrors, fmt.Errorf("failed to reload controller database: %w", err))
 	}
 
+	if err := c.reloadControllerRateLimits(newConf); err != nil {
+		reloadErrors = stderrors.Join(reloadErrors, fmt.Errorf("failed to reload controller api rate limits: %w", err))
+	}
+
 	if newConf != nil && c.worker != nil {
 		workerReloadErr := func() error {
 			if newConf.Controller != nil {
@@ -952,6 +958,13 @@ func (c *Command) reloadControllerDatabase(newConfig *config.Config) error {
 	return nil
 }
 
+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)
+}
+
 // acquireSchemaManager returns a schema manager and generally acquires a shared lock on
 // the database. This is done as a mechanism to disallow running migration commands
 // while the database is in use.

internal/daemon/controller/config.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/hashicorp/boundary/internal/cmd/base"
 	"github.com/hashicorp/boundary/internal/cmd/config"
+	"github.com/hashicorp/boundary/internal/ratelimit"
 )
 
 type Config struct {
@@ -21,4 +22,9 @@ type Config struct {
 	DisableAuthorizationFailures bool
 	// 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
 }

internal/daemon/controller/controller.go

@@ -35,6 +35,7 @@ import (
 	kmsjob "github.com/hashicorp/boundary/internal/kms/job"
 	"github.com/hashicorp/boundary/internal/plugin"
 	"github.com/hashicorp/boundary/internal/plugin/loopback"
+	"github.com/hashicorp/boundary/internal/ratelimit"
 	"github.com/hashicorp/boundary/internal/scheduler"
 	"github.com/hashicorp/boundary/internal/scheduler/cleaner"
 	"github.com/hashicorp/boundary/internal/scheduler/job"
@@ -120,7 +121,7 @@ type Controller struct {
 	apiGrpcServerListener grpcServerListener
 	apiGrpcGatewayTicket  string
 
-	rateLimiter *rate.Limiter
+	rateLimiter *atomic.Pointer[rate.Limiter]
 
 	// Repo factory methods
 	AuthTokenRepoFn           common.AuthTokenRepoFactory
@@ -172,6 +173,7 @@ 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 {
@@ -253,10 +255,11 @@ func New(ctx context.Context, conf *Config) (*Controller, error) {
 		return nil, fmt.Errorf("error parsing rate limit configuration: %w", err)
 	}
 
-	c.rateLimiter, err = rate.NewLimiter(rateLimits, conf.RawConfig.Controller.ApiRateLimiterMaxEntries)
+	rateLimiter, err := rate.NewLimiter(rateLimits, conf.RawConfig.Controller.ApiRateLimiterMaxEntries)
 	if 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 {
@@ -631,3 +634,37 @@ 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 := rate.NewLimiter(ratelimits, newMaxEntries)
+	if err != nil {
+		return err
+	}
+
+	old := c.rateLimiter.Swap(limiter)
+
+	c.conf.ApiRateLimits = newLimitConfigs
+	c.conf.ApiRateLimiterMaxEntries = newMaxEntries
+
+	return old.Shutdown()
+}

internal/daemon/controller/handler.go

@@ -74,7 +74,7 @@ func createMuxWithEndpoints(c *Controller, props HandlerProperties) (http.Handle
 	}
 
 	mux := http.NewServeMux()
-	mux.Handle("/v1/", ratelimit.Handler(c.baseContext, c.rateLimiter, grpcGwMux))
+	mux.Handle("/v1/", ratelimit.Handler(c.baseContext, c.getRateLimiter, grpcGwMux))
 	mux.Handle(uiPath, handleUi(c))
 
 	isUiRequest := func(req *http.Request) bool {

internal/ratelimit/config.go

@@ -6,6 +6,7 @@ package ratelimit
 import (
 	"context"
 	"fmt"
+	"reflect"
 	"sync"
 	"time"
 
@@ -108,6 +109,11 @@ type Config struct {
 // Configs is an ordered set of Config.
 type Configs []*Config
 
+// Equal checks if a set of Configs is equal to another set of Configs.
+func (c Configs) Equal(o Configs) bool {
+	return reflect.DeepEqual(c, o)
+}
+
 // Limits creates a slice of rate.Limit from the Configs. This will enumerate
 // every combination of resource+action, defining a Limit for each.
 func (c Configs) Limits(ctx context.Context) ([]*rate.Limit, error) {

internal/ratelimit/handler.go

@@ -112,14 +112,18 @@ func extractResourceAction(path, method string) (res, act string, err error) {
 	return
 }
 
+// LimiterFunc returns a rate.Limiter
+type LimiterFunc func() *rate.Limiter
+
 // Handler is an http middleware handler that checks if a request is allowed
-// using the provided rate limiter. If the request is allowed, the next handler
+// using the rate limiter returned by f. If the request is allowed, the next handler
 // is called. Otherwise a 429 is returned with the Retry-After response header
 // set to the number of seconds the client should wait to make it's next request.
-func Handler(ctx context.Context, l *rate.Limiter, next http.Handler) http.Handler {
+func Handler(ctx context.Context, f LimiterFunc, next http.Handler) http.Handler {
 	const op = "ratelimit.Handler"
-
 	return http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+		l := f()
+
 		reqInfo, ok := event.RequestInfoFromContext(req.Context())
 		if !ok || reqInfo == nil || reqInfo.ClientIp == "" {
 			rw.WriteHeader(http.StatusInternalServerError)

internal/ratelimit/handler_test.go

@@ -747,7 +747,7 @@ func TestHandler(t *testing.T) {
 
 						next.ServeHTTP(rw, req)
 					})
-				}(ratelimit.Handler(ctx, tc.limiter, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				}(ratelimit.Handler(ctx, func() *rate.Limiter { return tc.limiter }, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 					w.WriteHeader(http.StatusOK)
 					return
 				}))),
@@ -850,7 +850,7 @@ func TestHandlerErrors(t *testing.T) {
 
 						next.ServeHTTP(rw, req)
 					})
-				}(ratelimit.Handler(ctx, r, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				}(ratelimit.Handler(ctx, func() *rate.Limiter { return r }, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 					w.WriteHeader(http.StatusOK)
 					return
 				}))),