gohalt

README

Gohalt 👮‍♀🛑: Fast; Simple; Powerful; Go Throttler library

go get -u github.com/1pkg/gohalt

Introduction

Gohalt is simple and convenient yet powerful and efficient throttling go library. Gohalt provides various throttlers and surronding tools to build throttling pipelines and rate limiters of any complexity adjusted to your specific needs. Gohalt provides an easy way to integrate throttling and rate limiting with your infrastructure through built in middlewares.

Concepts

Gohalt uses Throttler as the core interface for all derived throttlers and surronding tools.

// Throttler defines core gohalt throttler abstraction and exposes pair of counterpart methods: `Acquire` and `Release`.
type Throttler interface {
	// Acquire takes a part of throttling quota or returns error if throttling quota is drained
	// it needs to be called right before shared resource acquire.
	Acquire(context.Context) error
	// Release puts a part of throttling quota back or returns error if this is not possible
	// it needs to be called just after shared resource release.
	Release(context.Context) error
}

Throttler interface exposes pair of counterpart methods: Acquire takes a part of throttling quota or returns error if throttling quota is drained and needs to be called right before shared resource acquire; Release puts a part of throttling quota back or returns error if this is not possible and needs to be called just after shared resource release; Note: all derived throttler implementations are thread safe, so they could be used concurrently without additional locking. Note: all acquired throttlers should be released exatly the same amount of times they have been acquired. Note: despite throttler Release method has the same signature as Acquire has, Release implementations should try to handle any internal error gracefully and return error back rarely, nevertheless all errors returned by Release should be handeled by client.

In Gohalt throtllers could be easily combined with each other to build complex pipelines. There are multiple composite throttlers (all, any, ring, pattern, not, generator, etc) as well as leaf throttlers (timed, latency, monitor, metric, percentile, etc) to work with in Gohalt. If you don't find in existing throttlers the one that fits your needs you can create custom throttler by implementing Throttler interface. Such custom throttler should work with existing Gohalt throttlers and tools out of box.

Gohalt includes multiple supporting surrounding tools to make throttling more sugary.

// Runnable defined by typical abstract async func signature.
// Runnable is used by `Runner` as a subject for execution.
type Runnable func(context.Context) error
// Runner defines abstraction to execute a set of `Runnable`
// and return possible execution error back.
// Runner is designed to simplify work with throttlers
// by managing `Acquire`/`Release` loop.
type Runner interface {
	// Run executes single prodived `Runnable` instance.
	Run(Runnable)
	// Result returns possible execution error back.
	Result() error
}

Runnable and Runner define slim abstraction for executable and executor in Gohalt. Runner insterface aims to provide similar interface as errgroup.Group does. So to run a single executable use Run to wait and get result use Result. There are two runners implementations in Gohalt:

• sync func NewRunnerSync(ctx context.Context, thr Throttler) Runner
• async func NewRunnerAsync(ctx context.Context, thr Throttler) Runner Both implementation accept throttler and context as input arguments and handle all throttling cycle internaly. This way client donesn't need to call neither Acquire nor Release manually, all this is done by the runner. This way the only thing that needs to be done to add throttling to existing code wrap existing executable by Runnable. The only difference between sync and async runner is that the async runner starts each new Runnable inside new goroutine and uses locks for its imternal state. Note: You can't use sync runner in async fashion with go syncr.Run(func(context.Context) error{}) this will cause data race, use async runner instead async.Run(func(context.Context) error{}).

Last but not least Gohalt uses context heavily inside and there are multiple helpers to provide data via context for throttles, see throttles list to know when to use them.

// WithTimestamp adds the provided timestamp to the provided context
// to determine latency between `Acquire` and `Release`.
// Resulted context is used by: `latency` and `percentile` throtttlers.
func WithTimestamp(ctx context.Context, ts time.Time) context.Context
// WithPriority adds the provided priority to the provided context
// to differ `Acquire` priority levels.
// Resulted context is used by: `priority` throtttler.
func WithPriority(ctx context.Context, priority uint8) context.Context
// WithWeight adds the provided weight to the provided context
// to differ `Acquire` weight levels.
// Resulted context is used by: `semaphore` and `cellrate` throtttlers.
func WithWeight(ctx context.Context, weight int64) context.Context
// WithKey adds the provided key to the provided context
// to add additional call identifier to context.
// Resulted context is used by: `before`, `after`, `timed`, `adaptive`, `semaphore`, `cellrate` and `bucket` throtttlers.
func WithKey(ctx context.Context, key string) context.Context
// WithMessage adds the provided message to the provided context
// to add additional message that need to be used to context.
// Resulted context is used by: `enqueue` throtttler.
func WithMessage(ctx context.Context, message interface{}) context.Context
// WithMarshaler adds the provided marshaler to the provided context
// to add additional message marshaler that need to be used to context.
// Resulted context is used by: `enqueue` throtttler.
// Used in pair with `WithMessage`.
func WithMarshaler(ctx context.Context, mrsh Marshaler) context.Context
// WithParams facade call that respectively calls:
// - `WithTimestamp`
// - `WithPriority`
// - `WithWeight`
// - `WithKey`
// - `WithMessage`
// - `WithMarshaler`
func WithParams(ctx context.Context, ts time.Time, priority uint8, weight int64, key string, message interface{}, marshaler Marshaler) context.Context

Also there is yet another throttling sugar func WithThrottler(ctx context.Context, thr Throttler, freq time.Duration) context.Context related to context. Which defines context implementation that uses parrent context plus throttler internally. Using it you can keep typical context patterns for cancelation handling and apply and combine it with throttling.

select {
	case <-ctx.Done():
		return ctx.Error()
	default:
}

If internal context throttler is throttling context done chanel will be closed respectively. Note such behavior is implemented by throttler long pooling with the specified frequency, so efficiently there will be additional throttling user in form of long pooling goroutine.

// complex throttler example
thr := NewThrottlerAll( // throttles only if all children throttle
	NewThrottlerPattern(
		Pattern{ // use throttler only if provided key matches `192.*.*.*` submask
			Pattern: regexp.MustCompile(`192\.[0-9]+\.[0-9]+\.[0-9]+`),
			Throttler: NewThrottlerAny( // throttles if any children throttles
				// throttles only if latency is above 50 millisecond
				NewThrottlerLatency(50*time.Millisecond, 5*time.Second),
				// throttles only if cpu usage is above 70%
				NewThrottlerMonitor(NewMonitorSystem(time.Minute), Stats{CPUUsage: 0.7}),
			),
		},
	),
	// throttles each not 3rd call
	NewThrottlerNot(NewThrottlerEach(3)),
	// enqueues provided message to queue
	NewThrottlerEnqueue(NewEnqueuerRabbit("amqp://user:pass@localhost:5672/vhost", "queue", time.Minute)),
)

In gohalt v0.4.0 breaking change is introduced to replace all untyped errors with two major error types:

• ErrorThreshold which defines error type that occurs if throttler reaches specified threshold.
• ErrorInternal which defines error type that occurs if throttler internal error happens. You can find list of returning error types for all existing throttlers in throttlers table bellow or in documentation.
  Note: not every gohalt throttler must return error; some throttlers might cause different side effects like logging or call to time.Sleep instead.

Throttlers

Throttler	Definition	Description
echo	func NewThrottlerEcho(err error) Throttler	Always throttles with the specified error back. - could return any specified error;
wait	func NewThrottlerWait(duration time.Duration) Throttler	Always waits for the specified duration.
square	func NewThrottlerSquare(duration time.Duration, limit time.Duration, reset bool) Throttler	Always waits for square growing [1, 4, 9, 16, ...] multiplier on the specified initial duration, up until the specified duration limit is reached. If reset is set then after throttler riches the specified duration limit next multiplier value will be reseted.
jitter	func NewThrottlerJitter(initial time.Duration, limit time.Duration, reset bool, jitter float64) Throttler	Waits accordingly to undelying square throttler but also adds the provided jitter delta distribution on top. Jitter value is normalized to [0.0, 1.0] range and defines which part of square delay could be randomized in percents. Implementation uses secure crypto/rand as PRNG function.
context	func NewThrottlerContext() Throttler	Always throttless on done context. - could return ErrorInternal;
panic	func NewThrottlerPanic() Throttler	Always panics with ErrorInternal.
each	func NewThrottlerEach(threshold uint64) Throttler	Throttles each periodic i-th call defined by the specified threshold. - could return ErrorThreshold;
before	func NewThrottlerBefore(threshold uint64) Throttler	Throttles each call below the i-th call defined by the specified threshold. Use WithWeight to override context call qunatity, 1 by default. - could return ErrorThreshold;
after	func NewThrottlerAfter(threshold uint64) Throttler	Throttles each call after the i-th call defined by the specified threshold. Use WithWeight to override context call qunatity, 1 by default. - could return ErrorThreshold;
past	func NewThrottlerPast(threshold time.Time) Throttler	Throttles each call befor timestamp defined by the specified UTC time threshold. - could return ErrorThreshold;
future	func NewThrottlerFuture(threshold time.Time) Throttler	Throttles each call after timestamp defined by the specified UTC time threshold. - could return ErrorThreshold;
chance	func NewThrottlerChance(threshold float64) Throttler	Throttles each call with the chance p defined by the specified threshold. Chance value is normalized to [0.0, 1.0] range. Implementation uses secure crypto/rand as PRNG function. - could return ErrorThreshold;
running	func NewThrottlerRunning(threshold uint64) Throttler	Throttles each call which exeeds the running quota acquired - release q defined by the specified threshold. - could return ErrorThreshold;
buffered	func NewThrottlerBuffered(threshold uint64) Throttler	Waits on call which exeeds the running quota acquired - release q defined by the specified threshold until the running quota is available again.
priority	func NewThrottlerPriority(threshold uint64, levels uint8) Throttler	Waits on call which exeeds the running quota acquired - release q defined by the specified threshold until the running quota is available again. Running quota is not equally distributed between n levels of priority defined by the specified levels. Use func WithPriority(ctx context.Context, priority uint8) context.Context to override context call priority, 1 by default.
timed	func NewThrottlerTimed(threshold uint64, interval time.Duration, quantum time.Duration) Throttler	Throttles each call which exeeds the running quota acquired - release q defined by the specified threshold in the specified interval. Periodically each specified interval the running quota number is reseted. If quantum is set then quantum will be used instead of interval to provide the running quota delta updates. Use WithWeight to override context call qunatity, 1 by default. - could return ErrorThreshold;
latency	func NewThrottlerLatency(threshold time.Duration, retention time.Duration) Throttler	Throttles each call after the call latency l defined by the specified threshold was exeeded once. If retention is set then throttler state will be reseted after retention duration. Use func WithTimestamp(ctx context.Context, ts time.Time) context.Context to specify running duration between throttler acquire and release. - could return ErrorThreshold;
percentile	func NewThrottlerPercentile(threshold time.Duration, capacity uint8, percentile float64, retention time.Duration) Throttler	Throttles each call after the call latency l defined by the specified threshold was exeeded once considering the specified percentile. Percentile values are kept in bounded buffer with capacity c defined by the specified capacity. If retention is set then throttler state will be reseted after retention duration. Use func WithTimestamp(ctx context.Context, ts time.Time) context.Context to specify running duration between throttler acquire and release. - could return ErrorThreshold;
monitor	func NewThrottlerMonitor(mnt Monitor, threshold Stats) Throttler	Throttles call if any of the stats returned by provided monitor exceeds any of the stats defined by the specified threshold or if any internal error occurred. Builtin Monitor implementations come with stats caching by default. Use builtin NewMonitorSystem to create go system monitor instance. - could return ErrorInternal; - could return ErrorThreshold;
metric	func NewThrottlerMetric(mtc Metric) Throttler	Throttles call if boolean metric defined by the specified boolean metric is reached or if any internal error occurred. Builtin Metric implementations come with boolean metric caching by default. Use builtin NewMetricPrometheus to create Prometheus metric instance. - could return ErrorInternal; - could return ErrorThreshold;
enqueuer	func NewThrottlerEnqueue(enq Enqueuer) Throttler	Always enqueues message to the specified queue throttles only if any internal error occurred. Use func WithMessage(ctx context.Context, message interface{}) context.Context to specify context message for enqueued message and func WithMarshaler(ctx context.Context, mrsh Marshaler) context.Context to specify context message marshaler. Builtin Enqueuer implementations come with connection reuse and retries by default. Use builtin func NewEnqueuerRabbit(url string, queue string, retries uint64) Enqueuer to create RabbitMQ enqueuer instance or func NewEnqueuerKafka(net string, url string, topic string, retries uint64) Enqueuer to create Kafka enqueuer instance. - could return ErrorInternal;
adaptive	func NewThrottlerAdaptive(threshold uint64, interval time.Duration, quantum time.Duration, step uint64, thr Throttler) Throttler	Throttles each call which exeeds the running quota acquired - release q defined by the specified threshold in the specified interval. Periodically each specified interval the running quota number is reseted. If quantum is set then quantum will be used instead of interval to provide the running quota delta updates. Provided adapted throttler adjusts the running quota of adapter throttler by changing the value by d defined by the specified step, it subtracts d^2 from the running quota if adapted throttler throttles or adds d to the running quota if it doesn't. Use WithWeight to override context call qunatity, 1 by default. - could return ErrorThreshold;
pattern	func NewThrottlerPattern(patterns ...Pattern) Throttler	Throttles if matching throttler from provided patterns throttles. Use func WithKey(ctx context.Context, key string) context.Context to specify key for regexp pattern throttler matching. Pattern defines a pair of regexp and related throttler. - could return ErrorInternal; - could return any underlying throttler error;
ring	func NewThrottlerRing(thrs ...Throttler) Throttler	Throttles if the i-th call throttler from provided list throttle. - could return ErrorInternal; - could return any underlying throttler error;
all	func NewThrottlerAll(thrs ...Throttler) Throttler	Throttles call if all provided throttlers throttle. - could return ErrorInternal;
any	func NewThrottlerAny(thrs ...Throttler) Throttler	Throttles call if any of provided throttlers throttle. - could return ErrorInternal;
not	func NewThrottlerNot(thr Throttler) Throttler	Throttles call if provided throttler doesn't throttle. - could return ErrorInternal;
suppress	func NewThrottlerSuppress(thr Throttler) Throttler	Suppresses provided throttler to never throttle.
retry	func NewThrottlerRetry(thr Throttler, retries uint64) Throttler	Retries provided throttler error up until the provided retries threshold. If provided onthreshold flag is set even ErrorThreshold errors will be retried. Internally retry uses square throttler with DefaultRetriedDuration initial duration. - could return any underlying throttler error;
cache	func NewThrottlerCache(thr Throttler, cache time.Duration) Throttler	Caches provided throttler calls for the provided cache duration, throttler release resulting resets cache. Only non throttling calls are cached for the provided cache duration. - could return any underlying throttler error;
generator	func NewThrottlerGenerator(gen Generator, capacity uint64, eviction float64) Throttler	Creates new throttler instance that throttles if found key matching throttler throttles. If no key matching throttler has been found generator used insted to provide new throttler that will be added to existing throttlers map. Generated throttlers are kept in bounded map with capacity c defined by the specified capacity and eviction rate e defined by specified eviction value is normalized to [0.0, 1.0], where eviction rate affects number of throttlers that will be removed from the map after bounds overflow. Use WithKey to specify key for throttler matching and generation. - could return ErrorInternal; - could return any underlying throttler error;
semaphore	func NewThrottlerSemaphore(weight int64) Throttler	Creates new throttler instance that throttles call if underlying semaphore throttles. Use WithWeight to override context call weight, 1 by default. - could return ErrorThreshold;
cellrate	func NewThrottlerCellRate(threshold uint64, interval time.Duration, monotone bool) Throttler	Creates new throttler instance that uses generic cell rate algorithm to throttles call within provided interval and threshold. If provided monotone flag is set class to release will have no effect on throttler. Use WithWeight to override context call qunatity, 1 by default. - could return ErrorThreshold;
bucket	func NewThrottlerBucket(threshold uint64, interval time.Duration, monotone bool) Throttler	Creates new throttler instance that leaky bucket algorithm to throttles call within provided interval and threshold. If provided monotone flag is set class to release will have no effect on throttler. Use WithWeight to override context call qunatity, 1 by default. - could return ErrorThreshold;

Integrations

Note: in gohalt v0.3.0 all integrations were moved to separate repository to make base gohalt repository dependencies footprint small.

go get -u github.com/1pkg/gohaltlib

Library	Adapter
gin	func NewMiddlewareGin(thr Throttler, with GinWith, on GinOn) gin.HandlerFunc
stdlib http handler	func NewMiddlewareStd(h http.Handler, thr Throttler, with StdWith, on StdOn) http.Handler
echo	func NewMiddlewareEcho(thr Throttler, with EchoWith, on EchoOn) echo.MiddlewareFunc
beego	func NewMiddlewareBeego(thr Throttler, with BeegoWith, on BeegoOn) beego.FilterFunc
kit	func NewMiddlewareKit(thr Throttler, with KitWith, on KitOn) endpoint.Middleware
mux	func NewMiddlewareMux(h http.Handler, thr Throttler, with MuxWith, on MuxOn) http.Handler
httprouter	func NewMiddlewareRouter(h http.Handler, thr Throttler, with RouterWith, on RouterOn) http.Handler
reveal	func NewMiddlewareRevel(thr Throttler, with RevealWith, on RevealOn) revel.Filter
iris	func NewMiddlewareIris(thr Throttler, with IrisWith, on IrisOn) iris.Handler
fasthttp	func NewMiddlewareFast(h fasthttp.RequestHandler, thr Throttler, with FastWith, on FastOn) fasthttp.RequestHandler
stdlib rt	func NewRoundTripperStd(rt http.RoundTripper, thr Throttler, with RoundTripperStdWith, on RoundTripperStdOn) http.RoundTripper
fasthttp rt	func NewRoundTripperFast(rt RoundTripperFast, thr Throttler, with RoundTripperFastWith, on RoundTripperFastOn) RoundTripperFast
stdlib rpc client coded	func NewRPCClientCodec(cc rpc.ClientCodec, thr Throttler, with RPCCodecWith, on RPCCodecOn) rpc.ClientCodec
stdlib rpc server coded	func NewRPCServerCodec(sc rpc.ServerCodec, thr Throttler, with RPCCodecWith, on RPCCodecOn) rpc.ServerCodec
grpc client stream	func NewGRPCClientStream(cs grpc.ClientStream, thr Throttler, with GRPCStreamWith, on GRPCStreamOn) grpc.ClientStream
grpc server stream	func NewGrpServerStream(ss grpc.ServerStream, thr Throttler, with GRPCStreamWith, on GRPCStreamOn) grpc.ServerStream
go-micro client	func NewMicroClient(thr Throttler, with MicroClientWith, on MicroOn) client.Wrapper
go-micro server	func NewMicroHandler(thr Throttler, with MicroServerWith, on MicroOn) server.HandlerWrapper
stdlib net conn	func NewNetConn(conn net.Conn, thr Throttler, with NetConnWith, on NetConnOn, mode NetConnMode) net.Conn
stdlib sql	func NewSQLClient(cli SQLClient, thr Throttler, with SQLClientWith, on SQLClientOn) SQLClient
stdlib io reader	func NewReader(r io.Reader, thr Throttler, with RWWith, on RWOn) io.Reader
stdlib io writer	func NewWriter(w io.Writer, thr Throttler, with RWWith, on RWOn) io.Writer

Licence

Gohalt is licensed under the MIT License.
See LICENSE for the full license text.

Documentation

Index

• Variables
• func WithKey(ctx context.Context, key string) context.Context
• func WithMarshaler(ctx context.Context, mrsh Marshaler) context.Context
• func WithMessage(ctx context.Context, message interface{}) context.Context
• func WithParams(ctx context.Context, ts time.Time, priority uint8, weight int64, key string, ...) context.Context
• func WithPriority(ctx context.Context, priority uint8) context.Context
• func WithThrottler(ctx context.Context, thr Throttler, freq time.Duration) context.Context
• func WithTimestamp(ctx context.Context, ts time.Time) context.Context
• func WithWeight(ctx context.Context, weight int64) context.Context
• type Enqueuer
  • func NewEnqueuerKafka(net string, url string, topic string, retries uint64) Enqueuer
  • func NewEnqueuerRabbit(url string, queue string, retries uint64) Enqueuer
• type ErrorInternal
  • func (err ErrorInternal) Error() string
• type ErrorThreshold
  • func (err ErrorThreshold) Error() string
• type Generator
• type Logger
• type Marshaler
• type Metric
  • func NewMetricPrometheus(url string, query string, cache time.Duration) Metric
• type Monitor
  • func NewMonitorSystem(cache time.Duration, tp time.Duration) Monitor
• type Pattern
• type Runnable
• type Runner
  • func NewRunnerAsync(ctx context.Context, thr Throttler) Runner
  • func NewRunnerSync(ctx context.Context, thr Throttler) Runner
• type Stats
  • func (s Stats) Compare(stats Stats) bool
• type Throttler
  • func NewThrottlerAdaptive(threshold uint64, interval time.Duration, quantum time.Duration, step uint64, ...) Throttler
  • func NewThrottlerAfter(threshold uint64) Throttler
  • func NewThrottlerAll(thrs ...Throttler) Throttler
  • func NewThrottlerAny(thrs ...Throttler) Throttler
  • func NewThrottlerBefore(threshold uint64) Throttler
  • func NewThrottlerBucket(threshold uint64, interval time.Duration, monotone bool) Throttler
  • func NewThrottlerBuffered(threshold uint64) Throttler
  • func NewThrottlerCache(thr Throttler, cache time.Duration) Throttler
  • func NewThrottlerCellRate(threshold uint64, interval time.Duration, monotone bool) Throttler
  • func NewThrottlerChance(threshold float64) Throttler
  • func NewThrottlerContext() Throttler
  • func NewThrottlerEach(threshold uint64) Throttler
  • func NewThrottlerEcho(err error) Throttler
  • func NewThrottlerEnqueue(enq Enqueuer) Throttler
  • func NewThrottlerFuture(threshold time.Time) Throttler
  • func NewThrottlerGenerator(gen Generator, capacity uint64, eviction float64) Throttler
  • func NewThrottlerJitter(initial time.Duration, limit time.Duration, reset bool, jitter float64) Throttler
  • func NewThrottlerLatency(threshold time.Duration, retention time.Duration) Throttler
  • func NewThrottlerMetric(mtc Metric) Throttler
  • func NewThrottlerMonitor(mnt Monitor, threshold Stats) Throttler
  • func NewThrottlerNot(thr Throttler) Throttler
  • func NewThrottlerPanic() Throttler
  • func NewThrottlerPast(threshold time.Time) Throttler
  • func NewThrottlerPattern(patterns ...Pattern) Throttler
  • func NewThrottlerPercentile(threshold time.Duration, capacity uint8, percentile float64, ...) Throttler
  • func NewThrottlerPriority(threshold uint64, levels uint8) Throttler
  • func NewThrottlerRetry(thr Throttler, retries uint64, onthreshold bool) Throttler
  • func NewThrottlerRing(thrs ...Throttler) Throttler
  • func NewThrottlerRunning(threshold uint64) Throttler
  • func NewThrottlerSemaphore(weight int64) Throttler
  • func NewThrottlerSquare(initial time.Duration, limit time.Duration, reset bool) Throttler
  • func NewThrottlerSuppress(thr Throttler) Throttler
  • func NewThrottlerTimed(threshold uint64, interval time.Duration, quantum time.Duration) Throttler
  • func NewThrottlerWait(duration time.Duration) Throttler

Variables

var DefaultRetriedDuration = 100 * time.Millisecond

DefaultRetriedDuration defines default initial duration for `square` throttler used by default. By default DefaultRetriedDuration is set to use `100 * time.Millisecond`.

Functions

func WithKey

func WithKey(ctx context.Context, key string) context.Context

WithKey adds the provided key to the provided context to add additional call identifier to context. Resulted context is used by: `pattern` and `generator` throtttlers.

func WithMarshaler

func WithMarshaler(ctx context.Context, mrsh Marshaler) context.Context

WithMarshaler adds the provided marshaler to the provided context to add additional message marshaler that need to be used to context. Resulted context is used by: `enqueue` throtttler. Used in pair with `WithMessage`.

func WithMessage

func WithMessage(ctx context.Context, message interface{}) context.Context

WithMessage adds the provided message to the provided context to add additional message that need to be used to context. Resulted context is used by: `enqueue` throtttler. Used in pair with `WithMarshaler`.

func WithParams

func WithParams(
	ctx context.Context,
	ts time.Time,
	priority uint8,
	weight int64,
	key string,
	message interface{},
	marshaler Marshaler,
) context.Context

WithParams facade call that respectively calls: - `WithTimestamp` - `WithPriority` - `WithWeight` - `WithKey` - `WithMessage` - `WithMarshaler`

func WithPriority

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

WithPriority adds the provided priority to the provided context to differ `Acquire` priority levels. Resulted context is used by: `priority` throtttler.

func WithThrottler

func WithThrottler(ctx context.Context, thr Throttler, freq time.Duration) context.Context

WithThrottler adds the provided thr to the provided context and defines context implementation that uses parrent context plus throttler internally that closes context done chanel if internal throttler throttles.

func WithTimestamp

func WithTimestamp(ctx context.Context, ts time.Time) context.Context

WithTimestamp adds the provided timestamp to the provided context to determine latency between `Acquire` and `Release`. Resulted context is used by: `latency` and `percentile` throtttlers.

func WithWeight

func WithWeight(ctx context.Context, weight int64) context.Context

WithWeight adds the provided weight to the provided context to differ `Acquire` weight levels. Resulted context is used by: `before`, `after`, `timed`, `adaptive`, `semaphore`, `cellrate` and `bucket` throtttlers.

Types

type Enqueuer

type Enqueuer interface {
	// Enqueue enqueues provided message or returns internal error if any happened.
	Enqueue(context.Context, []byte) error
}

Enqueuer defines abstract message enqueuing interface.

func NewEnqueuerKafka

func NewEnqueuerKafka(net string, url string, topic string, retries uint64) Enqueuer

NewEnqueuerKafka creates Kafka enqueuer instance with cached connection and failure retries which enqueues provided message to the specified topic. New unique message key `gohalt_enqueue_{{uuid}}` is created for each new message. Only successful connections are cached.

func NewEnqueuerRabbit

func NewEnqueuerRabbit(url string, queue string, retries uint64) Enqueuer

NewEnqueuerRabbit creates RabbitMQ enqueuer instance with cached connection and failure retries which enqueues provided message to the specified queue. New unique exchange `gohalt_exchange_{{uuid}}` is created for each new enqueuer, new unique message id `gohalt_enqueue_{{uuid}}` is created for each new message. Only successful connections are cached.

type ErrorInternal

type ErrorInternal struct {
	Throttler string
	Message   string
}

ErrorInternal defines error type that occurs if throttler internal error happens.

func (ErrorInternal) Error

func (err ErrorInternal) Error() string

type ErrorThreshold

type ErrorThreshold struct {
	Throttler string
	Threshold fmt.Stringer
}

ErrorThreshold defines error type that occurs if throttler reaches specified threshold.

func (ErrorThreshold) Error

func (err ErrorThreshold) Error() string

type Generator

type Generator func(string) (Throttler, error)

Generator defines func signature that is able to generate new throttlers by provided key.

type Logger

type Logger func(string, ...interface{})

Logger defined by typical logger func signature.

var DefaultLogger Logger = stdlog.Printf

DefaultLogger defines default logger value used for logging. By default DefaultLogger is set to use `log.Printf`. Loggign can be completely disabled by setting DefaultLogger to nil.

type Marshaler

type Marshaler func(interface{}) ([]byte, error)

Marshaler defined by typical marshaler func signature.

var DefaultMarshaler Marshaler = json.Marshal

DefaultMarshaler defines default marshaler value used by `WithMarshaler`. By default DefaultMarshaler is set to use `json.Marshal`.

type Metric

type Metric interface {
	// Query returns the result of the query or internal error if any happened.
	Query(context.Context) (bool, error)
}

Metric defines single metric querier interface that returns the metric query result.

func NewMetricPrometheus

func NewMetricPrometheus(url string, query string, cache time.Duration) Metric

NewMetricPrometheus creates prometheus metric querier instance with cache interval defined by the provided duration which executes provided prometheus boolean metric query and cache it. Only successful metric results are cached.

type Monitor

type Monitor interface {
	// Stats returns system stats or internal error if any happened.
	Stats(context.Context) (Stats, error)
}

Monitor defines system monitor interface that returns the system stats.

func NewMonitorSystem

func NewMonitorSystem(cache time.Duration, tp time.Duration) Monitor

NewMonitorSystem creates system monitor instance with cache interval defined by the provided duration and time to process CPU utilization. Only successful stats results are cached.

type Pattern

type Pattern struct {
	Pattern   *regexp.Regexp
	Throttler Throttler
}

Pattern defines a pair of regexp and related throttler.

type Runnable

type Runnable func(context.Context) error

Runnable defined by typical abstract async func signature. Runnable is used by `Runner` as a subject for execution.

type Runner

type Runner interface {
	// Run executes single prodived `Runnable` instance.
	Run(Runnable)
	// Result returns possible execution error back.
	Result() error
}

Runner defines abstraction to execute a set of `Runnable` and return possible execution error back. Runner is designed to simplify work with throttlers by managing `Acquire`/`Release` loop.

func NewRunnerAsync

func NewRunnerAsync(ctx context.Context, thr Throttler) Runner

NewRunnerAsync creates asynchronous runner instance that runs a set of `Runnable` simultaneously with regard to the provided context and throttler. First occurred error is returned from result.

func NewRunnerSync

func NewRunnerSync(ctx context.Context, thr Throttler) Runner

NewRunnerSync creates synchronous runner instance that runs a set of `Runnable` consecutively with regard to the provided context and throttler. First occurred error is returned from result.

type Stats

type Stats struct {
	MEMAlloc  uint64
	MEMSystem uint64
	CPUPause  uint64
	CPUUsage  float64
}

Stats defines typical set of metrics returned by system monitor: - MEMAlloc shows how many bytes are allocated by heap objects. - MEMSystem shows how many bytes are obtained from the OS. - CPUPause shows average GC stop-the-world pause in nanoseconds. - CPUUsage shows average CPU utilization in percents.

func (Stats) Compare

func (s Stats) Compare(stats Stats) bool

Compare checks if provided stats is below current stats.

type Throttler

type Throttler interface {
	// Acquire takes a part of throttling quota or returns error if throttling quota is drained
	// it needs to be called right before shared resource acquire.
	Acquire(context.Context) error
	// Release puts a part of throttling quota back or returns error if this is not possible
	// it needs to be called just after shared resource release.
	Release(context.Context) error
}

Throttler defines core gohalt throttler abstraction and exposes pair of counterpart methods: `Acquire` and `Release`.

func NewThrottlerAdaptive

func NewThrottlerAdaptive(
	threshold uint64,
	interval time.Duration,
	quantum time.Duration,
	step uint64,
	thr Throttler,
) Throttler

NewThrottlerAdaptive creates new throttler instance that throttles each call which exeeds the running quota acquired - release q defined by the specified threshold in the specified interval. Periodically each specified interval the running quota number is reseted. If quantum is set then quantum will be used instead of interval to provide the running quota delta updates. Provided adapted throttler adjusts the running quota of adapter throttler by changing the value by d defined by the specified step, it subtracts *d^2* from the running quota if adapted throttler throttles or adds *d* to the running quota if it doesn't. Use `WithWeight` to override context call qunatity, 1 by default. - could return `ErrorThreshold`;

func NewThrottlerAfter

func NewThrottlerAfter(threshold uint64) Throttler

NewThrottlerAfter creates new throttler instance that throttles each call after the i-th call defined by the specified threshold. Use `WithWeight` to override context call qunatity, 1 by default. - could return `ErrorThreshold`;

func NewThrottlerAll

func NewThrottlerAll(thrs ...Throttler) Throttler

NewThrottlerAll creates new throttler instance that throttles call if all provided throttlers throttle. - could return `ErrorInternal`;

func NewThrottlerAny

func NewThrottlerAny(thrs ...Throttler) Throttler

NewThrottlerAny creates new throttler instance that throttles call if any of provided throttlers throttle. - could return `ErrorInternal`;

func NewThrottlerBefore

func NewThrottlerBefore(threshold uint64) Throttler

NewThrottlerBefore creates new throttler instance that throttles each call below the i-th call defined by the specified threshold. Use `WithWeight` to override context call qunatity, 1 by default. - could return `ErrorThreshold`;

func NewThrottlerBucket

func NewThrottlerBucket(threshold uint64, interval time.Duration, monotone bool) Throttler

NewThrottlerBucket creates new throttler instance that uses leaky bucket algorithm to throttles call within provided interval and threshold. If provided monotone flag is set class to release will have no effect on throttler. Use `WithWeight` to override context call qunatity, 1 by default. - could return `ErrorThreshold`;

func NewThrottlerBuffered

func NewThrottlerBuffered(threshold uint64) Throttler

NewThrottlerBuffered creates new throttler instance that waits on call which exeeds the running quota acquired - release q defined by the specified threshold until the running quota is available again.

func NewThrottlerCache

func NewThrottlerCache(thr Throttler, cache time.Duration) Throttler

NewThrottlerCache creates new throttler instance that caches provided throttler calls for the provided cache duration, throttler release resulting resets cache. Only non throttling calls are cached for the provided cache duration. - could return any underlying throttler error;

func NewThrottlerCellRate

func NewThrottlerCellRate(threshold uint64, interval time.Duration, monotone bool) Throttler

NewThrottlerCellRate creates new throttler instance that uses generic cell rate algorithm to throttles call within provided interval and threshold. If provided monotone flag is set class to release will have no effect on throttler. Use `WithWeight` to override context call qunatity, 1 by default. - could return `ErrorThreshold`;

func NewThrottlerChance

func NewThrottlerChance(threshold float64) Throttler

NewThrottlerChance creates new throttler instance that throttles each call with the chance p defined by the specified threshold. Chance value is normalized to [0.0, 1.0] range. Implementation uses secure `crypto/rand` as PRNG function. - could return `ErrorThreshold`;

func NewThrottlerContext

func NewThrottlerContext() Throttler

NewThrottlerContext creates new throttler instance that always throttless on done context. - could return `ErrorInternal`;

func NewThrottlerEach

func NewThrottlerEach(threshold uint64) Throttler

NewThrottlerEach creates new throttler instance that throttles each periodic i-th call defined by the specified threshold. - could return `ErrorThreshold`;

func NewThrottlerEcho

func NewThrottlerEcho(err error) Throttler

NewThrottlerEcho creates new throttler instance that always throttles with the specified error back. - could return any specified error;

func NewThrottlerEnqueue

func NewThrottlerEnqueue(enq Enqueuer) Throttler

NewThrottlerEnqueue creates new throttler instance that always enqueues message to the specified queue throttles only if any internal error occurred. Use `WithMessage` to specify context message for enqueued message and `WithMarshaler` to specify context message marshaler. Builtin `Enqueuer` implementations come with connection reuse and retries by default. Use builtin `NewEnqueuerRabbit` to create RabbitMQ enqueuer instance or `NewEnqueuerKafka` to create Kafka enqueuer instance. - could return `ErrorInternal`;

func NewThrottlerFuture

func NewThrottlerFuture(threshold time.Time) Throttler

NewThrottlerFuture creates new throttler instance that throttles each call after timestamp defined by the specified UTC time threshold. - could return `ErrorThreshold`;

func NewThrottlerGenerator

func NewThrottlerGenerator(gen Generator, capacity uint64, eviction float64) Throttler

NewThrottlerGenerator creates new throttler instance that throttles if found key matching throttler throttles. If no key matching throttler has been found generator see `Generator` used insted to provide new throttler that will be added to existing throttlers map. Generated throttlers are kept in bounded map with capacity c defined by the specified capacity and eviction rate e defined by specified eviction value is normalized to [0.0, 1.0], where eviction rate affects number of throttlers that will be removed from the map after bounds overflow. Use `WithKey` to specify key for throttler matching and generation. - could return `ErrorInternal`; - could return any underlying throttler error;

func NewThrottlerJitter

func NewThrottlerJitter(initial time.Duration, limit time.Duration, reset bool, jitter float64) Throttler

NewThrottlerJitter creates new throttler instance that waits accordingly to undelying square throttler but also adds the provided jitter delta distribution on top. Jitter value is normalized to [0.0, 1.0] range and defines which part of square delay could be randomized in percents. Implementation uses secure `crypto/rand` as PRNG function.

func NewThrottlerLatency

func NewThrottlerLatency(threshold time.Duration, retention time.Duration) Throttler

NewThrottlerLatency creates new throttler instance that throttles each call after the call latency l defined by the specified threshold was exeeded once. If retention is set then throttler state will be reseted after retention duration. Use `WithTimestamp` to specify running duration between throttler acquire and release. - could return `ErrorThreshold`;

func NewThrottlerMetric

func NewThrottlerMetric(mtc Metric) Throttler

NewThrottlerMetric creates new throttler instance that throttles call ifboolean metric defined by the specified boolean metric is reached or if any internal error occurred. Builtin `Metric` implementations come with boolean metric caching by default. Use builtin `NewMetricPrometheus` to create Prometheus metric instance. - could return `ErrorInternal`; - could return `ErrorThreshold`;

func NewThrottlerMonitor

func NewThrottlerMonitor(mnt Monitor, threshold Stats) Throttler

NewThrottlerMonitor creates new throttler instance that throttles call if any of the stats returned by provided monitor exceeds any of the stats defined by the specified threshold or if any internal error occurred. Builtin `Monitor` implementations come with stats caching by default. Use builtin `NewMonitorSystem` to create go system monitor instance. - could return `ErrorInternal`; - could return `ErrorThreshold`;

func NewThrottlerNot

func NewThrottlerNot(thr Throttler) Throttler

NewThrottlerNot creates new throttler instance that throttles call if provided throttler doesn't throttle. - could return `ErrorInternal`;

func NewThrottlerPanic

func NewThrottlerPanic() Throttler

NewThrottlerPanic creates new throttler instance that always panics with `ErrorInternal`.

func NewThrottlerPast

func NewThrottlerPast(threshold time.Time) Throttler

NewThrottlerPast creates new throttler instance that throttles each call before timestamp defined by the specified UTC time threshold. - could return `ErrorThreshold`;

func NewThrottlerPattern

func NewThrottlerPattern(patterns ...Pattern) Throttler

NewThrottlerPattern creates new throttler instance that throttles if matching throttler from provided patterns throttles. Use `WithKey` to specify key for regexp pattern throttler matching. See `Pattern` which defines a pair of regexp and related throttler. - could return `ErrorInternal`; - could return any underlying throttler error;

func NewThrottlerPercentile

func NewThrottlerPercentile(
	threshold time.Duration,
	capacity uint8,
	percentile float64,
	retention time.Duration,
) Throttler

NewThrottlerPercentile creates new throttler instance that throttles each call after the call latency l defined by the specified threshold was exeeded once considering the specified percentile. Percentile values are kept in bounded buffer with capacity c defined by the specified capacity. If retention is set then throttler state will be reseted after retention duration. Use `WithTimestamp` to specify running duration between throttler acquire and release. - could return `ErrorThreshold`;

func NewThrottlerPriority

func NewThrottlerPriority(threshold uint64, levels uint8) Throttler

NewThrottlerPriority creates new throttler instance that waits on call which exeeds the running quota acquired - release q defined by the specified threshold until the running quota is available again. Running quota is not equally distributed between n levels of priority defined by the specified levels. Use `WithPriority` to override context call priority, 1 by default.

func NewThrottlerRetry

func NewThrottlerRetry(thr Throttler, retries uint64, onthreshold bool) Throttler

NewThrottlerRetry creates new throttler instance that retries provided throttler error up until the provided retries threshold. If provided onthreshold flag is set even `ErrorThreshold` errors will be retried. Internally retry uses square throttler with `DefaultRetriedDuration` initial duration. - could return any underlying throttler error;

func NewThrottlerRing

func NewThrottlerRing(thrs ...Throttler) Throttler

NewThrottlerRing creates new throttler instance that throttles if the i-th call throttler from provided list throttle. - could return `ErrorInternal`; - could return any underlying throttler error;

func NewThrottlerRunning

func NewThrottlerRunning(threshold uint64) Throttler

NewThrottlerRunning creates new throttler instance that throttles each call which exeeds the running quota acquired - release q defined by the specified threshold. - could return `ErrorThreshold`;

func NewThrottlerSemaphore

func NewThrottlerSemaphore(weight int64) Throttler

NewThrottlerSemaphore creates new throttler instance that throttles call if underlying semaphore throttles. Use `WithWeight` to override context call weight, 1 by default. - could return `ErrorThreshold`;

func NewThrottlerSquare

func NewThrottlerSquare(initial time.Duration, limit time.Duration, reset bool) Throttler

NewThrottlerSquare creates new throttler instance that always waits for square growing [1, 4, 9, 16, ...] multiplier on the specified initial duration, up until the specified duration limit is reached. If reset is set then after throttler riches the specified duration limit next multiplier value will be reseted.

func NewThrottlerSuppress

func NewThrottlerSuppress(thr Throttler) Throttler

NewThrottlerSuppress creates new throttler instance that suppresses provided throttler to never throttle.

func NewThrottlerTimed

func NewThrottlerTimed(threshold uint64, interval time.Duration, quantum time.Duration) Throttler

NewThrottlerTimed creates new throttler instance that throttles each call which exeeds the running quota acquired - release q defined by the specified threshold in the specified interval. Periodically each specified interval the running quota number is reseted. If quantum is set then quantum will be used instead of interval to provide the running quota delta updates. Use `WithWeight` to override context call qunatity, 1 by default. - could return `ErrorThreshold`;

func NewThrottlerWait

func NewThrottlerWait(duration time.Duration) Throttler

NewThrottlerWait creates new throttler instance that always waits for the specified duration.