Documentation
¶
Overview ¶
Package controllerutils holds controller helpers shared across services (backend, kube-applier, fleet, ...) that all build Cosmos-backed informer-driven controllers and want consistent cadence/gating behavior.
The cooldown gate exposed here was originally backend/pkg/utils/controllerutils' TimeBasedCooldownChecker; it lives in internal/ so other services can use the same implementation without re-importing the backend module.
Index ¶
- Constants
- func BuildReadDesire(resourceIDString string, managementCluster *azcorearm.ResourceID, ...) *kubeapplierapi.ReadDesire
- func NeedsUpdate(existing, desired any) bool
- func ReadDesireNeedsWork(existing, desired *kubeapplierapi.ReadDesire) bool
- func ResourceIDsEqual(a, b *azcorearm.ResourceID) bool
- type AfterEnqueuer
- type CooldownChecker
- type GenericSyncer
- type GenericWatchingController
- func (c *GenericWatchingController[T]) EnqueueAfter(keyObj any, duration time.Duration)
- func (c *GenericWatchingController[T]) EnqueueResourceIDAdd(resourceID *azcorearm.ResourceID, changed bool)
- func (c *GenericWatchingController[T]) EnqueueResourceIDAddWithMaxDepth(resourceID *azcorearm.ResourceID, changed bool, maxDepth int)
- func (c *GenericWatchingController[T]) QueueForInformers(resyncDuration time.Duration, notifiers ...Notifier) error
- func (c *GenericWatchingController[T]) QueueForInformersWithMaxDepth(resyncDuration time.Duration, maxDepth int, notifiers ...Notifier) error
- func (c *GenericWatchingController[T]) Run(ctx context.Context, threadiness int)
- func (c *GenericWatchingController[T]) SyncOnce(ctx context.Context, keyObj any) error
- type Notifier
- type SettableCooldownChecker
- type TimeBasedCooldownChecker
Constants ¶
const (
HcpClusterAzureResourceIdAnnotation = "azure.microsoft.com/hcp-cluster-azure-resource-id"
)
Variables ¶
This section is empty.
Functions ¶
func BuildReadDesire ¶
func BuildReadDesire(resourceIDString string, managementCluster *azcorearm.ResourceID, target kubeapplierapi.ResourceReference) *kubeapplierapi.ReadDesire
BuildReadDesire produces the desired-state ReadDesire for a resource. The status section is intentionally zero — the kube-applier owns status.
func NeedsUpdate ¶
NeedsUpdate reports whether `desired` differs from `existing` in any way that should cause us to write `desired` back to Cosmos. It is a strict-but-server-managed-fields-aware semantic equality check: all the fields that actually persist must match, but cosmos-managed values like the document etag are ignored, as are Go-level representation differences (RawExtension Raw vs Object, parent pointer chains in ResourceID, etc.).
func ReadDesireNeedsWork ¶
func ReadDesireNeedsWork(existing, desired *kubeapplierapi.ReadDesire) bool
ReadDesireNeedsWork reports whether existing matches desired in the fields the controller writes (Spec.ManagementCluster, Spec.TargetItem). A nil existing means "doesn't exist yet" — work is required.
func ResourceIDsEqual ¶
func ResourceIDsEqual(a, b *azcorearm.ResourceID) bool
ResourceIDsEqual compares two *azcorearm.ResourceID for equality by their canonical string form. Both may be nil; non-nil values are compared by String(), so independently-parsed instances with different parent pointer chains still compare equal when they represent the same ARM ID.
Types ¶
type AfterEnqueuer ¶
AfterEnqueuer allows scheduling a workqueue item for processing after an explicit delay. Validation controllers use this to implement EarliestRetryAfter semantics.
type CooldownChecker ¶
CooldownChecker decides whether a key may be (re-)queued.
Implementations must be safe to call concurrently — informer event handlers, periodic resyncs, and worker goroutines may all consult the same checker.
CanSync takes a context so implementations may emit logs or use context-bound services; the time-based variant ignores it. A true return is taken as "the caller WILL sync this key" and the implementation is free to record state (e.g. stamp a next-allowed time) on that basis.
type GenericSyncer ¶
type GenericSyncer[T comparable] interface { SyncOnce(ctx context.Context, keyObj T) error CooldownChecker() CooldownChecker MakeKey(resourceID *azcorearm.ResourceID) T }
type GenericWatchingController ¶
type GenericWatchingController[T comparable] struct { // contains filtered or unexported fields }
func NewGenericWatchingController ¶
func NewGenericWatchingController[T comparable](name string, resourceType azcorearm.ResourceType, syncer GenericSyncer[T], reconcileTotal *prometheus.CounterVec) *GenericWatchingController[T]
NewGenericWatchingController creates a controller that watches Cosmos-backed informers and delegates reconciliation to syncer.
func (*GenericWatchingController[T]) EnqueueAfter ¶
func (c *GenericWatchingController[T]) EnqueueAfter(keyObj any, duration time.Duration)
func (*GenericWatchingController[T]) EnqueueResourceIDAdd ¶
func (c *GenericWatchingController[T]) EnqueueResourceIDAdd(resourceID *azcorearm.ResourceID, changed bool)
EnqueueResourceIDAdd is equivalent to calling EnqueueResourceIDAddWithMaxDepth with a maxDepth of -1. See EnqueueResourceIDAddWithMaxDepth for more details. It is exposed so that individual controllers can add other items to requeue based on easily.
func (*GenericWatchingController[T]) EnqueueResourceIDAddWithMaxDepth ¶
func (c *GenericWatchingController[T]) EnqueueResourceIDAddWithMaxDepth(resourceID *azcorearm.ResourceID, changed bool, maxDepth int)
EnqueueResourceIDAddWithMaxDepth traverses resourceID and its parents according to maxDepth until it finds a resourceID that is of the resource type of c.resourceType and adds it if found. Each walk to Parent consumes one level. maxDepth is the maximum number of parent hops to traverse when searching for a resourceID of type c.resourceType. maxDepth 0 means only the resourceID itself is considered. maxDepth -1 (or any negative value) means no limit. The parent walk continues until a match or nil parent is reached. It is exposed so that individual controllers can add other items to requeue based on easily. When there's a match of resourceType: when changed is true, the resourceID is added to the queue immediately. Otherwise, the resourceID is added to the queue only if the cooldown checker allows it.
func (*GenericWatchingController[T]) QueueForInformers ¶
func (c *GenericWatchingController[T]) QueueForInformers(resyncDuration time.Duration, notifiers ...Notifier) error
QueueForInformers is equivalent to calling QueueForInformersWithMaxDepth with maxDepth of -1. See QueueForInformersWithMaxDepth for more details.
func (*GenericWatchingController[T]) QueueForInformersWithMaxDepth ¶
func (c *GenericWatchingController[T]) QueueForInformersWithMaxDepth(resyncDuration time.Duration, maxDepth int, notifiers ...Notifier) error
QueueForInformersWithMaxDepth adds event handlers to the notifiers for the controller with a given max depth. maxDepth is the maximum number of parent hops to traverse when searching for a resourceID whose type is c.resourceType. Each walk to Parent consumes one level. maxDepth 0 means only the resourceID itself is considered. maxDepth -1 (or any negative value) means no limit. The parent walk continues until a match or nil parent is reached. It is exposed so that individual controllers can add other items to requeue based on easily.
type Notifier ¶
type Notifier interface {
AddEventHandlerWithOptions(handler cache.ResourceEventHandler, options cache.HandlerOptions) (cache.ResourceEventHandlerRegistration, error)
}
type SettableCooldownChecker ¶
type SettableCooldownChecker struct {
// contains filtered or unexported fields
}
SettableCooldownChecker is a cooldown gate where the per-key cooldown duration is set explicitly by the caller via SetCooldown, rather than being fixed at construction time. A key with no cooldown set is always allowed.
func NewSettableCooldownChecker ¶
func NewSettableCooldownChecker() *SettableCooldownChecker
func (*SettableCooldownChecker) CanSync ¶
func (c *SettableCooldownChecker) CanSync(_ context.Context, key any) bool
func (*SettableCooldownChecker) SetClock ¶
func (c *SettableCooldownChecker) SetClock(clock utilsclock.PassiveClock)
func (*SettableCooldownChecker) SetCooldown ¶
func (c *SettableCooldownChecker) SetCooldown(key any, duration time.Duration)
SetCooldown records that the given key should not be re-synced until now+duration has elapsed.
func (*SettableCooldownChecker) TimeUntilReady ¶
func (c *SettableCooldownChecker) TimeUntilReady(key any) time.Duration
TimeUntilReady returns the duration until the key's cooldown expires. Returns 0 if the key has no cooldown set or the cooldown has already expired.
type TimeBasedCooldownChecker ¶
type TimeBasedCooldownChecker struct {
// contains filtered or unexported fields
}
TimeBasedCooldownChecker is a fixed-interval cooldown gate: after CanSync returns true for a key, subsequent calls for the same key return false until cooldownDuration has elapsed since the allowed call.
The next-exec map is an LRU rather than an unbounded map so that a long-running process whose keys come and go does not leak memory. 1M entries is far above any realistic management cluster's resource count.
func NewTimeBasedCooldownChecker ¶
func NewTimeBasedCooldownChecker(cooldownDuration time.Duration) *TimeBasedCooldownChecker
NewTimeBasedCooldownChecker constructs a checker bound to the real wall-clock and a 1M-entry LRU. Tests should call SetClock to inject a fake clock.
func (*TimeBasedCooldownChecker) CanSync ¶
func (c *TimeBasedCooldownChecker) CanSync(_ context.Context, key any) bool
CanSync stamps now+cooldownDuration on a true return so subsequent calls within the cooldown window return false. A key with no record is always allowed. ctx is part of the CooldownChecker interface but unused here.
func (*TimeBasedCooldownChecker) SetClock ¶
func (c *TimeBasedCooldownChecker) SetClock(clock utilsclock.PassiveClock)
SetClock substitutes the time source used to evaluate the cooldown. Intended for tests; production code should use the real clock from the constructor.
Source Files