throttle

README

Throttle plugin

It discards the events if pipeline throughput gets higher than a configured threshold.

Config params

throttle_field cfg.FieldSelector

The event field which will be used as a key for throttling. It means that throttling will work separately for events with different keys. If not set, it's assumed that all events have the same key.

time_field cfg.FieldSelector default=time

The event field which defines the time when event was fired. It is used to detect the event throughput in a particular time range. If not set, the current time will be taken.

time_field_format string default=rfc3339nano options=ansic|unixdate|rubydate|rfc822|rfc822z|rfc850|rfc1123|rfc1123z|rfc3339|rfc3339nano|kitchen|stamp|stampmilli|stampmicro|stampnano|unixtime|nginx_errorlog

It defines how to parse the time field format.

default_limit int64 default=5000

The default events limit that plugin allows per bucket_interval.

limit_kind string default=count options=count|size

It defines subject of limiting: number of messages or total size of the messages.

limiter_backend string default=memory options=memory|redis

Defines kind of backend. When redis backend is chosen and if by any reason plugin cannot connect to redis, limiters will not start syncing with redis until successful reconnect.

redis_backend_config RedisBackendConfig

It contains redis settings

buckets_count int default=60

How much time buckets to hold in the memory. E.g. if buckets_count is 60 and bucket_interval is 5m, then 5 hours will be covered. Events with time later than now() - 5h will be dropped even if threshold isn't exceeded.

bucket_interval cfg.Duration default=1m

Time interval to check event throughput.

rules []RuleConfig

Rules to override the default_limit for different group of event. It's a list of objects. Each object has the limit and conditions fields.

• limit – the value which will override the default_limit, if conditions are met.
• limit_kind – the type of a limit: count - number of messages, size - total size from all messages
• conditions – the map of event field name => event field value. The conditions are checked using AND operator.

limiter_expiration cfg.Duration default=30m

Time interval after which unused limiters are removed.

endpoint string

password string

Password to redis server.

sync_interval cfg.Duration default=5s

Defines sync interval between global and local limiters.

worker_count int default=32

Defines num of parallel workers that will sync limits.

timeout cfg.Duration default=1s

Defines redis timeout.

max_retries int default=3

Defines redis maximum number of retries. If set to 0, no retries will happen.

min_retry_backoff cfg.Duration default=8ms

Defines redis minimum backoff between each retry. If set to 0, sets default 8ms. If set to -1, disables backoff.

max_retry_backoff cfg.Duration default=512ms

Defines redis maximum backoff between each retry. If set to 0, sets default 512ms. If set to -1, disables backoff.

limiter_key_field cfg.FieldSelector

Defines the event field from which values are used as limiter keys. Serves as an override of the default limiter keys naming pattern. If not set limiter keys are formed using pipeline name, throttle field and throttle field value.

limiter_value_field string

Defines field with limit inside json object stored in value (e.g. if set to "limit", values must be of kind {"limit":"<int>",...}). If not set limiter values are considered as non-json data.

Generated using insane-doc

Documentation

Index

• func NewInMemoryLimiter(interval time.Duration, bucketCount int, limit complexLimit, ...) *inMemoryLimiter
• func NewRedisLimiter(ctx context.Context, redis redisClient, ...) *redisLimiter
• func NewRule(conditions map[string]string, limit complexLimit, ruleNum int) *rule
• type Config
• type Plugin
  • func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult
  • func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.ActionPluginParams)
  • func (p *Plugin) Stop()
• type RedisBackendConfig
• type RuleConfig

Functions

func NewInMemoryLimiter

func NewInMemoryLimiter(interval time.Duration, bucketCount int, limit complexLimit, nowFn func() time.Time) *inMemoryLimiter

NewInMemoryLimiter returns limiter instance.

func NewRedisLimiter

func NewRedisLimiter(
	ctx context.Context,
	redis redisClient,
	pipelineName, throttleFieldName, throttleFieldValue string,
	bucketInterval time.Duration,
	bucketCount int,
	limit complexLimit,
	keyLimitOverride string,
	valField string,
	nowFn func() time.Time,
) *redisLimiter

NewRedisLimiter return instance of redis limiter.

func NewRule

func NewRule(conditions map[string]string, limit complexLimit, ruleNum int) *rule

NewRule returns new Limit instance.

Types

type Config

type Config struct {
	// > @3@4@5@6
	// >
	// > The event field which will be used as a key for throttling.
	// > It means that throttling will work separately for events with different keys.
	// > If not set, it's assumed that all events have the same key.
	ThrottleField  cfg.FieldSelector `json:"throttle_field" default:"" parse:"selector"` // *
	ThrottleField_ []string

	// > @3@4@5@6
	// >
	// > The event field which defines the time when event was fired.
	// > It is used to detect the event throughput in a particular time range.
	// > If not set, the current time will be taken.
	TimeField  cfg.FieldSelector `json:"time_field" default:"time" parse:"selector"` // *
	TimeField_ []string

	// > @3@4@5@6
	// >
	// > It defines how to parse the time field format.
	TimeFieldFormat string `` // *
	/* 209-byte string literal not displayed */

	// > @3@4@5@6
	// >
	// > The default events limit that plugin allows per `bucket_interval`.
	DefaultLimit int64 `json:"default_limit" default:"5000"` // *

	// > @3@4@5@6
	// >
	// > It defines subject of limiting: number of messages or total size of the messages.
	LimitKind string `json:"limit_kind" default:"count" options:"count|size"` // *

	// > @3@4@5@6
	// >
	// > Defines kind of backend. When redis backend is chosen and if by any reason plugin cannot connect to redis,
	// > limiters will not start syncing with redis until successful reconnect.
	LimiterBackend string `json:"limiter_backend" default:"memory" options:"memory|redis"` // *

	// > @3@4@5@6
	// >
	// > It contains redis settings
	RedisBackendCfg RedisBackendConfig `json:"redis_backend_config" child:"true"` // *

	// > @3@4@5@6
	// >
	// > How much time buckets to hold in the memory. E.g. if `buckets_count` is `60` and `bucket_interval` is `5m`,
	// > then `5 hours` will be covered. Events with time later than `now() - 5h` will be dropped even if threshold isn't exceeded.
	BucketsCount int `json:"buckets_count" default:"60"` // *

	// > @3@4@5@6
	// >
	// > Time interval to check event throughput.
	BucketInterval  cfg.Duration `json:"bucket_interval" parse:"duration" default:"1m"` // *
	BucketInterval_ time.Duration

	// > @3@4@5@6
	// >
	// > Rules to override the `default_limit` for different group of event. It's a list of objects.
	// > Each object has the `limit` and `conditions` fields.
	// > * `limit` – the value which will override the `default_limit`, if `conditions` are met.
	// > * `limit_kind` – the type of a limit: `count` - number of messages, `size` - total size from all messages
	// > * `conditions` – the map of `event field name => event field value`. The conditions are checked using `AND` operator.
	Rules []RuleConfig `json:"rules" default:"" slice:"true"` // *

	// > @3@4@5@6
	// >
	// > Time interval after which unused limiters are removed.
	LimiterExpiration  cfg.Duration `json:"limiter_expiration" parse:"duration" default:"30m"` // *
	LimiterExpiration_ time.Duration
}

! config-params ^ config-params

type Plugin

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

func (*Plugin) Do

func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult

func (*Plugin) Start

func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.ActionPluginParams)

func (*Plugin) Stop

func (p *Plugin) Stop()

Stop ends plugin activity.

type RedisBackendConfig

type RedisBackendConfig struct {
	// > @3@4@5@6
	// >
	// Аddress of redis server. Format: HOST:PORT.
	Endpoint string `json:"endpoint"` // *

	// > @3@4@5@6
	// >
	// > Password to redis server.
	Password string `json:"password"` // *

	// > @3@4@5@6
	// >
	// > Defines sync interval between global and local limiters.
	SyncInterval  cfg.Duration `json:"sync_interval" parse:"duration" default:"5s"` // *
	SyncInterval_ time.Duration

	// > @3@4@5@6
	// >
	// > Defines num of parallel workers that will sync limits.
	WorkerCount int `json:"worker_count" default:"32"` // *

	// > @3@4@5@6
	// >
	// > Defines redis timeout.
	Timeout  cfg.Duration `json:"timeout" parse:"duration" default:"1s"` // *
	Timeout_ time.Duration

	// > @3@4@5@6
	// >
	// > Defines redis maximum number of retries. If set to 0, no retries will happen.
	MaxRetries int `json:"max_retries" default:"3"` // *

	// > @3@4@5@6
	// >
	// > Defines redis minimum backoff between each retry. If set to 0, sets default 8ms. If set to -1, disables backoff.
	MinRetryBackoff  cfg.Duration `json:"min_retry_backoff" parse:"duration" default:"8ms"` // *
	MinRetryBackoff_ time.Duration

	// > @3@4@5@6
	// >
	// > Defines redis maximum backoff between each retry. If set to 0, sets default 512ms. If set to -1, disables backoff.
	MaxRetryBackoff  cfg.Duration `json:"max_retry_backoff" parse:"duration" default:"512ms"` // *
	MaxRetryBackoff_ time.Duration

	// > @3@4@5@6
	// >
	// > Defines the event field from which values are used as limiter keys. Serves as an override of the default limiter keys naming pattern.
	// > If not set limiter keys are formed using pipeline name, throttle field and throttle field value.
	LimiterKeyField  cfg.FieldSelector `json:"limiter_key_field" default:"" parse:"selector"` // *
	LimiterKeyField_ []string

	// > @3@4@5@6
	// >
	// > Defines field with limit inside json object stored in value
	// > (e.g. if set to "limit", values must be of kind `{"limit":"<int>",...}`).
	// > If not set limiter values are considered as non-json data.
	LimiterValueField string `json:"limiter_value_field" default:""` // *
}

type RuleConfig

type RuleConfig struct {
	Limit      int64             `json:"limit"`
	LimitKind  string            `json:"limit_kind" default:"count" options:"count|size"`
	Conditions map[string]string `json:"conditions"`
}