dynamiccontroller

type Config ¶

type Config struct {
	// Workers specifies the number of workers processing items from the queue
	Workers int
	// ResyncPeriod defines the interval at which the controller will re list
	// the resources, even if there haven't been any changes.
	ResyncPeriod time.Duration
	// QueueMaxRetries is the maximum number of retries for an item in the queue
	// will be retried before being dropped.
	//
	// NOTE(a-hilaly): I'm not very sure how useful is this, i'm trying to avoid
	// situations where reconcile errors exhaust the queue.
	QueueMaxRetries int
	// MinRetryDelay is the minimum delay before retrying an item in the queue
	MinRetryDelay time.Duration
	// MaxRetryDelay is the maximum delay before retrying an item in the queue
	MaxRetryDelay time.Duration
	// RateLimit is the maximum number of events processed per second
	RateLimit int
	// BurstLimit is the maximum number of events in a burst
	BurstLimit int
	// QueueShutdownTimeout is the maximum time to wait for the queue to drain before shutting down.
	QueueShutdownTimeout time.Duration
}

Config holds the configuration for DynamicController

type DynamicController ¶

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

DynamicController (DC) is a single controller capable of managing multiple different kubernetes resources (GVRs) in parallel. It can safely start watching new resources and stop watching others at runtime - hence the term "dynamic". This flexibility allows us to accept and manage various resources in a Kubernetes cluster without requiring restarts or pod redeployments.

It is mainly inspired by native Kubernetes controllers but designed for more flexible and lightweight operation. DC serves as the core component of kro's dynamic resource management system. Its primary purpose is to create and manage "micro" controllers for custom resources defined by users at runtime (via the ResourceGraphDefinition CRs).

The DynamicController uses a layered architecture:

1. WatchManager: manages informer lifecycle per GVR (owner-set based).

2. WatchCoordinator: aggregates watch requests from all instance reconcilers, maintains reverse indexes for event routing, and manages shared watches on the WatchManager.

3. DynamicController: orchestrates parent watches (one per RGD), the work queue, and handler dispatch. Child/external resource watches are handled by the coordinator based on requests from instance reconcilers.

func NewDynamicController(
	log logr.Logger,
	config Config,
	kubeClient k8smetadata.Interface,
	mapper meta.RESTMapper,
) *DynamicController

func (dc *DynamicController) Coordinator() *WatchCoordinator

func (dc *DynamicController) Deregister(_ context.Context, parent schema.GroupVersionResource) error

func (dc *DynamicController) Register(
	_ context.Context,
	parent schema.GroupVersionResource,
	instanceHandler Handler,
) error

func (dc *DynamicController) Start(ctx context.Context) error

type EnqueueFunc ¶ added in v0.9.0

type EnqueueFunc func(parentGVR schema.GroupVersionResource, instance types.NamespacedName)

EnqueueFunc is called by the coordinator to enqueue an instance for re-reconciliation when one of its watched resources changes.

type Event ¶ added in v0.9.0

type Event struct {
	Type      EventType
	GVR       schema.GroupVersionResource
	Name      string
	Namespace string
	Labels    map[string]string
	// OldLabels holds the labels from the previous version of the object
	// (populated only for update events). Used by collection watches to detect
	// label changes that cause an object to enter or leave a selector match.
	OldLabels map[string]string
}

Event is a normalized watch event emitted by the WatchManager. Consumers decide what to act on -- no old/new comparison or generation filtering is performed by the watch layer.

type EventHandler ¶ added in v0.9.0

type EventHandler func(event Event)

EventHandler processes a single watch event.

type EventType ¶ added in v0.9.0

type EventType string

EventType identifies the kind of change that triggered an event.

const (
	EventAdd    EventType = "add"
	EventUpdate EventType = "update"
	EventDelete EventType = "delete"
)

type Handler ¶

type Handler func(ctx context.Context, req ctrl.Request) error

Handler is used to actually perform the reconciliation logic for an instance GVR and will operate on a single instance of the resource received from the queue

type InstanceWatcher ¶ added in v0.9.0

type InstanceWatcher interface {
	// Watch requests that the instance be re-reconciled when the specified
	// resource changes. Call this for every resource (managed or external)
	// the instance cares about.
	//
	// For scalar resources: set Name + Namespace.
	// For collections: set Selector + Namespace.
	//
	// Call Watch() BEFORE operating on the resource to avoid event gaps.
	Watch(req WatchRequest) error

	// Done signals that all Watch() calls for this reconciliation cycle
	// are complete. Any watch requests from the previous cycle that were
	// NOT re-requested are automatically cleaned up. If commit is false, the
	// current cycle is discarded and the previously committed watch set stays
	// active.
	Done(commit bool)
}

InstanceWatcher is the interface the instance reconciler uses to request watches. It is scoped to a single instance and obtained via WatchCoordinator.ForInstance().

type NoopInstanceWatcher ¶ added in v0.9.0

type NoopInstanceWatcher struct{}

NoopInstanceWatcher is a no-op implementation of InstanceWatcher for use in tests or when no coordinator is available.

func (NoopInstanceWatcher) Done(bool)

func (NoopInstanceWatcher) Watch(_ WatchRequest) error

type ObjectIdentifiers ¶

type ObjectIdentifiers struct {
	types.NamespacedName
	GVR schema.GroupVersionResource
}

ObjectIdentifiers holds the key and GVR of the object to reconcile.

type WatchCoordinator ¶ added in v0.9.0

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

WatchCoordinator aggregates watch requests from all instances, manages shared watches via WatchManager, and routes events back to the correct instances.

func NewWatchCoordinator(watches *WatchManager, enqueue EnqueueFunc, log logr.Logger) *WatchCoordinator

func (c *WatchCoordinator) ForInstance(parentGVR schema.GroupVersionResource, instance types.NamespacedName) InstanceWatcher

func (c *WatchCoordinator) InstanceWatchCount() int

func (c *WatchCoordinator) RemoveInstance(parentGVR schema.GroupVersionResource, instance types.NamespacedName)

func (c *WatchCoordinator) RemoveParentGVR(parentGVR schema.GroupVersionResource)

func (c *WatchCoordinator) RouteEvent(event Event)

func (c *WatchCoordinator) WatchRequestCount() (scalar, collection int)

type WatchManager ¶ added in v0.9.0

type WatchManager struct {

	// SyncTimeout is the maximum time to wait for cache sync in EnsureWatch.
	// Zero means use the default (30s).
	SyncTimeout time.Duration
	// contains filtered or unexported fields
}

WatchManager manages informer lifecycle per GVR. Informers start lazily on first use and stay alive until all owners release them or Shutdown() is called.

func NewWatchManager(client metadata.Interface, resync time.Duration, onEvent EventHandler, log logr.Logger) *WatchManager

func (m *WatchManager) ActiveWatchCount() int

func (m *WatchManager) EnsureWatch(gvr schema.GroupVersionResource, ownerID string) error

func (m *WatchManager) GetInformer(gvr schema.GroupVersionResource) cache.SharedIndexInformer

func (m *WatchManager) ReleaseWatch(gvr schema.GroupVersionResource, ownerID string)

func (m *WatchManager) Shutdown()

type WatchRequest ¶ added in v0.9.0

type WatchRequest struct {
	// NodeID is the graph node ID (for debugging/metrics).
	NodeID string
	// GVR is the GroupVersionResource to watch.
	GVR schema.GroupVersionResource
	// Name is the specific resource name (scalar watches).
	Name string
	// Namespace is the resource namespace. Empty for cluster-scoped resources.
	Namespace string
	// Selector is a label selector for collection watches. nil means scalar watch.
	Selector labels.Selector
}

WatchRequest describes a resource the instance reconciler wants to watch. For scalar resources, set Name + Namespace. For collections, set Selector + Namespace. Selector supports both matchLabels and matchExpressions (the full metav1.LabelSelector spec).