Merge pull request #761 from ecordell/multiple-upgrades

Subscription steps through multiple upgrades

Author: openshift-merge-robot

Makefile

@@ -71,7 +71,7 @@ setup-bare: clean e2e.namespace
 	. ./scripts/install_bare.sh $(shell cat ./e2e.namespace) test/e2e/resources
 
 e2e:
-	go test -v -failfast -timeout 50m ./test/e2e/... -namespace=openshift-operators -kubeconfig=${KUBECONFIG} -olmNamespace=openshift-operator-lifecycle-manager
+	go test -v -failfast -timeout 70m ./test/e2e/... -namespace=openshift-operators -kubeconfig=${KUBECONFIG} -olmNamespace=openshift-operator-lifecycle-manager
 
 e2e-local:
 	. ./scripts/build_local.sh
@@ -98,7 +98,8 @@ container:
 
 clean-e2e:
 	kubectl delete crds --all
-	kubectl delete apiservices.apiregistration.k8s.io v1alpha1.packages.apps.redhat.com
+	kubectl delete apiservices.apiregistration.k8s.io v1alpha1.packages.apps.redhat.com || true
+	for i in {1..40}; do kubectl delete namespace "ns-$i" || true; done
 	kubectl delete -f test/e2e/resources/0000_50_olm_00-namespace.yaml
 
 clean:

go.sum

@@ -30,6 +30,20 @@ var uncopiableReasons = map[ConditionReason]struct{}{
 	CSVReasonCannotModifyStaticOperatorGroupProvidedAPIs: {},
 }
 
+// safeToAnnotateOperatorGroupReasons are the set of reasons that it's safe to attempt to update the operatorgroup
+// annotations
+var safeToAnnotateOperatorGroupReasons = map[ConditionReason]struct{}{
+	CSVReasonOwnerConflict:                               {},
+	CSVReasonInstallSuccessful:                           {},
+	CSVReasonInvalidInstallModes:                         {},
+	CSVReasonNoTargetNamespaces:                          {},
+	CSVReasonUnsupportedOperatorGroup:                    {},
+	CSVReasonNoOperatorGroup:                             {},
+	CSVReasonTooManyOperatorGroups:                       {},
+	CSVReasonInterOperatorGroupOwnerConflict:             {},
+	CSVReasonCannotModifyStaticOperatorGroupProvidedAPIs: {},
+}
+
 // SetPhaseWithEventIfChanged emits a Kubernetes event with details of a phase change and sets the current phase if phase, reason, or message would changed
 func (c *ClusterServiceVersion) SetPhaseWithEventIfChanged(phase ClusterServiceVersionPhase, reason ConditionReason, message string, now metav1.Time, recorder record.EventRecorder) {
 	if c.Status.Phase == phase && c.Status.Reason == reason && c.Status.Message == message {
@@ -120,6 +134,11 @@ func (c *ClusterServiceVersion) IsUncopiable() bool {
 	return ok
 }
 
+func (c *ClusterServiceVersion) IsSafeToUpdateOperatorGroupAnnotations() bool {
+	_, ok := safeToAnnotateOperatorGroupReasons[c.Status.Reason]
+	return ok
+}
+
 // NewInstallModeSet returns an InstallModeSet instantiated from the given list of InstallModes.
 // If the given list is not a set, an error is returned.
 func NewInstallModeSet(modes []InstallMode) (InstallModeSet, error) {

pkg/api/apis/operators/v1alpha1/clusterserviceversion.go

@@ -75,7 +75,9 @@ func (i *StrategyDeploymentInstaller) installDeployments(deps []StrategyDeployme
 		dep.Spec.Template.SetAnnotations(annotations)
 
 		ownerutil.AddNonBlockingOwner(dep, i.owner)
-		ownerutil.AddOwnerLabels(dep, i.owner)
+		if err := ownerutil.AddOwnerLabels(dep, i.owner); err != nil {
+			return err
+		}
 		if _, err := i.strategyClient.CreateOrUpdateDeployment(dep); err != nil {
 			return err
 		}

pkg/controller/install/deployment.go

@@ -89,10 +89,12 @@ func NewOperator(kubeconfigPath string, logger *logrus.Logger, wakeupInterval ti
 	// Create an informer for each watched namespace.
 	ipSharedIndexInformers := []cache.SharedIndexInformer{}
 	subSharedIndexInformers := []cache.SharedIndexInformer{}
+	csvSharedIndexInformers := []cache.SharedIndexInformer{}
 	for _, namespace := range watchedNamespaces {
 		nsInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(crClient, wakeupInterval, externalversions.WithNamespace(namespace))
 		ipSharedIndexInformers = append(ipSharedIndexInformers, nsInformerFactory.Operators().V1alpha1().InstallPlans().Informer())
 		subSharedIndexInformers = append(subSharedIndexInformers, nsInformerFactory.Operators().V1alpha1().Subscriptions().Informer())
+		csvSharedIndexInformers = append(csvSharedIndexInformers, nsInformerFactory.Operators().V1alpha1().ClusterServiceVersions().Informer())
 
 		// resolver needs subscription and csv listers
 		lister.OperatorsV1alpha1().RegisterSubscriptionLister(namespace, nsInformerFactory.Operators().V1alpha1().Subscriptions().Lister())
@@ -230,6 +232,11 @@ func NewOperator(kubeconfigPath string, logger *logrus.Logger, wakeupInterval ti
 	op.lister.CoreV1().RegisterNamespaceLister(namespaceInformer.Lister())
 	op.namespaceResolveQueue = resolvingNamespaceQueue
 
+	// Register CSV informers to fill cache
+	for _, informer := range csvSharedIndexInformers {
+		op.RegisterInformer(informer)
+	}
+
 	return op, nil
 }
 
@@ -381,7 +388,7 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
 
 			return nil
 		}
-		
+
 		logger.Debug("catsrc configmap state good, checking registry pod")
 	}
 
@@ -391,7 +398,6 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
 		return fmt.Errorf("no reconciler for source type %s", catsrc.Spec.SourceType)
 	}
 
-
 	// if registry pod hasn't been created or hasn't been updated since the last configmap update, recreate it
 	if catsrc.Status.RegistryServiceStatus == nil || catsrc.Status.RegistryServiceStatus.CreatedAt.Before(&catsrc.Status.LastSync) {
 		logger.Debug("registry server scheduled recheck")
@@ -538,7 +544,7 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 
 	subs, err := o.lister.OperatorsV1alpha1().SubscriptionLister().Subscriptions(namespace).List(labels.Everything())
 	if err != nil {
-		logger.WithError(err).Debug("couldn't list subscriptions")	
+		logger.WithError(err).Debug("couldn't list subscriptions")
 		return err
 	}
 
@@ -745,7 +751,7 @@ func (o *Operator) ensureSubscriptionCSVState(logger *logrus.Entry, sub *v1alpha
 		} else {
 			out.Status.State = v1alpha1.SubscriptionStateAtLatest
 		}
-		
+
 		out.Status.InstalledCSV = sub.Status.CurrentCSV
 	}
 

pkg/controller/operators/catalog/operator.go

@@ -93,6 +93,7 @@ func (a *Operator) checkAPIServiceResources(csv *v1alpha1.ClusterServiceVersion,
 
 		// Check if the APIService is adoptable
 		if !ownerutil.AdoptableLabels(apiService.GetLabels(), true, owners...) {
+			logger.WithFields(log.Fields{"obj": "apiService", "labels": apiService.GetLabels()}).Debug("adoption failed")
 			err := olmerrors.NewUnadoptableError("", apiServiceName)
 			logger.WithError(err).Warn("found unadoptable apiservice")
 			errs = append(errs, err)
@@ -554,7 +555,10 @@ func (a *Operator) installAPIServiceRequirements(desc v1alpha1.APIServiceDescrip
 	if err == nil {
 		// Check if the only owners are this CSV or in this CSV's replacement chain.
 		if ownerutil.AdoptableLabels(existingAuthDelegatorClusterRoleBinding.GetLabels(), true, csv) {
-			ownerutil.AddOwnerLabels(authDelegatorClusterRoleBinding, csv)
+			logger.WithFields(log.Fields{"obj": "authDelegatorCRB", "labels": existingAuthDelegatorClusterRoleBinding.GetLabels()}).Debug("adopting")
+			if err := ownerutil.AddOwnerLabels(authDelegatorClusterRoleBinding, csv); err != nil {
+				return nil, err
+			}
 		}
 
 		// Attempt an update.
@@ -564,7 +568,9 @@ func (a *Operator) installAPIServiceRequirements(desc v1alpha1.APIServiceDescrip
 		}
 	} else if k8serrors.IsNotFound(err) {
 		// Create the role.
-		ownerutil.AddOwnerLabels(authDelegatorClusterRoleBinding, csv)
+		if err := ownerutil.AddOwnerLabels(authDelegatorClusterRoleBinding, csv); err != nil {
+			return nil, err
+		}
 		_, err = a.OpClient.CreateClusterRoleBinding(authDelegatorClusterRoleBinding)
 		if err != nil {
 			log.Warnf("could not create auth delegator clusterrolebinding %s", authDelegatorClusterRoleBinding.GetName())
@@ -597,7 +603,10 @@ func (a *Operator) installAPIServiceRequirements(desc v1alpha1.APIServiceDescrip
 	if err == nil {
 		// Check if the only owners are this CSV or in this CSV's replacement chain.
 		if ownerutil.AdoptableLabels(existingAuthReaderRoleBinding.GetLabels(), true, csv) {
-			ownerutil.AddOwnerLabels(authReaderRoleBinding, csv)
+			logger.WithFields(log.Fields{"obj": "existingAuthReaderRB", "labels": existingAuthReaderRoleBinding.GetLabels()}).Debug("adopting")
+			if err := ownerutil.AddOwnerLabels(authReaderRoleBinding, csv); err != nil {
+				return nil, err
+			}
 		}
 		// Attempt an update.
 		if _, err := a.OpClient.UpdateRoleBinding(authReaderRoleBinding); err != nil {
@@ -606,7 +615,9 @@ func (a *Operator) installAPIServiceRequirements(desc v1alpha1.APIServiceDescrip
 		}
 	} else if k8serrors.IsNotFound(err) {
 		// Create the role.
-		ownerutil.AddOwnerLabels(authReaderRoleBinding, csv)
+		if err := ownerutil.AddOwnerLabels(authReaderRoleBinding, csv); err != nil {
+			return nil, err
+		}
 		_, err = a.OpClient.CreateRoleBinding(authReaderRoleBinding)
 		if err != nil {
 			log.Warnf("could not create auth reader role binding %s", authReaderRoleBinding.GetName())
@@ -706,12 +717,15 @@ func (a *Operator) installAPIServiceRequirements(desc v1alpha1.APIServiceDescrip
 
 		// check if the APIService is adoptable
 		if !ownerutil.AdoptableLabels(apiService.GetLabels(), true, owners...) {
+			logger.WithFields(log.Fields{"obj": "apiService", "labels": apiService.GetLabels()}).Debug("adoption failed")
 			return nil, fmt.Errorf("pre-existing APIService %s is not adoptable", apiServiceName)
 		}
 	}
 
 	// Add the CSV as an owner
-	ownerutil.AddOwnerLabels(apiService, csv)
+	if err := ownerutil.AddOwnerLabels(apiService, csv); err != nil {
+		return nil, err
+	}
 
 	// update the ServiceReference
 	apiService.Spec.Service = &apiregistrationv1.ServiceReference{

pkg/controller/operators/olm/apiservices.go

@@ -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 gcing 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,34 +485,35 @@ 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 {
-	return a.client.OperatorsV1alpha1().ClusterServiceVersions(csv.GetNamespace()).Delete(csv.GetName(), &metav1.DeleteOptions{})
+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.NewDeleteOptions(0))
 }
 
 // syncClusterServiceVersion is the method that gets called when we see a CSV event in the cluster
@@ -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
 	}
 
@@ -732,7 +753,6 @@ func (a *Operator) transitionCSVState(in v1alpha1.ClusterServiceVersion) (out *v
 		return
 	}
 
-	logger.Info("updated annotations to match current operatorgroup")
 	if err := a.ensureDeploymentAnnotations(logger, out); err != nil {
 		return nil, err
 	}
@@ -1324,6 +1344,10 @@ func (a *Operator) cleanupCSVDeployments(logger *logrus.Entry, csv *v1alpha1.Clu
 }
 
 func (a *Operator) ensureDeploymentAnnotations(logger *logrus.Entry, csv *v1alpha1.ClusterServiceVersion) error {
+	if !csv.IsSafeToUpdateOperatorGroupAnnotations() {
+		return nil
+	}
+
 	// Get csv operatorgroup annotations
 	annotations := a.copyOperatorGroupAnnotations(&csv.ObjectMeta)
 
@@ -1355,7 +1379,6 @@ func (a *Operator) ensureDeploymentAnnotations(logger *logrus.Entry, csv *v1alph
 	for _, d := range existingDeployments {
 		existingMap[d.GetName()] = d
 	}
-
 	updateErrs := []error{}
 	for _, dep := range existingMap {
 		if dep.Spec.Template.Annotations == nil {
@@ -1368,5 +1391,7 @@ func (a *Operator) ensureDeploymentAnnotations(logger *logrus.Entry, csv *v1alph
 			updateErrs = append(updateErrs, err)
 		}
 	}
+	logger.Info("updated annotations to match current operatorgroup")
+
 	return utilerrors.NewAggregate(updateErrs)
 }

pkg/controller/operators/olm/operator.go

@@ -2987,7 +2987,7 @@ func TestSyncOperatorGroups(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:            "csv-role",
 			Namespace:       operatorNamespace,
-			Labels:          ownerutil.OwnerLabel(operatorCSV),
+			Labels:          ownerutil.OwnerLabel(operatorCSV, v1alpha1.ClusterServiceVersionKind),
 			OwnerReferences: []metav1.OwnerReference{ownerutil.NonBlockingOwner(operatorCSV)},
 		},
 		Rules: permissions[0].Rules,
@@ -3001,7 +3001,7 @@ func TestSyncOperatorGroups(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:            "csv-rolebinding",
 			Namespace:       operatorNamespace,
-			Labels:          ownerutil.OwnerLabel(operatorCSV),
+			Labels:          ownerutil.OwnerLabel(operatorCSV, v1alpha1.ClusterServiceVersionKind),
 			OwnerReferences: []metav1.OwnerReference{ownerutil.NonBlockingOwner(operatorCSV)},
 		},
 		Subjects: []rbacv1.Subject{