Bug 1762769: Prioritize APIs from same CatSrc

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 {