workqueue

Documentation

Index

• func DefaultCDDaemonRateLimiter() workqueue.TypedRateLimiter[any]
• func DefaultControllerRateLimiter() workqueue.TypedRateLimiter[any]
• func DefaultPrepUnprepRateLimiter() workqueue.TypedRateLimiter[any]
• func NewJitterRateLimiter[T comparable](inner workqueue.TypedRateLimiter[T], factor float64) workqueue.TypedRateLimiter[T]
• type JitterRL
  • func (j *JitterRL[T]) Forget(item T)
  • func (j *JitterRL[T]) NumRequeues(item T) int
  • func (j *JitterRL[T]) When(item T) time.Duration
• type WorkItem
• type WorkQueue
  • func New(r workqueue.TypedRateLimiter[any]) *WorkQueue
  • func (q *WorkQueue) Enqueue(obj any, callback func(ctx context.Context, obj any) error)
  • func (q *WorkQueue) EnqueueRaw(obj any, callback func(ctx context.Context, obj any) error)
  • func (q *WorkQueue) EnqueueRawWithKey(obj any, key string, callback func(ctx context.Context, obj any) error)
  • func (q *WorkQueue) EnqueueWithKey(obj any, key string, callback func(ctx context.Context, obj any) error)
  • func (q *WorkQueue) Run(ctx context.Context)

Functions

func DefaultCDDaemonRateLimiter

func DefaultCDDaemonRateLimiter() workqueue.TypedRateLimiter[any]

func DefaultControllerRateLimiter

func DefaultControllerRateLimiter() workqueue.TypedRateLimiter[any]

func DefaultPrepUnprepRateLimiter

func DefaultPrepUnprepRateLimiter() workqueue.TypedRateLimiter[any]

Return composite rate limiter that combines both per-item exponential backoff and an overall token bucket rate-limiting strategy. It calculates the exponential backoff for the individual item (based on its personal retry history), checks the global rate against the token bucket, and picks the longest delay from either strategy, ensuring that both per-item and overall queue health are respected.

func NewJitterRateLimiter

func NewJitterRateLimiter[T comparable](inner workqueue.TypedRateLimiter[T], factor float64) workqueue.TypedRateLimiter[T]

Types

type JitterRL

type JitterRL[T comparable] struct {
	// contains filtered or unexported fields
}

Jitter relative to the delay yielded by the inner limiter. Example: a factor of 0.1 translates to a jitter interval with a width of 10 % compared to the inner delay, and centered around the inner delay time (resulting in +/- 5 % deviation compared to the inner delay time).

func (*JitterRL[T]) Forget

func (j *JitterRL[T]) Forget(item T)

func (*JitterRL[T]) NumRequeues

func (j *JitterRL[T]) NumRequeues(item T) int

func (*JitterRL[T]) When

func (j *JitterRL[T]) When(item T) time.Duration

type WorkItem

type WorkItem struct {
	Key      string
	Object   any
	Callback func(ctx context.Context, obj any) error
}

type WorkQueue

type WorkQueue struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func New

func New(r workqueue.TypedRateLimiter[any]) *WorkQueue

func (*WorkQueue) Enqueue

func (q *WorkQueue) Enqueue(obj any, callback func(ctx context.Context, obj any) error)

func (*WorkQueue) EnqueueRaw

func (q *WorkQueue) EnqueueRaw(obj any, callback func(ctx context.Context, obj any) error)

func (*WorkQueue) EnqueueRawWithKey

func (q *WorkQueue) EnqueueRawWithKey(obj any, key string, callback func(ctx context.Context, obj any) error)

func (*WorkQueue) EnqueueWithKey

func (q *WorkQueue) EnqueueWithKey(obj any, key string, callback func(ctx context.Context, obj any) error)

func (*WorkQueue) Run

func (q *WorkQueue) Run(ctx context.Context)