Identify fail forward in csvSources (#2743)

This commit undos changes to the Resolve and ResolveSteps methods and
updats the csvSourceProvider to infer whether or not fail forward is
enabled in a namespace.

Signed-off-by: Alexander Greene <[email protected]>

Author: awgreene

pkg/controller/operators/catalog/operator.go

@@ -42,7 +42,6 @@ import (
 	"k8s.io/client-go/util/workqueue"
 
 	"github.com/operator-framework/api/pkg/operators/reference"
-	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/clientset/versioned"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/informers/externalversions"
@@ -903,20 +902,6 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
 	return
 }
 
-func (o *Operator) isFailForwardEnabled(namespace string) (bool, error) {
-	ogs, err := o.lister.OperatorsV1().OperatorGroupLister().OperatorGroups(namespace).List(labels.Everything())
-	if err != nil {
-		o.logger.Debugf("failed to list operatorgroups in the %s namespace: %v", namespace, err)
-		// Couldn't list operatorGroups, assuming default upgradeStrategy
-		// so existing behavior is observed for failed CSVs.
-		return false, nil
-	}
-	if len(ogs) != 1 {
-		return false, fmt.Errorf("found %d operatorGroups in namespace %s, expected 1", len(ogs), namespace)
-	}
-	return ogs[0].UpgradeStrategy() == operatorsv1.UpgradeStrategyUnsafeFailForward, nil
-}
-
 func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 	ns, ok := obj.(*corev1.Namespace)
 	if !ok {
@@ -943,7 +928,7 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 		return err
 	}
 
-	failForwardEnabled, err := o.isFailForwardEnabled(namespace)
+	failForwardEnabled, err := resolver.IsFailForwardEnabled(o.lister.OperatorsV1().OperatorGroupLister().OperatorGroups(namespace))
 	if err != nil {
 		return err
 	}
@@ -998,7 +983,7 @@ func (o *Operator) syncResolvingNamespace(obj interface{}) error {
 	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, bundleLookups, updatedSubs, err := o.resolver.ResolveSteps(namespace, failForwardEnabled)
+	steps, bundleLookups, updatedSubs, err := o.resolver.ResolveSteps(namespace)
 	if err != nil {
 		go o.recorder.Event(ns, corev1.EventTypeWarning, "ResolutionFailed", err.Error())
 		// If the error is constraints not satisfiable, then simply project the

pkg/controller/operators/catalog/operator_test.go

@@ -1254,7 +1254,7 @@ func TestSyncResolvingNamespace(t *testing.T) {
 
 			o.sourcesLastUpdate.Set(tt.fields.sourcesLastUpdate.Time)
 			o.resolver = &fakes.FakeStepResolver{
-				ResolveStepsStub: func(string, bool) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
+				ResolveStepsStub: func(string) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
 					return nil, nil, nil, tt.fields.resolveErr
 				},
 			}

pkg/controller/operators/catalog/subscriptions_test.go

@@ -1031,7 +1031,7 @@ func TestSyncSubscriptions(t *testing.T) {
 
 			o.sourcesLastUpdate.Set(tt.fields.sourcesLastUpdate.Time)
 			o.resolver = &fakes.FakeStepResolver{
-				ResolveStepsStub: func(string, bool) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
+				ResolveStepsStub: func(string) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
 					return tt.fields.resolveSteps, tt.fields.bundleLookups, tt.fields.resolveSubs, tt.fields.resolveErr
 				},
 			}
@@ -1168,7 +1168,7 @@ func BenchmarkSyncResolvingNamespace(b *testing.B) {
 				},
 			},
 			resolver: &fakes.FakeStepResolver{
-				ResolveStepsStub: func(string, bool) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
+				ResolveStepsStub: func(string) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
 					steps := []*v1alpha1.Step{
 						{
 							Resolving: "csv.v.2",

pkg/controller/registry/resolver/fail_forward.go

@@ -0,0 +1,103 @@
+package resolver
+
+import (
+	"fmt"
+
+	operatorsv1 "github.com/operator-framework/api/pkg/operators/v1"
+	operatorsv1alpha1 "github.com/operator-framework/api/pkg/operators/v1alpha1"
+	v1listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1"
+	"k8s.io/apimachinery/pkg/labels"
+)
+
+// IsFailForwardEnabled takes a namespaced operatorGroup lister and returns
+// True if an operatorGroup exists in the namespace and its upgradeStrategy
+// is set to UnsafeFailForward and false otherwise. An error is returned if
+// an more than one operatorGroup exists in the namespace.
+// No error is returned if no OperatorGroups are found to keep the resolver
+// backwards compatible.
+func IsFailForwardEnabled(ogLister v1listers.OperatorGroupNamespaceLister) (bool, error) {
+	ogs, err := ogLister.List(labels.Everything())
+	if err != nil || len(ogs) == 0 {
+		return false, nil
+	}
+	if len(ogs) != 1 {
+		return false, fmt.Errorf("found %d operatorGroups, expected 1", len(ogs))
+	}
+	return ogs[0].UpgradeStrategy() == operatorsv1.UpgradeStrategyUnsafeFailForward, nil
+}
+
+type walkOption func(csv *operatorsv1alpha1.ClusterServiceVersion) error
+
+// WithCSVPhase returns an error if the CSV is not in the given phase.
+func WithCSVPhase(phase operatorsv1alpha1.ClusterServiceVersionPhase) walkOption {
+	return func(csv *operatorsv1alpha1.ClusterServiceVersion) error {
+		if csv == nil || csv.Status.Phase != phase {
+			return fmt.Errorf("csv %s/%s in phase %s instead of %s", csv.GetNamespace(), csv.GetName(), csv.Status.Phase, phase)
+		}
+		return nil
+	}
+}
+
+// WithUniqueCSVs returns an error if the CSV has been seen before.
+func WithUniqueCSVs() walkOption {
+	visited := map[string]struct{}{}
+	return func(csv *operatorsv1alpha1.ClusterServiceVersion) error {
+		// Check if we have visited the CSV before
+		if _, ok := visited[csv.GetName()]; ok {
+			return fmt.Errorf("csv %s/%s has already been seen", csv.GetNamespace(), csv.GetName())
+		}
+
+		visited[csv.GetName()] = struct{}{}
+		return nil
+	}
+}
+
+// WalkReplacementChain walks along the chain of clusterServiceVersions being replaced and returns
+// the last clusterServiceVersions in the replacement chain. An error is returned if any of the
+// clusterServiceVersions before the last is not in the replaces phase or if an infinite replacement
+// chain is detected.
+func WalkReplacementChain(csv *operatorsv1alpha1.ClusterServiceVersion, csvToReplacement map[string]*operatorsv1alpha1.ClusterServiceVersion, options ...walkOption) (*operatorsv1alpha1.ClusterServiceVersion, error) {
+	if csv == nil {
+		return nil, fmt.Errorf("csv cannot be nil")
+	}
+
+	for {
+		// Check if there is a CSV that replaces this CSVs
+		next, ok := csvToReplacement[csv.GetName()]
+		if !ok {
+			break
+		}
+
+		// Check walk options
+		for _, o := range options {
+			if err := o(csv); err != nil {
+				return nil, err
+			}
+		}
+
+		// Move along replacement chain.
+		csv = next
+	}
+	return csv, nil
+}
+
+// isReplacementChainThatEndsInFailure returns true if the last CSV in the chain is in the failed phase and all other
+// CSVs are in the replacing phase.
+func isReplacementChainThatEndsInFailure(csv *operatorsv1alpha1.ClusterServiceVersion, csvToReplacement map[string]*operatorsv1alpha1.ClusterServiceVersion) (bool, error) {
+	lastCSV, err := WalkReplacementChain(csv, csvToReplacement, WithCSVPhase(operatorsv1alpha1.CSVPhaseReplacing), WithUniqueCSVs())
+	if err != nil {
+		return false, err
+	}
+	return (lastCSV != nil && lastCSV.Status.Phase == operatorsv1alpha1.CSVPhaseFailed), nil
+}
+
+// ReplacementMapping takes a list of CSVs and returns a map that maps a CSV's name to the CSV that replaces it.
+func ReplacementMapping(csvs []*operatorsv1alpha1.ClusterServiceVersion) map[string]*operatorsv1alpha1.ClusterServiceVersion {
+	replacementMapping := map[string]*operatorsv1alpha1.ClusterServiceVersion{}
+	for _, csv := range csvs {
+		if csv.Spec.Replaces != "" {
+			replacementMapping[csv.Spec.Replaces] = csv
+		}
+	}
+	return replacementMapping
+}

pkg/controller/registry/resolver/instrumented_resolver.go

@@ -22,9 +22,9 @@ func NewInstrumentedResolver(resolver StepResolver, successMetricsEmitter, failu
 	}
 }
 
-func (ir *InstrumentedResolver) ResolveSteps(namespace string, failForwardEnabled bool) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
+func (ir *InstrumentedResolver) ResolveSteps(namespace string) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
 	start := time.Now()
-	steps, lookups, subs, err := ir.resolver.ResolveSteps(namespace, failForwardEnabled)
+	steps, lookups, subs, err := ir.resolver.ResolveSteps(namespace)
 	if err != nil {
 		ir.failureMetricsEmitter(time.Since(start))
 	} else {

pkg/controller/registry/resolver/instrumented_resolver_test.go

@@ -17,11 +17,11 @@ const (
 type fakeResolverWithError struct{}
 type fakeResolverWithoutError struct{}
 
-func (r *fakeResolverWithError) ResolveSteps(namespace string, failForwardEnabled bool) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
+func (r *fakeResolverWithError) ResolveSteps(namespace string) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
 	return nil, nil, nil, errors.New("Fake error")
 }
 
-func (r *fakeResolverWithoutError) ResolveSteps(namespace string, failForwardEnabled bool) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
+func (r *fakeResolverWithoutError) ResolveSteps(namespace string) ([]*v1alpha1.Step, []v1alpha1.BundleLookup, []*v1alpha1.Subscription, error) {
 	return nil, nil, nil, nil
 }
 
@@ -45,7 +45,7 @@ func TestInstrumentedResolverFailure(t *testing.T) {
 	}
 
 	instrumentedResolver := NewInstrumentedResolver(newFakeResolverWithError(), changeToSuccess, changeToFailure)
-	instrumentedResolver.ResolveSteps("", false)
+	instrumentedResolver.ResolveSteps("")
 	require.Equal(t, len(result), 1)     // check that only one call was made to a change function
 	require.Equal(t, result[0], failure) // check that the call was made to changeToFailure function
 }
@@ -62,7 +62,7 @@ func TestInstrumentedResolverSuccess(t *testing.T) {
 	}
 
 	instrumentedResolver := NewInstrumentedResolver(newFakeResolverWithoutError(), changeToSuccess, changeToFailure)
-	instrumentedResolver.ResolveSteps("", false)
+	instrumentedResolver.ResolveSteps("")
 	require.Equal(t, len(result), 1)     // check that only one call was made to a change function
 	require.Equal(t, result[0], success) // check that the call was made to changeToSuccess function
 }

pkg/controller/registry/resolver/resolver.go

@@ -55,7 +55,7 @@ func (w *debugWriter) Write(b []byte) (int, error) {
 	return n, nil
 }
 
-func (r *Resolver) Resolve(namespaces []string, subs []*v1alpha1.Subscription, existingEntryPredicates ...cache.Predicate) ([]*cache.Entry, error) {
+func (r *Resolver) Resolve(namespaces []string, subs []*v1alpha1.Subscription) ([]*cache.Entry, error) {
 	var errs []error
 
 	variables := make(map[solver.Identifier]solver.Variable)
@@ -72,8 +72,7 @@ func (r *Resolver) Resolve(namespaces []string, subs []*v1alpha1.Subscription, e
 	}
 
 	preferredNamespace := namespaces[0]
-	existingEntryPredicates = append(existingEntryPredicates, cache.True())
-	_, existingVariables, err := r.getBundleVariables(preferredNamespace, namespacedCache.Catalog(cache.NewVirtualSourceKey(preferredNamespace)).Find(existingEntryPredicates...), namespacedCache, visited)
+	_, existingVariables, err := r.getBundleVariables(preferredNamespace, namespacedCache.Catalog(cache.NewVirtualSourceKey(preferredNamespace)).Find(cache.True()), namespacedCache, visited)
 	if err != nil {
 		return nil, err
 	}

pkg/controller/registry/resolver/resolver_test.go

@@ -191,85 +191,6 @@ func TestSolveOperators_WithSystemConstraints(t *testing.T) {
 	}
 }
 
-func WithInstalledCSV(sub *v1alpha1.Subscription, csvName string) *v1alpha1.Subscription {
-	sub.Status.InstalledCSV = csvName
-	return sub
-}
-
-func TestSolveOperators_WithFailForward(t *testing.T) {
-	const namespace = "test-namespace"
-	catalog := cache.SourceKey{Name: "test-catalog", Namespace: namespace}
-
-	packageASubV2 := newSub(namespace, "packageA", "alpha", catalog)
-	APISet := cache.APISet{opregistry.APIKey{Group: "g", Version: "v", Kind: "k", Plural: "ks"}: struct{}{}}
-
-	// packageA provides an API
-	packageAV1 := genEntry("packageA.v1", "0.0.1", "", "packageA", "alpha", catalog.Name, catalog.Namespace, nil, APISet, nil, "", false)
-	packageAV2 := genEntry("packageA.v2", "0.0.2", "packageA.v1", "packageA", "alpha", catalog.Name, catalog.Namespace, nil, APISet, nil, "", false)
-	packageAV3 := genEntry("packageA.v3", "0.0.3", "packageA.v2", "packageA", "alpha", catalog.Name, catalog.Namespace, nil, APISet, nil, "", false)
-
-	existingPackageAV1 := existingOperator(namespace, "packageA.v1", "packageA", "alpha", "", APISet, nil, nil, nil)
-	existingPackageAV2 := existingOperator(namespace, "packageA.v2", "packageA", "alpha", "packageA.v1", APISet, nil, nil, nil)
-
-	testCases := []struct {
-		title                 string
-		expectedOperators     []*cache.Entry
-		csvs                  []*v1alpha1.ClusterServiceVersion
-		subs                  []*v1alpha1.Subscription
-		snapshotEntries       []*cache.Entry
-		failForwardPredicates []cache.Predicate
-		err                   string
-	}{
-		{
-			title:             "Resolver fails if v1 and v2 provide the same APIs and v1 is not omitted from the resolver",
-			snapshotEntries:   []*cache.Entry{packageAV1, packageAV2},
-			expectedOperators: nil,
-			csvs:              []*v1alpha1.ClusterServiceVersion{existingPackageAV1, existingPackageAV2},
-			subs:              []*v1alpha1.Subscription{WithInstalledCSV(packageASubV2, existingPackageAV2.Name)},
-			err:               "provide k (g/v)",
-		},
-		{
-			title:                 "Resolver succeeds if v1 and v2 provide the same APIs and v1 is omitted from the resolver",
-			snapshotEntries:       []*cache.Entry{packageAV1, packageAV2},
-			expectedOperators:     nil,
-			csvs:                  []*v1alpha1.ClusterServiceVersion{existingPackageAV1, existingPackageAV2},
-			subs:                  []*v1alpha1.Subscription{WithInstalledCSV(packageASubV2, existingPackageAV2.Name)},
-			failForwardPredicates: []cache.Predicate{cache.Not(cache.CSVNamePredicate("packageA.v1"))},
-			err:                   "",
-		},
-		{
-			title:                 "Resolver succeeds if v1 and v2 provide the same APIs, v1 is omitted from the resolver, and an upgrade for v2 exists",
-			snapshotEntries:       []*cache.Entry{packageAV1, packageAV2, packageAV3},
-			expectedOperators:     []*cache.Entry{packageAV3},
-			csvs:                  []*v1alpha1.ClusterServiceVersion{existingPackageAV1, existingPackageAV2},
-			subs:                  []*v1alpha1.Subscription{WithInstalledCSV(packageASubV2, existingPackageAV2.Name)},
-			failForwardPredicates: []cache.Predicate{cache.Not(cache.CSVNamePredicate("packageA.v1"))},
-			err:                   "",
-		},
-	}
-
-	for _, testCase := range testCases {
-		resolver := Resolver{
-			cache: cache.New(cache.StaticSourceProvider{
-				catalog: &cache.Snapshot{
-					Entries: testCase.snapshotEntries,
-				},
-				cache.NewVirtualSourceKey(namespace): csvSnapshotOrPanic(namespace, testCase.subs, testCase.csvs...),
-			}),
-			log: logrus.New(),
-		}
-		operators, err := resolver.Resolve([]string{namespace}, testCase.subs, testCase.failForwardPredicates...)
-
-		if testCase.err != "" {
-			require.Error(t, err)
-			require.Containsf(t, err.Error(), testCase.err, "Test %s failed", testCase.title)
-		} else {
-			require.NoErrorf(t, err, "Test %s failed", testCase.title)
-		}
-		require.ElementsMatch(t, testCase.expectedOperators, operators, "Test %s failed", testCase.title)
-	}
-}
-
 func TestDisjointChannelGraph(t *testing.T) {
 	const namespace = "test-namespace"
 	catalog := cache.SourceKey{Name: "test-catalog", Namespace: namespace}
@@ -1521,6 +1442,7 @@ func TestSolveOperators_TransferApiOwnership(t *testing.T) {
 						key:       cache.NewVirtualSourceKey(namespace),
 						csvLister: &csvs,
 						subLister: fakeSubscriptionLister(p.subs),
+						ogLister:  fakeOperatorGroupLister{},
 						logger:    logger,
 					},
 				}),

pkg/controller/registry/resolver/source_csvs.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/blang/semver/v4"
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	v1listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1"
 	v1alpha1listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver/cache"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver/projection"
@@ -20,6 +21,7 @@ import (
 type csvSourceProvider struct {
 	csvLister v1alpha1listers.ClusterServiceVersionLister
 	subLister v1alpha1listers.SubscriptionLister
+	ogLister  v1listers.OperatorGroupLister
 	logger    logrus.StdLogger
 }
 
@@ -30,6 +32,7 @@ func (csp *csvSourceProvider) Sources(namespaces ...string) map[cache.SourceKey]
 			key:       cache.NewVirtualSourceKey(namespace),
 			csvLister: csp.csvLister.ClusterServiceVersions(namespace),
 			subLister: csp.subLister.Subscriptions(namespace),
+			ogLister:  csp.ogLister.OperatorGroups(namespace),
 			logger:    csp.logger,
 		}
 		break // first ns is assumed to be the target ns, todo: make explicit
@@ -41,6 +44,7 @@ type csvSource struct {
 	key       cache.SourceKey
 	csvLister v1alpha1listers.ClusterServiceVersionNamespaceLister
 	subLister v1alpha1listers.SubscriptionNamespaceLister
+	ogLister  v1listers.OperatorGroupNamespaceLister
 	logger    logrus.StdLogger
 }
 
@@ -55,6 +59,11 @@ func (s *csvSource) Snapshot(ctx context.Context) (*cache.Snapshot, error) {
 		return nil, err
 	}
 
+	failForwardEnabled, err := IsFailForwardEnabled(s.ogLister)
+	if err != nil {
+		return nil, err
+	}
+
 	// build a catalog snapshot of CSVs without subscriptions
 	csvSubscriptions := make(map[*v1alpha1.ClusterServiceVersion]*v1alpha1.Subscription)
 	for _, sub := range subs {
@@ -75,6 +84,17 @@ func (s *csvSource) Snapshot(ctx context.Context) (*cache.Snapshot, error) {
 		if csv.IsCopied() {
 			continue
 		}
+
+		if failForwardEnabled {
+			replacementChainEndsInFailure, err := isReplacementChainThatEndsInFailure(csv, ReplacementMapping(csvs))
+			if err != nil {
+				return nil, err
+			}
+			if csv.Status.Phase == v1alpha1.CSVPhaseReplacing && replacementChainEndsInFailure {
+				continue
+			}
+		}
+
 		entry, err := newEntryFromV1Alpha1CSV(csv)
 		if err != nil {
 			return nil, err