Annotate CRDs that are installed alongside CSVs.

In the absence of a way to correlate installed resources back to their
bundle of origin, it's possible for upgrades to complete
early. Upgrades begin when the catalog operator creates a new
CSV (with .spec.replaces set) as part of InstallPlan execution, and
complete when the olm operator sees that the new CSV has .status.phase
Succeeded. When both versions of the operator being upgraded own the
same CRD, the new CSV may use the existing CRD to satisfy its CRD
ownership requirement, thereby transitioning to Succeeded and deleting
the old CSV.

This problem can be avoided for namespace-scoped bundle resources,
which perform the transfer by adding the new CSV as an owner reference
and allowing garbage collection to remove the old owner reference
post-upgrade. This is not possible for CRDs because cluster-scoped
resources cannot have owner references to a namespace-scoped resource
like ClusterServiceVersion.

It may be necessary to make a similar change affecting other
cluster-scoped resources, but CRDs are a common culprit of early
upgrade completion. Ideally, a first-class, cluster-scoped Bundle
resource will obsolete the need to use an annotation to track bundle
of origin.

Signed-off-by: Ben Luddy <[email protected]>

Author: benluddy

pkg/controller/operators/catalog/operator.go

@@ -1642,7 +1642,7 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 		o.logger.Errorf("failed to get a client for plan execution- %v", err)
 		return err
 	}
-	b := newBuilder(builderKubeClient, builderDynamicClient, o.csvProvidedAPIsIndexer, r, o.logger)
+	b := newBuilder(plan, o.lister.OperatorsV1alpha1().ClusterServiceVersionLister(), builderKubeClient, builderDynamicClient, o.csvProvidedAPIsIndexer, r, o.logger)
 
 	for i, step := range plan.Status.Plan {
 		doStep := true

pkg/controller/operators/catalog/step.go

@@ -3,22 +3,27 @@ package catalog
 import (
 	"context"
 	"fmt"
+	"strings"
 
-	"k8s.io/client-go/dynamic"
-	"k8s.io/client-go/tools/cache"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
 
-	"github.com/operator-framework/api/pkg/operators/v1alpha1"
-	crdlib "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/crd"
-	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
-	errorwrap "github.com/pkg/errors"
-	logger "github.com/sirupsen/logrus"
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	apiextensionsv1beta1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 	apiextensionsv1client "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1"
 	apiextensionsv1beta1client "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1beta1"
-
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/util/yaml"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	listersv1alpha1 "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1alpha1"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/internal/alongside"
+	crdlib "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/crd"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
 )
 
 // Stepper manages cluster interactions based on the step.
@@ -36,16 +41,22 @@ func (s StepperFunc) Status() (v1alpha1.StepStatus, error) {
 // Builder holds clients and data structures required for the StepBuilder to work
 // Builder attributes are not to meant to be accessed outside the StepBuilder method
 type builder struct {
+	plan              *v1alpha1.InstallPlan
+	csvLister         listersv1alpha1.ClusterServiceVersionLister
 	opclient          operatorclient.ClientInterface
 	dynamicClient     dynamic.Interface
 	manifestResolver  ManifestResolver
 	csvToProvidedAPIs map[string]cache.Indexer
-	logger            logger.FieldLogger
+	logger            logrus.FieldLogger
+
+	annotator alongside.Annotator
 }
 
-func newBuilder(opclient operatorclient.ClientInterface, dynamicClient dynamic.Interface, csvToProvidedAPIs map[string]cache.Indexer,
-	manifestResolver ManifestResolver, logger logger.FieldLogger) *builder {
+func newBuilder(plan *v1alpha1.InstallPlan, csvLister listersv1alpha1.ClusterServiceVersionLister, opclient operatorclient.ClientInterface, dynamicClient dynamic.Interface, csvToProvidedAPIs map[string]cache.Indexer,
+	manifestResolver ManifestResolver, logger logrus.FieldLogger) *builder {
 	return &builder{
+		plan:              plan,
+		csvLister:         csvLister,
 		opclient:          opclient,
 		dynamicClient:     dynamicClient,
 		manifestResolver:  manifestResolver,
@@ -62,6 +73,29 @@ func (n notSupportedStepperErr) Error() string {
 	return n.message
 }
 
+func setInstalledAlongsideAnnotation(a alongside.Annotator, o metav1.Object, namespace string, name string, lister listersv1alpha1.ClusterServiceVersionLister) {
+	var (
+		nns []alongside.NamespacedName
+	)
+
+	// Only keep references to existing and non-copied CSVs to
+	// avoid unbounded growth.
+	for _, nn := range a.FromObject(o) {
+		if csv, err := lister.ClusterServiceVersions(nn.Namespace).Get(nn.Name); k8serrors.IsNotFound(err) {
+			continue
+		} else if err == nil && csv.IsCopied() {
+			continue
+		}
+		// CSV exists and is not copied OR err is non-nil, but
+		// not 404. Safer to assume it exists in unhandled
+		// error cases and try again next time.
+		nns = append(nns, nn)
+	}
+	nns = append(nns, alongside.NamespacedName{Namespace: namespace, Name: name})
+
+	a.ToObject(o, nns)
+}
+
 // step is a factory that creates StepperFuncs based on the install plan step Kind.
 func (b *builder) create(step v1alpha1.Step) (Stepper, error) {
 	manifest, err := b.manifestResolver.ManifestForStep(&step)
@@ -75,6 +109,20 @@ func (b *builder) create(step v1alpha1.Step) (Stepper, error) {
 		if err != nil {
 			return nil, err
 		}
+
+		if step.Resolving != "" {
+			var u unstructured.Unstructured
+			if err := yaml.NewYAMLOrJSONDecoder(strings.NewReader(manifest), 32).Decode(&u); err != nil {
+				return nil, err
+			}
+			setInstalledAlongsideAnnotation(b.annotator, &u, b.plan.GetNamespace(), step.Resolving, b.csvLister)
+			if b, err := u.MarshalJSON(); err != nil {
+				return nil, err
+			} else {
+				manifest = string(b)
+			}
+		}
+
 		switch version {
 		case crdlib.V1Version:
 			return b.NewCRDV1Step(b.opclient.ApiextensionsInterface().ApiextensionsV1(), &step, manifest), nil
@@ -98,7 +146,7 @@ func (b *builder) NewCRDV1Step(client apiextensionsv1client.ApiextensionsV1Inter
 				if k8serrors.IsNotFound(err) {
 					return v1alpha1.StepStatusNotPresent, nil
 				} else {
-					return v1alpha1.StepStatusNotPresent, errorwrap.Wrapf(err, "error finding the %s CRD", crd.Name)
+					return v1alpha1.StepStatusNotPresent, errors.Wrapf(err, "error finding the %s CRD", crd.Name)
 				}
 			}
 			established, namesAccepted := false, false
@@ -128,23 +176,23 @@ func (b *builder) NewCRDV1Step(client apiextensionsv1client.ApiextensionsV1Inter
 				currentCRD, _ := client.CustomResourceDefinitions().Get(context.TODO(), crd.GetName(), metav1.GetOptions{})
 				crd.SetResourceVersion(currentCRD.GetResourceVersion())
 				if err = validateV1CRDCompatibility(b.dynamicClient, currentCRD, crd); err != nil {
-					return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "error validating existing CRs against new CRD's schema: %s", step.Resource.Name)
+					return v1alpha1.StepStatusUnknown, errors.Wrapf(err, "error validating existing CRs against new CRD's schema: %s", step.Resource.Name)
 				}
 
 				// check to see if stored versions changed and whether the upgrade could cause potential data loss
 				safe, err := crdlib.SafeStorageVersionUpgrade(currentCRD, crd)
 				if !safe {
 					b.logger.Errorf("risk of data loss updating %s: %s", step.Resource.Name, err)
-					return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "risk of data loss updating %s", step.Resource.Name)
+					return v1alpha1.StepStatusUnknown, errors.Wrapf(err, "risk of data loss updating %s", step.Resource.Name)
 				}
 				if err != nil {
-					return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "checking CRD for potential data loss updating %s", step.Resource.Name)
+					return v1alpha1.StepStatusUnknown, errors.Wrapf(err, "checking CRD for potential data loss updating %s", step.Resource.Name)
 				}
 
 				// Update CRD to new version
 				_, err = client.CustomResourceDefinitions().Update(context.TODO(), crd, metav1.UpdateOptions{})
 				if err != nil {
-					return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "error updating CRD: %s", step.Resource.Name)
+					return v1alpha1.StepStatusUnknown, errors.Wrapf(err, "error updating CRD: %s", step.Resource.Name)
 				}
 				// If it already existed, mark the step as Present.
 				// they were equal - mark CRD as present
@@ -173,7 +221,7 @@ func (b *builder) NewCRDV1Beta1Step(client apiextensionsv1beta1client.Apiextensi
 				if k8serrors.IsNotFound(err) {
 					return v1alpha1.StepStatusNotPresent, nil
 				} else {
-					return v1alpha1.StepStatusNotPresent, errorwrap.Wrapf(err, "error finding the %s CRD", crd.Name)
+					return v1alpha1.StepStatusNotPresent, errors.Wrapf(err, "error finding the %s CRD", crd.Name)
 				}
 			}
 			established, namesAccepted := false, false
@@ -204,23 +252,23 @@ func (b *builder) NewCRDV1Beta1Step(client apiextensionsv1beta1client.Apiextensi
 				crd.SetResourceVersion(currentCRD.GetResourceVersion())
 
 				if err = validateV1Beta1CRDCompatibility(b.dynamicClient, currentCRD, crd); err != nil {
-					return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "error validating existing CRs against new CRD's schema: %s", step.Resource.Name)
+					return v1alpha1.StepStatusUnknown, errors.Wrapf(err, "error validating existing CRs against new CRD's schema: %s", step.Resource.Name)
 				}
 
 				// check to see if stored versions changed and whether the upgrade could cause potential data loss
 				safe, err := crdlib.SafeStorageVersionUpgrade(currentCRD, crd)
 				if !safe {
 					b.logger.Errorf("risk of data loss updating %s: %s", step.Resource.Name, err)
-					return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "risk of data loss updating %s", step.Resource.Name)
+					return v1alpha1.StepStatusUnknown, errors.Wrapf(err, "risk of data loss updating %s", step.Resource.Name)
 				}
 				if err != nil {
-					return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "checking CRD for potential data loss updating %s", step.Resource.Name)
+					return v1alpha1.StepStatusUnknown, errors.Wrapf(err, "checking CRD for potential data loss updating %s", step.Resource.Name)
 				}
 
 				// Update CRD to new version
 				_, err = client.CustomResourceDefinitions().Update(context.TODO(), crd, metav1.UpdateOptions{})
 				if err != nil {
-					return v1alpha1.StepStatusUnknown, errorwrap.Wrapf(err, "error updating CRD: %s", step.Resource.Name)
+					return v1alpha1.StepStatusUnknown, errors.Wrapf(err, "error updating CRD: %s", step.Resource.Name)
 				}
 				// If it already existed, mark the step as Present.
 				// they were equal - mark CRD as present

pkg/controller/operators/catalog/step_test.go

@@ -0,0 +1,134 @@
+package catalog
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"github.com/operator-framework/api/pkg/operators/v1alpha1"
+	v1alpha1listers "github.com/operator-framework/operator-lifecycle-manager/pkg/api/client/listers/operators/v1alpha1"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/internal/alongside"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorlister/operatorlisterfakes"
+)
+
+func TestSetInstalledAlongsideAnnotation(t *testing.T) {
+	for _, tc := range []struct {
+		Name         string
+		NewNamespace string
+		NewName      string
+		CSVs         []v1alpha1.ClusterServiceVersion
+		Before       []alongside.NamespacedName
+		After        []alongside.NamespacedName
+	}{
+		{
+			Name:         "object annotated with bundle name",
+			NewNamespace: "test-namespace",
+			NewName:      "test-name",
+			After: []alongside.NamespacedName{
+				{Namespace: "test-namespace", Name: "test-name"},
+			},
+		},
+		{
+			Name:         "annotations referencing missing bundles removed",
+			NewNamespace: "test-namespace",
+			NewName:      "test-name",
+			Before: []alongside.NamespacedName{
+				{Namespace: "missing-namespace", Name: "missing-name"},
+			},
+			After: []alongside.NamespacedName{
+				{Namespace: "test-namespace", Name: "test-name"},
+			},
+		},
+		{
+			Name:         "annotations referencing copied csv removed",
+			NewNamespace: "test-namespace",
+			NewName:      "test-name",
+			CSVs: []v1alpha1.ClusterServiceVersion{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "copied-namespace",
+						Name:      "copied-name",
+					},
+					Status: v1alpha1.ClusterServiceVersionStatus{
+						Reason: v1alpha1.CSVReasonCopied,
+					},
+				},
+			},
+			Before: []alongside.NamespacedName{
+				{Namespace: "copied-namespace", Name: "copied-name"},
+			},
+			After: []alongside.NamespacedName{
+				{Namespace: "test-namespace", Name: "test-name"},
+			},
+		},
+		{
+			Name:         "annotations referencing found bundles preserved",
+			NewNamespace: "test-namespace",
+			NewName:      "test-name",
+			CSVs: []v1alpha1.ClusterServiceVersion{
+				{
+					ObjectMeta: metav1.ObjectMeta{
+						Namespace: "found-namespace",
+						Name:      "found-name",
+					},
+				},
+			},
+			Before: []alongside.NamespacedName{
+				{Namespace: "found-namespace", Name: "found-name"},
+			},
+			After: []alongside.NamespacedName{
+				{Namespace: "found-namespace", Name: "found-name"},
+				{Namespace: "test-namespace", Name: "test-name"},
+			},
+		},
+	} {
+		t.Run(tc.Name, func(t *testing.T) {
+			csvsByNamespace := make(map[string][]*v1alpha1.ClusterServiceVersion)
+			for _, csv := range tc.CSVs {
+				csvsByNamespace[csv.GetNamespace()] = append(csvsByNamespace[csv.GetNamespace()], csv.DeepCopy())
+			}
+
+			nsListers := make(map[string]v1alpha1listers.ClusterServiceVersionNamespaceLister)
+			for ns, csvs := range csvsByNamespace {
+				ns := ns
+				csvs := csvs
+				nslister := &operatorlisterfakes.FakeClusterServiceVersionNamespaceLister{}
+				nslister.GetCalls(func(name string) (*v1alpha1.ClusterServiceVersion, error) {
+					for _, csv := range csvs {
+						if csv.GetName() == name {
+							return csv, nil
+						}
+					}
+					return nil, errors.NewNotFound(schema.GroupResource{}, name)
+				})
+				nsListers[ns] = nslister
+			}
+
+			emptyLister := &operatorlisterfakes.FakeClusterServiceVersionNamespaceLister{}
+			emptyLister.GetCalls(func(name string) (*v1alpha1.ClusterServiceVersion, error) {
+				return nil, errors.NewNotFound(schema.GroupResource{}, name)
+			})
+
+			csvLister := &operatorlisterfakes.FakeClusterServiceVersionLister{}
+			csvLister.ClusterServiceVersionsCalls(func(namespace string) v1alpha1listers.ClusterServiceVersionNamespaceLister {
+				if lister, ok := nsListers[namespace]; ok {
+					return lister
+				}
+				return emptyLister
+			})
+
+			var (
+				o metav1.ObjectMeta
+				a alongside.Annotator
+			)
+			a.ToObject(&o, tc.Before)
+			setInstalledAlongsideAnnotation(a, &o, tc.NewNamespace, tc.NewName, csvLister)
+			after := a.FromObject(&o)
+			assert.Equal(t, tc.After, after)
+		})
+	}
+}