(fix) registry pods do not come up again after node failure

[PR 3201](operator-framework#3201) attempted to solve for the
issue by deleting the pods stuck in Terminating due to unreachable node.
However, the logic to do that was included in `EnsureRegistryServer`, which only
gets executed in polling in requested by the user.

This PR fixes the issue by modifying the `RegistryReconciler` interface with the
introduction of a new component for the interface: `RegistryCleaner`.
This promotes the job of cleaning up the pods that are stuck (and any other resources
that may need to be cleaned) to a first class status. The `RegistryCleaner` is then
called as the first step in the Catalog Operator registry reconclier, so that the pods
stuck are cleaned up before the rest of the reconciler logic is executed.
The PR provides implementation of `RegistryCleaner` for the `GrpcReconciler`,
`ConfigMapReconciler` and the `GrpcAddressRegistryReconciler` implementations of
`RegistryReconciler` interface.

Author: anik120

pkg/controller/operators/catalog/operator.go

@@ -992,6 +992,12 @@ func (o *Operator) syncRegistryServer(logger *logrus.Entry, in *v1alpha1.Catalog
 		out.SetError(v1alpha1.CatalogSourceRegistryServerError, syncError)
 		return
 	}
+	err := srcReconciler.CleanRegistryServer(logger, in)
+	if err != nil {
+		syncError = fmt.Errorf("could not clean dead resources: %s", err)
+		out.SetError(v1alpha1.CatalogSourceRegistryServerError, syncError)
+		return
+	}
 
 	healthy, err := srcReconciler.CheckRegistryServer(logger, in)
 	if err != nil {

pkg/controller/registry/reconciler/configmap.go

@@ -3,18 +3,20 @@ package reconciler
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/install"
 	hashutil "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/kubernetes/pkg/util/hash"
-	"github.com/pkg/errors"
+	pkgerrors "github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/utils/ptr"
 
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
@@ -192,6 +194,7 @@ type ConfigMapRegistryReconciler struct {
 
 var _ RegistryEnsurer = &ConfigMapRegistryReconciler{}
 var _ RegistryChecker = &ConfigMapRegistryReconciler{}
+var _ RegistryCleaner = &ConfigMapRegistryReconciler{}
 var _ RegistryReconciler = &ConfigMapRegistryReconciler{}
 
 func (c *ConfigMapRegistryReconciler) currentService(source configMapCatalogSourceDecorator) (*corev1.Service, error) {
@@ -327,27 +330,27 @@ func (c *ConfigMapRegistryReconciler) EnsureRegistryServer(logger *logrus.Entry,
 
 	//TODO: if any of these error out, we should write a status back (possibly set RegistryServiceStatus to nil so they get recreated)
 	if err := c.ensureServiceAccount(source, overwrite); err != nil {
-		return errors.Wrapf(err, "error ensuring service account: %s", source.serviceAccountName())
+		return pkgerrors.Wrapf(err, "error ensuring service account: %s", source.serviceAccountName())
 	}
 	if err := c.ensureRole(source, overwrite); err != nil {
-		return errors.Wrapf(err, "error ensuring role: %s", source.roleName())
+		return pkgerrors.Wrapf(err, "error ensuring role: %s", source.roleName())
 	}
 	if err := c.ensureRoleBinding(source, overwrite); err != nil {
-		return errors.Wrapf(err, "error ensuring rolebinding: %s", source.RoleBinding().GetName())
+		return pkgerrors.Wrapf(err, "error ensuring rolebinding: %s", source.RoleBinding().GetName())
 	}
 	pod, err := source.Pod(image, defaultPodSecurityConfig)
 	if err != nil {
 		return err
 	}
 	if err := c.ensurePod(source, defaultPodSecurityConfig, overwritePod); err != nil {
-		return errors.Wrapf(err, "error ensuring pod: %s", pod.GetName())
+		return pkgerrors.Wrapf(err, "error ensuring pod: %s", pod.GetName())
 	}
 	service, err := source.Service()
 	if err != nil {
 		return err
 	}
 	if err := c.ensureService(source, overwrite); err != nil {
-		return errors.Wrapf(err, "error ensuring service: %s", service.GetName())
+		return pkgerrors.Wrapf(err, "error ensuring service: %s", service.GetName())
 	}
 
 	if overwritePod {
@@ -420,15 +423,15 @@ func (c *ConfigMapRegistryReconciler) ensurePod(source configMapCatalogSourceDec
 		}
 		for _, p := range currentPods {
 			if err := c.OpClient.KubernetesInterface().CoreV1().Pods(pod.GetNamespace()).Delete(context.TODO(), p.GetName(), *metav1.NewDeleteOptions(1)); err != nil && !apierrors.IsNotFound(err) {
-				return errors.Wrapf(err, "error deleting old pod: %s", p.GetName())
+				return pkgerrors.Wrapf(err, "error deleting old pod: %s", p.GetName())
 			}
 		}
 	}
 	_, err = c.OpClient.KubernetesInterface().CoreV1().Pods(pod.GetNamespace()).Create(context.TODO(), pod, metav1.CreateOptions{})
 	if err == nil {
 		return nil
 	}
-	return errors.Wrapf(err, "error creating new pod: %s", pod.GetGenerateName())
+	return pkgerrors.Wrapf(err, "error creating new pod: %s", pod.GetGenerateName())
 }
 
 func (c *ConfigMapRegistryReconciler) ensureService(source configMapCatalogSourceDecorator, overwrite bool) error {
@@ -515,3 +518,31 @@ func (c *ConfigMapRegistryReconciler) CheckRegistryServer(logger *logrus.Entry,
 	healthy = true
 	return
 }
+
+// CleanRegistryServer attempts to force delete registry client pods that are stale.
+func (c *ConfigMapRegistryReconciler) CleanRegistryServer(logger *logrus.Entry, catalogSource *v1alpha1.CatalogSource) error {
+	source := configMapCatalogSourceDecorator{catalogSource, c.createPodAsUser}
+	defaultPodSecurityConfig, err := getDefaultPodContextConfig(c.OpClient, catalogSource.GetNamespace())
+	if err != nil {
+		return err
+	}
+	currentPods, err := c.currentPods(source, c.Image, defaultPodSecurityConfig)
+	if err != nil {
+		return err
+	}
+	var forceDeleteErrs []error
+	for _, pod := range currentPods {
+		if isPodDead(pod) {
+			logger.WithFields(logrus.Fields{"pod.namespace": source.GetNamespace(), "pod.name": pod.GetName()}).Info("force deleting dead pod")
+			if err := c.OpClient.KubernetesInterface().CoreV1().Pods(source.GetNamespace()).Delete(context.TODO(), pod.GetName(), metav1.DeleteOptions{
+				GracePeriodSeconds: ptr.To[int64](0),
+			}); err != nil && !apierrors.IsNotFound(err) {
+				forceDeleteErrs = append(forceDeleteErrs, pkgerrors.Wrapf(err, "error deleting old pod: %s", pod.GetName()))
+			}
+		}
+	}
+	if len(forceDeleteErrs) > 0 {
+		return errors.Join(forceDeleteErrs...)
+	}
+	return nil
+}

pkg/controller/registry/reconciler/grpc.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"slices"
 	"strings"
 	"time"
 
@@ -346,32 +345,13 @@ func isRegistryServiceStatusValid(source *grpcCatalogSourceDecorator) (bool, err
 }
 
 func (c *GrpcRegistryReconciler) ensurePod(logger *logrus.Entry, source grpcCatalogSourceDecorator, serviceAccount *corev1.ServiceAccount, defaultPodSecurityConfig v1alpha1.SecurityConfig, overwrite bool) error {
-	// currentPods refers to the current pod instances of the catalog source
-	currentPods := c.currentPods(logger, source)
-
-	var forceDeleteErrs []error
-	currentPods = slices.DeleteFunc(currentPods, func(pod *corev1.Pod) bool {
-		if !isPodDead(pod) {
-			logger.WithFields(logrus.Fields{"pod.namespace": source.GetNamespace(), "pod.name": pod.GetName()}).Debug("pod is alive")
-			return false
-		}
-		logger.WithFields(logrus.Fields{"pod.namespace": source.GetNamespace(), "pod.name": pod.GetName()}).Info("force deleting dead pod")
-		if err := c.OpClient.KubernetesInterface().CoreV1().Pods(source.GetNamespace()).Delete(context.TODO(), pod.GetName(), metav1.DeleteOptions{
-			GracePeriodSeconds: ptr.To[int64](0),
-		}); err != nil && !apierrors.IsNotFound(err) {
-			forceDeleteErrs = append(forceDeleteErrs, pkgerrors.Wrapf(err, "error deleting old pod: %s", pod.GetName()))
-		}
-		return true
-	})
-	if len(forceDeleteErrs) > 0 {
-		return errors.Join(forceDeleteErrs...)
-	}
-
-	if len(currentPods) > 0 {
+	// currentLivePods refers to the currently live instances of the catalog source
+	currentLivePods := c.currentPods(logger, source)
+	if len(currentLivePods) > 0 {
 		if !overwrite {
 			return nil
 		}
-		for _, p := range currentPods {
+		for _, p := range currentLivePods {
 			logger.WithFields(logrus.Fields{"pod.namespace": source.GetNamespace(), "pod.name": p.GetName()}).Info("deleting current pod")
 			if err := c.OpClient.KubernetesInterface().CoreV1().Pods(source.GetNamespace()).Delete(context.TODO(), p.GetName(), *metav1.NewDeleteOptions(1)); err != nil && !apierrors.IsNotFound(err) {
 				return pkgerrors.Wrapf(err, "error deleting old pod: %s", p.GetName())
@@ -711,3 +691,24 @@ func podHashMatch(existing, new *corev1.Pod) bool {
 
 	return true
 }
+
+// CleanRegistryServer attempts to force delete registry client pods that are stale.
+func (c *GrpcRegistryReconciler) CleanRegistryServer(logger *logrus.Entry, catalogSource *v1alpha1.CatalogSource) error {
+	source := grpcCatalogSourceDecorator{CatalogSource: catalogSource, createPodAsUser: c.createPodAsUser, opmImage: c.opmImage, utilImage: c.utilImage}
+	currentPods := c.currentPods(logger, source)
+	var forceDeleteErrs []error
+	for _, pod := range currentPods {
+		if isPodDead(pod) {
+			logger.WithFields(logrus.Fields{"pod.namespace": source.GetNamespace(), "pod.name": pod.GetName()}).Info("force deleting dead pod")
+			if err := c.OpClient.KubernetesInterface().CoreV1().Pods(source.GetNamespace()).Delete(context.TODO(), pod.GetName(), metav1.DeleteOptions{
+				GracePeriodSeconds: ptr.To[int64](0),
+			}); err != nil && !apierrors.IsNotFound(err) {
+				forceDeleteErrs = append(forceDeleteErrs, pkgerrors.Wrapf(err, "error deleting old pod: %s", pod.GetName()))
+			}
+		}
+	}
+	if len(forceDeleteErrs) > 0 {
+		return errors.Join(forceDeleteErrs...)
+	}
+	return nil
+}

pkg/controller/registry/reconciler/grpc_address.go

@@ -11,6 +11,7 @@ type GrpcAddressRegistryReconciler struct {
 
 var _ RegistryEnsurer = &GrpcAddressRegistryReconciler{}
 var _ RegistryChecker = &GrpcAddressRegistryReconciler{}
+var _ RegistryCleaner = &GrpcAddressRegistryReconciler{}
 var _ RegistryReconciler = &GrpcAddressRegistryReconciler{}
 
 // EnsureRegistryServer ensures a registry server exists for the given CatalogSource.
@@ -30,3 +31,7 @@ func (g *GrpcAddressRegistryReconciler) CheckRegistryServer(logger *logrus.Entry
 	healthy = true
 	return
 }
+
+func (g *GrpcAddressRegistryReconciler) CleanRegistryServer(logger *logrus.Entry, catalogSource *v1alpha1.CatalogSource) error {
+	return nil
+}

pkg/controller/registry/reconciler/reconciler.go

@@ -46,10 +46,17 @@ type RegistryChecker interface {
 	CheckRegistryServer(logger *logrus.Entry, catalogSource *operatorsv1alpha1.CatalogSource) (healthy bool, err error)
 }
 
+// RegistryCleaner force deletes stale resources.
+type RegistryCleaner interface {
+	// CleanRegistryServer attempts to force delete registry clients that are stale.
+	CleanRegistryServer(logger *logrus.Entry, catalogSource *operatorsv1alpha1.CatalogSource) error
+}
+
 // RegistryReconciler knows how to reconcile a registry.
 type RegistryReconciler interface {
 	RegistryChecker
 	RegistryEnsurer
+	RegistryCleaner
 }
 
 // RegistryReconcilerFactory describes factory methods for RegistryReconcilers.