Change index func to return list of ns instead

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

Author: dinhxuanvu

pkg/controller/operators/catalog/operator.go

@@ -348,11 +348,11 @@ func (o *Operator) syncSourceState(state grpc.SourceState) {
 	switch state.State {
 	case connectivity.Ready:
 		if o.namespace == state.Key.Namespace {
-			subs, err := index.CatalogSubscriberNamespaces(o.catalogSubscriberIndexer,
+			namespaces, err := index.CatalogSubscriberNamespaces(o.catalogSubscriberIndexer,
 				state.Key.Name, state.Key.Namespace)
 
 			if err == nil {
-				for _, ns := range subs {
+				for ns := range namespaces {
 					o.nsResolveQueue.Add(ns)
 				}
 			}

pkg/lib/index/catalog.go

@@ -16,23 +16,22 @@ const (
 // PresentCatalogIndexFunc returns index from CatalogSource/CatalogSourceNamespace
 // of the given object (Subscription)
 func PresentCatalogIndexFunc(obj interface{}) ([]string, error) {
-	indicies := []string{}
-
 	sub, ok := obj.(*v1alpha1.Subscription)
 	if !ok {
-		return indicies, fmt.Errorf("invalid object of type: %T", obj)
+		return []string{""}, fmt.Errorf("invalid object of type: %T", obj)
 	}
 
-	indicies = append(indicies, fmt.Sprintf("%s/%s", sub.Spec.CatalogSource,
-		sub.Spec.CatalogSourceNamespace))
+	if sub.Spec.CatalogSource != "" && sub.Spec.CatalogSourceNamespace != "" {
+		return []string{sub.Spec.CatalogSource + "/" + sub.Spec.CatalogSourceNamespace}, nil
+	}
 
-	return indicies, nil
+	return []string{""}, nil
 }
 
-// CatalogSubscriberNamespaces returns the list of Suscriptions' name and namespace
-// (name/namespace as key and namespace as value) that uses the given CatalogSource (name/namespace)
-func CatalogSubscriberNamespaces(indexers map[string]cache.Indexer, name, namespace string) (map[string]string, error) {
-	nsSet := map[string]string{}
+// CatalogSubscriberNamespaces returns the list of namespace (as a map with namespace as key)
+// which has Suscriptions(s) that subscribe(s) to a given CatalogSource (name/namespace)
+func CatalogSubscriberNamespaces(indexers map[string]cache.Indexer, name, namespace string) (map[string]struct{}, error) {
+	nsSet := map[string]struct{}{}
 	index := fmt.Sprintf("%s/%s", name, namespace)
 
 	for _, indexer := range indexers {
@@ -46,8 +45,7 @@ func CatalogSubscriberNamespaces(indexers map[string]cache.Indexer, name, namesp
 				continue
 			}
 			// Add to set
-			key := fmt.Sprintf("%s/%s", s.GetName(), s.GetNamespace())
-			nsSet[key] = s.GetNamespace()
+			nsSet[s.GetNamespace()] = struct{}{}
 		}
 	}
 

test/e2e/catalog_e2e_test.go

@@ -138,17 +138,27 @@ func TestGlobalCatalogUpdateTriggersSubscriptionSync(t *testing.T) {
 		Package:                mainPackageName,
 		Channel:                stableChannel,
 		StartingCSV:            mainCSV.GetName(),
-		InstallPlanApproval:    v1alpha1.ApprovalAutomatic,
+		InstallPlanApproval:    v1alpha1.ApprovalManual,
 	}
 
 	// Create Subscription
 	subscriptionName := genName("sub-")
 	createSubscriptionForCatalogWithSpec(t, crc, testNamespace, subscriptionName, subscriptionSpec)
 
-	subscription, err := fetchSubscription(t, crc, testNamespace, subscriptionName, subscriptionStateAtLatestChecker)
+	subscription, err := fetchSubscription(t, crc, testNamespace, subscriptionName, subscriptionHasInstallPlanChecker)
 	require.NoError(t, err)
 	require.NotNil(t, subscription)
-	_, err = fetchCSV(t, crc, subscription.Status.CurrentCSV, testNamespace, buildCSVConditionChecker(v1alpha1.CSVPhaseSucceeded))
+
+	installPlanName := subscription.Status.Install.Name
+	requiresApprovalChecker := buildInstallPlanPhaseCheckFunc(v1alpha1.InstallPlanPhaseRequiresApproval)
+	fetchedInstallPlan, err := fetchInstallPlan(t, crc, installPlanName, requiresApprovalChecker)
+	require.NoError(t, err)
+
+	fetchedInstallPlan.Spec.Approved = true
+	_, err = crc.OperatorsV1alpha1().InstallPlans(testNamespace).Update(fetchedInstallPlan)
+	require.NoError(t, err)
+
+	_, err = awaitCSV(t, crc, testNamespace, mainCSV.GetName(), csvSucceededChecker)
 	require.NoError(t, err)
 
 	// Update manifest
@@ -166,19 +176,10 @@ func TestGlobalCatalogUpdateTriggersSubscriptionSync(t *testing.T) {
 	updateInternalCatalog(t, c, crc, mainCatalogName, globalNS, []apiextensions.CustomResourceDefinition{mainCRD}, []v1alpha1.ClusterServiceVersion{mainCSV, replacementCSV}, mainManifests)
 
 	// Get updated catalogsource
-	fetchedUpdatedCatalog, err := fetchCatalogSource(t, crc, mainCatalogName, globalNS, func(catalog *v1alpha1.CatalogSource) bool {
-		registry := catalog.Status.RegistryServiceStatus
-		connState := catalog.Status.GRPCConnectionState
-		if registry != nil && connState != nil && connState.LastObservedState == "READY" && !connState.LastConnectTime.IsZero() {
-			fmt.Printf("catalog %s pod with address %s\n", catalog.GetName(), registry.Address())
-			return registryPodHealthy(registry.Address())
-		}
-		fmt.Printf("waiting for catalog pod %v to be available (for sync)\n", catalog.GetName())
-		return false
-	})
+	fetchedUpdatedCatalog, err := fetchCatalogSource(t, crc, mainCatalogName, globalNS, catalogSourceRegistryPodSynced)
 	require.NoError(t, err)
 
-	subscription, err = fetchSubscription(t, crc, testNamespace, subscriptionName, subscriptionStateUpgradeAvailableChecker)
+	subscription, err = fetchSubscription(t, crc, testNamespace, subscriptionName, subscriptionStateUpgradePendingChecker)
 	require.NoError(t, err)
 	require.NotNil(t, subscription)