README
¶
pkg/queue
queue is the buffering layer between informer events and the reconciler. Every
CRD gets its own Workqueue. The kordinator reads from these queues and calls
the reconciler for each item. The queue package has no knowledge of what is
being reconciled — it holds keys and routes them.
What lives here
| File | Role |
|---|---|
queue.go |
Workqueue — per-CRD rate-limiting queue with atomic depth limit |
registry.go |
QueueRegistry — maps GVK strings to their Workqueue; started as an Orkestra component |
Core concepts
One queue per CRD. Isolation means a CRD with large CRs or slow external calls cannot delay reconciles for other CRDs. Workers for a CRD read only from that CRD's queue.
Rate-limiting. The underlying workqueue.TypedRateLimitingInterface applies
exponential backoff on re-queued items after reconcile failures. The first
failure retries immediately; subsequent failures back off up to a maximum.
Forget(item) resets the backoff counter on success.
Depth limit. Workqueue.Enqueue enforces an atomic int32 depth limit.
When the limit is non-zero and the queue is at or beyond it, new items are
dropped with a warning log. This is the mechanism by which the autoscaler's
do.queueDepth override takes effect. 0 means unlimited (the default).
Deduplication. The workqueue deduplicates by key — if the same key is enqueued multiple times while a worker holds it, only one copy queues behind. This makes concurrent resync sources (informer + autoscaler resync goroutine) safe.
Developer documentation
| I want to understand… | Go to |
|---|---|
| The Workqueue type and how Enqueue enforces depth limits | 01 — Workqueue |
| The QueueRegistry and how queues are registered and looked up | 02 — Registry |
Key types at a glance
| Type | File | Role |
|---|---|---|
Workqueue |
queue.go |
Per-CRD queue: Enqueue, Depth, SetMaxQueueDepth |
QueueRegistry |
registry.go |
GVK → Workqueue map; Orkestra component lifecycle |
QueueItem |
queue.go |
{Key, GVK} — the unit of work passed between informer and worker |
Documentation
¶
Overview ¶
pkg/queue/registry.go
Index ¶
- type QueueItem
- type QueueRegistry
- func (qr *QueueRegistry) Depth(gvk string) int
- func (qr *QueueRegistry) For(gvk string) (*Workqueue, bool)
- func (qr *QueueRegistry) Name() string
- func (qr *QueueRegistry) Register(gvk string, maxQueueDepth int) *Workqueue
- func (qr *QueueRegistry) Shutdown(ctx context.Context)
- func (qr *QueueRegistry) ShutdownQueue(gvkStr string) error
- func (qr *QueueRegistry) Start(ctx context.Context) error
- func (qr *QueueRegistry) Started() bool
- type Workqueue
- func (q *Workqueue) Depth() int
- func (q *Workqueue) Enqueue(obj interface{}, gvk string)
- func (q *Workqueue) GetWithContext(ctx context.Context) (QueueItem, bool)
- func (q *Workqueue) MaxQueueDepth() int
- func (q *Workqueue) Name() string
- func (q *Workqueue) SetMaxQueueDepth(n int)
- func (q *Workqueue) Shutdown(ctx context.Context)
- func (q *Workqueue) Start(ctx context.Context) error
- func (q *Workqueue) Started() bool
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type QueueRegistry ¶
type QueueRegistry struct {
// contains filtered or unexported fields
}
func NewQueueRegistry ¶
func NewQueueRegistry() *QueueRegistry
NewQueueRegistry returns a new queue registry At this stage only the queues map is created Register uses the map created to register all CRDs For returns a registered CRD
func (*QueueRegistry) Depth ¶
func (qr *QueueRegistry) Depth(gvk string) int
Depth returns the queue depth for a given GVK
func (*QueueRegistry) For ¶
func (qr *QueueRegistry) For(gvk string) (*Workqueue, bool)
For returns the workqueue for a given GVK
func (*QueueRegistry) Name ¶
func (qr *QueueRegistry) Name() string
Name returns the name of the queue registry
func (*QueueRegistry) Register ¶
func (qr *QueueRegistry) Register(gvk string, maxQueueDepth int) *Workqueue
Register registers each CRD with their respective GVK and maximum queue depth
func (*QueueRegistry) Shutdown ¶
func (qr *QueueRegistry) Shutdown(ctx context.Context)
Shutdown drains all registered workqueues This is called by orkestra.Shutdown() for graceful degradation
func (*QueueRegistry) ShutdownQueue ¶
func (qr *QueueRegistry) ShutdownQueue(gvkStr string) error
ShutdownQueue drains the queue of a given CRD
func (*QueueRegistry) Start ¶
func (qr *QueueRegistry) Start(ctx context.Context) error
Called by orkestra.Start() to start the workqueue registry
func (*QueueRegistry) Started ¶
func (qr *QueueRegistry) Started() bool
Started is a status check for all orkestra komponents
type Workqueue ¶
type Workqueue struct {
Queue workqueue.TypedRateLimitingInterface[QueueItem]
// contains filtered or unexported fields
}
func NewWorkqueue ¶
func NewWorkqueue() *Workqueue
func (*Workqueue) Enqueue ¶
Enqueue adds the object's key to the workqueue. When a non-zero maxQueueDepth is set, new items are dropped (with a warning) once the queue is at or beyond that limit. Items already in the queue are not evicted — only incoming enqueues are rejected.
func (*Workqueue) GetWithContext ¶
GetWithContext returns an item from the work queue. Context (e.g., timeout, cancellation). Future use
func (*Workqueue) MaxQueueDepth ¶
MaxQueueDepth returns the current maximum queue depth (0 = unlimited).
func (*Workqueue) SetMaxQueueDepth ¶
SetMaxQueueDepth adjusts the queue depth limit at runtime. 0 means unlimited. Safe to call concurrently with Enqueue.
func (*Workqueue) Shutdown ¶
Shutdown drains the default workqueue This is called by orkestra.Shutdown() for graceful degradation
Source Files