controller.go

package controller

import (
	"context"
	"fmt"
	"sync"
	"time"

	metricsserver "github.com/krateoplatformops/unstructured-runtime/pkg/metrics/server"

	"github.com/google/go-cmp/cmp"
	"github.com/krateoplatformops/plumbing/kubeutil/event"
	"github.com/krateoplatformops/plumbing/shortid"
	ctrlevent "github.com/krateoplatformops/unstructured-runtime/pkg/controller/event"
	"github.com/krateoplatformops/unstructured-runtime/pkg/controller/objectref"
	"github.com/krateoplatformops/unstructured-runtime/pkg/controller/priorityqueue"
	"github.com/krateoplatformops/unstructured-runtime/pkg/listwatcher"
	"github.com/krateoplatformops/unstructured-runtime/pkg/logging"
	"github.com/krateoplatformops/unstructured-runtime/pkg/meta"
	"github.com/krateoplatformops/unstructured-runtime/pkg/pluralizer"
	"github.com/krateoplatformops/unstructured-runtime/pkg/telemetry"
	"github.com/prometheus/client_golang/prometheus"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime/schema"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"k8s.io/apimachinery/pkg/util/wait"
	"k8s.io/client-go/dynamic"
	"k8s.io/client-go/tools/cache"
	"k8s.io/client-go/util/workqueue"
)

const (
	reasonReconciliationPaused event.Reason = "ReconciliationPaused"
)

const LowPriority = -100 //Low priority for the priorityqueue
const NormalPriority = 0 //Normal priority for the priorityqueue
const HighPriority = 100 //High priority for the priorityqueue

// An ExternalClient manages the lifecycle of an external resource.
// None of the calls here should be blocking. All of the calls should be
// idempotent. For example, Create call should not return AlreadyExists error
// if it's called again with the same parameters or Delete call should not
// return error if there is an ongoing deletion or resource does not exist.
type ExternalClient interface {
	Observe(ctx context.Context, mg *unstructured.Unstructured) (ExternalObservation, error)
	Create(ctx context.Context, mg *unstructured.Unstructured) error
	Update(ctx context.Context, mg *unstructured.Unstructured) error
	Delete(ctx context.Context, mg *unstructured.Unstructured) error
}

// An ExternalObservation is the result of an observation of an external resource.
type ExternalObservation struct {
	// ResourceExists must be true if a corresponding external resource exists
	// for the managed resource.
	ResourceExists bool

	// ResourceUpToDate should be true if the corresponding external resource
	// appears to be up-to-date - i.e. updating the external resource to match
	// the desired state of the managed resource would be a no-op.
	ResourceUpToDate bool
}

type ListWatcherConfiguration struct {
	LabelSelector *string
	FieldSelector *string
}

type Options struct {
	Client            dynamic.Interface
	GVR               schema.GroupVersionResource
	Namespace         string
	ResyncInterval    time.Duration
	Recorder          event.Recorder
	ThrottledRecorder event.Recorder
	Logger            logging.Logger
	Metrics           *telemetry.Metrics
	ListWatcher       ListWatcherConfiguration
	Pluralizer        pluralizer.PluralizerInterface
	GlobalRateLimiter workqueue.TypedRateLimiter[any]
	MetricsServer     metricsserver.Server
	WatchAnnotations  ctrlevent.AnnotationEvents
	MaxRetries        int
	ActionsEvent      ctrlevent.ActionsEvent
}

func (o Options) validate() error {
	if o.Client == nil {
		return fmt.Errorf("client is required")
	}
	if o.GVR.Empty() {
		return fmt.Errorf("GVR is required")
	}
	if o.Recorder == nil {
		return fmt.Errorf("recorder is required")
	}
	if o.Logger == nil {
		return fmt.Errorf("logger is required")
	}
	if o.Pluralizer == nil {
		return fmt.Errorf("pluralizer is required")
	}
	if o.GlobalRateLimiter == nil {
		return fmt.Errorf("global rate limiter is required")
	}
	if o.ResyncInterval <= 0 {
		return fmt.Errorf("resync interval must be greater than 0")
	}
	if o.MaxRetries < 0 {
		return fmt.Errorf("max retries must be greater than or equal to 0")
	}
	return nil
}

type Controller struct {
	metricsServer     metricsserver.Server
	pluralizer        pluralizer.PluralizerInterface
	dynamicClient     dynamic.Interface
	gvr               schema.GroupVersionResource
	queue             priorityqueue.PriorityQueue[any]
	items             *sync.Map
	informer          cache.Controller
	recorder          event.Recorder
	throttledRecorder event.Recorder
	logger            logging.Logger
	metrics           *telemetry.Metrics
	externalClient    ExternalClient
	maxRetries        int
}

var (
	reconcileTotal = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "controller_reconcile_total",
			Help: "Total number of reconciliations",
		},
		[]string{"kind", "namespace", "result"},
	)

	reconcileDuration = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name: "controller_reconcile_duration_seconds",
			Help: "Time spent reconciling",
		},
		[]string{"kind", "namespace"},
	)
)

func init() {
	prometheus.MustRegister(reconcileTotal)
	prometheus.MustRegister(reconcileDuration)
}

func New(sid *shortid.Shortid, opts Options) (*Controller, error) {
	if err := opts.validate(); err != nil {
		opts.Logger.Error(err, "Invalid controller options")
		return nil, fmt.Errorf("invalid controller options: %w", err)
	}
	queue := priorityqueue.New("controller", func(o *priorityqueue.Opts[any]) {
		o.RateLimiter = opts.GlobalRateLimiter
	})

	// Wrap the queue with metrics instrumentation if metrics are available
	var finalQueue priorityqueue.PriorityQueue[any] = queue
	if opts.Metrics != nil {
		finalQueue = NewInstrumentedQueue(queue, opts.Metrics)
	}
	items := &sync.Map{}

	lw, err := listwatcher.Create(listwatcher.CreateOption{
		Client:        opts.Client,
		GVR:           opts.GVR,
		LabelSelector: opts.ListWatcher.LabelSelector,
		FieldSelector: opts.ListWatcher.FieldSelector,
		Namespace:     opts.Namespace,
	})
	if err != nil {
		opts.Logger.Error(err, "Failed to create listwatcher.")
		return nil, fmt.Errorf("failed to create listwatcher: %w", err)
	}

	_, informer := cache.NewInformerWithOptions(cache.InformerOptions{
		ListerWatcher: lw,
		ObjectType:    &unstructured.Unstructured{},
		ResyncPeriod:  opts.ResyncInterval,
		Indexers:      cache.Indexers{},
		Handler: cache.ResourceEventHandlerFuncs{
			AddFunc: func(obj interface{}) {
				log := opts.Logger
				el, ok := obj.(*unstructured.Unstructured)
				if !ok {
					log.Warn("Object is not an unstructured.")
					return
				}

				item := ctrlevent.Event{
					EventType: opts.ActionsEvent.GetEventType(ctrlevent.CRCreated),
					ObjectRef: objectref.ObjectRef{
						APIVersion: el.GetAPIVersion(),
						Kind:       el.GetKind(),
						Name:       el.GetName(),
						Namespace:  el.GetNamespace(),
					},
					QueuedAt: time.Now(),
				}
				dig := ctrlevent.DigestForEvent(item)

				// Checking if the object is already being processed
				priority := NormalPriority
				annotations := el.GetAnnotations()
				if annotations == nil {
					priority = NormalPriority
				}
				_, ok = annotations[meta.AnnotationKeyExternalCreateFailed]
				_, ok_pending := annotations[meta.AnnotationKeyExternalCreatePending]
				_, ok_succedeed := annotations[meta.AnnotationKeyExternalCreateSucceeded]
				if ok || ok_pending || ok_succedeed {
					priority = LowPriority //These are events that are already being processed, so we can lower the priority
				}

				if _, loaded := items.LoadOrStore(dig, struct{}{}); !loaded {
					log.WithValues(
						"kind", item.ObjectRef.Kind,
						"apiVersion", item.ObjectRef.APIVersion,
						"name", item.ObjectRef.Name,
						"namespace", item.ObjectRef.Namespace,
						"queuedAt", item.QueuedAt,
					).Debug("Adding Observe event to queue", "priority", priority)
					// Generally a user event, occurs when a user creates a resource or the resource is first seen
					// by the controller. We want to process these events as soon as possible, so we add them to the front of the queue.
					// However, if the resource has an annotation that indicates it is being processed, we lower the priority.
					queue.AddWithOpts(priorityqueue.AddOpts{
						RateLimited: false,
						Priority:    priority,
					}, item)
				}
			},
			UpdateFunc: func(old, new interface{}) {
				log := opts.Logger
				oldUns, ok := old.(*unstructured.Unstructured)
				if !ok {
					log.Warn("Object is not an unstructured.")
					return
				}

				newUns, ok := new.(*unstructured.Unstructured)
				if !ok {
					log.Warn("Object is not an unstructured.")
					return
				}

				if meta.WasDeleted(newUns) {
					log.Debug(fmt.Sprintf("Object %s/%s is being deleted", newUns.GetNamespace(), newUns.GetName()))

					item := ctrlevent.Event{
						EventType: opts.ActionsEvent.GetEventType(ctrlevent.CRDeleted),
						ObjectRef: objectref.ObjectRef{
							APIVersion: newUns.GetAPIVersion(),
							Kind:       newUns.GetKind(),
							Name:       newUns.GetName(),
							Namespace:  newUns.GetNamespace(),
						},
						QueuedAt: time.Now(),
					}

					dig := ctrlevent.DigestForEvent(item)

					if _, loaded := items.LoadOrStore(dig, struct{}{}); !loaded {
						log.WithValues(
							"kind", item.ObjectRef.Kind,
							"apiVersion", item.ObjectRef.APIVersion,
							"name", item.ObjectRef.Name,
							"namespace", item.ObjectRef.Namespace,
							"queuedAt", item.QueuedAt,
						).Debug("Adding Delete event to queue", "priority", HighPriority)
						// Generally this is a pending delete event, where the resource has a deletion timestamp but was not effectively deleted yet.
						// We want to process these events as soon as possible, so we add the event with high priority.
						// This is a user event, so we want to process it quickly.
						queue.AddWithOpts(priorityqueue.AddOpts{
							RateLimited: false,
							Priority:    HighPriority,
						}, item)
					}
					return
				}

				if len(opts.WatchAnnotations) > 0 {
					// Check if any of the annotations we are watching have changed
					for _, event := range opts.WatchAnnotations {
						oldValue, oldExists := oldUns.GetAnnotations()[event.Annotation]
						newValue, newExists := newUns.GetAnnotations()[event.Annotation]

						deletedCondition := !newExists && oldExists
						createdCondition := newExists && !oldExists
						changedCondition := oldExists && newExists && !cmp.Equal(oldValue, newValue)
						anyConditions := deletedCondition || createdCondition || changedCondition

						trigger := false
						action := event.OnAction
						if action == ctrlevent.OnDelete && deletedCondition ||
							action == ctrlevent.OnCreate && createdCondition ||
							action == ctrlevent.OnChange && changedCondition ||
							action == ctrlevent.OnAny && anyConditions {
							trigger = true
						}
						if trigger {
							item := ctrlevent.Event{
								EventType: event.EventType,
								ObjectRef: objectref.ObjectRef{
									APIVersion: newUns.GetAPIVersion(),
									Kind:       newUns.GetKind(),
									Name:       newUns.GetName(),
									Namespace:  newUns.GetNamespace(),
								},
								QueuedAt: time.Now(),
							}

							log.WithValues(
								"kind", item.ObjectRef.Kind,
								"apiVersion", item.ObjectRef.APIVersion,
								"name", item.ObjectRef.Name,
								"namespace", item.ObjectRef.Namespace,
								"queuedAt", item.QueuedAt,
							).Debug("Adding event to queue for annotation change", "priority", NormalPriority, "eventType", event.EventType, "annotation", event.Annotation)
							// Generally a user event, occurs when a user changes an annotation we are watching. We want to process these events as soon as possible, so we add them to the front of the queue.
							// We use normal priority because these are user events that should be processed quickly, but they are not as urgent as create or delete events.
							// This is a user event, so we want to process it quickly.
							queue.AddWithOpts(priorityqueue.AddOpts{
								RateLimited: false,
								Priority:    NormalPriority,
							}, item)

							return
						}
					}
				}

				// Check if ResourceVersion changed to distinguish between:
				// 1. Periodic resync (same ResourceVersion) - should queue Observe
				// 2. Status-only update (different ResourceVersion, same spec) - should IGNORE
				// 3. Spec change (different ResourceVersion, different spec) - should queue Update
				oldResourceVersion := oldUns.GetResourceVersion()
				newResourceVersion := newUns.GetResourceVersion()

				newSpec, _, err := unstructured.NestedMap(newUns.Object, "spec")
				if err != nil {
					log.Error(err, "getting new object spec")
					return
				}

				oldSpec, _, err := unstructured.NestedMap(oldUns.Object, "spec")
				if err != nil {
					log.Error(err, "getting old object spec")
					return
				}

				diff := cmp.Diff(newSpec, oldSpec)

				if len(diff) > 0 {
					// Spec changed - user-initiated update
					item := ctrlevent.Event{
						EventType: opts.ActionsEvent.GetEventType(ctrlevent.CRUpdated),
						ObjectRef: objectref.ObjectRef{
							APIVersion: newUns.GetAPIVersion(),
							Kind:       newUns.GetKind(),
							Name:       newUns.GetName(),
							Namespace:  newUns.GetNamespace(),
						},
						QueuedAt: time.Now(),
					}

					dig := ctrlevent.DigestForEvent(item)

					if _, loaded := items.LoadOrStore(dig, struct{}{}); !loaded {
						log.WithValues(
							"kind", item.ObjectRef.Kind,
							"apiVersion", item.ObjectRef.APIVersion,
							"name", item.ObjectRef.Name,
							"namespace", item.ObjectRef.Namespace,
							"queuedAt", item.QueuedAt,
						).Debug("Adding Update event to queue (spec changed)", "priority", HighPriority)
						queue.AddWithOpts(priorityqueue.AddOpts{
							RateLimited: false,
							Priority:    HighPriority,
						}, item)
					}
				} else if oldResourceVersion == newResourceVersion {
					// Periodic resync from informer - ResourceVersion unchanged
					item := ctrlevent.Event{
						EventType: opts.ActionsEvent.GetEventType(ctrlevent.CRObserved),
						ObjectRef: objectref.ObjectRef{
							APIVersion: newUns.GetAPIVersion(),
							Kind:       newUns.GetKind(),
							Name:       newUns.GetName(),
							Namespace:  newUns.GetNamespace(),
						},
						QueuedAt: time.Now(),
					}

					dig := ctrlevent.DigestForEvent(item)

					if _, loaded := items.LoadOrStore(dig, struct{}{}); !loaded {
						log.WithValues(
							"kind", item.ObjectRef.Kind,
							"apiVersion", item.ObjectRef.APIVersion,
							"name", item.ObjectRef.Name,
							"namespace", item.ObjectRef.Namespace,
							"queuedAt", item.QueuedAt,
						).Debug("Adding Observe event to queue (periodic resync)", "priority", LowPriority)
						queue.AddWithOpts(priorityqueue.AddOpts{
							RateLimited: false,
							Priority:    LowPriority,
						}, item)
					}
				} else {
					// ResourceVersion changed but spec didn't - this is a status-only update from the controller itself
					// IGNORE to prevent self-triggering loop
					log.WithValues(
						"kind", newUns.GetKind(),
						"name", newUns.GetName(),
						"namespace", newUns.GetNamespace(),
						"oldResourceVersion", oldResourceVersion,
						"newResourceVersion", newResourceVersion,
					).Debug("Ignoring status-only update")
				}

			},
			DeleteFunc: func(obj interface{}) {
				log := opts.Logger

				// Attempt to cast the object to *unstructured.Unstructured
				el, ok := obj.(*unstructured.Unstructured)
				if !ok {
					// If the cast fails, check if it's a Tombstone (DeletedFinalStateUnknown)
					tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
					if !ok {
						log.Warn("Failed to recover object from DeleteFunc: unknown type")
						return
					}
					// Recover the last known state of the object from the tombstone
					el, ok = tombstone.Obj.(*unstructured.Unstructured)
					if !ok {
						log.Warn("Tombstone does not contain an unstructured object")
						return
					}
				}

				if el.GetDeletionTimestamp() == nil {
					log.WithValues(
						"name", el.GetName(),
						"apiVersion", el.GetAPIVersion(),
						"kind", el.GetKind(),
						"namespace", el.GetNamespace(),
					).Info("Object exited controller control without deletion request. Skipping external resource cleanup.")
					return
				}

				log.Debug(fmt.Sprintf("Deleting object %s/%s", el.GetNamespace(), el.GetName()))

				item := ctrlevent.Event{
					EventType: opts.ActionsEvent.GetEventType(ctrlevent.CRDeleted),
					ObjectRef: objectref.ObjectRef{
						APIVersion: el.GetAPIVersion(),
						Kind:       el.GetKind(),
						Name:       el.GetName(),
						Namespace:  el.GetNamespace(),
					},
					QueuedAt: time.Now(),
				}

				log.WithValues(
					"kind", item.ObjectRef.Kind,
					"apiVersion", item.ObjectRef.APIVersion,
					"name", item.ObjectRef.Name,
					"namespace", item.ObjectRef.Namespace,
					"queuedAt", item.QueuedAt,
				).Debug("Adding Delete event to queue")
				// Generally this is a delete event where the resource is already gone. We want to process these events as soon as possible, so we add the event with high priority.
				// This is a user event, so we want to process it quickly.
				queue.AddWithOpts(priorityqueue.AddOpts{
					RateLimited: false,
					Priority:    HighPriority,
				}, item)
			},
		},
	})
	return &Controller{
		dynamicClient:     opts.Client,
		gvr:               opts.GVR,
		items:             items,
		recorder:          opts.Recorder,
		throttledRecorder: opts.ThrottledRecorder,
		logger:            opts.Logger,
		metrics:           opts.Metrics,
		informer:          informer,
		queue:             finalQueue,
		pluralizer:        opts.Pluralizer,
		metricsServer:     opts.MetricsServer,
		maxRetries:        opts.MaxRetries,
	}, nil
}

func (c *Controller) SetExternalClient(ec ExternalClient) {
	c.externalClient = ec
}

// Run begins watching and syncing.
func (c *Controller) Run(ctx context.Context, numWorkers int) error {
	defer utilruntime.HandleCrash()
	defer c.queue.ShutDown()

	c.logger.Info("Starting controller")
	go c.informer.Run(ctx.Done())

	// Start metrics server in goroutine so it doesn't block
	if c.metricsServer != nil {
		go func() {
			if err := c.metricsServer.WithLogger(c.logger).Start(ctx); err != nil {
				c.logger.Error(err, "metrics server failed")
			}
		}()
	}

	// Wait for all involved caches to be synced, before
	// processing items from the queue is started
	c.logger.Info("waiting for informer caches to sync")
	if !cache.WaitForCacheSync(ctx.Done(), c.informer.HasSynced) {
		err := fmt.Errorf("failed to wait for informers caches to sync")
		utilruntime.HandleError(err)
		return err
	}

	c.logger.Info(fmt.Sprintf("Starting workers: %d", numWorkers))
	for i := 0; i < numWorkers; i++ {
		go wait.Until(func() {
			c.runWorker(ctx)
		}, 2*time.Second, ctx.Done())
	}
	c.logger.Info("Controller ready.")

	<-ctx.Done()
	c.logger.Info("Stopping controller.")

	return nil
}