fix(installplan): fix bug where too many installplans can be created

Author: ecordell

pkg/controller/operators/catalog/operator.go

@@ -56,16 +56,17 @@ var timeNow = func() metav1.Time { return metav1.NewTime(time.Now().UTC()) }
 // resolving dependencies in a catalog.
 type Operator struct {
 	*queueinformer.Operator
-	client            versioned.Interface
-	lister            operatorlister.OperatorLister
-	namespace         string
-	sources           map[resolver.CatalogKey]resolver.SourceRef
-	sourcesLock       sync.RWMutex
-	sourcesLastUpdate metav1.Time
-	resolver          resolver.Resolver
-	subQueue          workqueue.RateLimitingInterface
-	catSrcQueueSet    queueinformer.ResourceQueueSet
-	reconciler        reconciler.ReconcilerFactory
+	client                versioned.Interface
+	lister                operatorlister.OperatorLister
+	namespace             string
+	sources               map[resolver.CatalogKey]resolver.SourceRef
+	sourcesLock           sync.RWMutex
+	sourcesLastUpdate     metav1.Time
+	resolver              resolver.Resolver
+	subQueue              workqueue.RateLimitingInterface
+	catSrcQueueSet        queueinformer.ResourceQueueSet
+	namespaceResolveQueue workqueue.RateLimitingInterface
+	reconciler            reconciler.ReconcilerFactory
 }
 
 // NewOperator creates a new Catalog Operator.
@@ -209,6 +210,23 @@ func NewOperator(kubeconfigPath string, logger *logrus.Logger, wakeupInterval ti
 		OpClient:             op.OpClient,
 		Lister:               op.lister,
 	}
+
+	namespaceInformer := informers.NewSharedInformerFactory(op.OpClient.KubernetesInterface(), wakeupInterval).Core().V1().Namespaces()
+	resolvingNamespaceQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "resolver")
+	namespaceQueueInformer := queueinformer.NewInformer(
+		resolvingNamespaceQueue,
+		namespaceInformer.Informer(),
+		op.syncResolvingNamespace,
+		nil,
+		"resolver",
+		metrics.NewMetricsNil(),
+		logger,
+	)
+
+	op.RegisterQueueInformer(namespaceQueueInformer)
+	op.lister.CoreV1().RegisterNamespaceLister(namespaceInformer.Lister())
+	op.namespaceResolveQueue = resolvingNamespaceQueue
+
 	return op, nil
 }
 
@@ -456,9 +474,8 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
 		return err
 	}
 
-	// Sync any dependent Subscriptions
-	// TODO: this should go away, we should resync the namespace instead
-	o.syncDependentSubscriptions(logger, out.GetName(), out.GetNamespace())
+	// Trigger a resolve, will pick up any subscriptions that depend on the catalog
+	o.namespaceResolveQueue.AddRateLimited(out.GetNamespace())
 
 	return nil
 }
@@ -487,37 +504,22 @@ func (o *Operator) syncDependentSubscriptions(logger *logrus.Entry, catalogSourc
 	}
 }
 
-func (o *Operator) syncSubscriptions(obj interface{}) error {
-	sub, ok := obj.(*v1alpha1.Subscription)
+func (o *Operator) syncResolvingNamespace(obj interface{}) error {
+	ns, ok := obj.(*corev1.Namespace)
 	if !ok {
 		o.Log.Debugf("wrong type: %#v", obj)
-		return fmt.Errorf("casting Subscription failed")
+		return fmt.Errorf("casting Namespace failed")
 	}
-	namespace := sub.GetNamespace()
-
+	namespace := ns.GetName()
 	logger := o.Log.WithFields(logrus.Fields{
-		"sub":       sub.GetName(),
-		"namespace": sub.GetNamespace(),
-		"source":    sub.Spec.CatalogSource,
-		"pkg":       sub.Spec.Package,
-		"channel":   sub.Spec.Channel,
+		"namespace": namespace,
 	})
-
-	// record the current state of the desired corresponding CSV in the status. no-op if we don't know the csv yet.
-	sub, err := o.ensureSubscriptionCSVState(logger, sub)
-	if err != nil {
-		return err
-	}
-
-	// return early if the subscription is up to date
-	if o.nothingToUpdate(logger, sub) {
-		return nil
-	}
-
 	// get the set of sources that should be used for resolution and best-effort get their connections working
-	logger.Debugf("resolving sources for %s", namespace)
+	logger.Debug("resolving sources")
 	resolverSources := o.ensureResolverSources(logger, namespace)
 
+	logger.Debug("resolving subscriptions in namespace")
+
 	// resolve a set of steps to apply to a cluster, a set of subscriptions to create/update, and any errors
 	steps, subs, err := o.resolver.ResolveSteps(namespace, resolver.NewNamespaceSourceQuerier(resolverSources))
 	if err != nil {
@@ -533,7 +535,6 @@ func (o *Operator) syncSubscriptions(obj interface{}) error {
 			break
 		}
 	}
-
 	installplanReference, err := o.createInstallPlan(namespace, subs, installPlanApproval, steps)
 	if err != nil {
 		logger.WithError(err).Debug("error creating installplan")
@@ -547,6 +548,37 @@ func (o *Operator) syncSubscriptions(obj interface{}) error {
 	return nil
 }
 
+func (o *Operator) syncSubscriptions(obj interface{}) error {
+	sub, ok := obj.(*v1alpha1.Subscription)
+	if !ok {
+		o.Log.Debugf("wrong type: %#v", obj)
+		return fmt.Errorf("casting Subscription failed")
+	}
+
+	logger := o.Log.WithFields(logrus.Fields{
+		"sub":       sub.GetName(),
+		"namespace": sub.GetNamespace(),
+		"source":    sub.Spec.CatalogSource,
+		"pkg":       sub.Spec.Package,
+		"channel":   sub.Spec.Channel,
+	})
+
+	// record the current state of the desired corresponding CSV in the status. no-op if we don't know the csv yet.
+	sub, err := o.ensureSubscriptionCSVState(logger, sub)
+	if err != nil {
+		return err
+	}
+
+	// return early if the subscription is up to date
+	if o.nothingToUpdate(logger, sub) {
+		return nil
+	}
+
+	o.namespaceResolveQueue.AddRateLimited(sub.GetNamespace())
+
+	return nil
+}
+
 func (o *Operator) ensureResolverSources(logger *logrus.Entry, namespace string) map[resolver.CatalogKey]registryclient.Interface {
 	// TODO: record connection status onto an object
 	resolverSources := make(map[resolver.CatalogKey]registryclient.Interface, 0)

test/e2e/catalog_e2e_test.go

@@ -220,6 +220,10 @@ func TestConfigMapUpdateTriggersRegistryPodRollout(t *testing.T) {
 	require.NotNil(t, subscription)
 	_, err = fetchCSV(t, crc, subscription.Status.CurrentCSV, testNamespace, buildCSVConditionChecker(v1alpha1.CSVPhaseSucceeded))
 	require.NoError(t, err)
+
+	ipList, err := crc.OperatorsV1alpha1().InstallPlans(testNamespace).List(metav1.ListOptions{})
+	require.NoError(t, err)
+	require.Equal(t, 1, len(ipList.Items))
 }
 
 func TestConfigMapReplaceTriggersRegistryPodRollout(t *testing.T) {