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,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 {
@@ -2047,7 +2035,8 @@ func (o *Operator) syncInstallPlans(obj interface{}) (syncError error) {
 		defer o.ipQueueSet.RequeueAfter(outInstallPlan.GetNamespace(), outInstallPlan.GetName(), time.Second*5)
 	}
 
-	defer o.requeueSubscriptionForInstallPlan(plan, logger)
+	// TODO: do we need this?
+	// defer o.requeueSubscriptionForInstallPlan(plan, logger)
 
 	// Update InstallPlan with status of transition. Log errors if we can't write them to the status.
 	if _, err := o.client.OperatorsV1alpha1().InstallPlans(plan.GetNamespace()).UpdateStatus(context.TODO(), outInstallPlan, metav1.UpdateOptions{}); err != nil {
@@ -2075,22 +2064,22 @@ func hasBundleLookupFailureCondition(bundleLookups []v1alpha1.BundleLookup) (boo
 	return false, nil
 }
 
-func (o *Operator) requeueSubscriptionForInstallPlan(plan *v1alpha1.InstallPlan, logger *logrus.Entry) {
-	// Notify subscription loop of installplan changes
-	owners := ownerutil.GetOwnersByKind(plan, v1alpha1.SubscriptionKind)
-
-	if len(owners) == 0 {
-		logger.Trace("no installplan owner subscriptions found to requeue")
-		return
-	}
-
-	for _, owner := range owners {
-		logger.WithField("owner", owner).Debug("requeueing installplan owner")
-		if err := o.subQueueSet.Requeue(plan.GetNamespace(), owner.Name); err != nil {
-			logger.WithError(err).Warn("error requeuing installplan owner")
-		}
-	}
-}
+//func (o *Operator) requeueSubscriptionForInstallPlan(plan *v1alpha1.InstallPlan, logger *logrus.Entry) {
+//	// Notify subscription loop of installplan changes
+//	owners := ownerutil.GetOwnersByKind(plan, v1alpha1.SubscriptionKind)
+//
+//	if len(owners) == 0 {
+//		logger.Trace("no installplan owner subscriptions found to requeue")
+//		return
+//	}
+//
+//	for _, owner := range owners {
+//		logger.WithField("owner", owner).Debug("requeueing installplan owner")
+//		if err := o.subQueueSet.Requeue(plan.GetNamespace(), owner.Name); err != nil {
+//			logger.WithError(err).Warn("error requeuing installplan owner")
+//		}
+//	}
+//}
 
 func (o *Operator) setInstallPlanInstalledCond(ip *v1alpha1.InstallPlan, reason v1alpha1.InstallPlanConditionReason, message string, logger *logrus.Entry) (*v1alpha1.InstallPlan, error) {
 	now := o.now()
@@ -2850,31 +2839,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)