Skip to content

Commit

Permalink
Merge pull request #290 from awgreene/revert-resolver-changes
Browse files Browse the repository at this point in the history
Identify fail forward in csvSources (#2743)
  • Loading branch information
openshift-merge-robot authored Apr 19, 2022
2 parents cd2e995 + 72d6b3e commit f70f4d6
Show file tree
Hide file tree
Showing 19 changed files with 338 additions and 388 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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 {
Expand All @@ -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
}
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
},
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
},
}
Expand Down Expand Up @@ -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",
Expand Down
Original file line number Diff line number Diff line change
@@ -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
}
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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
}

Expand All @@ -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
}
Expand All @@ -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
}
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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}
Expand Down Expand Up @@ -1521,6 +1442,7 @@ func TestSolveOperators_TransferApiOwnership(t *testing.T) {
key: cache.NewVirtualSourceKey(namespace),
csvLister: &csvs,
subLister: fakeSubscriptionLister(p.subs),
ogLister: fakeOperatorGroupLister{},
logger: logger,
},
}),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -20,6 +21,7 @@ import (
type csvSourceProvider struct {
csvLister v1alpha1listers.ClusterServiceVersionLister
subLister v1alpha1listers.SubscriptionLister
ogLister v1listers.OperatorGroupLister
logger logrus.StdLogger
}

Expand All @@ -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
Expand All @@ -41,6 +44,7 @@ type csvSource struct {
key cache.SourceKey
csvLister v1alpha1listers.ClusterServiceVersionNamespaceLister
subLister v1alpha1listers.SubscriptionNamespaceLister
ogLister v1listers.OperatorGroupNamespaceLister
logger logrus.StdLogger
}

Expand All @@ -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 {
Expand All @@ -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
Expand Down
Loading

0 comments on commit f70f4d6

Please sign in to comment.