Merge pull request #838 from ecordell/fix-dependent-subs

fix(resolver): fixes a bug where resolved dependent subscriptions don't

Author: openshift-merge-robot

pkg/controller/operators/catalog/operator.go

@@ -736,7 +736,7 @@ func (o *Operator) nothingToUpdate(logger *logrus.Entry, sub *v1alpha1.Subscript
 }
 
 func (o *Operator) ensureSubscriptionInstallPlanState(logger *logrus.Entry, sub *v1alpha1.Subscription) (*v1alpha1.Subscription, bool, error) {
-	if sub.Status.Install != nil {
+	if sub.Status.InstallPlanRef != nil {
 		return sub, false, nil
 	}
 
@@ -765,6 +765,8 @@ func (o *Operator) ensureSubscriptionInstallPlanState(logger *logrus.Entry, sub
 	out.Status.InstallPlanRef = ref
 	out.Status.Install = v1alpha1.NewInstallPlanReference(ref)
 	out.Status.State = v1alpha1.SubscriptionStateUpgradePending
+	out.Status.CurrentCSV = out.Spec.StartingCSV
+	out.Status.LastUpdated = timeNow()
 
 	updated, err := o.client.OperatorsV1alpha1().Subscriptions(sub.GetNamespace()).UpdateStatus(out)
 	if err != nil {

pkg/controller/operators/catalog/operator_test.go

@@ -142,23 +142,23 @@ func TestExecutePlan(t *testing.T) {
 						Resource: v1alpha1.StepResource{
 							CatalogSource:          "catalog",
 							CatalogSourceNamespace: namespace,
-							Group:                  "",
-							Version:                "v1",
-							Kind:                   "Service",
-							Name:                   "service",
-							Manifest:               toManifest(service("service", namespace)),
+							Group:    "",
+							Version:  "v1",
+							Kind:     "Service",
+							Name:     "service",
+							Manifest: toManifest(service("service", namespace)),
 						},
 						Status: v1alpha1.StepStatusUnknown,
 					},
 					&v1alpha1.Step{
 						Resource: v1alpha1.StepResource{
 							CatalogSource:          "catalog",
 							CatalogSourceNamespace: namespace,
-							Group:                  "operators.coreos.com",
-							Version:                "v1alpha1",
-							Kind:                   "ClusterServiceVersion",
-							Name:                   "csv",
-							Manifest:               toManifest(csv("csv", namespace, nil, nil)),
+							Group:    "operators.coreos.com",
+							Version:  "v1alpha1",
+							Kind:     "ClusterServiceVersion",
+							Name:     "csv",
+							Manifest: toManifest(csv("csv", namespace, nil, nil)),
 						},
 						Status: v1alpha1.StepStatusUnknown,
 					},

pkg/controller/operators/olm/operator.go

@@ -607,11 +607,6 @@ func (a *Operator) syncCopyCSV(obj interface{}) (syncError error) {
 		return
 	}
 
-	if len(operatorGroup.Status.Namespaces) == 1 && operatorGroup.Status.Namespaces[0] == operatorGroup.GetNamespace() {
-		logger.Debug("skipping copy for OwnNamespace operatorgroup")
-		return
-	}
-
 	logger.WithFields(logrus.Fields{
 		"targetNamespaces": strings.Join(operatorGroup.Status.Namespaces, ","),
 	}).Debug("copying csv to targets")
@@ -773,12 +768,6 @@ func (a *Operator) transitionCSVState(in v1alpha1.ClusterServiceVersion) (out *v
 		*out = *updated
 	}
 
-	// Check if the current CSV is being replaced, return with replacing status if so
-	if err := a.checkReplacementsAndUpdateStatus(out); err != nil {
-		logger.WithError(err).Info("replacement check")
-		return
-	}
-
 	// Verify CSV operatorgroup (and update annotations if needed)
 	operatorGroup, err := a.operatorGroupForCSV(out, logger)
 	if operatorGroup == nil {
@@ -889,6 +878,15 @@ func (a *Operator) transitionCSVState(in v1alpha1.ClusterServiceVersion) (out *v
 		}
 		out.SetRequirementStatus(statuses)
 
+		// Check if we need to requeue the previous
+		if prev := a.isReplacing(out); prev != nil {
+			if prev.Status.Phase == v1alpha1.CSVPhaseSucceeded {
+				if err := a.csvQueueSet.Requeue(prev.GetName(), prev.GetNamespace()); err != nil {
+					a.Log.WithError(err).Warn("error requeueing previous")
+				}
+			}
+		}
+
 		if !met {
 			logger.Info("requirements were not met")
 			out.SetPhaseWithEventIfChanged(v1alpha1.CSVPhasePending, v1alpha1.CSVReasonRequirementsNotMet, "one or more requirements couldn't be found", now, a.recorder)
@@ -912,6 +910,13 @@ func (a *Operator) transitionCSVState(in v1alpha1.ClusterServiceVersion) (out *v
 			return
 		}
 
+		// Check if we're not ready to install part of the replacement chain yet
+		if prev := a.isReplacing(out); prev != nil {
+			if prev.Status.Phase != v1alpha1.CSVPhaseReplacing {
+				return
+			}
+		}
+
 		logger.Info("scheduling ClusterServiceVersion for install")
 		out.SetPhaseWithEvent(v1alpha1.CSVPhaseInstallReady, v1alpha1.CSVReasonRequirementsMet, "all requirements found, attempting install", now, a.recorder)
 	case v1alpha1.CSVPhaseInstallReady:
@@ -955,6 +960,12 @@ func (a *Operator) transitionCSVState(in v1alpha1.ClusterServiceVersion) (out *v
 		}
 
 	case v1alpha1.CSVPhaseSucceeded:
+		// Check if the current CSV is being replaced, return with replacing status if so
+		if err := a.checkReplacementsAndUpdateStatus(out); err != nil {
+			logger.WithError(err).Info("replacement check")
+			return
+		}
+
 		installer, strategy, _ := a.parseStrategiesAndUpdateStatus(out)
 		if strategy == nil {
 			return
@@ -1073,6 +1084,14 @@ func (a *Operator) transitionCSVState(in v1alpha1.ClusterServiceVersion) (out *v
 			return
 		}
 
+		// If we are a leaf, we should requeue the replacement for processing
+		if next := a.isBeingReplaced(out, a.csvSet(out.GetNamespace(), v1alpha1.CSVPhaseAny)); next != nil {
+			err := a.csvQueueSet.Requeue(next.GetName(), next.GetNamespace())
+			if err != nil {
+				a.Log.WithError(err).Warn("error requeuing replacement")
+			}
+		}
+
 		// If we can find a newer version that's successfully installed, we're safe to mark all intermediates
 		for _, csv := range a.findIntermediatesForDeletion(out) {
 			// we only mark them in this step, in case some get deleted but others fail and break the replacement chain
@@ -1093,6 +1112,10 @@ func (a *Operator) transitionCSVState(in v1alpha1.ClusterServiceVersion) (out *v
 		}
 	case v1alpha1.CSVPhaseDeleting:
 		var immediate int64 = 0
+
+		if err := a.csvQueueSet.Remove(out.GetNamespace(), out.GetName()); err != nil {
+			logger.WithError(err).Debug("error removing from queue")
+		}
 		syncError = a.client.OperatorsV1alpha1().ClusterServiceVersions(out.GetNamespace()).Delete(out.GetName(), &metav1.DeleteOptions{GracePeriodSeconds: &immediate})
 		if syncError != nil {
 			logger.Debugf("unable to get delete csv marked for deletion: %s", syncError.Error())
@@ -1158,7 +1181,7 @@ func (a *Operator) checkReplacementsAndUpdateStatus(csv *v1alpha1.ClusterService
 	}
 	if replacement := a.isBeingReplaced(csv, a.csvSet(csv.GetNamespace(), v1alpha1.CSVPhaseAny)); replacement != nil {
 		a.Log.Infof("newer ClusterServiceVersion replacing %s, no-op", csv.SelfLink)
-		msg := fmt.Sprintf("being replaced by csv: %s", replacement.SelfLink)
+		msg := fmt.Sprintf("being replaced by csv: %s", replacement.GetName())
 		csv.SetPhaseWithEvent(v1alpha1.CSVPhaseReplacing, v1alpha1.CSVReasonBeingReplaced, msg, timeNow(), a.recorder)
 		metrics.CSVUpgradeCount.Inc()
 
@@ -1237,10 +1260,15 @@ func (a *Operator) parseStrategiesAndUpdateStatus(csv *v1alpha1.ClusterServiceVe
 	return installer, strategy, previousStrategy
 }
 
-func (a *Operator) crdOwnerConflicts(in *v1alpha1.ClusterServiceVersion, csvs map[string]*v1alpha1.ClusterServiceVersion) error {
-	for _, crd := range in.Spec.CustomResourceDefinitions.Owned {
-		for name, csv := range csvs {
-			if name != in.GetName() && in.Spec.Replaces != name && csv.OwnsCRD(crd.Name) {
+func (a *Operator) crdOwnerConflicts(in *v1alpha1.ClusterServiceVersion, csvsInNamespace map[string]*v1alpha1.ClusterServiceVersion) error {
+	csvsInChain := a.getReplacementChain(in, csvsInNamespace)
+	// find csvs in the namespace that are not part of the replacement chain
+	for name, csv := range csvsInNamespace {
+		if _, ok := csvsInChain[name]; ok {
+			continue
+		}
+		for _, crd := range in.Spec.CustomResourceDefinitions.Owned {
+			if name != in.GetName() && csv.OwnsCRD(crd.Name) {
 				return ErrCRDOwnerConflict
 			}
 		}
@@ -1249,6 +1277,53 @@ func (a *Operator) crdOwnerConflicts(in *v1alpha1.ClusterServiceVersion, csvs ma
 	return nil
 }
 
+func (a *Operator) getReplacementChain(in *v1alpha1.ClusterServiceVersion, csvsInNamespace map[string]*v1alpha1.ClusterServiceVersion) map[string]struct{} {
+	current := in.GetName()
+	csvsInChain := map[string]struct{}{
+		current: {},
+	}
+
+	replacement := func(csvName string) *string {
+		for _, csv := range csvsInNamespace {
+			if csv.Spec.Replaces == csvName {
+				name := csv.GetName()
+				return &name
+			}
+		}
+		return nil
+	}
+
+	replaces := func(replaces string) *string {
+		for _, csv := range csvsInNamespace {
+			name := csv.GetName()
+			if name == replaces {
+				rep := csv.Spec.Replaces
+				return &rep
+			}
+		}
+		return nil
+	}
+
+	next := replacement(current)
+	for next != nil {
+		csvsInChain[*next] = struct{}{}
+		current = *next
+		next = replacement(current)
+	}
+
+	current = in.Spec.Replaces
+	prev := replaces(current)
+	if prev != nil {
+		csvsInChain[current] = struct{}{}
+	}
+	for prev != nil && *prev != "" {
+		current = *prev
+		csvsInChain[current] = struct{}{}
+		prev = replaces(current)
+	}
+	return csvsInChain
+}
+
 func (a *Operator) apiServiceOwnerConflicts(csv *v1alpha1.ClusterServiceVersion) error {
 	// Get replacing CSV if exists
 	replacing, err := a.lister.OperatorsV1alpha1().ClusterServiceVersionLister().ClusterServiceVersions(csv.GetNamespace()).Get(csv.Spec.Replaces)
@@ -1297,7 +1372,9 @@ func (a *Operator) isReplacing(in *v1alpha1.ClusterServiceVersion) *v1alpha1.Clu
 	if in.Spec.Replaces == "" {
 		return nil
 	}
-	previous, err := a.lister.OperatorsV1alpha1().ClusterServiceVersionLister().ClusterServiceVersions(in.GetNamespace()).Get(in.Spec.Replaces)
+
+	// using the client instead of a lister; missing an object because of a cache sync can cause upgrades to fail
+	previous, err := a.client.OperatorsV1alpha1().ClusterServiceVersions(in.GetNamespace()).Get(in.Spec.Replaces, metav1.GetOptions{})
 	if err != nil {
 		a.Log.WithField("replacing", in.Spec.Replaces).WithError(err).Debugf("unable to get previous csv")
 		return nil