bug(dependencies) Prioritize APIs from same CatSrc

Problem: When OLM installs an operator that requires dependencies that
are provided via multiple operators in different CatalogSources, the API
is pulled from any of the CatalogSources that provide the API.

Solution: This commit introduces a change so that OLM will generate a
list of operators that depend on the API, randomly select one of their
sources, and prioritize checking in that CatalogSource for the API prior
to checking the remaining CatalogSource for the API.

Author: awgreene

pkg/controller/registry/resolver/evolver.go

@@ -112,8 +112,15 @@ func (e *NamespaceGenerationEvolver) queryForRequiredAPIs() error {
 		}
 		e.gen.MarkAPIChecked(*api)
 
+		// identify the initialSource
+		initialSource := CatalogKey{}
+		for _, operator := range e.gen.MissingAPIs()[*api] {
+			initialSource = operator.SourceInfo().Catalog
+			break
+		}
+
 		// attempt to find a bundle that provides that api
-		if bundle, key, err := e.querier.FindProvider(*api); err == nil {
+		if bundle, key, err := e.querier.FindProvider(*api, initialSource); 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

@@ -22,7 +22,7 @@ type SourceRef struct {
 }
 
 type SourceQuerier interface {
-	FindProvider(api opregistry.APIKey) (*opregistry.Bundle, *CatalogKey, error)
+	FindProvider(api opregistry.APIKey, initialSource CatalogKey) (*opregistry.Bundle, *CatalogKey, error)
 	FindBundle(pkgName, channelName, bundleName string, initialSource CatalogKey) (*opregistry.Bundle, *CatalogKey, error)
 	FindLatestBundle(pkgName, channelName string, initialSource CatalogKey) (*opregistry.Bundle, *CatalogKey, error)
 	FindReplacement(currentVersion *semver.Version, bundleName, pkgName, channelName string, initialSource CatalogKey) (*opregistry.Bundle, *CatalogKey, error)
@@ -48,7 +48,18 @@ func (q *NamespaceSourceQuerier) Queryable() error {
 	return nil
 }
 
-func (q *NamespaceSourceQuerier) FindProvider(api opregistry.APIKey) (*opregistry.Bundle, *CatalogKey, error) {
+func (q *NamespaceSourceQuerier) FindProvider(api opregistry.APIKey, initialSource CatalogKey) (*opregistry.Bundle, *CatalogKey, error) {
+	if initialSource.Name != "" && initialSource.Namespace != "" {
+		source, ok := q.sources[initialSource]
+		if ok {
+			if bundle, err := source.GetBundleThatProvides(context.TODO(), api.Group, api.Version, api.Kind); err == nil {
+				return bundle, &initialSource, nil
+			}
+			if bundle, err := source.GetBundleThatProvides(context.TODO(), api.Plural+"."+api.Group, api.Version, api.Kind); 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 {
 			return bundle, &key, nil

pkg/controller/registry/resolver/querier_test.go

@@ -106,20 +106,33 @@ func TestNamespaceSourceQuerier_Queryable(t *testing.T) {
 
 func TestNamespaceSourceQuerier_FindProvider(t *testing.T) {
 	fakeSource := fakes.FakeInterface{}
+	fakeSource2 := fakes.FakeInterface{}
 	sources := map[CatalogKey]client.Interface{
 		CatalogKey{"test", "ns"}: &fakeSource,
+		CatalogKey{"test2", "ns"}: &fakeSource2,
 	}
 
 	bundle := opregistry.NewBundle("test", "testPkg", "testChannel")
+	bundle2 := opregistry.NewBundle("test2", "testPkg2", "testChannel2")
 	fakeSource.GetBundleThatProvidesStub = func(ctx context.Context, group, version, kind string) (*opregistry.Bundle, error) {
+		if group != "group" || version != "version" || kind != "kind" {
+			return nil, fmt.Errorf("Not Found")
+		}
 		return bundle, nil
 	}
+	fakeSource2.GetBundleThatProvidesStub = func(ctx context.Context, group, version, kind string) (*opregistry.Bundle, error) {
+		if group != "group2" || version != "version2" || kind != "kind2" {
+			return nil, fmt.Errorf("Not Found")
+		}
+		return bundle2, nil
+	}
 
 	type fields struct {
 		sources map[CatalogKey]client.Interface
 	}
 	type args struct {
 		api opregistry.APIKey
+		catalogKey CatalogKey
 	}
 	type out struct {
 		bundle *opregistry.Bundle
@@ -138,6 +151,7 @@ func TestNamespaceSourceQuerier_FindProvider(t *testing.T) {
 			},
 			args: args{
 				api: opregistry.APIKey{"group", "version", "kind", "plural"},
+				catalogKey: CatalogKey{},
 			},
 			out: out{
 				bundle: bundle,
@@ -151,23 +165,52 @@ func TestNamespaceSourceQuerier_FindProvider(t *testing.T) {
 			},
 			args: args{
 				api: opregistry.APIKey{"group", "version", "kind", "plural"},
+				catalogKey: CatalogKey{},
 			},
 			out: out{
 				bundle: nil,
 				key:    nil,
 				err:    fmt.Errorf("group/version/kind (plural) not provided by a package in any CatalogSource"),
 			},
 		},
+		{
+			fields: fields{
+				sources: sources,
+			},
+			args: args{
+				api: opregistry.APIKey{"group2", "version2", "kind2", "plural2"},
+				catalogKey: CatalogKey{Name: "test2", Namespace: "ns"},
+			},
+			out: out{
+				bundle: bundle2,
+				key:    &CatalogKey{Name: "test2", Namespace: "ns"},
+				err:    nil,
+			},
+		},
+		{
+			fields: fields{
+				sources: sources,
+			},
+			args: args{
+				api: opregistry.APIKey{"group2", "version2", "kind2", "plural2"},
+				catalogKey: CatalogKey{Name: "test3", Namespace: "ns"},
+			},
+			out: out{
+				bundle: bundle2,
+				key:    &CatalogKey{Name: "test2", Namespace: "ns"},
+				err:    nil,
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			q := &NamespaceSourceQuerier{
 				sources: tt.fields.sources,
 			}
-			bundle, key, err := q.FindProvider(tt.args.api)
-			require.Equal(t, err, tt.out.err)
-			require.Equal(t, bundle, tt.out.bundle)
-			require.Equal(t, key, tt.out.key)
+			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)
 		})
 	}
 }

test/e2e/subscription_e2e_test.go

@@ -1362,6 +1362,48 @@ func TestCreateNewSubscriptionWithPodConfig(t *testing.T) {
 	checkDeploymentWithPodConfiguration(t, kubeClient, csv, podConfig.Env, podConfig.Volumes, podConfig.VolumeMounts)
 }
 
+
+func TestCreateNewSubscriptionWithDependencies(t *testing.T) {
+	defer cleaner.NotifyTestComplete(t, true)
+
+	kubeClient := newKubeClient(t)
+	crClient := newCRClient(t)
+
+	permissions := deploymentPermissions(t)
+
+	catsrc, subSpec, catsrcCleanup := newCatalogSourceWithDependencies(t, kubeClient, crClient, "podconfig", testNamespace, permissions)
+	defer catsrcCleanup()
+
+	// Ensure that the catalog source is resolved before we create a subscription.
+	_, err := fetchCatalogSource(t, crClient, catsrc.GetName(), testNamespace, catalogSourceRegistryPodSynced)
+	require.NoError(t, err)
+
+	// Create duplicates of the CatalogSource
+	for i := 0; i < 10; i++ {
+		duplicateCatsrc, _, duplicateCatSrcCleanup := newCatalogSourceWithDependencies(t, kubeClient, crClient, "podconfig", testNamespace, permissions)
+		defer duplicateCatSrcCleanup()
+
+		// Ensure that the catalog source is resolved before we create a subscription.
+		_, err = fetchCatalogSource(t, crClient, duplicateCatsrc.GetName(), testNamespace, catalogSourceRegistryPodSynced)
+		require.NoError(t, err)
+	}
+
+	// Create a subscription that has a dependency
+	subscriptionName := genName("podconfig-sub-")
+	cleanupSubscription := createSubscriptionForCatalogWithSpec(t, crClient, testNamespace, subscriptionName, subSpec)
+	defer cleanupSubscription()
+
+	subscription, err := fetchSubscription(t, crClient, testNamespace, subscriptionName, subscriptionStateAtLatestChecker)
+	require.NoError(t, err)
+	require.NotNil(t, subscription)
+
+	// Check that a single catalog source was used to resolve the InstallPlan
+	installPlan, err:= fetchInstallPlan(t, crClient, subscription.Status.InstallPlanRef.Name, buildInstallPlanPhaseCheckFunc(v1alpha1.InstallPlanPhaseComplete))
+	require.NoError(t, err)
+	require.Len(t, installPlan.Status.CatalogSources,1)
+
+}
+
 func checkDeploymentWithPodConfiguration(t *testing.T, client operatorclient.ClientInterface, csv *v1alpha1.ClusterServiceVersion, envVar []corev1.EnvVar, volumes []corev1.Volume, volumeMounts []corev1.VolumeMount) {
 	resolver := install.StrategyResolver{}
 

test/e2e/user_defined_sa_test.go

@@ -310,6 +310,75 @@ func newCatalogSource(t *testing.T, kubeclient operatorclient.ClientInterface, c
 	return
 }
 
+
+func newCatalogSourceWithDependencies(t *testing.T, kubeclient operatorclient.ClientInterface, crclient versioned.Interface, prefix, namespace string, permissions []install.StrategyDeploymentPermissions) (catsrc *v1alpha1.CatalogSource, subscriptionSpec *v1alpha1.SubscriptionSpec, cleanup cleanupFunc) {
+	crdPlural := genName("ins")
+	crdName := crdPlural + ".cluster.com"
+
+	crd := apiextensions.CustomResourceDefinition{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: crdName,
+		},
+		Spec: apiextensions.CustomResourceDefinitionSpec{
+			Group:   "cluster.com",
+			Version: "v1alpha1",
+			Names: apiextensions.CustomResourceDefinitionNames{
+				Plural:   crdPlural,
+				Singular: crdPlural,
+				Kind:     crdPlural,
+				ListKind: "list" + crdPlural,
+			},
+			Scope: "Namespaced",
+		},
+	}
+
+	prefixFunc := func(s string) string {
+		return fmt.Sprintf("%s-%s-", prefix, s)
+	}
+
+	// Create CSV
+	packageName1 := genName(prefixFunc("package"))
+	packageName2 := genName(prefixFunc("package"))
+	stableChannel := "stable"
+
+	namedStrategy := newNginxInstallStrategy(genName(prefixFunc("dep")), permissions, nil)
+	csvA := newCSV("nginx-req-dep", namespace, "", semver.MustParse("0.1.0"), nil, []apiextensions.CustomResourceDefinition{crd}, namedStrategy)
+	csvB := newCSV("nginx-dependency", namespace, "", semver.MustParse("0.1.0"), []apiextensions.CustomResourceDefinition{crd}, nil, namedStrategy)
+
+	// Create PackageManifests
+	manifests := []registry.PackageManifest{
+		{
+			PackageName: packageName1,
+			Channels: []registry.PackageChannel{
+				{Name: stableChannel, CurrentCSVName: csvA.GetName()},
+			},
+			DefaultChannelName: stableChannel,
+		},
+		{
+			PackageName: packageName2,
+			Channels: []registry.PackageChannel{
+				{Name: stableChannel, CurrentCSVName: csvB.GetName()},
+			},
+			DefaultChannelName: stableChannel,
+		},
+	}
+
+	catalogSourceName := genName(prefixFunc("catsrc"))
+	catsrc, cleanup = createInternalCatalogSource(t, kubeclient, crclient, catalogSourceName, namespace, manifests, []apiextensions.CustomResourceDefinition{crd}, []v1alpha1.ClusterServiceVersion{csvA, csvB})
+	require.NotNil(t, catsrc)
+	require.NotNil(t, cleanup)
+
+	subscriptionSpec = &v1alpha1.SubscriptionSpec{
+		CatalogSource:          catsrc.GetName(),
+		CatalogSourceNamespace: catsrc.GetNamespace(),
+		Package:                packageName1,
+		Channel:                stableChannel,
+		StartingCSV:            csvA.GetName(),
+		InstallPlanApproval:    v1alpha1.ApprovalAutomatic,
+	}
+	return
+}
+
 func mustHaveCondition(t *testing.T, ip *v1alpha1.InstallPlan, conditionType v1alpha1.InstallPlanConditionType) (condition *v1alpha1.InstallPlanCondition) {
 	for i := range ip.Status.Conditions {
 		if ip.Status.Conditions[i].Type == conditionType {