fix(resolver): don't pick operator from same package when resolve requiredAPI

When resolver looks for an operator that provides requiredAPIs, it
doesn't exclude operators from the same package that the original
operator that requires those APIs. As a result, if there is an
existing version of that operator now provides those requiredAPIs
(due to API ownership transfer), resolver will select an older or
newer version of the same operator to fulfill that requiredAPIs
which leads to API ownership conflicts on providedAPIs.

The resolver will look up all latest channel entries that provides
the necessary requiredAPIs and filter out those from the same package
before go on with the operator selection.

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

Author: dinhxuanvu

pkg/controller/registry/grpc/source.go

@@ -190,8 +190,9 @@ func (s *SourceStore) Remove(key resolver.CatalogKey) error {
 	return source.Conn.Close()
 }
 
-func (s *SourceStore) AsClients(namespaces ...string) map[resolver.CatalogKey]client.Interface {
-	refs := map[resolver.CatalogKey]client.Interface{}
+func (s *SourceStore) AsClients(namespaces ...string) (map[resolver.CatalogKey]client.Interface, map[resolver.CatalogKey]*client.Client) {
+	refsInterface := map[resolver.CatalogKey]client.Interface{}
+	refsClient := map[resolver.CatalogKey]*client.Client{}
 	s.sourcesLock.RLock()
 	defer s.sourcesLock.RUnlock()
 	for key, source := range s.sources {
@@ -200,11 +201,12 @@ func (s *SourceStore) AsClients(namespaces ...string) map[resolver.CatalogKey]cl
 		}
 		for _, namespace := range namespaces {
 			if key.Namespace == namespace {
-				refs[key] = client.NewClientFromConn(source.Conn)
+				refsInterface[key] = client.NewClientFromConn(source.Conn)
+				refsClient[key] = client.NewClientFromConn(source.Conn)
 			}
 		}
 	}
 
 	// TODO : remove unhealthy
-	return refs
+	return refsInterface, refsClient
 }

pkg/controller/registry/resolver/evolver.go

@@ -134,7 +134,7 @@ func (e *NamespaceGenerationEvolver) queryForRequiredAPIs() error {
 		}
 
 		// attempt to find a bundle that provides that api
-		if bundle, key, err := e.querier.FindProvider(*api, initialSource.Catalog); err == nil {
+		if bundle, key, err := e.querier.FindProvider(*api, initialSource.Catalog, initialSource.Package); err == nil {
 			// add a bundle that provides the api to the generation
 			o, err := NewOperatorFromBundle(bundle, "", *key)
 			if err != nil {

pkg/controller/registry/resolver/querier.go

@@ -4,12 +4,14 @@ package resolver
 import (
 	"context"
 	"fmt"
+	"io"
 
 	"github.com/blang/semver"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/errors"
 
 	"github.com/operator-framework/operator-registry/pkg/api"
+	registryapi "github.com/operator-framework/operator-registry/pkg/api"
 	"github.com/operator-framework/operator-registry/pkg/client"
 	opregistry "github.com/operator-framework/operator-registry/pkg/registry"
 )
@@ -24,7 +26,7 @@ type SourceRef struct {
 }
 
 type SourceQuerier interface {
-	FindProvider(api opregistry.APIKey, initialSource CatalogKey) (*api.Bundle, *CatalogKey, error)
+	FindProvider(api opregistry.APIKey, initialSource CatalogKey, pkgName string) (*api.Bundle, *CatalogKey, error)
 	FindBundle(pkgName, channelName, bundleName string, initialSource CatalogKey) (*api.Bundle, *CatalogKey, error)
 	FindLatestBundle(pkgName, channelName string, initialSource CatalogKey) (*api.Bundle, *CatalogKey, error)
 	FindReplacement(currentVersion *semver.Version, bundleName, pkgName, channelName string, initialSource CatalogKey) (*api.Bundle, *CatalogKey, error)
@@ -33,13 +35,46 @@ type SourceQuerier interface {
 
 type NamespaceSourceQuerier struct {
 	sources map[CatalogKey]client.Interface
+	clients map[CatalogKey]*client.Client
 }
 
 var _ SourceQuerier = &NamespaceSourceQuerier{}
 
-func NewNamespaceSourceQuerier(sources map[CatalogKey]client.Interface) *NamespaceSourceQuerier {
+type ChannelEntryStream interface {
+	Recv() (*api.ChannelEntry, error)
+}
+
+type ChannelEntryIterator struct {
+	stream ChannelEntryStream
+	error  error
+}
+
+func NewChannelEntryIterator(stream ChannelEntryStream) *ChannelEntryIterator {
+	return &ChannelEntryIterator{stream: stream}
+}
+
+func (ceit *ChannelEntryIterator) Next() *registryapi.ChannelEntry {
+	if ceit.error != nil {
+		return nil
+	}
+	next, err := ceit.stream.Recv()
+	if err == io.EOF {
+		return nil
+	}
+	if err != nil {
+		ceit.error = err
+	}
+	return next
+}
+
+func (ceit *ChannelEntryIterator) Error() error {
+	return ceit.error
+}
+
+func NewNamespaceSourceQuerier(sources map[CatalogKey]client.Interface, clients map[CatalogKey]*client.Client) *NamespaceSourceQuerier {
 	return &NamespaceSourceQuerier{
 		sources: sources,
+		clients: clients,
 	}
 }
 
@@ -50,23 +85,23 @@ func (q *NamespaceSourceQuerier) Queryable() error {
 	return nil
 }
 
-func (q *NamespaceSourceQuerier) FindProvider(api opregistry.APIKey, initialSource CatalogKey) (*api.Bundle, *CatalogKey, error) {
+func (q *NamespaceSourceQuerier) FindProvider(api opregistry.APIKey, initialSource CatalogKey, pkgName string) (*registryapi.Bundle, *CatalogKey, error) {
 	if initialSource.Name != "" && initialSource.Namespace != "" {
-		source, ok := q.sources[initialSource]
+		client, ok := q.clients[initialSource]
 		if ok {
-			if bundle, err := source.GetBundleThatProvides(context.TODO(), api.Group, api.Version, api.Kind); err == nil {
+			if bundle, err := FindBundleThatProvides(context.TODO(), client, api.Group, api.Version, api.Kind, pkgName); err == nil {
 				return bundle, &initialSource, nil
 			}
-			if bundle, err := source.GetBundleThatProvides(context.TODO(), api.Plural+"."+api.Group, api.Version, api.Kind); err == nil {
+			if bundle, err := FindBundleThatProvides(context.TODO(), client, api.Plural+"."+api.Group, api.Version, api.Kind, pkgName); err == nil {
 				return bundle, &initialSource, nil
 			}
 		}
 	}
-	for key, source := range q.sources {
-		if bundle, err := source.GetBundleThatProvides(context.TODO(), api.Group, api.Version, api.Kind); err == nil {
+	for key, client := range q.clients {
+		if bundle, err := FindBundleThatProvides(context.TODO(), client, api.Group, api.Version, api.Kind, pkgName); err == nil {
 			return bundle, &key, nil
 		}
-		if bundle, err := source.GetBundleThatProvides(context.TODO(), api.Plural+"."+api.Group, api.Version, api.Kind); err == nil {
+		if bundle, err := FindBundleThatProvides(context.TODO(), client, api.Plural+"."+api.Group, api.Version, api.Kind, pkgName); err == nil {
 			return bundle, &key, nil
 		}
 	}
@@ -191,3 +226,51 @@ func (q *NamespaceSourceQuerier) findChannelHead(currentVersion *semver.Version,
 	}
 	return nil, nil
 }
+
+// GetLatestChannelEntriesThatProvide uses registry client to get a list of
+// latest channel entries that provide the requested API (via an iterator)
+func GetLatestChannelEntriesThatProvide(ctx context.Context, c *client.Client, group, version, kind string) (*ChannelEntryIterator, error) {
+	stream, err := c.Registry.GetLatestChannelEntriesThatProvide(ctx, &registryapi.GetLatestProvidersRequest{Group: group, Version: version, Kind: kind})
+	if err != nil {
+		return nil, err
+	}
+	return NewChannelEntryIterator(stream), nil
+}
+
+// FilterChannelEntries filters out a channel entries that provide the requested
+// API and come from the same package with original operator and returns the
+// first entry on the list
+func FilterChannelEntries(it *ChannelEntryIterator, pkgName string) *opregistry.ChannelEntry {
+	var entry *opregistry.ChannelEntry
+	for e := it.Next(); e != nil; e = it.Next() {
+		if e.PackageName != pkgName {
+			entry = &opregistry.ChannelEntry{
+				PackageName: e.PackageName,
+				ChannelName: e.ChannelName,
+				BundleName:  e.BundleName,
+				Replaces:    e.Replaces,
+			}
+			break
+		}
+	}
+	return entry
+}
+
+// FindBundleThatProvides returns a bundle that provides the request API and
+// doesn't belong to the provided package
+func FindBundleThatProvides(ctx context.Context, client *client.Client, group, version, kind, pkgName string) (*api.Bundle, error) {
+	it, err := GetLatestChannelEntriesThatProvide(ctx, client, group, version, kind)
+	if err != nil {
+		return nil, err
+	}
+
+	entry := FilterChannelEntries(it, pkgName)
+	if entry != nil {
+		return nil, fmt.Errorf("Unable to find a channel entry which doesn't belong to package %s", pkgName)
+	}
+	bundle, err := client.Registry.GetBundle(ctx, &registryapi.GetBundleRequest{PkgName: entry.PackageName, ChannelName: entry.ChannelName, CsvName: entry.BundleName})
+	if err != nil {
+		return nil, err
+	}
+	return bundle, nil
+}

pkg/controller/registry/resolver/querier_test.go

@@ -13,9 +13,9 @@ import (
 	"github.com/stretchr/testify/require"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"github.com/operator-framework/api/pkg/lib/version"
 	"github.com/operator-framework/api/pkg/operators/v1alpha1"
 	"github.com/operator-framework/operator-lifecycle-manager/pkg/controller/registry/resolver/fakes"
-	"github.com/operator-framework/api/pkg/lib/version"
 )
 
 func TestNewNamespaceSourceQuerier(t *testing.T) {
@@ -207,7 +207,7 @@ func TestNamespaceSourceQuerier_FindProvider(t *testing.T) {
 			q := &NamespaceSourceQuerier{
 				sources: tt.fields.sources,
 			}
-			bundle, key, err := q.FindProvider(tt.args.api, tt.args.catalogKey)
+			bundle, key, err := q.FindProvider(tt.args.api, tt.args.catalogKey, "")
 			require.Equal(t, tt.out.err, err)
 			require.Equal(t, tt.out.bundle, bundle)
 			require.Equal(t, tt.out.key, key)