bulwark

Documentation

Index

• Constants
• Variables
• func DefaultRejectedErrorFunc(err error) bool
• func RejectedError(err error) error
• func Throttle[T any](ctx context.Context, at *AdaptiveThrottle, defaultPriority Priority, ...) (res T, err error)
• func WithPriority(ctx context.Context, priority Priority) context.Context
• type AdaptiveThrottle
  • func NewAdaptiveThrottle(priorities int, options ...AdaptiveThrottleOption) *AdaptiveThrottle
  • func (t *AdaptiveThrottle) Throttle(ctx context.Context, defaultPriority Priority, fn throttledFn, ...) error
• type AdaptiveThrottleOption
  • func WithAdaptiveThrottleMinimumRate(x float64) AdaptiveThrottleOption
  • func WithAdaptiveThrottleRatio(k float64) AdaptiveThrottleOption
  • func WithAdaptiveThrottleWindow(d time.Duration) AdaptiveThrottleOption
  • func WithClientSideRejectionError(err error) AdaptiveThrottleOption
  • func WithNow(fn func() time.Time) AdaptiveThrottleOption
  • func WithPriorityValidator(fn func(p Priority, priorities int) (Priority, error)) AdaptiveThrottleOption
  • func WithRandomSource(src rand.Source) AdaptiveThrottleOption
  • func WithRejectedErrorFunc(fn func(error) bool) AdaptiveThrottleOption
• type Priority
  • func AssertValidPriority(p Priority, priorities int) (Priority, error)
  • func ClampInvalidPriority(p Priority, priorities int) (Priority, error)
  • func PriorityFromContext(ctx context.Context, defaultPriority Priority) Priority
  • func RejectInvalidPriority(p Priority, priorities int) (Priority, error)

Constants

const (
	// K is the default accept multiplier, which is used to determine the number
	// of requests that are allowed to reach the backend.
	//
	// A value of 2 means that the throttle will allow twice as many requests to
	// actually reach the backend as it believes will succeed.
	K = 2
	// MinRPS is the minimum number of requests per second that the adaptive
	// throttle will allow (approximately) through to the upstream, even if every
	// request is failing.
	MinRPS = 1
)

const StandardPriorities = 4

StandardPriorities is the number of priority levels that are available. This value should be used when creating a new AdaptiveThrottle when the default Priority constants are used.

throttler := bulwark.NewAdaptiveThrottle(bulwark.StandardPriorities)
_, err := bulwark.Throttle(ctx, throttler, bulwark.High, throttledFn)
if err != nil {
	// handle the error
}

Variables

var ErrClientSideRejection = errors.New("bulwark: client-side rejection")

ErrClientSideRejection is the default error returned when the throttle rejects a request on the client side without forwarding it to the backend. Use WithClientSideRejectionError to override this per instance.

Functions

func DefaultRejectedErrorFunc

func DefaultRejectedErrorFunc(err error) bool

DefaultRejectedErrorFunc is the default function used to classify errors as rejections. It treats Unavailable and ResourceExhausted errors as rejections. Use WithRejectedErrorFunc to override this per instance.

func RejectedError

func RejectedError(err error) error

RejectedError wraps an error to indicate that the error should be considered for the throttling.

Any error that indicates that the backend is unhealthy should be wrapped with `RejectedError`. But other errors, such as bad requests, authentication failures, pre-condition failures, etc., should not be wrapped with `RejectedError`.

A nil error is returned unchanged: wrapping "no error" as a rejection is meaningless. This also keeps errRejected's invariant that inner is never nil.

func Throttle

func Throttle[T any](
	ctx context.Context,
	at *AdaptiveThrottle,
	defaultPriority Priority,
	throttledFn throttledArgsFn[T],
	fallbackFn ...fallbackArgsFn[T],
) (res T, err error)

Throttle executes throttledFn through the given AdaptiveThrottle and returns the result. It is the generic counterpart to AdaptiveThrottle.Throttle for functions that return a value.

The default priority is used when the given `ctx` does not have a priority set. The `ctx` can set the priority using `WithPriority`.

func WithPriority

func WithPriority(ctx context.Context, priority Priority) context.Context

WithPriority attaches the given `Priority` to the context. It is good practice to call the adaptive throttle this way:

`bulwark.WithAdaptiveThrottle(at, bulwark.PriorityFromContext(ctx, priority), f)`

Then requests should have a priority attached to them, so all throttles can adapt their behaviour accordingly.

Types

type AdaptiveThrottle

type AdaptiveThrottle struct {
	// contains filtered or unexported fields
}

AdaptiveThrottle is used in a client to throttle requests to a backend as it becomes unhealthy to help it recover from overload more quickly. Because backends must expend resources to reject requests over their capacity it is vital for clients to ease off on sending load when they are in trouble, lest the backend spend all of its resources on rejecting requests and have none left over to actually serve any.

The adaptive throttle works by tracking the success rate of requests over some time interval (usually a minute or so), and randomly rejecting requests without sending them to avoid sending too much more than the rate that are expected to actually be successful. Some slop is included, because even if the backend is serving zero requests successfully, we do need to occasionally send it requests to learn when it becomes healthy again.

More on adaptive throttles in https://sre.google/sre-book/handling-overload/

func NewAdaptiveThrottle

func NewAdaptiveThrottle(priorities int, options ...AdaptiveThrottleOption) *AdaptiveThrottle

NewAdaptiveThrottle returns an AdaptiveThrottle.

priorities is the number of priorities that the throttle will accept. Giving a priority outside of `[0, priorities)` will panic.

func (*AdaptiveThrottle) Throttle

func (t *AdaptiveThrottle) Throttle(
	ctx context.Context, defaultPriority Priority, fn throttledFn, fallbackFn ...fallbackFn,
) error

Throttle sends a request to the backend when the adaptive throttle allows it. The request is throttled based on the priority of the request.

The default priority is used when the given `ctx` does not have a priority set. The `ctx` can set the priority using `WithPriority`.

When `throttledFn` returns an error, the error is considered as a rejection when `WithRejectedErrorFunc` returns true or when the error is wrapped in a `RejectedError`.

If there are enough rejections within a given time window, further calls to `Throttle` may begin returning `ErrClientSideRejection` immediately without invoking `throttledFn`. Lower-priority requests are preferred to be rejected first.

type AdaptiveThrottleOption

type AdaptiveThrottleOption struct {
	// contains filtered or unexported fields
}

Additional options for the AdaptiveThrottle type. These options do not frequently need to be tuned as the defaults work in a majority of cases.

func WithAdaptiveThrottleMinimumRate

func WithAdaptiveThrottleMinimumRate(x float64) AdaptiveThrottleOption

WithAdaptiveThrottleMinimumRate sets the minimum number of requests per second that the adaptive throttle will allow (approximately) through to the upstream, even if every request is failing. This is important because this is how the adaptive throttle 'learns' when the upstream becomes healthy again.

func WithAdaptiveThrottleRatio

func WithAdaptiveThrottleRatio(k float64) AdaptiveThrottleOption

WithAdaptiveThrottleRatio sets the ratio of the measured success rate and the rate that the throttle will admit. For example, when k is 2 the throttle will allow twice as many requests to actually reach the backend as it believes will succeed. Higher values of k mean that the throttle will react more slowly when a backend becomes unhealthy, but react more quickly when it becomes healthy again, and will allow more load to an unhealthy backend. k=2 is usually a good place to start, but backends that serve "cheap" requests (e.g. in-memory caches) may need a lower value.

func WithAdaptiveThrottleWindow

func WithAdaptiveThrottleWindow(d time.Duration) AdaptiveThrottleOption

WithAdaptiveThrottleWindow sets the time window over which the throttle remembers requests for use in figuring out the success rate.

func WithClientSideRejectionError

func WithClientSideRejectionError(err error) AdaptiveThrottleOption

WithClientSideRejectionError sets the per-instance error returned when the throttle rejects a request on the client side without forwarding it to the backend. Defaults to ErrClientSideRejection.

func WithNow

func WithNow(fn func() time.Time) AdaptiveThrottleOption

WithNow sets the per-instance time source. This is primarily useful in tests to control the clock without affecting other AdaptiveThrottle instances.

func WithPriorityValidator

func WithPriorityValidator(fn func(p Priority, priorities int) (Priority, error)) AdaptiveThrottleOption

WithPriorityValidator sets the function that validates input priority values.

The function should return the validated priority value. If the priority is invalid, the function should return an error.

func WithRandomSource

func WithRandomSource(src rand.Source) AdaptiveThrottleOption

WithRandomSource sets the per-instance random source used to sample the rejection probability. This is primarily useful in tests to produce deterministic behaviour: a source that always returns 0 will shed every request whose rejection probability is greater than zero, and a source that always returns math.MaxUint64 will never shed.

The provided source must be safe for concurrent use if the AdaptiveThrottle is used concurrently. rand.NewPCG and rand.NewChaCha8 are not concurrent-safe by default.

func WithRejectedErrorFunc

func WithRejectedErrorFunc(fn func(error) bool) AdaptiveThrottleOption

WithRejectedErrorFunc sets the per-instance function that determines whether an error returned by the throttled function should be counted as a rejection. Defaults to DefaultRejectedErrorFunc.

type Priority

type Priority int8

Priority determines the importance of a request in ascending order. e.g. priority 0 is more important than priority 1.

When a system reaches its capacity, it will sort requests by their priority and process them. Lower-priority requests can either be delayed or dropped.

const (
	// Use High when for requests that are critical to the overall experience.
	High Priority = 0
	// Use Important for requests that are important, but not critical.
	Important Priority = 1
	// Use Medium for noncritical requests where an elevated latency or
	// failure rate would not significantly impact the experience.
	Medium Priority = 2
	// Use Low for trivial requests and good for any system that can retry
	// later when the system has spare capacity.
	Low Priority = 3
)

These are pre-defined priority levels that can be used, but any int value can be used as a priority.

func AssertValidPriority

func AssertValidPriority(p Priority, priorities int) (Priority, error)

AssertValidPriority panics when a priority is out of range. A priority is out of range when it is less than 0 or greater than or equal to priorities.

func ClampInvalidPriority

func ClampInvalidPriority(p Priority, priorities int) (Priority, error)

ClampInvalidPriority clamps any out-of-range priority to the lowest valid priority (priorities-1). This applies to both negative values and values that exceed the configured number of priorities, preventing invalid or malicious input from being promoted to a higher-importance tier.

func PriorityFromContext

func PriorityFromContext(ctx context.Context, defaultPriority Priority) Priority

PriorityFromContext returns the `Priority` attached to the context. If no priority is attached, it returns the default priority.

It is good practive to attach a global priority to requests, so all throttles can adapt their behaviour accordingly.

func RejectInvalidPriority

func RejectInvalidPriority(p Priority, priorities int) (Priority, error)

RejectInvalidPriority returns an error when a priority is out of range. A priority is out of range when it is less than 0 or greater than or equal to priorities.