kubeapplier

Documentation

Overview

Package kubeapplier contains union informers for the kube-applier *Desire types, peering the union listers in internal/database/unionlisters/kubeapplier. Each union holds a set of per-management-cluster cache.SharedIndexInformers keyed by management cluster resourceID; event handlers registered on the union are propagated to every current sub-informer and to any sub-informer Added later.

The union does not own sub-informer lifecycle: callers start each SharedIndexInformer themselves (typically against a per-MC ctx) and call Remove when that ctx is cancelled. The pattern fits a backend reactor that wants a single informer surface spanning every management cluster's container as MCs come and go.

Index

• type ManagementClusterKey
  • func (k ManagementClusterKey) AddLoggerValues(logger logr.Logger) logr.Logger
  • func (k ManagementClusterKey) GetResourceID() *azcorearm.ResourceID
• type PerMCKubeApplierInformerFactory
  • func NewKubeApplierInformerFactory(clients kubeappliercosmosstorage.KubeApplierDBClients, ...) PerMCKubeApplierInformerFactory
• type UnionDesireInformer
  • func NewUnionDesireInformer() *UnionDesireInformer
  • func (u *UnionDesireInformer) Add(managementClusterResourceID *azcorearm.ResourceID, ...) error
  • func (u *UnionDesireInformer) AddEventHandler(handler cache.ResourceEventHandler) (cache.ResourceEventHandlerRegistration, error)
  • func (u *UnionDesireInformer) AddEventHandlerWithOptions(handler cache.ResourceEventHandler, options cache.HandlerOptions) (cache.ResourceEventHandlerRegistration, error)
  • func (u *UnionDesireInformer) AddEventHandlerWithResyncPeriod(handler cache.ResourceEventHandler, resyncPeriod time.Duration) (cache.ResourceEventHandlerRegistration, error)
  • func (u *UnionDesireInformer) HasSynced() bool
  • func (u *UnionDesireInformer) Remove(managementClusterResourceID *azcorearm.ResourceID)
  • func (u *UnionDesireInformer) RemoveEventHandler(reg cache.ResourceEventHandlerRegistration) error
• type UnionKubeApplierInformers
  • func NewUnionKubeApplierInformers() *UnionKubeApplierInformers
  • func (u *UnionKubeApplierInformers) Add(managementClusterResourceID *azcorearm.ResourceID, ...) error
  • func (u *UnionKubeApplierInformers) ApplyDesires() (*UnionDesireInformer, kubeapplierlisters.ApplyDesireLister)
  • func (u *UnionKubeApplierInformers) HasSynced() bool
  • func (u *UnionKubeApplierInformers) ReadDesires() (*UnionDesireInformer, kubeapplierlisters.ReadDesireLister)
  • func (u *UnionKubeApplierInformers) Remove(managementClusterResourceID *azcorearm.ResourceID)
• type UnionKubeApplierInformersController
  • func NewUnionKubeApplierInformersController(mcInformer cache.SharedIndexInformer, ...) *UnionKubeApplierInformersController
  • func (c *UnionKubeApplierInformersController) Run(ctx context.Context, threadiness int)
  • func (c *UnionKubeApplierInformersController) SyncOnce(ctx context.Context, key ManagementClusterKey) error
  • func (c *UnionKubeApplierInformersController) Union() *UnionKubeApplierInformers

Types

type ManagementClusterKey

type ManagementClusterKey struct {
	StampIdentifier       string `json:"stampIdentifier"`
	ManagementClusterName string `json:"managementClusterName"`
}

ManagementClusterKey identifies one management cluster for the controller's workqueue. Today a stamp hosts a single management cluster (the "default" singleton, see fleetapi.ManagementClusterResourceName), so StampIdentifier alone is the identity in practice; ManagementClusterName is kept on the key because the cosmos resourceID encodes both segments and we don't want callers reaching into resourceID-string surgery. Together they reconstruct the full resourceID:

/providers/microsoft.redhatopenshift/stamps/{StampIdentifier}/managementClusters/{ManagementClusterName}

func (ManagementClusterKey) AddLoggerValues

func (k ManagementClusterKey) AddLoggerValues(logger logr.Logger) logr.Logger

AddLoggerValues enriches logger with the standard resource-id key/value pairs derived from this management cluster's resourceID. Matches the shape of the other controller keys in backend/pkg/utils/controllerutils (HCPClusterKey, HCPNodePoolKey, etc.).

func (ManagementClusterKey) GetResourceID

func (k ManagementClusterKey) GetResourceID() *azcorearm.ResourceID

GetResourceID returns the management-cluster resourceID for this key.

type PerMCKubeApplierInformerFactory

type PerMCKubeApplierInformerFactory interface {
	NewKubeApplierInformers(ctx context.Context, managementClusterResourceID *azcorearm.ResourceID) kubeapplierinformers.KubeApplierInformers
}

PerMCKubeApplierInformerFactory builds a single management cluster's KubeApplierInformers on demand. The factory is decoupled from the controller so production wires it to a KubeApplierDBClients while tests can supply hand-rolled fakes.

The returned KubeApplierInformers must be unstarted: the controller owns the lifecycle and will call RunWithContext on a child context. Returning nil signals "no per-MC informers available for this resourceID" — the controller silently skips this sync and waits for the next event for that MC.

func NewKubeApplierInformerFactory

func NewKubeApplierInformerFactory(
	clients kubeappliercosmosstorage.KubeApplierDBClients,
	relistDuration *time.Duration,
) PerMCKubeApplierInformerFactory

NewKubeApplierInformerFactory wires a kubeappliercosmosstorage.KubeApplierDBClients into the PerMCKubeApplierInformerFactory shape the controller expects. It is the production wiring used by the backend; the integration test wires up an equivalent factory inline against the in-memory mock registry.

relistDuration is forwarded to the per-MC informer factory; pass nil to use the default cadence from internal/database/informers.

type UnionDesireInformer

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

UnionDesireInformer fans event-handler registration out to a set of per-management-cluster SharedIndexInformers and reports the worst-case sync state. It is the informer peer of unionlisters/kubeapplier.UnionXxxDesireLister and is type-agnostic at this layer because cache.SharedIndexInformer is.

All operations are thread-safe. Reads (HasSynced) snapshot the sub set under RLock; mutations (Add/Remove and (Add|Remove)EventHandler) hold the write lock for the duration of cache.SharedIndexInformer handler calls, which are expected to be quick.

func NewUnionDesireInformer

func NewUnionDesireInformer() *UnionDesireInformer

NewUnionDesireInformer returns an empty union; call Add to register per-management-cluster sub-informers and AddEventHandler to register handlers that fan out to them.

func (*UnionDesireInformer) Add

func (u *UnionDesireInformer) Add(managementClusterResourceID *azcorearm.ResourceID, sub cache.SharedIndexInformer) error

Add registers a SharedIndexInformer under the given management cluster's resourceID. Every handler previously registered on the union is installed on the new sub-informer before Add returns. A second Add under the same resourceID replaces the previous sub-informer, deregistering the union's handlers from it first. A nil resourceID is a programming error and is ignored.

The caller is responsible for the sub-informer's lifecycle (Run/stop); Add only wires handler propagation.

func (*UnionDesireInformer) AddEventHandler

func (u *UnionDesireInformer) AddEventHandler(handler cache.ResourceEventHandler) (cache.ResourceEventHandlerRegistration, error)

AddEventHandler installs handler on every currently-registered sub-informer and remembers it so future Adds also receive it. The returned registration is opaque; pass it to RemoveEventHandler to detach the handler from all sub-informers (current and future).

func (*UnionDesireInformer) AddEventHandlerWithOptions

func (u *UnionDesireInformer) AddEventHandlerWithOptions(
	handler cache.ResourceEventHandler, options cache.HandlerOptions,
) (cache.ResourceEventHandlerRegistration, error)

AddEventHandlerWithOptions is the options-bearing variant; it lets callers pass cache.HandlerOptions (logger, resync period) through to every sub-informer. This is the API the backend's generic watching-controller machinery (controllerutils.Notifier) consumes, so providing it lets the union plug into cluster-watching controllers verbatim.

func (*UnionDesireInformer) AddEventHandlerWithResyncPeriod

func (u *UnionDesireInformer) AddEventHandlerWithResyncPeriod(
	handler cache.ResourceEventHandler, resyncPeriod time.Duration,
) (cache.ResourceEventHandlerRegistration, error)

AddEventHandlerWithResyncPeriod is the resync-period variant; semantics are otherwise identical to AddEventHandler. The resync period is recorded so future Adds also install the handler with the same resync period.

func (*UnionDesireInformer) HasSynced

func (u *UnionDesireInformer) HasSynced() bool

HasSynced returns true only when every currently-registered sub-informer reports HasSynced. An empty union is considered synced (vacuously true); callers that need a populated union should gate on their MC count.

func (*UnionDesireInformer) Remove

func (u *UnionDesireInformer) Remove(managementClusterResourceID *azcorearm.ResourceID)

Remove deregisters every union-installed handler from the sub-informer registered under the given management cluster's resourceID and drops it from the union. The sub-informer itself is not stopped — that is the caller's responsibility. A nil or unregistered resourceID is a no-op.

func (*UnionDesireInformer) RemoveEventHandler

func (u *UnionDesireInformer) RemoveEventHandler(reg cache.ResourceEventHandlerRegistration) error

RemoveEventHandler detaches the handler from every current sub-informer and forgets it so future Adds will not install it again. Passing a registration that did not originate from this union is a no-op (mirrors the cache.SharedIndexInformer contract).

type UnionKubeApplierInformers

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

UnionKubeApplierInformers is the union peer of kubeapplierinformers.KubeApplierInformers. It exposes one (UnionDesireInformer, lister) pair per *Desire type, but each pair fans out across every per-management-cluster KubeApplierInformers that has been Added.

The aggregator does not own sub-informer lifecycle: the caller starts each per-MC KubeApplierInformers with a child ctx (e.g. via RunWithContext) and pairs cancellation of that ctx with a Remove call. That layering keeps this type a pure registry — perfect substrate for a higher-level reactor that drives Add/Remove from a ManagementCluster informer.

Add and Remove are serialized by mu so two concurrent registry mutations can't interleave their four sub-registrations (two informers, two listers). Readers on the individual union informers/listers do NOT take mu — each underlying surface has its own internal lock, so each surface presents a consistent view in isolation, but a reader that consults multiple surfaces during an in-flight Add/Remove may see the sub registered on some surfaces and not on others. The reactor consuming this type (UnionKubeApplierInformersController) doesn't rely on cross-surface atomicity, which is why we keep the cheaper locking discipline.

func NewUnionKubeApplierInformers

func NewUnionKubeApplierInformers() *UnionKubeApplierInformers

NewUnionKubeApplierInformers returns an empty aggregator. Call Add to register per-management-cluster KubeApplierInformers.

func (*UnionKubeApplierInformers) Add

func (u *UnionKubeApplierInformers) Add(managementClusterResourceID *azcorearm.ResourceID, sub kubeapplierinformers.KubeApplierInformers) error

Add registers a per-management-cluster KubeApplierInformers under the given resourceID. The sub's two informers and two listers are wired into the matching union informers and listers under u.mu, so concurrent callers see the registration atomically.

Add does not start the sub-informer; the caller is responsible for that (typically by calling sub.RunWithContext on a per-MC ctx). Re-Add under the same resourceID replaces the previous registration on all four union surfaces.

A nil resourceID or nil sub is a no-op. The error is non-nil only when a sub-informer rejects the union's previously-registered handlers; in that case the partial registration on the lister side is rolled back.

func (*UnionKubeApplierInformers) ApplyDesires

func (u *UnionKubeApplierInformers) ApplyDesires() (*UnionDesireInformer, kubeapplierlisters.ApplyDesireLister)

ApplyDesires returns the union ApplyDesire informer and lister. Event handlers registered on the returned informer fan out to every per-management-cluster sub-informer (current and future).

func (*UnionKubeApplierInformers) HasSynced

func (u *UnionKubeApplierInformers) HasSynced() bool

HasSynced returns true only when every sub-informer (across both *Desire types and every registered management cluster) has synced. An empty union is vacuously synced.

func (*UnionKubeApplierInformers) ReadDesires

func (u *UnionKubeApplierInformers) ReadDesires() (*UnionDesireInformer, kubeapplierlisters.ReadDesireLister)

ReadDesires returns the union ReadDesire informer and lister.

func (*UnionKubeApplierInformers) Remove

func (u *UnionKubeApplierInformers) Remove(managementClusterResourceID *azcorearm.ResourceID)

Remove deregisters the sub-informers and sublisters previously Added for the given resourceID across all four union surfaces. The sub-informers themselves are not stopped — the caller owns that lifecycle. Unknown or nil resourceID is a no-op.

type UnionKubeApplierInformersController

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

UnionKubeApplierInformersController owns a UnionKubeApplierInformers and keeps it in sync with the set of management clusters reported by the configured management-cluster informer and lister.

Event handlers enqueue a ManagementClusterKey onto an internal workqueue. Worker goroutines pull keys and call SyncOnce, which looks the management cluster up in the lister: if found, the worker ensures a per-MC sub-informer is registered with the union; if not found, the worker removes any existing registration. Lookup happens in SyncOnce, not in the event handler, so we always act on the current lister state rather than the stale snapshot embedded in the event payload.

The controller does not own the management-cluster informer's lifecycle — the caller starts it. The controller installs an event handler in Run and removes it before returning.

func NewUnionKubeApplierInformersController

func NewUnionKubeApplierInformersController(
	mcInformer cache.SharedIndexInformer,
	mcLister fleetlisters.ManagementClusterLister,
	factory PerMCKubeApplierInformerFactory,
) *UnionKubeApplierInformersController

NewUnionKubeApplierInformersController returns a stopped controller. Call Run to start watching the management-cluster informer and wiring up per-MC sub-informers.

func (*UnionKubeApplierInformersController) Run

func (c *UnionKubeApplierInformersController) Run(ctx context.Context, threadiness int)

Run installs an event handler on the management-cluster informer, runs `threadiness` worker goroutines that process the workqueue, and blocks until ctx is cancelled. The caller is responsible for starting the management-cluster informer; Run only registers handlers on it.

On exit Run shuts down the workqueue (which unblocks the workers), waits for the workers to stop, removes the event handler, and cancels every per-MC sub-informer the controller started.

func (*UnionKubeApplierInformersController) SyncOnce

func (c *UnionKubeApplierInformersController) SyncOnce(ctx context.Context, key ManagementClusterKey) error

SyncOnce reconciles one management cluster: it looks the cluster up in the lister by full resourceID match and either ensures a per-MC sub-informer is registered (if the MC is present) or removed (if the MC is absent). The lister is queried via List rather than Get because there is no canonical management-cluster name we can pass to Get.

func (*UnionKubeApplierInformersController) Union

func (c *UnionKubeApplierInformersController) Union() *UnionKubeApplierInformers

Union returns the managed UnionKubeApplierInformers. Callers wire their event handlers and listers from this — it is updated dynamically as MCs come and go.