Merge pull request operator-framework#380 from alecmerdler/ALM-639

Add CatalogSource Namespace to Subscription Objects

pkg/api/apis/subscription/v1alpha1/types.go

@@ -26,11 +26,12 @@ const (
 
 // SubscriptionSpec defines an Application that can be installed
 type SubscriptionSpec struct {
-	CatalogSource       string            `json:"source"`
-	Package             string            `json:"name"`
-	Channel             string            `json:"channel,omitempty"`
-	StartingCSV         string            `json:"startingCSV,omitempty"`
-	InstallPlanApproval v1alpha1.Approval `json:"installPlanApproval,omitempty"`
+	CatalogSource          string            `json:"source"`
+	CatalogSourceNamespace string            `json:"sourceNamespace"`
+	Package                string            `json:"name"`
+	Channel                string            `json:"channel,omitempty"`
+	StartingCSV            string            `json:"startingCSV,omitempty"`
+	InstallPlanApproval    v1alpha1.Approval `json:"installPlanApproval,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/controller/operators/catalog/operator.go

@@ -34,13 +34,18 @@ const (
 //for test stubbing and for ensuring standardization of timezones to UTC
 var timeNow = func() metav1.Time { return metav1.NewTime(time.Now().UTC()) }
 
+type sourceKey struct {
+	name      string
+	namespace string
+}
+
 // Operator represents a Kubernetes operator that executes InstallPlans by
 // resolving dependencies in a catalog.
 type Operator struct {
 	*queueinformer.Operator
 	client            versioned.Interface
 	namespace         string
-	sources           map[string]registry.Source
+	sources           map[sourceKey]registry.Source
 	sourcesLock       sync.Mutex
 	sourcesLastUpdate metav1.Time
 }
@@ -85,7 +90,7 @@ func NewOperator(kubeconfigPath string, wakeupInterval time.Duration, operatorNa
 		Operator:  queueOperator,
 		client:    crClient,
 		namespace: operatorNamespace,
-		sources:   make(map[string]registry.Source),
+		sources:   make(map[sourceKey]registry.Source),
 	}
 	// Register CatalogSource informers.
 	catsrcQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "catalogsources")
@@ -140,7 +145,7 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
 
 	o.sourcesLock.Lock()
 	defer o.sourcesLock.Unlock()
-	o.sources[catsrc.GetName()] = src
+	o.sources[sourceKey{name: catsrc.GetName(), namespace: catsrc.GetNamespace()}] = src
 	o.sourcesLastUpdate = timeNow()
 	return nil
 }
@@ -268,16 +273,16 @@ func (o *Operator) ResolvePlan(plan *v1alpha1.InstallPlan) error {
 	defer o.sourcesLock.Unlock()
 
 	var notFoundErr error
-	for sourceName, source := range o.sources {
-		log.Debugf("resolving against source %v", sourceName)
-		plan.EnsureCatalogSource(sourceName)
-		notFoundErr = resolveInstallPlan(sourceName, source, plan)
+	for key, source := range o.sources {
+		log.Debugf("resolving against source %v", key)
+		plan.EnsureCatalogSource(key.name)
+		notFoundErr = resolveInstallPlan(key.name, source, plan)
 		if notFoundErr != nil {
 			continue
 		}
 
 		// Look up the CatalogSource.
-		catsrc, err := o.client.CatalogsourceV1alpha1().CatalogSources(o.namespace).Get(sourceName, metav1.GetOptions{})
+		catsrc, err := o.client.CatalogsourceV1alpha1().CatalogSources(key.namespace).Get(key.name, metav1.GetOptions{})
 		if err != nil {
 			return err
 		}

pkg/controller/operators/catalog/subscriptions.go

@@ -40,9 +40,13 @@ func (o *Operator) syncSubscription(sub *v1alpha1.Subscription) (*v1alpha1.Subsc
 	o.sourcesLock.Lock()
 	defer o.sourcesLock.Unlock()
 
-	catalog, ok := o.sources[sub.Spec.CatalogSource]
+	catalogNamespace := sub.Spec.CatalogSourceNamespace
+	if catalogNamespace == "" {
+		catalogNamespace = o.namespace
+	}
+	catalog, ok := o.sources[sourceKey{name: sub.Spec.CatalogSource, namespace: catalogNamespace}]
 	if !ok {
-		return sub, fmt.Errorf("unknown catalog source %s", sub.Spec.CatalogSource)
+		return sub, fmt.Errorf("unknown catalog source %s in namespace %s", sub.Spec.CatalogSource, catalogNamespace)
 	}
 
 	// Find latest CSV if no CSVs are installed already

pkg/controller/operators/catalog/subscriptions_test.go

@@ -112,7 +112,7 @@ func TestSyncSubscription(t *testing.T) {
 					CatalogSource: "flying-unicorns",
 				},
 			}},
-			expected: expected{err: "unknown catalog source flying-unicorns"},
+			expected: expected{err: "unknown catalog source flying-unicorns in namespace ns"},
 		},
 		{
 			name:    "no updates",
@@ -934,8 +934,8 @@ func TestSyncSubscription(t *testing.T) {
 			op := &Operator{
 				client:    clientFake,
 				namespace: "ns",
-				sources: map[string]registry.Source{
-					tt.initial.catalogName: catalogFake,
+				sources: map[sourceKey]registry.Source{
+					sourceKey{name: tt.initial.catalogName, namespace: "ns"}: catalogFake,
 				},
 				sourcesLastUpdate: tt.initial.sourcesLastUpdate,
 			}