Add IndexFunc using ProvidedAPIs as key to look up CSVs

Signed-off-by: Vu Dinh <[email protected]>

Author: dinhxuanvu

pkg/controller/operators/catalog/operator.go

@@ -33,6 +33,7 @@ import (
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/operators/catalog/subscription"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/reconciler"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver"
+	index "github.com/operator-framework/operator-lifecycle-manager/pkg/lib/index"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorclient"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/operatorlister"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/lib/ownerutil"
@@ -57,21 +58,22 @@ const (
 type Operator struct {
 	queueinformer.Operator
 
-	logger            *logrus.Logger
-	clock             utilclock.Clock
-	opClient          operatorclient.ClientInterface
-	client            versioned.Interface
-	lister            operatorlister.OperatorLister
-	catsrcQueueSet    *queueinformer.ResourceQueueSet
-	subQueueSet       *queueinformer.ResourceQueueSet
-	ipQueueSet        *queueinformer.ResourceQueueSet
-	nsResolveQueue    workqueue.RateLimitingInterface
-	namespace         string
-	sources           map[resolver.CatalogKey]resolver.SourceRef
-	sourcesLock       sync.RWMutex
-	sourcesLastUpdate metav1.Time
-	resolver          resolver.Resolver
-	reconciler        reconciler.RegistryReconcilerFactory
+	logger                 *logrus.Logger
+	clock                  utilclock.Clock
+	opClient               operatorclient.ClientInterface
+	client                 versioned.Interface
+	lister                 operatorlister.OperatorLister
+	catsrcQueueSet         *queueinformer.ResourceQueueSet
+	subQueueSet            *queueinformer.ResourceQueueSet
+	ipQueueSet             *queueinformer.ResourceQueueSet
+	nsResolveQueue         workqueue.RateLimitingInterface
+	namespace              string
+	sources                map[resolver.CatalogKey]resolver.SourceRef
+	sourcesLock            sync.RWMutex
+	sourcesLastUpdate      metav1.Time
+	resolver               resolver.Resolver
+	reconciler             reconciler.RegistryReconcilerFactory
+	csvProvidedAPIsIndexer map[string]cache.Indexer
 }
 
 // NewOperator creates a new Catalog Operator.
@@ -123,6 +125,10 @@ func NewOperator(ctx context.Context, kubeconfigPath string, clock utilclock.Clo
 		op.lister.OperatorsV1alpha1().RegisterClusterServiceVersionLister(namespace, csvInformer.Lister())
 		op.RegisterInformer(csvInformer.Informer())
 
+		csvInformer.Informer().AddIndexers(cache.Indexers{index.ProvidedAPIsIndexFuncKey: index.ProvidedAPIsIndexFunc})
+		csvIndexer := csvInformer.Informer().GetIndexer()
+		op.csvProvidedAPIsIndexer[namespace] = csvIndexer
+
 		// TODO: Add namespace resolve sync
 
 		// Wire InstallPlans
@@ -1040,8 +1046,14 @@ func (o *Operator) ExecutePlan(plan *v1alpha1.InstallPlan) error {
 					if !reflect.DeepEqual(crd, *currentCRD) {
 						// Verify CRD ownership, only attempt to update if
 						// CRD has only one owner
-						if len(existingCRDOwners[currentCRD.GetName()]) == 1 {
+						// Example: provided=database.coreos.com/v1alpha1/EtcdCluster
+						matchedCSV, err := index.APIsIndexValues(o.csvProvidedAPIsIndexer, crd)
+						if err != nil {
+							return errorwrap.Wrapf(err, "error find matched CSV: %s", step.Resource.Name)
+						}
+						if len(matchedCSV) == 1 {
 							// Attempt to update CRD
+							crd.SetResourceVersion(currentCRD.GetResourceVersion())
 							_, err = o.OpClient.ApiextensionsV1beta1Interface().ApiextensionsV1beta1().CustomResourceDefinitions().Update(&crd)
 							if err != nil {
 								return errorwrap.Wrapf(err, "error update CRD: %s", step.Resource.Name)

pkg/lib/index/api.go

@@ -0,0 +1,66 @@
+package indexer
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/operators/v1alpha1"
+	v1beta1ext "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
+	"k8s.io/client-go/tools/cache"
+)
+
+const (
+	// ProvidedAPIsIndexFuncKey is the recommended key to use for registering the index func with an indexer.
+	ProvidedAPIsIndexFuncKey string = "providedAPIs"
+)
+
+func ProvidedAPIsIndexFunc(obj interface{}) ([]string, error) {
+	indicies := []string{}
+
+	csv, ok := obj.(*v1alpha1.ClusterServiceVersion)
+	if !ok {
+		return indicies, fmt.Errorf("invalid object of type: %T", obj)
+	}
+
+	for _, crd := range csv.Spec.CustomResourceDefinitions.Owned {
+		parts := strings.SplitN(crd.Name, ".", 2)
+		if len(parts) < 2 {
+			return indicies, fmt.Errorf("couldn't parse plural.group from crd name: %s", crd.Name)
+		}
+		indicies = append(indicies, fmt.Sprintf("provided=%s/%s/%s", parts[1], crd.Version, crd.Kind))
+	}
+	for _, api := range csv.Spec.APIServiceDefinitions.Owned {
+		indicies = append(indicies, fmt.Sprintf("provided=%s/%s/%s", api.Group, api.Version, api.Kind))
+	}
+
+	return indicies, nil
+}
+
+// APIsIndexValues returns the names of CSVs that own the given CRD
+func APIsIndexValues(indexers map[string]cache.Indexer, crd v1beta1ext.CustomResourceDefinition) (map[string]struct{}, error) {
+	csvSet := map[string]struct{}{}
+	crdSpec := map[string]struct{}{}
+	for _, v := range crd.Spec.Versions {
+		crdSpec[fmt.Sprintf("provided=%s/%s/%s", crd.Spec.Group, v.Name, crd.Spec.Names.Kind)] = struct{}{}
+	}
+	if crd.Spec.Version != "" {
+		crdSpec[fmt.Sprintf("provided=%s/%s/%s", crd.Spec.Group, crd.Spec.Version, crd.Spec.Names.Kind)] = struct{}{}
+	}
+	for _, indexer := range indexers {
+		for key, _ := range crdSpec {
+			csvs, err := indexer.ByIndex(ProvidedAPIsIndexFuncKey, key)
+			if err != nil {
+				return nil, err
+			}
+			for _, csv := range csvs {
+				csv, ok := csv.(*v1alpha1.ClusterServiceVersion)
+				if !ok {
+					continue
+				}
+				// Add to set
+				csvSet[csv.GetName()] = struct{}{}
+			}
+		}
+	}
+	return csvSet, nil
+}