scottk/vault
1
0
Fork 0
mirror of https://github.com/optim-enterprises-bv/vault.git synced 2025-10-29 17:52:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Request Limiter (#25093)

Browse Source 
This commit introduces two new adaptive concurrency limiters in Vault,
which should handle overloading of the server during periods of
untenable request rate. The limiter adjusts the number of allowable
in-flight requests based on latency measurements performed across the
request duration. This approach allows us to reject entire requests
prior to doing any work and prevents clients from exceeding server
capacity.

The limiters intentionally target two separate vectors that have been
proven to lead to server over-utilization.

- Back pressure from the storage backend, resulting in bufferbloat in
  the WAL system. (enterprise)
- Back pressure from CPU over-utilization via PKI issue requests
  (specifically for RSA keys), resulting in failed heartbeats.

Storage constraints can be accounted for by limiting logical requests
according to their http.Method. We only limit requests with write-based
methods, since these will result in storage Puts and exhibit the
aforementioned bufferbloat.

CPU constraints are accounted for using the same underlying library and
technique; however, they require special treatment. The maximum number
of concurrent pki/issue requests found in testing (again, specifically
for RSA keys) is far lower than the minimum tolerable write request
rate. Without separate limiting, we would artificially impose limits on
tolerable request rates for non-PKI requests. To specifically target PKI
issue requests, we add a new PathsSpecial field, called limited,
allowing backends to specify a list of paths which should get
special-case request limiting.

For the sake of code cleanliness and future extensibility, we introduce
the concept of a LimiterRegistry. The registry proposed in this PR has
two entries, corresponding with the two vectors above. Each Limiter
entry has its own corresponding maximum and minimum concurrency,
allowing them to react to latency deviation independently and handle
high volumes of requests to targeted bottlenecks (CPU and storage).

In both cases, utilization will be effectively throttled before Vault
reaches any degraded state. The resulting 503 - Service Unavailable is a
retryable HTTP response code, which can be handled to gracefully retry
and eventually succeed. Clients should handle this by retrying with
jitter and exponential backoff. This is done within Vault's API, using
the go-retryablehttp library.

Limiter testing was performed via benchmarks of mixed workloads and
across a deployment of agent pods with great success.
This commit is contained in:
Mike Palmiotto
2024-01-26 14:26:21 -05:00
committed by GitHub
parent 3ba802d8dc
commit 43be9fc18a
21 changed files with 1166 additions and 586 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
builtin/logical/pki/backend.go
View File

@@ -147,6 +147,11 @@ func Backend(conf *logical.BackendConfig) *backend {
unifiedDeltaWALPath,
},
Limited: []string{
"issue",
"issue/*",
},
Binary: []string{
"ocsp", // OCSP POST
"ocsp/*", // OCSP GET

5
changelog/25093.txt Normal file
View File

@@ -0,0 +1,5 @@
```release-note:feature
**Request Limiter**: Add adaptive concurrency limits to write-based HTTP
methods and special-case `pki/issue` requests to prevent overloading the Vault
server.
```

1
go.mod
View File

@@ -191,6 +191,7 @@ require (
github.com/patrickmn/go-cache v2.1.0+incompatible
github.com/pires/go-proxyproto v0.6.1
github.com/pkg/errors v0.9.1
github.com/platinummonkey/go-concurrency-limits v0.7.0
github.com/posener/complete v1.2.3
github.com/pquerna/otp v1.2.1-0.20191009055518-468c2dd2b58d
github.com/prometheus/client_golang v1.14.0

5
go.sum
View File

@@ -1039,6 +1039,7 @@ github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym
github.com/DataDog/datadog-go v2.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
github.com/DataDog/datadog-go v3.2.0+incompatible h1:qSG2N4FghB1He/r2mFrWKCaL7dXCilEuNEeAn20fdD4=
github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
github.com/DataDog/datadog-go/v5 v5.0.2/go.mod h1:ZI9JFB4ewXbw1sBnF4sxsR2k1H3xjV+PUAOUsHvKpcU=
github.com/HdrHistogram/hdrhistogram-go v1.1.2/go.mod h1:yDgFjdqOqDEKOvasDdhWNXYg9BVp4O+o5f6V/ehm6Oo=
github.com/Jeffail/gabs v1.1.1 h1:V0uzR08Hj22EX8+8QMhyI9sX2hwRu+/RJhJUmnwda/E=
github.com/Jeffail/gabs v1.1.1/go.mod h1:6xMvQMK4k33lb7GUUpaAPh6nKMmemQeg5d4gn7/bOXc=
@@ -1063,6 +1064,7 @@ github.com/Microsoft/go-winio v0.4.16/go.mod h1:XB6nPKklQyQ7GC9LdcBEcBl8PF76WugX
github.com/Microsoft/go-winio v0.4.17-0.20210211115548-6eac466e5fa3/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
github.com/Microsoft/go-winio v0.4.17-0.20210324224401-5516f17a5958/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
github.com/Microsoft/go-winio v0.4.17/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
github.com/Microsoft/go-winio v0.5.0/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
github.com/Microsoft/go-winio v0.5.1/go.mod h1:JPGBdM1cNvN/6ISo+n8V5iA4v8pBzdOpzfwIujj1a84=
github.com/Microsoft/go-winio v0.5.2/go.mod h1:WpS1mjBmmwHBEWmogvA2mj8546UReBk4v8QkMxJ6pZY=
github.com/Microsoft/go-winio v0.6.0/go.mod h1:cTAf44im0RAYeL23bpB+fzCyDH2MJiz2BO69KH/soAE=
@@ -2879,6 +2881,8 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZI=
github.com/pkg/sftp v1.13.1/go.mod h1:3HaPG6Dq1ILlpPZRO0HVMrsydcdLt6HRDccSgb87qRg=
github.com/platinummonkey/go-concurrency-limits v0.7.0 h1:Bl9E74+67BrlRLBeryHOaFy0e1L3zD9g436/3vo6akQ=
github.com/platinummonkey/go-concurrency-limits v0.7.0/go.mod h1:Xxr6BywMVH3QyLyd0PanLnkkkmByTTPET3azMpdfmng=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
@@ -2947,6 +2951,7 @@ github.com/prometheus/procfs v0.8.0/go.mod h1:z7EfXMXOkbkqb9IINtpCn86r/to3BnA0ua
github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
github.com/rboyer/safeio v0.2.1 h1:05xhhdRNAdS3apYm7JRjOqngf4xruaW959jmRxGDuSU=
github.com/rboyer/safeio v0.2.1/go.mod h1:Cq/cEPK+YXFn622lsQ0K4KsPZSPtaptHHEldsy7Fmig=
github.com/rcrowley/go-metrics v0.0.0-20180503174638-e2704e165165/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
github.com/remyoudompheng/bigfft v0.0.0-20200410134404-eec4a21b6bb0/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=

38
http/handler.go
View File

@@ -30,6 +30,7 @@ import (
"github.com/hashicorp/go-uuid"
"github.com/hashicorp/vault/helper/namespace"
"github.com/hashicorp/vault/internalshared/configutil"
"github.com/hashicorp/vault/limits"
"github.com/hashicorp/vault/sdk/helper/consts"
"github.com/hashicorp/vault/sdk/helper/jsonutil"
"github.com/hashicorp/vault/sdk/helper/pathmanager"
@@ -908,10 +909,47 @@ func forwardRequest(core *vault.Core, w http.ResponseWriter, r *http.Request) {
w.Write(retBytes)
}
func acquireLimiterListener(core *vault.Core, rawReq *http.Request, r *logical.Request) (*limits.RequestListener, bool) {
lim := &limits.RequestLimiter{}
if r.PathLimited {
lim = core.GetRequestLimiter(limits.SpecialPathLimiter)
} else {
switch rawReq.Method {
case http.MethodGet, http.MethodHead, http.MethodTrace, http.MethodOptions:
// We're only interested in the inverse, so do nothing here.
default:
lim = core.GetRequestLimiter(limits.WriteLimiter)
}
}
return lim.Acquire(rawReq.Context())
}
// request is a helper to perform a request and properly exit in the
// case of an error.
func request(core *vault.Core, w http.ResponseWriter, rawReq *http.Request, r *logical.Request) (*logical.Response, bool, bool) {
lsnr, ok := acquireLimiterListener(core, rawReq, r)
if !ok {
resp := &logical.Response{}
logical.RespondWithStatusCode(resp, r, http.StatusServiceUnavailable)
respondError(w, http.StatusServiceUnavailable, limits.ErrCapacity)
return resp, false, false
}
// To guard against leaking RequestListener slots, we should ignore Limiter
// measurements on panic. OnIgnore will check to see if a RequestListener
// slot has been acquired and not released, which could happen on
// recoverable panics.
defer lsnr.OnIgnore()
resp, err := core.HandleRequest(rawReq.Context(), r)
// Do the limiter measurement
if err != nil {
lsnr.OnDropped()
} else {
lsnr.OnSuccess()
}
if r.LastRemoteWAL() > 0 && !core.EntWaitUntilWALShipped(rawReq.Context(), r.LastRemoteWAL()) {
if resp == nil {
resp = &logical.Response{}

9
http/logical.go
View File

@@ -7,6 +7,7 @@ import (
"bufio"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"mime"
@@ -18,6 +19,7 @@ import (
"github.com/hashicorp/go-uuid"
"github.com/hashicorp/vault/helper/namespace"
"github.com/hashicorp/vault/limits"
"github.com/hashicorp/vault/sdk/helper/consts"
"github.com/hashicorp/vault/sdk/logical"
"github.com/hashicorp/vault/vault"
@@ -211,6 +213,10 @@ func buildLogicalRequestNoAuth(perfStandby bool, ra *vault.RouterAccess, w http.
Headers: r.Header,
}
if ra != nil && ra.IsLimitedPath(r.Context(), path) {
req.PathLimited = true
}
if passHTTPReq {
req.HTTPRequest = r
}
@@ -378,6 +384,9 @@ func handleLogicalInternal(core *vault.Core, injectDataIntoTopLevel bool, noForw
// success.
resp, ok, needsForward := request(core, w, r, req)
switch {
case errors.Is(resp.Error(), limits.ErrCapacity):
respondError(w, http.StatusServiceUnavailable, limits.ErrCapacity)
return
case needsForward && noForward:
respondError(w, http.StatusBadRequest, vault.ErrCannotForwardLocalOnly)
return

191
limits/limiter.go Normal file
View File

@@ -0,0 +1,191 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: BUSL-1.1
package limits
import (
"context"
"errors"
"fmt"
"math"
"sync/atomic"
"github.com/armon/go-metrics"
"github.com/hashicorp/go-hclog"
"github.com/platinummonkey/go-concurrency-limits/core"
"github.com/platinummonkey/go-concurrency-limits/limit"
"github.com/platinummonkey/go-concurrency-limits/limiter"
"github.com/platinummonkey/go-concurrency-limits/strategy"
)
var (
// ErrCapacity is a new error type to indicate that Vault is not accepting new
// requests. This should be handled by callers in request paths to return
// http.StatusServiceUnavailable to the client.
ErrCapacity = errors.New("Vault server temporarily overloaded")
// DefaultDebugLogger opts out of the go-concurrency-limits internal Debug
// logger, since it's rather noisy. We're generating logs of interest in
// Vault.
DefaultDebugLogger limit.Logger = nil
// DefaultMetricsRegistry opts out of the go-concurrency-limits internal
// metrics because we're tracking what we care about in Vault.
DefaultMetricsRegistry core.MetricRegistry = core.EmptyMetricRegistryInstance
)
const (
// Smoothing adjusts how heavily we weight newer high-latency detection.
// Higher values (>1) place more emphasis on recent measurements. We set
// this below 1 to better tolerate short-lived spikes in request rate.
DefaultSmoothing = .1
// DefaultLongWindow is chosen as a minimum of 1000 samples. longWindow
// defines sliding window size used for the Exponential Moving Average.
DefaultLongWindow = 1000
)
// RequestLimiter is a thin wrapper for limiter.DefaultLimiter.
type RequestLimiter struct {
*limiter.DefaultLimiter
}
// Acquire consults the underlying RequestLimiter to see if a new
// RequestListener can be acquired.
//
// The return values are a *RequestListener, which the caller can use to perform
// latency measurements, and a bool to indicate whether or not a RequestListener
// was acquired.
//
// The returned RequestListener is short-lived and eventually garbage-collected;
// however, the RequestLimiter keeps track of in-flight concurrency using a
// token bucket implementation. The caller must release the resulting Limiter
// token by conducting a measurement.
//
// There are three return cases:
//
// 1) If Request Limiting is disabled, we return an empty RequestListener so all
// measurements are no-ops.
//
// 2) If the request limit has been exceeded, we will not acquire a
// RequestListener and instead return nil, false. No measurement is required,
// since we immediately return from callers with ErrCapacity.
//
// 3) If we have not exceeded the request limit, the caller must call one of
// OnSuccess(), OnDropped(), or OnIgnore() to return a measurement and release
// the underlying Limiter token.
func (l *RequestLimiter) Acquire(ctx context.Context) (*RequestListener, bool) {
// Transparently handle the case where the limiter is disabled.
if l == nil || l.DefaultLimiter == nil {
return &RequestListener{}, true
}
lsnr, ok := l.DefaultLimiter.Acquire(ctx)
if !ok {
metrics.IncrCounter(([]string{"limits", "concurrency", "service_unavailable"}), 1)
// If the token acquisition fails, we've reached capacity and we won't
// get a listener, so just return nil.
return nil, false
}
return &RequestListener{
DefaultListener: lsnr.(*limiter.DefaultListener),
released: new(atomic.Bool),
}, true
}
// concurrencyChanger adjusts the current allowed concurrency with an
// exponential backoff as we approach the max limit.
func concurrencyChanger(limit int) int {
change := math.Sqrt(float64(limit))
if change < 1.0 {
change = 1.0
}
return int(change)
}
var (
// DefaultWriteLimiterFlags have a less conservative MinLimit to prevent
// over-optimizing the request latency, which would result in
// under-utilization and client starvation.
DefaultWriteLimiterFlags = LimiterFlags{
Name: WriteLimiter,
MinLimit: 100,
MaxLimit: 5000,
}
// DefaultSpecialPathLimiterFlags have a conservative MinLimit to allow more
// aggressive concurrency throttling for CPU-bound workloads such as
// `pki/issue`.
DefaultSpecialPathLimiterFlags = LimiterFlags{
Name: SpecialPathLimiter,
MinLimit: 5,
MaxLimit: 5000,
}
)
// LimiterFlags establish some initial configuration for a new request limiter.
type LimiterFlags struct {
// Name specifies the limiter Name for registry lookup and logging.
Name string
// MinLimit defines the minimum concurrency floor to prevent over-throttling
// requests during periods of high traffic.
MinLimit int
// MaxLimit defines the maximum concurrency ceiling to prevent skewing to a
// point of no return.
//
// We set this to a high value (5000) with the expectation that systems with
// high-performing specs will tolerate higher limits, while the algorithm
// will find its own steady-state concurrency well below this threshold in
// most cases.
MaxLimit int
// InitialLimit defines the starting concurrency limit prior to any
// measurements.
//
// If we start this value off too high, Vault could become
// overloaded before the algorithm has a chance to adapt. Setting the value
// to the minimum is a safety measure which could result in early request
// rejection; however, the adaptive nature of the algorithm will prevent
// this from being a prolonged state as the allowed concurrency will
// increase during normal operation.
InitialLimit int
}
// NewRequestLimiter is a basic constructor for the RequestLimiter wrapper. It
// is responsible for setting up the Gradient2 Limit and instantiating a new
// wrapped DefaultLimiter.
func NewRequestLimiter(logger hclog.Logger, flags LimiterFlags) (*RequestLimiter, error) {
logger.Info("setting up new request limiter",
"initialLimit", flags.InitialLimit,
"maxLimit", flags.MaxLimit,
"minLimit", flags.MinLimit,
)
// NewGradient2Limit is the algorithm which drives request limiting
// decisions. It gathers latency measurements and calculates an Exponential
// Moving Average to determine whether latency deviation warrants a change
// in the current concurrency limit.
lim, err := limit.NewGradient2Limit(flags.Name,
flags.InitialLimit,
flags.MaxLimit,
flags.MinLimit,
concurrencyChanger,
DefaultSmoothing,
DefaultLongWindow,
DefaultDebugLogger,
DefaultMetricsRegistry,
)
if err != nil {
return nil, fmt.Errorf("failed to create gradient2 limit: %w", err)
}
strategy := strategy.NewSimpleStrategy(flags.InitialLimit)
defLimiter, err := limiter.NewDefaultLimiter(lim, 1e9, 1e9, 10, 100, strategy, nil, DefaultMetricsRegistry)
if err != nil {
return &RequestLimiter{}, err
}
return &RequestLimiter{defLimiter}, nil
}

51
limits/listener.go Normal file
View File

@@ -0,0 +1,51 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: BUSL-1.1
package limits
import (
"sync/atomic"
"github.com/armon/go-metrics"
"github.com/platinummonkey/go-concurrency-limits/limiter"
)
// RequestListener is a thin wrapper for limiter.DefaultLimiter to handle the
// case where request limiting is turned off.
type RequestListener struct {
*limiter.DefaultListener
released *atomic.Bool
}
// OnSuccess is called as a notification that the operation succeeded and
// internally measured latency should be used as an RTT sample.
func (l *RequestListener) OnSuccess() {
if l.DefaultListener != nil {
metrics.IncrCounter(([]string{"limits", "concurrency", "success"}), 1)
l.DefaultListener.OnSuccess()
l.released.Store(true)
}
}
// OnDropped is called to indicate the request failed and was dropped due to an
// internal server error. Note that this does not include ErrCapacity.
func (l *RequestListener) OnDropped() {
if l.DefaultListener != nil {
metrics.IncrCounter(([]string{"limits", "concurrency", "dropped"}), 1)
l.DefaultListener.OnDropped()
l.released.Store(true)
}
}
// OnIgnore is called to indicate the operation failed before any meaningful RTT
// measurement could be made and should be ignored to not introduce an
// artificially low RTT. It also provides an extra layer of protection against
// leaks of the underlying StrategyToken during recoverable panics in the
// request handler. We treat these as Ignored, discard the measurement, and mark
// the listener as released.
func (l *RequestListener) OnIgnore() {
if l.DefaultListener != nil && l.released.Load() != true {
metrics.IncrCounter(([]string{"limits", "concurrency", "ignored"}), 1)
l.DefaultListener.OnIgnore()
l.released.Store(true)
}
}

203
limits/registry.go Normal file
View File

@@ -0,0 +1,203 @@
// Copyright (c) HashiCorp, Inc.
// SPDX-License-Identifier: BUSL-1.1
package limits
import (
"os"
"strconv"
"sync"
"github.com/hashicorp/go-hclog"
)
const (
WriteLimiter = "write"
SpecialPathLimiter = "special-path"
LimitsBadEnvVariable = "failed to process limiter environment variable, using default"
)
// NOTE: Great care should be taken when setting any of these variables to avoid
// adverse affects in optimal request servicing. It is strongly advised that
// these variables not be used unless there is a very good reason. These are
// intentionally undocumented environment variables that may be removed in
// future versions of Vault.
const (
// EnvVaultDisableWriteLimiter is used to turn off the
// RequestLimiter for write-based HTTP methods.
EnvVaultDisableWriteLimiter = "VAULT_DISABLE_WRITE_LIMITER"
// EnvVaultWriteLimiterMin is used to modify the minimum
// concurrency limit for write-based HTTP methods.
EnvVaultWriteLimiterMin = "VAULT_WRITE_LIMITER_MIN"
// EnvVaultWriteLimiterMax is used to modify the maximum
// concurrency limit for write-based HTTP methods.
EnvVaultWriteLimiterMax = "VAULT_WRITE_LIMITER_MAX"
// EnvVaultDisablePathBasedRequestLimiting is used to turn off the
// RequestLimiter for special-cased paths, specified in
// Backend.PathsSpecial.
EnvVaultDisableSpecialPathLimiter = "VAULT_DISABLE_SPECIAL_PATH_LIMITER"
// EnvVaultSpecialPathLimiterMin is used to modify the minimum
// concurrency limit for write-based HTTP methods.
EnvVaultSpecialPathLimiterMin = "VAULT_SPECIAL_PATH_LIMITER_MIN"
// EnvVaultSpecialPathLimiterMax is used to modify the maximum
// concurrency limit for write-based HTTP methods.
EnvVaultSpecialPathLimiterMax = "VAULT_SPECIAL_PATH_LIMITER_MAX"
)
// LimiterRegistry holds the map of RequestLimiters mapped to keys.
type LimiterRegistry struct {
Limiters map[string]*RequestLimiter
Logger hclog.Logger
Enabled bool
sync.RWMutex
}
// NewLimiterRegistry is a basic LimiterRegistry constructor.
func NewLimiterRegistry(logger hclog.Logger) *LimiterRegistry {
return &LimiterRegistry{
Limiters: make(map[string]*RequestLimiter),
Logger: logger,
}
}
// processEnvVars consults Limiter-specific environment variables and tells the
// caller if the Limiter should be disabled. If not, it adjusts the passed-in
// limiterFlags as appropriate.
func (r *LimiterRegistry) processEnvVars(flags *LimiterFlags, envDisabled, envMin, envMax string) bool {
envFlagsLogger := r.Logger.With("name", flags.Name)
if disabledRaw := os.Getenv(envDisabled); disabledRaw != "" {
disabled, err := strconv.ParseBool(disabledRaw)
if err != nil {
envFlagsLogger.Warn(LimitsBadEnvVariable,
"env", envDisabled,
"val", disabledRaw,
"default", false,
"error", err,
)
}
if disabled {
envFlagsLogger.Warn("limiter disabled by environment variable", "env", envDisabled, "val", disabledRaw)
return true
}
}
envFlags := &LimiterFlags{}
if minRaw := os.Getenv(envMin); minRaw != "" {
min, err := strconv.Atoi(minRaw)
if err != nil {
envFlagsLogger.Warn(LimitsBadEnvVariable,
"env", envMin,
"val", minRaw,
"default", flags.MinLimit,
"error", err,
)
} else {
envFlags.MinLimit = min
}
}
if maxRaw := os.Getenv(envMax); maxRaw != "" {
max, err := strconv.Atoi(maxRaw)
if err != nil {
envFlagsLogger.Warn(LimitsBadEnvVariable,
"env", envMax,
"val", maxRaw,
"default", flags.MaxLimit,
"error", err,
)
} else {
envFlags.MaxLimit = max
}
}
switch {
case envFlags.MinLimit == 0:
// Assume no environment variable was provided.
case envFlags.MinLimit > 0:
flags.MinLimit = envFlags.MinLimit
default:
r.Logger.Warn("min limit must be greater than zero, falling back to defaults", "minLimit", flags.MinLimit)
}
switch {
case envFlags.MaxLimit == 0:
// Assume no environment variable was provided.
case envFlags.MaxLimit > flags.MinLimit:
flags.MaxLimit = envFlags.MaxLimit
default:
r.Logger.Warn("max limit must be greater than min, falling back to defaults", "maxLimit", flags.MaxLimit)
}
return false
}
// Enable sets up a new LimiterRegistry and marks it Enabled.
func (r *LimiterRegistry) Enable() {
r.Lock()
defer r.Unlock()
if r.Enabled {
return
}
r.Logger.Info("enabling request limiters")
r.Limiters = map[string]*RequestLimiter{}
r.Register(DefaultWriteLimiterFlags)
r.Register(DefaultSpecialPathLimiterFlags)
r.Enabled = true
}
// Register creates a new request limiter and assigns it a slot in the
// LimiterRegistry. Locking should be done in the caller.
func (r *LimiterRegistry) Register(flags LimiterFlags) {
var disabled bool
switch flags.Name {
case WriteLimiter:
disabled = r.processEnvVars(&flags,
EnvVaultDisableWriteLimiter,
EnvVaultWriteLimiterMin,
EnvVaultWriteLimiterMax,
)
if disabled {
return
}
case SpecialPathLimiter:
disabled = r.processEnvVars(&flags,
EnvVaultDisableSpecialPathLimiter,
EnvVaultSpecialPathLimiterMin,
EnvVaultSpecialPathLimiterMax,
)
if disabled {
return
}
default:
r.Logger.Warn("skipping invalid limiter type", "key", flags.Name)
return
}
// Always set the initial limit to min so the system can find its own
// equilibrium, since max might be too high.
flags.InitialLimit = flags.MinLimit
limiter, err := NewRequestLimiter(r.Logger.Named(flags.Name), flags)
if err != nil {
r.Logger.Error("failed to register limiter", "name", flags.Name, "error", err)
return
}
r.Limiters[flags.Name] = limiter
}
// GetLimiter looks up a RequestLimiter by key in the LimiterRegistry.
func (r *LimiterRegistry) GetLimiter(key string) *RequestLimiter {
r.RLock()
defer r.RUnlock()
return r.Limiters[key]
}

11
sdk/logical/logical.go
View File

@@ -163,6 +163,17 @@ type Paths struct {
// Binary paths are those whose request bodies should not be assumed to
// be JSON encoded, and for which the backend will decode values for auditing
Binary []string
// Limited paths are storage paths that require special-cased request
// limiting.
//
// This was initially added to separate limiting of "write" requests
// (limits.WriteLimiter) from limiting for CPU-bound pki/issue requests
// (limits.SpecialPathLimiter). Other plugins might also choose to mark
// paths if they don't follow a typical resource usage pattern.
//
// For more details, consult limits/registry.go.
Limited []string
}
type Auditor interface {

4
sdk/logical/request.go
View File

@@ -194,6 +194,10 @@ type Request struct {
// accessible.
Unauthenticated bool `json:"unauthenticated" structs:"unauthenticated" mapstructure:"unauthenticated"`
// PathLimited indicates that the request path is marked for special-case
// request limiting.
PathLimited bool `json:"path_limited" structs:"path_limited" mapstructure:"path_limited"`
// MFACreds holds the parsed MFA information supplied over the API as part of
// X-Vault-MFA header
MFACreds MFACreds `json:"mfa_creds" structs:"mfa_creds" mapstructure:"mfa_creds" sentinel:""`

2
sdk/logical/response_util.go
View File

@@ -207,7 +207,7 @@ func RespondErrorAndData(w http.ResponseWriter, status int, data interface{}, er
type ErrorAndDataResponse struct {
Errors []string `json:"errors"`
Data interface{} `json:"data""`
Data interface{} `json:"data"`
}
resp := &ErrorAndDataResponse{Errors: make([]string, 0, 1)}
if err != nil {

1
sdk/plugin/grpc_backend_server.go
View File

@@ -198,6 +198,7 @@ func (b *backendGRPCPluginServer) SpecialPaths(ctx context.Context, args *pb.Emp
SealWrapStorage: paths.SealWrapStorage,
WriteForwardedStorage: paths.WriteForwardedStorage,
Binary: paths.Binary,
Limited: paths.Limited,
},
}, nil
}

1063
sdk/plugin/pb/backend.pb.go
View File

File diff suppressed because it is too large Load Diff

5
sdk/plugin/pb/backend.proto
View File

@@ -63,6 +63,11 @@ message Paths {
//
// See note in /sdk/logical/logical.go.
repeated string binary = 6;
// Limited paths are storage paths that require special-case request limiting.
//
// See note in /sdk/logical/logical.go.
repeated string limited = 7;
}
message Request {

19
vault/core.go
View File

@@ -49,6 +49,7 @@ import (
"github.com/hashicorp/vault/helper/metricsutil"
"github.com/hashicorp/vault/helper/namespace"
"github.com/hashicorp/vault/helper/osutil"
"github.com/hashicorp/vault/limits"
"github.com/hashicorp/vault/physical/raft"
"github.com/hashicorp/vault/sdk/helper/certutil"
"github.com/hashicorp/vault/sdk/helper/consts"
@@ -707,6 +708,9 @@ type Core struct {
periodicLeaderRefreshInterval time.Duration
clusterAddrBridge *raft.ClusterAddrBridge
limiterRegistry *limits.LimiterRegistry
limiterRegistryLock sync.Mutex
}
func (c *Core) ActiveNodeClockSkewMillis() int64 {
@@ -717,6 +721,10 @@ func (c *Core) EchoDuration() time.Duration {
return c.echoDuration.Load()
}
func (c *Core) GetRequestLimiter(key string) *limits.RequestLimiter {
return c.limiterRegistry.GetLimiter(key)
}
// c.stateLock needs to be held in read mode before calling this function.
func (c *Core) HAState() consts.HAState {
switch {
@@ -882,6 +890,8 @@ type CoreConfig struct {
PeriodicLeaderRefreshInterval time.Duration
ClusterAddrBridge *raft.ClusterAddrBridge
LimiterRegistry *limits.LimiterRegistry
}
// GetServiceRegistration returns the config's ServiceRegistration, or nil if it does
@@ -984,6 +994,10 @@ func CreateCore(conf *CoreConfig) (*Core, error) {
}
}
if conf.LimiterRegistry == nil {
conf.LimiterRegistry = limits.NewLimiterRegistry(conf.Logger.Named("limits"))
}
// Use imported logging deadlock if requested
var stateLock locking.RWMutex
stateLock = &locking.SyncRWMutex{}
@@ -1284,6 +1298,11 @@ func NewCore(conf *CoreConfig) (*Core, error) {
return nil, err
}
c.limiterRegistry = conf.LimiterRegistry
c.limiterRegistryLock.Lock()
c.limiterRegistry.Enable()
c.limiterRegistryLock.Unlock()
// Version history
if c.versionHistory == nil {
c.logger.Info("Initializing version history cache for core")

15
vault/core_metrics.go
View File

@@ -13,6 +13,7 @@ import (
"github.com/armon/go-metrics"
"github.com/hashicorp/vault/helper/metricsutil"
"github.com/hashicorp/vault/helper/namespace"
"github.com/hashicorp/vault/limits"
"github.com/hashicorp/vault/physical/raft"
"github.com/hashicorp/vault/sdk/helper/consts"
"github.com/hashicorp/vault/sdk/logical"
@@ -79,6 +80,20 @@ func (c *Core) metricsLoop(stopCh chan struct{}) {
c.metricSink.SetGaugeWithLabels([]string{"core", "replication", "write_undo_logs"}, 0, nil)
}
writeLimiter := c.GetRequestLimiter(limits.WriteLimiter)
if writeLimiter != nil {
c.metricSink.SetGaugeWithLabels([]string{
"core", "limits", "concurrency", limits.WriteLimiter,
}, float32(writeLimiter.EstimatedLimit()), nil)
}
pathLimiter := c.GetRequestLimiter(limits.SpecialPathLimiter)
if pathLimiter != nil {
c.metricSink.SetGaugeWithLabels([]string{
"core", "limits", "concurrency", limits.SpecialPathLimiter,
}, float32(pathLimiter.EstimatedLimit()), nil)
}
// Refresh the standby gauge, on all nodes
if haState != consts.Active {
c.metricSink.SetGaugeWithLabels([]string{"core", "active"}, 0, nil)

105
vault/router.go
View File

@@ -68,6 +68,7 @@ type routeEntry struct {
rootPaths atomic.Value
loginPaths atomic.Value
binaryPaths atomic.Value
limitedPaths atomic.Value
l sync.RWMutex
}
@@ -78,12 +79,16 @@ type wildcardPath struct {
isPrefix bool
}
// loginPathsEntry is used to hold the routeEntry loginPaths
type loginPathsEntry struct {
// specialPathsEntry is used to hold the routeEntry specialPaths
type specialPathsEntry struct {
paths *radix.Tree
wildcardPaths []wildcardPath
}
// specialPathsLookupFunc is used by (*Router).specialPath to look up a
// specialPathsEntry corresponding to loginPath, binaryPath, or limitedPath.
type specialPathsLookupFunc func(re *routeEntry) *specialPathsEntry
type ValidateMountResponse struct {
MountType string `json:"mount_type" structs:"mount_type" mapstructure:"mount_type"`
MountAccessor string `json:"mount_accessor" structs:"mount_accessor" mapstructure:"mount_accessor"`
@@ -204,12 +209,19 @@ func (r *Router) Mount(backend logical.Backend, prefix string, mountEntry *Mount
return err
}
re.loginPaths.Store(loginPathsEntry)
binaryPathsEntry, err := parseUnauthenticatedPaths(paths.Binary)
if err != nil {
return err
}
re.binaryPaths.Store(binaryPathsEntry)
limitedPathsEntry, err := parseUnauthenticatedPaths(paths.Limited)
if err != nil {
return err
}
re.limitedPaths.Store(limitedPathsEntry)
switch {
case prefix == "":
return fmt.Errorf("missing prefix to be used for router entry; mount_path: %q, mount_type: %q", re.mountEntry.Path, re.mountEntry.Type)
@@ -886,11 +898,37 @@ func (r *Router) RootPath(ctx context.Context, path string) bool {
}
// LoginPath checks if the given path is used for logins
func (r *Router) LoginPath(ctx context.Context, path string) bool {
return r.specialPath(ctx, path,
func(re *routeEntry) *specialPathsEntry {
return re.loginPaths.Load().(*specialPathsEntry)
})
}
// BinaryPath checks if the given path is used for binary requests
func (r *Router) BinaryPath(ctx context.Context, path string) bool {
return r.specialPath(ctx, path,
func(re *routeEntry) *specialPathsEntry {
return re.binaryPaths.Load().(*specialPathsEntry)
})
}
// LimitedPath checks if the given path uses limited requests
func (r *Router) LimitedPath(ctx context.Context, path string) bool {
return r.specialPath(ctx, path,
func(re *routeEntry) *specialPathsEntry {
return re.limitedPaths.Load().(*specialPathsEntry)
})
}
// specialPath is a common method for checking if the given path has a matching
// PathsSpecial entry. This is used for Login, Binary, and Limited PathsSpecial
// fields.
// Matching Priority
// 1. prefix
// 2. exact
// 3. wildcard
func (r *Router) LoginPath(ctx context.Context, path string) bool {
func (r *Router) specialPath(ctx context.Context, path string, lookup specialPathsLookupFunc) bool {
ns, err := namespace.FromContext(ctx)
if err != nil {
return false
@@ -909,8 +947,8 @@ func (r *Router) LoginPath(ctx context.Context, path string) bool {
// Trim to get remaining path
remain := strings.TrimPrefix(adjustedPath, mount)
// Check the loginPaths of this backend
pe := re.loginPaths.Load().(*loginPathsEntry)
// Check the specialPath of this backend as specified by the caller.
pe := lookup(re)
match, raw, ok := pe.paths.LongestPrefix(remain)
if !ok && len(pe.wildcardPaths) == 0 {
// no match found
@@ -939,57 +977,6 @@ func (r *Router) LoginPath(ctx context.Context, path string) bool {
return false
}
// BinaryPath checks if the given path uses binary requests
func (r *Router) BinaryPath(ctx context.Context, path string) bool {
ns, err := namespace.FromContext(ctx)
if err != nil {
return false
}
adjustedPath := ns.Path + path
r.l.RLock()
mount, raw, ok := r.root.LongestPrefix(adjustedPath)
r.l.RUnlock()
if !ok {
return false
}
re := raw.(*routeEntry)
// Trim to get remaining path
remain := strings.TrimPrefix(adjustedPath, mount)
// Check the binaryPaths of this backend
// Check the loginPaths of this backend
pe := re.binaryPaths.Load().(*loginPathsEntry)
match, raw, ok := pe.paths.LongestPrefix(remain)
if !ok && len(pe.wildcardPaths) == 0 {
// no match found
return false
}
if ok {
prefixMatch := raw.(bool)
// Handle the prefix match case
if prefixMatch && strings.HasPrefix(remain, match) {
return true
}
if match == remain {
// Handle the exact match case
return true
}
}
// check Login Paths containing wildcards
reqPathParts := strings.Split(remain, "/")
for _, w := range pe.wildcardPaths {
if pathMatchesWildcardPath(reqPathParts, w.segments, w.isPrefix) {
return true
}
}
return false
}
// pathMatchesWildcardPath returns true if the path made up of the path slice
// matches the given wildcard path slice
func pathMatchesWildcardPath(path, wcPath []string, isPrefix bool) bool {
@@ -1038,8 +1025,8 @@ func isValidUnauthenticatedPath(path string) (bool, error) {
}
// parseUnauthenticatedPaths converts a list of special paths to a
// loginPathsEntry
func parseUnauthenticatedPaths(paths []string) (*loginPathsEntry, error) {
// specialPathsEntry
func parseUnauthenticatedPaths(paths []string) (*specialPathsEntry, error) {
var tempPaths []string
tempWildcardPaths := make([]wildcardPath, 0)
for _, path := range paths {
@@ -1065,7 +1052,7 @@ func parseUnauthenticatedPaths(paths []string) (*loginPathsEntry, error) {
}
}
return &loginPathsEntry{
return &specialPathsEntry{
paths: pathsToRadix(tempPaths),
wildcardPaths: tempWildcardPaths,
}, nil

4
vault/router_access.go
View File

@@ -23,3 +23,7 @@ func (r *RouterAccess) StoragePrefixByAPIPath(ctx context.Context, path string)
func (r *RouterAccess) IsBinaryPath(ctx context.Context, path string) bool {
return r.c.router.BinaryPath(ctx, path)
}
func (r *RouterAccess) IsLimitedPath(ctx context.Context, path string) bool {
return r.c.router.LimitedPath(ctx, path)
}

2
vault/router_test.go
View File

@@ -581,7 +581,7 @@ func TestParseUnauthenticatedPaths(t *testing.T) {
{segments: []string{"+", "begin", ""}, isPrefix: true},
{segments: []string{"middle", "+", "bar"}, isPrefix: true},
}
expected := &loginPathsEntry{
expected := &specialPathsEntry{
paths: pathsToRadix(paths),
wildcardPaths: wildcardPathsEntry,
}

13
vault/testing.go
View File

@@ -49,6 +49,7 @@ import (
"github.com/hashicorp/vault/helper/testhelpers/corehelpers"
"github.com/hashicorp/vault/helper/testhelpers/pluginhelpers"
"github.com/hashicorp/vault/internalshared/configutil"
"github.com/hashicorp/vault/limits"
"github.com/hashicorp/vault/sdk/framework"
"github.com/hashicorp/vault/sdk/helper/consts"
"github.com/hashicorp/vault/sdk/helper/logging"
@@ -1130,6 +1131,8 @@ type TestClusterOptions struct {
// ABCDLoggerNames names the loggers according to our ABCD convention when generating 4 clusters
ABCDLoggerNames bool
LimiterRegistry *limits.LimiterRegistry
}
type TestPluginConfig struct {
@@ -1420,6 +1423,7 @@ func NewTestCluster(t testing.T, base *CoreConfig, opts *TestClusterOptions) *Te
EnableUI: true,
EnableRaw: true,
BuiltinRegistry: corehelpers.NewMockBuiltinRegistry(),
LimiterRegistry: limits.NewLimiterRegistry(testCluster.Logger),
}
if base != nil {
@@ -1507,6 +1511,11 @@ func NewTestCluster(t testing.T, base *CoreConfig, opts *TestClusterOptions) *Te
coreConfig.ExpirationRevokeRetryBase = base.ExpirationRevokeRetryBase
coreConfig.PeriodicLeaderRefreshInterval = base.PeriodicLeaderRefreshInterval
coreConfig.ClusterAddrBridge = base.ClusterAddrBridge
if base.LimiterRegistry != nil {
coreConfig.LimiterRegistry = base.LimiterRegistry
}
testApplyEntBaseConfig(coreConfig, base)
}
if coreConfig.ClusterName == "" {
@@ -1900,6 +1909,10 @@ func (testCluster *TestCluster) newCore(t testing.T, idx int, coreConfig *CoreCo
localConfig.NumExpirationWorkers = numExpirationWorkersTest
if opts != nil && opts.LimiterRegistry != nil {
localConfig.LimiterRegistry = opts.LimiterRegistry
}
c, err := NewCore(&localConfig)
if err != nil {
t.Fatalf("err: %v", err)