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

@@ -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,11 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 		return err
 	}
 
+	// Remove any sticky ResolutionFailure condition
+	for _, sub := range subs {
+		sub.Status.RemoveConditions(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 {
@@ -1499,16 +1504,9 @@ 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")
-	}
-
+func (o *Operator) syncSubscriptions(_ context.Context, sub *v1alpha1.Subscription) (*v1alpha1.Subscription, error) {
 	o.nsResolveQueue.Add(sub.GetNamespace())
-
-	return nil
+	return sub, nil
 }
 
 // syncOperatorGroups requeues the namespace resolution queue on changes to an operatorgroup
@@ -2850,31 +2848,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)