catalog-operator: Delete Pods that were evicted (operator-framework#3459)

This change adds another reason why a Pod could be detected as "dead",
namely when it was evicted by the kubelet. This can happen when there is
resource pressure on the Node. Then the reason will be
"TerminationByKubelet".

This addresses the issue described in
https://issues.redhat.com/browse/OCPBUGS-45490

Signed-off-by: Simon Krenger <[email protected]>
Signed-off-by: Per Goncalves da Silva <[email protected]>

Author: simonkrenger

pkg/controller/operators/catalog/operator.go

@@ -238,6 +238,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 						Namespace:   csv.Namespace,
 						Labels:      csv.Labels,
 						Annotations: csv.Annotations,
+						UID:         csv.UID,
 					},
 					Spec: v1alpha1.ClusterServiceVersionSpec{
 						CustomResourceDefinitions: csv.Spec.CustomResourceDefinitions,
@@ -745,16 +746,21 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 	// Namespace sync for resolving subscriptions
 	namespaceInformer := informers.NewSharedInformerFactory(op.opClient.KubernetesInterface(), resyncPeriod()).Core().V1().Namespaces()
 	op.lister.CoreV1().RegisterNamespaceLister(namespaceInformer.Lister())
-	op.nsResolveQueue = workqueue.NewTypedRateLimitingQueueWithConfig[any](workqueue.DefaultTypedControllerRateLimiter[any](),
+	op.nsResolveQueue = workqueue.NewTypedRateLimitingQueueWithConfig[any](
+		workqueue.NewTypedItemExponentialFailureRateLimiter[any](1*time.Second, 30*time.Second),
 		workqueue.TypedRateLimitingQueueConfig[any]{
 			Name: "resolve",
 		})
+	//op.nsResolveQueue = workqueue.NewTypedRateLimitingQueueWithConfig[any](workqueue.DefaultTypedControllerRateLimiter[any](),
+	//	workqueue.TypedRateLimitingQueueConfig[any]{
+	//		Name: "resolve",
+	//	})
 	namespaceQueueInformer, err := queueinformer.NewQueueInformer(
 		ctx,
 		queueinformer.WithLogger(op.logger),
 		queueinformer.WithQueue(op.nsResolveQueue),
 		queueinformer.WithInformer(namespaceInformer.Informer()),
-		queueinformer.WithSyncer(queueinformer.LegacySyncHandler(op.syncResolvingNamespace).ToSyncer()),
+		queueinformer.WithSyncer(queueinformer.LegacySyncHandler(op.syncResolvingNamespace).ToSyncerWithDelete(op.handleDeletion)),
 	)
 	if err != nil {
 		return nil, err
@@ -1313,6 +1319,9 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 		// from users/admins. Resyncing the namespace again is unlikely to resolve
 		// not-satisfiable error
 		if _, ok := err.(solver.NotSatisfiable); ok {
+			if err := o.ResyncInformers(); err != nil {
+				logger.WithError(err).Infof("error resyncing informers")
+			}
 			logger.WithError(err).Debug("resolution failed")
 			_, updateErr := o.updateSubscriptionStatuses(
 				o.setSubsCond(subs, v1alpha1.SubscriptionCondition{
@@ -1325,7 +1334,7 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 				logger.WithError(updateErr).Debug("failed to update subs conditions")
 				return updateErr
 			}
-			return nil
+			return err
 		}
 
 		_, updateErr := o.updateSubscriptionStatuses(
@@ -1736,7 +1745,8 @@ func (o *Operator) setSubsCond(subs []*v1alpha1.Subscription, cond v1alpha1.Subs
 
 	for _, sub := range subs {
 		subCond := sub.Status.GetCondition(cond.Type)
-		if subCond.Equals(cond) {
+
+		if subCond.Type == cond.Type && subCond.Status == cond.Status && subCond.Reason == cond.Reason {
 			continue
 		}
 		sub.Status.LastUpdated = lastUpdated

pkg/controller/registry/reconciler/grpc.go

@@ -531,6 +531,7 @@ func imageChanged(logger *logrus.Entry, updatePod *corev1.Pod, servingPods []*co
 func isPodDead(pod *corev1.Pod) bool {
 	for _, check := range []func(*corev1.Pod) bool{
 		isPodDeletedByTaintManager,
+		isPodTerminatedByKubelet,
 	} {
 		if check(pod) {
 			return true
@@ -551,6 +552,19 @@ func isPodDeletedByTaintManager(pod *corev1.Pod) bool {
 	return false
 }
 
+// This reason is set when the Pod was evicted due to resource pressure on the Node
+func isPodTerminatedByKubelet(pod *corev1.Pod) bool {
+	if pod.DeletionTimestamp == nil {
+		return false
+	}
+	for _, condition := range pod.Status.Conditions {
+		if condition.Type == corev1.DisruptionTarget && condition.Reason == "TerminationByKubelet" && condition.Status == corev1.ConditionTrue {
+			return true
+		}
+	}
+	return false
+}
+
 // imageID returns the ImageID of the primary catalog source container or an empty string if the image ID isn't available yet.
 // Note: the pod must be running and the container in a ready status to return a valid ImageID.
 func imageID(pod *corev1.Pod) string {

pkg/controller/registry/resolver/source_csvs.go

@@ -40,6 +40,9 @@ func (csp *csvSourceProvider) Sources(namespaces ...string) map[cache.SourceKey]
 			listSubscriptions: func(ctx context.Context) (*v1alpha1.SubscriptionList, error) {
 				return csp.client.OperatorsV1alpha1().Subscriptions(namespace).List(ctx, metav1.ListOptions{})
 			},
+			//getCSV: func(ctx context.Context, namespace string, name string) (*v1alpha1.ClusterServiceVersion, error) {
+			//	return csp.client.OperatorsV1alpha1().ClusterServiceVersions(namespace).Get(ctx, name, metav1.GetOptions{})
+			//},
 		}
 		break // first ns is assumed to be the target ns, todo: make explicit
 	}
@@ -54,6 +57,7 @@ type csvSource struct {
 	logger    logrus.StdLogger
 
 	listSubscriptions func(context.Context) (*v1alpha1.SubscriptionList, error)
+	// getCSV            func(ctx context.Context, namespace string, name string) (*v1alpha1.ClusterServiceVersion, error)
 }
 
 func (s *csvSource) Snapshot(ctx context.Context) (*cache.Snapshot, error) {
@@ -93,19 +97,26 @@ func (s *csvSource) Snapshot(ctx context.Context) (*cache.Snapshot, error) {
 			continue
 		}
 
-		if cachedSubscription, ok := csvSubscriptions[csv]; !ok || cachedSubscription == nil {
-			// we might be in an incoherent state, so let's check with live clients to make sure
-			realSubscriptions, err := s.listSubscriptions(ctx)
-			if err != nil {
-				return nil, fmt.Errorf("failed to list subscriptions: %w", err)
-			}
-			for _, realSubscription := range realSubscriptions.Items {
-				if realSubscription.Status.InstalledCSV == csv.Name {
-					// oops, live cluster state is coherent
-					return nil, fmt.Errorf("lister caches incoherent for CSV %s/%s - found owning Subscription %s/%s", csv.Namespace, csv.Name, realSubscription.Namespace, realSubscription.Name)
-				}
-			}
-		}
+		//if cachedSubscription, ok := csvSubscriptions[csv]; !ok || cachedSubscription == nil {
+		//	// we might be in an incoherent state, so let's check with live clients to make sure
+		//	realSubscriptions, err := s.listSubscriptions(ctx)
+		//	if err != nil {
+		//		return nil, fmt.Errorf("failed to list subscriptions: %w", err)
+		//	}
+		//	for _, realSubscription := range realSubscriptions.Items {
+		//		if realSubscription.Status.InstalledCSV == csv.Name {
+		//			// oops, live cluster state is coherent
+		//			return nil, fmt.Errorf("lister caches incoherent for CSV %s/%s - found owning Subscription %s/%s", csv.Namespace, csv.Name, realSubscription.Namespace, realSubscription.Name)
+		//		}
+		//	}
+		//	realCsv, err := s.getCSV(ctx, csv.Namespace, csv.Name)
+		//	if err != nil {
+		//		return nil, fmt.Errorf("lister caches might be incoherent for CSV %s/%s: %w", csv.Namespace, csv.Name, err)
+		//	}
+		//	if realCsv.GetUID() != csv.GetUID() {
+		//		return nil, fmt.Errorf("lister caches incoherent for CSV %s/%s: differing UIDs (%s != %s)", csv.Namespace, csv.Name, csv.UID)
+		//	}
+		//}
 
 		if failForwardEnabled {
 			replacementChainEndsInFailure, err := isReplacementChainThatEndsInFailure(csv, ReplacementMapping(csvs))

pkg/controller/registry/resolver/step_resolver.go

@@ -91,7 +91,7 @@ func NewOperatorStepResolver(lister operatorlister.OperatorLister, client versio
 }
 
 func (r *OperatorStepResolver) ResolveSteps(namespace string) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
-	subs, err := r.listSubscriptions(namespace)
+	subs, err := r.subLister.Subscriptions(namespace).List(labels.Everything())
 	if err != nil {
 		return nil, nil, nil, err
 	}

pkg/lib/queueinformer/queueinformer.go

@@ -2,7 +2,6 @@ package queueinformer
 
 import (
 	"context"
-
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"k8s.io/client-go/tools/cache"

pkg/lib/queueinformer/queueinformer_operator.go

@@ -54,6 +54,8 @@ type Operator interface {
 	// RunInformers starts the Operator's underlying Informers.
 	RunInformers(ctx context.Context)
 
+	ResyncInformers() error
+
 	// Run starts the Operator and its underlying Informers.
 	Run(ctx context.Context)
 }
@@ -177,6 +179,17 @@ func (o *operator) RunInformers(ctx context.Context) {
 	})
 }
 
+func (o *operator) ResyncInformers() error {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	for _, informer := range o.informers {
+		if err := informer.GetStore().Resync(); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 // Run starts the operator's control loops.
 func (o *operator) Run(ctx context.Context) {
 	o.reconcileOnce.Do(func() {
@@ -314,7 +327,7 @@ func (o *operator) processNextWorkItem(ctx context.Context, loop *QueueInformer)
 
 	// Sync and requeue on error (throw out failed deletion syncs)
 	err := loop.Sync(ctx, event)
-	if requeues := queue.NumRequeues(item); err != nil && requeues < 8 && event.Type() != kubestate.ResourceDeleted {
+	if requeues := queue.NumRequeues(item); err != nil && requeues < 15 && event.Type() != kubestate.ResourceDeleted {
 		logger.WithField("requeues", requeues).Trace("requeuing with rate limiting")
 		utilruntime.HandleError(errors.Wrap(err, fmt.Sprintf("sync %q failed", item)))
 		queue.AddRateLimited(item)