Refactor subscription reconciler away from kubestate

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

Author: Per Goncalves da Silva

pkg/controller/operators/catalog/operator.go

@@ -2,6 +2,7 @@ package catalog
 
 import (
 	"context"
+	"crypto/sha256"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -47,7 +48,6 @@ import (
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/pager"
 	"k8s.io/client-go/tools/record"
-	"k8s.io/client-go/util/retry"
 	"k8s.io/client-go/util/workqueue"
 	utilclock "k8s.io/utils/clock"
 
@@ -332,6 +332,27 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		return nil, err
 	}
 
+	// 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](),
+		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()),
+	)
+	if err != nil {
+		return nil, err
+	}
+	if err := op.RegisterQueueInformer(namespaceQueueInformer); err != nil {
+		return nil, err
+	}
+
 	// Wire Subscriptions
 	subInformer := crInformerFactory.Operators().V1alpha1().Subscriptions()
 	op.lister.OperatorsV1alpha1().RegisterSubscriptionLister(metav1.NamespaceAll, subInformer.Lister())
@@ -355,7 +376,7 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		subscription.WithCatalogInformer(catsrcInformer.Informer()),
 		subscription.WithInstallPlanInformer(ipInformer.Informer()),
 		subscription.WithSubscriptionQueue(subQueue),
-		subscription.WithAppendedReconcilers(subscription.ReconcilerFromLegacySyncHandler(op.syncSubscriptions, nil)),
+		subscription.WithNamespaceResolveQueue(op.nsResolveQueue),
 		subscription.WithRegistryReconcilerFactory(op.reconciler),
 		subscription.WithGlobalCatalogNamespace(op.namespace),
 		subscription.WithSourceProvider(op.resolverSourceProvider),
@@ -742,27 +763,6 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		return nil, err
 	}
 
-	// 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](),
-		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()),
-	)
-	if err != nil {
-		return nil, err
-	}
-	if err := op.RegisterQueueInformer(namespaceQueueInformer); err != nil {
-		return nil, err
-	}
-
 	op.sources.Start(context.Background())
 
 	return op, nil
@@ -1342,6 +1342,9 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 		return err
 	}
 
+	// Remove resolutionfailed condition from subscriptions
+	o.removeSubsCond(subs, v1alpha1.SubscriptionResolutionFailed)
+
 	// Attempt to unpack bundles before installing
 	// Note: This should probably use the attenuated client to prevent users from resolving resources they otherwise don't have access to.
 	if len(bundleLookups) > 0 {
@@ -1469,9 +1472,6 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 	// Remove BundleUnpackFailed condition from subscriptions
 	o.removeSubsCond(subs, v1alpha1.SubscriptionBundleUnpackFailed)
 
-	// Remove resolutionfailed condition from subscriptions
-	o.removeSubsCond(subs, v1alpha1.SubscriptionResolutionFailed)
-
 	newSub := true
 	for _, updatedSub := range updatedSubs {
 		updatedSub.Status.RemoveConditions(v1alpha1.SubscriptionResolutionFailed)
@@ -1499,18 +1499,6 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 	return nil
 }
 
-func (o *Operator) syncSubscriptions(obj interface{}) error {
-	sub, ok := obj.(*v1alpha1.Subscription)
-	if !ok {
-		o.logger.Infof("wrong type: %#v", obj)
-		return fmt.Errorf("casting Subscription failed")
-	}
-
-	o.nsResolveQueue.Add(sub.GetNamespace())
-
-	return nil
-}
-
 // syncOperatorGroups requeues the namespace resolution queue on changes to an operatorgroup
 // This is because the operatorgroup is now an input to resolution via the global catalog exclusion annotation
 func (o *Operator) syncOperatorGroups(obj interface{}) error {
@@ -1672,7 +1660,20 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 		return nil, nil
 	}
 
-	csvNames := []string{}
+	sha, err := func() (string, error) {
+		jsonBytes, err := json.Marshal(steps)
+		if err != nil {
+			return "", err
+		}
+		hash := sha256.Sum256(jsonBytes)
+		return fmt.Sprintf("%x", hash), nil
+	}()
+
+	if err != nil {
+		return nil, err
+	}
+
+	var csvNames []string
 	catalogSourceMap := map[string]struct{}{}
 	for _, s := range steps {
 		if s.Resource.Kind == "ClusterServiceVersion" {
@@ -1681,7 +1682,7 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 		catalogSourceMap[s.Resource.CatalogSource] = struct{}{}
 	}
 
-	catalogSources := []string{}
+	var catalogSources []string
 	for s := range catalogSourceMap {
 		catalogSources = append(catalogSources, s)
 	}
@@ -1692,8 +1693,8 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 	}
 	ip := &v1alpha1.InstallPlan{
 		ObjectMeta: metav1.ObjectMeta{
-			GenerateName: "install-",
-			Namespace:    namespace,
+			Name:      fmt.Sprintf("install-%s", sha),
+			Namespace: namespace,
 		},
 		Spec: v1alpha1.InstallPlanSpec{
 			ClusterServiceVersionNames: csvNames,
@@ -1707,7 +1708,8 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 	}
 
 	res, err := o.client.OperatorsV1alpha1().InstallPlans(namespace).Create(context.TODO(), ip, metav1.CreateOptions{})
-	if err != nil {
+	if err != nil && !apierrors.IsAlreadyExists(err) {
+
 		return nil, err
 	}
 
@@ -1717,11 +1719,10 @@ func (o *Operator) createInstallPlan(namespace string, gen int, subs []*v1alpha1
 		CatalogSources: catalogSources,
 		BundleLookups:  bundleLookups,
 	}
-	res, err = o.client.OperatorsV1alpha1().InstallPlans(namespace).UpdateStatus(context.TODO(), res, metav1.UpdateOptions{})
-	if err != nil {
+
+	if _, err = o.client.OperatorsV1alpha1().InstallPlans(namespace).UpdateStatus(context.TODO(), res, metav1.UpdateOptions{}); err != nil {
 		return nil, err
 	}
-
 	return reference.GetReference(res)
 }
 
@@ -1765,7 +1766,6 @@ func (o *Operator) updateSubscriptionStatuses(subs []*v1alpha1.Subscription) ([]
 		errs       []error
 		mu         sync.Mutex
 		wg         sync.WaitGroup
-		getOpts    = metav1.GetOptions{}
 		updateOpts = metav1.UpdateOptions{}
 	)
 
@@ -1774,22 +1774,29 @@ func (o *Operator) updateSubscriptionStatuses(subs []*v1alpha1.Subscription) ([]
 		go func(sub *v1alpha1.Subscription) {
 			defer wg.Done()
 
-			update := func() error {
-				// Update the status of the latest revision
-				latest, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).Get(context.TODO(), sub.GetName(), getOpts)
-				if err != nil {
-					return err
-				}
-				latest.Status = sub.Status
-				*sub = *latest
-				_, err = o.client.OperatorsV1alpha1().Subscriptions(sub.Namespace).UpdateStatus(context.TODO(), latest, updateOpts)
-				return err
-			}
-			if err := retry.RetryOnConflict(retry.DefaultRetry, update); err != nil {
+			_, err := o.client.OperatorsV1alpha1().Subscriptions(sub.Namespace).UpdateStatus(context.TODO(), sub, updateOpts)
+			if err != nil {
 				mu.Lock()
 				defer mu.Unlock()
 				errs = append(errs, err)
 			}
+
+			//update := func() error {
+			//	// Update the status of the latest revision
+			//	latest, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).Get(context.TODO(), sub.GetName(), getOpts)
+			//	if err != nil {
+			//		return err
+			//	}
+			//	latest.Status = sub.Status
+			//	*sub = *latest
+			//	_, err = o.client.OperatorsV1alpha1().Subscriptions(sub.Namespace).UpdateStatus(context.TODO(), latest, updateOpts)
+			//	return err
+			//}
+			//if err := retry.RetryOnConflict(retry.DefaultRetry, update); err != nil {
+			//	mu.Lock()
+			//	defer mu.Unlock()
+			//	errs = append(errs, err)
+			//}
 		}(sub)
 	}
 	wg.Wait()
@@ -2850,31 +2857,11 @@ func (o *Operator) getUpdatedOwnerReferences(refs []metav1.OwnerReference, names
 }
 
 func (o *Operator) listSubscriptions(namespace string) (subs []*v1alpha1.Subscription, err error) {
-	list, err := o.client.OperatorsV1alpha1().Subscriptions(namespace).List(context.TODO(), metav1.ListOptions{})
-	if err != nil {
-		return
-	}
-
-	subs = make([]*v1alpha1.Subscription, 0)
-	for i := range list.Items {
-		subs = append(subs, &list.Items[i])
-	}
-
-	return
+	return o.lister.OperatorsV1alpha1().SubscriptionLister().Subscriptions(namespace).List(labels.Everything())
 }
 
 func (o *Operator) listInstallPlans(namespace string) (ips []*v1alpha1.InstallPlan, err error) {
-	list, err := o.client.OperatorsV1alpha1().InstallPlans(namespace).List(context.TODO(), metav1.ListOptions{})
-	if err != nil {
-		return
-	}
-
-	ips = make([]*v1alpha1.InstallPlan, 0)
-	for i := range list.Items {
-		ips = append(ips, &list.Items[i])
-	}
-
-	return
+	return o.lister.OperatorsV1alpha1().InstallPlanLister().InstallPlans(namespace).List(labels.Everything())
 }
 
 // competingCRDOwnersExist returns true if there exists a CSV that owns at least one of the given CSVs owned CRDs (that's not the given CSV)