Reconciler Logic

// AppInformerConfig contains configuration for the App's internal operator.InformerController
type AppInformerConfig struct {
	// InformerOptions are the options for the informer.
	InformerOptions operator.InformerOptions
	// RetryPolicy is the policy for retrying events.
	RetryPolicy operator.RetryPolicy
	// RetryDequeuePolicy is the policy for dequeuing events.
	RetryDequeuePolicy operator.RetryDequeuePolicy
	// FinalizerSupplier is used to generate the finalizer for the kind.
	FinalizerSupplier operator.FinalizerSupplier
	// InProgressFinalizerSupplier is used to generate the "in-progress" finalizer used by opinionated adds,
	// before the "normal" finalizer (provided by FinalizerSupplier) is applied when the add completes successfully.
	// By default, this is "<app name>-wip"
	InProgressFinalizerSupplier operator.FinalizerSupplier
	// InformerSupplier can be set to specify a function for creating informers for kinds.
	// If left unset, DefaultInformerSupplier will be used.
	InformerSupplier InformerSupplier
}

// InformerOptions are generic options for all Informer implementations.
type InformerOptions struct {
	// ListWatchOptions are the options for filtering the watch based on namespace and other compatible filters.
	ListWatchOptions ListWatchOptions
	// CacheResyncInterval is the interval at which the informer will emit CacheResync events for all resources in the cache.
	// This is distinct from a full resync, as no information is fetched from the API server.
	// Changes to this value after run() is called will not take effect.
	// An empty value will disable cache resyncs.
	CacheResyncInterval time.Duration
	// EventTimeout is the timeout for an event to be processed.
	// If an event is not processed within this timeout, it will be dropped.
	// The timeout cannot be larger than the cache resync interval, if it is,
	// the cache resync interval will be used instead.
	// An empty value will disable event timeouts.
	EventTimeout time.Duration
	// ErrorHandler is called if the informer encounters an error which does not stop the informer from running,
	// but may stop it from processing a given event.
	ErrorHandler func(context.Context, error)
	// HealthCheckIgnoreSync will set the KubernetesBasedInformer HealthCheck to return ok once the informer is started,
	// rather than waiting for the informer to finish with its initial list sync.
	// You may want to set this to `true` if you have a particularly long initial sync period and don't want readiness checks failing.
	HealthCheckIgnoreSync bool
	// UseWatchList if turned on instructs the reflector to open a stream to bring data from the API server.
	// Streaming has the primary advantage of using fewer server's resources to fetch data.
	//
	// The old behavior establishes a LIST request which gets data in chunks.
	// Paginated list is less efficient and depending on the actual size of objects
	// might result in an increased memory consumption of the APIServer.
	//
	// Defaults to false. Requires Kubernetes 1.27+ when enabled.
	// See https://github.com/kubernetes/enhancements/tree/master/keps/sig-api-machinery/3157-watch-list#design-details
	UseWatchList bool
	// WatchListPageSize is the requested chunk size for paginated LIST operations.
	// This significantly reduces memory usage when watching large numbers of objects (>10K).
	// Recommended values: 5000-10000 for most use cases.
	// Note: This only affects traditional LIST operations. It does NOT apply to watch-list streaming (UseWatchList).
	// An empty value (0) will use client-go's default pagination behavior based on resource version.
	WatchListPageSize int64
	// MaxConcurrentWorkers is the maximum number of concurrent workers for event processing in ConcurrentInformer.
	// Each worker maintains a queue of events which are processed sequentially.
	// Events for a particular object are assigned to the same worker to maintain in-order delivery per object.
	// An empty value (0) will use the default of 10 workers.
	MaxConcurrentWorkers uint64
}

// RetryPolicy is a function that defines whether an event should be retried, based on the error and number of attempts.
// It returns a boolean indicating whether another attempt should be made, and a time.Duration after which that attempt should be made again.
type RetryPolicy func(err error, attempt int) (bool, time.Duration)

// RetryDequeuePolicy is a function that defines when a retry should be dequeued when a new action is taken on a resource.
// It accepts information about the new action being taken, and information about the current queued retry,
// and returns `true` if the retry should be dequeued.
// A RetryDequeuePolicy may be called multiple times for the same action, depending on the number of pending retries for the object.
type RetryDequeuePolicy func(newAction ResourceAction, newObject resource.Object, retryAction ResourceAction, retryObject resource.Object, retryError error) bool

// OpinionatedRetryDequeuePolicy is a RetryDequeuePolicy which has the following logic:
// 1. If the newAction is a delete, dequeue the retry
// 2. If the newAction and retryAction are different, keep the retry (for example, a queued create retry, and a received update action)
// 3. If the generation of newObject and retryObject is the same, keep the retry
// 4. Otherwise, dequeue the retry
var OpinionatedRetryDequeuePolicy = func(newAction ResourceAction, newObject resource.Object, retryAction ResourceAction, retryObject resource.Object, _ error) bool {
	if newAction == ResourceActionDelete {
		return true
	}
	if newAction != retryAction {
		return false
	}
	if newObject.GetGeneration() == retryObject.GetGeneration() {
		return false
	}
	return true
}

// FinalizerSupplier represents a function that creates string finalizer from provider schema.
type FinalizerSupplier func(sch resource.Schema) string

// DefaultFinalizerSupplier crates finalizer following to pattern `operator.{version}.{kind}.{group}`.
func DefaultFinalizerSupplier(sch resource.Schema) string {
	return fmt.Sprintf("operator.%s.%s.%s", sch.Version(), sch.Kind(), sch.Group())
}

func InProgressFinalizerSupplier(sch resource.Schema) string {
	return fmt.Sprintf("wip.%s.%s.%s", sch.Version(), sch.Kind(), sch.Group())
}

// InformerSupplier is a function which creates an operator.Informer for a kind, given a ClientGenerator and ListWatchOptions
type InformerSupplier func(
	kind resource.Kind, clients resource.ClientGenerator, options operator.InformerOptions,
) (operator.Informer, error)

// DefaultInformerSupplier is a default InformerSupplier function which creates a basic operator.KubernetesBasedInformer.
// Note: This supplier does NOT respect WatchListPageSize or UseWatchList configuration.
// For memory-optimized informers that support these options, use OptimizedInformerSupplier instead.
var DefaultInformerSupplier = func(
	kind resource.Kind, clients resource.ClientGenerator, options operator.InformerOptions,
) (operator.Informer, error) {
	client, err := clients.ClientFor(kind)
	if err != nil {
		return nil, err
	}

	inf, err := operator.NewKubernetesBasedInformer(kind, client, options)
	if err != nil {
		return nil, err
	}

	return operator.NewConcurrentInformerFromOptions(inf, options)
}

simple.AppConfig{
    InformerConfig: simple.AppInformerConfig{
        InformerSupplier: simple.OptimizedInformerSupplier,
        InformerOptions: operator.InformerOptions{
            UseWatchList:      true,  // streaming, requires Kubernetes 1.27+
            WatchListPageSize: 5000,  // chunk size for paginated LIST fallback
        },
    },
    // ...
}