Skip to content

Bug 2076187: Identify fail forward in csvSources #2743

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 2 additions & 17 deletions pkg/controller/operators/catalog/operator.go
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
2 changes: 1 addition & 1 deletion pkg/controller/operators/catalog/operator_test.go
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
4 changes: 2 additions & 2 deletions pkg/controller/operators/catalog/subscriptions_test.go
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
103 changes: 103 additions & 0 deletions pkg/controller/registry/resolver/fail_forward.go
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.
Comment on lines +12 to +17
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit:

Suggested change
// 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.
// 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
// 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
}
4 changes: 2 additions & 2 deletions pkg/controller/registry/resolver/instrumented_resolver.go
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
}
5 changes: 2 additions & 3 deletions pkg/controller/registry/resolver/resolver.go
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
80 changes: 1 addition & 79 deletions pkg/controller/registry/resolver/resolver_test.go
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) {
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Removed since the changes are not done on the resolver level anymore.

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
20 changes: 20 additions & 0 deletions pkg/controller/registry/resolver/source_csvs.go
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