fix(operatorgroups): gc copied csvs from a separate queue

Author: ecordell

go.sum

@@ -58,6 +58,7 @@ type Operator struct {
 	lister           operatorlister.OperatorLister
 	recorder         record.EventRecorder
 	copyQueueIndexer *queueinformer.QueueIndexer
+	gcQueueIndexer   *queueinformer.QueueIndexer
 }
 
 func NewOperator(logger *logrus.Logger, crClient versioned.Interface, opClient operatorclient.ClientInterface, strategyResolver install.StrategyResolverInterface, wakeupInterval time.Duration, namespaces []string) (*Operator, error) {
@@ -261,6 +262,12 @@ func NewOperator(logger *logrus.Logger, crClient versioned.Interface, opClient o
 	op.RegisterQueueIndexer(csvQueueIndexer)
 	op.copyQueueIndexer = csvQueueIndexer
 
+	// Register separate queue for copying csvs
+	csvGCQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "csvGC")
+	csvGCQueueIndexer := queueinformer.NewQueueIndexer(csvGCQueue, csvIndexes, op.syncGcCsv, "csvGC", logger, metrics.NewMetricsNil())
+	op.RegisterQueueIndexer(csvGCQueueIndexer)
+	op.gcQueueIndexer = csvGCQueueIndexer
+
 	// Set up watch on deployments
 	depHandlers := &cache.ResourceEventHandlerFuncs{
 		// TODO: pass closure that forgets queue item after calling custom deletion handler.
@@ -435,7 +442,7 @@ func (a *Operator) handleClusterServiceVersionDeletion(obj interface{}) {
 	for _, namespace := range namespaces {
 		if namespace != operatorNamespace {
 			logger.WithField("targetNamespace", namespace).Debug("requeueing child csv for deletion")
-			a.csvQueueSet.Requeue(clusterServiceVersion.GetName(), namespace)
+			a.gcQueueIndexer.Add(fmt.Sprintf("%s/%s", namespace, clusterServiceVersion.GetName()))
 		}
 	}
 
@@ -478,33 +485,34 @@ func (a *Operator) removeDanglingChildCSVs(csv *v1alpha1.ClusterServiceVersion)
 	operatorNamespace, ok := csv.Annotations[v1alpha2.OperatorGroupNamespaceAnnotationKey]
 	if !ok {
 		logger.Debug("missing operator namespace annotation on copied CSV")
-		return a.deleteChild(csv)
+		return a.deleteChild(csv, logger)
 	}
 
 	logger = logger.WithField("parentNamespace", operatorNamespace)
 	parent, err := a.lister.OperatorsV1alpha1().ClusterServiceVersionLister().ClusterServiceVersions(operatorNamespace).Get(csv.GetName())
 	if k8serrors.IsNotFound(err) || k8serrors.IsGone(err) || parent == nil {
 		logger.Debug("deleting copied CSV since parent is missing")
-		return a.deleteChild(csv)
+		return a.deleteChild(csv, logger)
 	}
 
 	if parent.Status.Phase == v1alpha1.CSVPhaseFailed && parent.Status.Reason == v1alpha1.CSVReasonInterOperatorGroupOwnerConflict {
 		logger.Debug("deleting copied CSV since parent has intersecting operatorgroup conflict")
-		return a.deleteChild(csv)
+		return a.deleteChild(csv, logger)
 	}
 
 	if annotations := parent.GetAnnotations(); annotations != nil {
 		if !resolver.NewNamespaceSetFromString(annotations[v1alpha2.OperatorGroupTargetsAnnotationKey]).Contains(csv.GetNamespace()) {
 			logger.WithField("parentTargets", annotations[v1alpha2.OperatorGroupTargetsAnnotationKey]).
 				Debug("deleting copied CSV since parent no longer lists this as a target namespace")
-			return a.deleteChild(csv)
+			return a.deleteChild(csv, logger)
 		}
 	}
 
 	return nil
 }
 
-func (a *Operator) deleteChild(csv *v1alpha1.ClusterServiceVersion) error {
+func (a *Operator) deleteChild(csv *v1alpha1.ClusterServiceVersion, logger *logrus.Entry) error {
+	logger.Debug("gcing csv")
 	return a.client.OperatorsV1alpha1().ClusterServiceVersions(csv.GetNamespace()).Delete(csv.GetName(), &metav1.DeleteOptions{})
 }
 
@@ -609,6 +617,19 @@ func (a *Operator) syncCopyCSV(obj interface{}) (syncError error) {
 	return
 }
 
+func (a *Operator) syncGcCsv(obj interface{}) (syncError error) {
+	clusterServiceVersion, ok := obj.(*v1alpha1.ClusterServiceVersion)
+	if !ok {
+		a.Log.Debugf("wrong type: %#v", obj)
+		return fmt.Errorf("casting ClusterServiceVersion failed")
+	}
+	if clusterServiceVersion.IsCopied() {
+		syncError = a.removeDanglingChildCSVs(clusterServiceVersion)
+		return
+	}
+	return
+}
+
 // operatorGroupFromAnnotations returns the OperatorGroup for the CSV only if the CSV is active one in the group
 func (a *Operator) operatorGroupFromAnnotations(logger *logrus.Entry, csv *v1alpha1.ClusterServiceVersion) *v1alpha2.OperatorGroup {
 	annotations := csv.GetAnnotations()
@@ -712,8 +733,8 @@ func (a *Operator) transitionCSVState(in v1alpha1.ClusterServiceVersion) (out *v
 	now := timeNow()
 
 	if out.IsCopied() {
-		logger.Debug("skipping copied csv transition")
-		syncError = a.removeDanglingChildCSVs(out)
+		logger.Debug("skipping copied csv transition, schedule for gc check")
+		a.gcQueueIndexer.Enqueue(out)
 		return
 	}
 

pkg/controller/operators/olm/operator.go

@@ -33,6 +33,7 @@ var (
 	AdminVerbs     = []string{"*"}
 	EditVerbs      = []string{"create", "update", "patch", "delete"}
 	ViewVerbs      = []string{"get", "list", "watch"}
+	Suffices       = []string{AdminSuffix, EditSuffix, ViewSuffix}
 	VerbsForSuffix = map[string][]string{
 		AdminSuffix: AdminVerbs,
 		EditSuffix:  EditVerbs,
@@ -294,8 +295,11 @@ func (a *Operator) ensureRBACInTargetNamespace(csv *v1alpha1.ClusterServiceVersi
 	}
 	ruleChecker := install.NewCSVRuleChecker(a.lister.RbacV1().RoleLister(), a.lister.RbacV1().RoleBindingLister(), a.lister.RbacV1().ClusterRoleLister(), a.lister.RbacV1().ClusterRoleBindingLister(), csv)
 
+	logger := a.Log.WithField("opgroup", operatorGroup.GetName()).WithField("csv", csv.GetName())
+
 	// if OperatorGroup is global (all namespaces) we generate cluster roles / cluster role bindings instead
 	if len(targetNamespaces) == 1 && targetNamespaces[0] == corev1.NamespaceAll {
+		logger.Debug("opgroup is global")
 
 		// synthesize cluster permissions to verify rbac
 		for _, p := range strategyDetailsDeployment.Permissions {
@@ -309,8 +313,10 @@ func (a *Operator) ensureRBACInTargetNamespace(csv *v1alpha1.ClusterServiceVersi
 
 		// operator already has access at the cluster scope
 		if permMet {
+			logger.Debug("global operator has correct global permissions")
 			return nil
 		}
+		logger.Debug("lift roles/rolebindings to clusterroles/rolebindings")
 		if err := a.ensureSingletonRBAC(operatorGroup.GetNamespace(), csv); err != nil {
 			return err
 		}
@@ -347,6 +353,7 @@ func (a *Operator) ensureSingletonRBAC(operatorNamespace string, csv *v1alpha1.C
 	}
 
 	for _, r := range ownedRoles {
+		a.Log.Debug("processing role")
 		_, err := a.lister.RbacV1().ClusterRoleLister().Get(r.GetName())
 		if err != nil {
 			clusterRole := &rbacv1.ClusterRole{
@@ -363,6 +370,7 @@ func (a *Operator) ensureSingletonRBAC(operatorNamespace string, csv *v1alpha1.C
 			if _, err := a.OpClient.CreateClusterRole(clusterRole); err != nil {
 				return err
 			}
+			a.Log.Debug("created cluster role")
 		}
 	}
 
@@ -578,24 +586,19 @@ func (a *Operator) copyToNamespace(csv *v1alpha1.ClusterServiceVersion, namespac
 	return nil
 }
 
-// TODO: do we want to do this? or just let the dangling CSV clean it up
 func (a *Operator) pruneFromNamespace(operatorGroupName, namespace string) error {
 	fetchedCSVs, err := a.lister.OperatorsV1alpha1().ClusterServiceVersionLister().ClusterServiceVersions(namespace).List(labels.Everything())
 	if err != nil {
 		return err
 	}
 
-	errlist := []error{}
 	for _, csv := range fetchedCSVs {
 		if csv.IsCopied() && csv.GetAnnotations()[v1alpha2.OperatorGroupAnnotationKey] == operatorGroupName {
 			a.Log.Debugf("Found CSV '%v' in namespace %v to delete", csv.GetName(), namespace)
-			err := a.client.OperatorsV1alpha1().ClusterServiceVersions(namespace).Delete(csv.GetName(), &metav1.DeleteOptions{})
-			if err != nil {
-				errlist = append(errlist, err)
-			}
+			a.gcQueueIndexer.Enqueue(csv)
 		}
 	}
-	return errors.NewAggregate(errlist)
+	return nil
 }
 
 func (a *Operator) setOperatorGroupAnnotations(obj *metav1.ObjectMeta, op *v1alpha2.OperatorGroup, addTargets bool) {

pkg/controller/operators/olm/operatorgroup.go

@@ -38,6 +38,10 @@ func (q *QueueIndexer) Enqueue(obj interface{}) {
 	q.queue.Add(key)
 }
 
+func (q *QueueIndexer) Add(key string) {
+	q.queue.Add(key)
+}
+
 // keyFunc turns an object into a key for the queue. In the future will use a (name, namespace) struct as key
 func (q *QueueIndexer) keyFunc(obj interface{}) (string, bool) {
 	k, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)

pkg/lib/queueinformer/queueindexer.go

@@ -140,10 +140,12 @@ func (o *Operator) Run(stopc <-chan struct{}) (ready, done chan struct{}, atLeve
 		o.Log.Info("starting workers...")
 		for _, queueInformer := range o.queueInformers {
 			go o.worker(queueInformer)
+			go o.worker(queueInformer)
 		}
 
 		for _, queueIndexer := range o.queueIndexers {
 			go o.indexerWorker(queueIndexer)
+			go o.indexerWorker(queueIndexer)
 		}
 		ready <- struct{}{}
 		<-stopc
@@ -204,7 +206,7 @@ func (o *Operator) sync(loop *QueueInformer, key string) error {
 	return loop.syncHandler(obj)
 }
 
-// This provides the same function as above, but for indexes and queues not fed by informers.
+// This provides the same function as above, but for queues that are not auto-fed by informers.
 // indexerWorker runs a worker thread that just dequeues items, processes them, and marks them done.
 // It enforces that the syncHandler is never invoked concurrently with the same key.
 func (o *Operator) indexerWorker(loop *QueueIndexer) {

pkg/lib/queueinformer/queueinformer_operator.go

@@ -130,6 +130,7 @@ func TestOperatorGroup(t *testing.T) {
 	// Verify the copied CSV transitions to FAILED
 	// Delete CSV
 	// Verify copied CVS is deleted
+	defer cleaner.NotifyTestComplete(t, true)
 
 	c := newKubeClient(t)
 	crc := newCRClient(t)
@@ -464,6 +465,8 @@ func TestOperatorGroupInstallModeSupport(t *testing.T) {
 	// Update csvA to have AllNamespaces supported=true
 	// Ensure csvA transitions to Pending
 
+	defer cleaner.NotifyTestComplete(t, true)
+
 	// Generate namespaceA and namespaceB
 	nsA := genName("a")
 	nsB := genName("b")
@@ -695,6 +698,8 @@ func TestOperatorGroupIntersection(t *testing.T) {
 	// Wait for operatorGroupB to have providedAPI annotation with crdB's Kind.version.group
 	// Wait for csvB to have a CSV with a copied status in namespace C
 
+	defer cleaner.NotifyTestComplete(t, true)
+
 	// Create a catalog for csvA, csvB, and csvD
 	pkgA := genName("a-")
 	pkgB := genName("b-")
@@ -927,6 +932,8 @@ func TestStaticProviderOperatorGroup(t *testing.T) {
 	// Wait for KindA.version.group providedAPI annotation to be removed from operatorGroupC's providedAPIs annotation
 	// Ensure KindA.version.group providedAPI annotation on operatorGroupA
 
+	defer cleaner.NotifyTestComplete(t, true)
+
 	// Create a catalog for csvA, csvB
 	pkgA := genName("a-")
 	pkgB := genName("b-")
@@ -1118,6 +1125,7 @@ func TestStaticProviderOperatorGroup(t *testing.T) {
 // TODO: Test Subscriptions with depedencies and transitive dependencies in intersecting OperatorGroups
 // TODO: Test Subscription upgrade paths with + and - providedAPIs
 func TestCSVCopyWatchingAllNamespaces(t *testing.T) {
+	defer cleaner.NotifyTestComplete(t, true)
 	c := newKubeClient(t)
 	crc := newCRClient(t)
 	csvName := genName("another-csv-") // must be lowercase for DNS-1123 validation
@@ -1298,7 +1306,7 @@ func TestCSVCopyWatchingAllNamespaces(t *testing.T) {
 		require.NoError(t, err)
 	}()
 
-	err = wait.Poll(pollInterval, pollDuration, func() (bool, error) {
+	err = wait.Poll(pollInterval, 2*pollDuration, func() (bool, error) {
 		_, fetchErr := crc.OperatorsV1alpha1().ClusterServiceVersions(otherNamespaceName).Get(csvName, metav1.GetOptions{})
 		if fetchErr != nil {
 			if errors.IsNotFound(fetchErr) {

test/e2e/operator_groups_e2e_test.go

@@ -45,6 +45,8 @@ func fetchPackageManifest(t *testing.T, pmc pmversioned.Interface, namespace, na
 }
 
 func TestPackageManifestLoading(t *testing.T) {
+	defer cleaner.NotifyTestComplete(t, true)
+
 	// create a simple catalogsource
 	packageName := genName("nginx")
 	stableChannel := "stable"