Merge pull request #362 from ecordell/catalog-fixes

Fix memory usage in catalog operator

pkg/controller/operators/catalog/operator.go

@@ -58,9 +58,7 @@ func NewOperator(kubeconfigPath string, wakeupInterval time.Duration, operatorNa
 		return nil, err
 	}
 
-	sharedInformerFactory := externalversions.NewSharedInformerFactory(crClient, wakeupInterval)
-
-	// Create an informer for each namespace.
+	// Create an informer for each watched namespace.
 	ipSharedIndexInformers := []cache.SharedIndexInformer{}
 	subSharedIndexInformers := []cache.SharedIndexInformer{}
 	for _, namespace := range watchedNamespaces {
@@ -69,6 +67,13 @@ func NewOperator(kubeconfigPath string, wakeupInterval time.Duration, operatorNa
 		subSharedIndexInformers = append(subSharedIndexInformers, nsInformerFactory.Subscription().V1alpha1().Subscriptions().Informer())
 	}
 
+	// Create an informer for each catalog namespace
+	catsrcSharedIndexInformers := []cache.SharedIndexInformer{}
+	for _, namespace := range []string{operatorNamespace} {
+		nsInformerFactory := externalversions.NewFilteredSharedInformerFactory(crClient, wakeupInterval, namespace, nil)
+		catsrcSharedIndexInformers = append(catsrcSharedIndexInformers, nsInformerFactory.Catalogsource().V1alpha1().CatalogSources().Informer())
+	}
+
 	// Create a new queueinformer-based operator.
 	queueOperator, err := queueinformer.NewOperator(kubeconfigPath)
 	if err != nil {
@@ -86,9 +91,7 @@ func NewOperator(kubeconfigPath string, wakeupInterval time.Duration, operatorNa
 	catsrcQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "catalogsources")
 	catsrcQueueInformer := queueinformer.New(
 		catsrcQueue,
-		[]cache.SharedIndexInformer{
-			sharedInformerFactory.Catalogsource().V1alpha1().CatalogSources().Informer(),
-		},
+		catsrcSharedIndexInformers,
 		op.syncCatalogSources,
 		nil,
 	)
@@ -139,7 +142,7 @@ func (o *Operator) syncCatalogSources(obj interface{}) (syncError error) {
 	defer o.sourcesLock.Unlock()
 	o.sources[catsrc.GetName()] = src
 	o.sourcesLastUpdate = timeNow()
-	return err
+	return nil
 }
 
 func (o *Operator) syncSubscriptions(obj interface{}) (syncError error) {

pkg/controller/registry/configmap_loader.go

@@ -8,8 +8,8 @@ import (
 	log "github.com/sirupsen/logrus"
 	"k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1beta1"
 
-	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/clusterserviceversion/v1alpha1"
 	"github.com/coreos-inc/tectonic-operators/operator-client/pkg/client"
+	"github.com/operator-framework/operator-lifecycle-manager/pkg/api/apis/clusterserviceversion/v1alpha1"
 	"k8s.io/api/core/v1"
 )
 
@@ -21,23 +21,22 @@ const (
 
 // ConfigMapCatalogResourceLoader loads a ConfigMap of resources into the in-memory catalog
 type ConfigMapCatalogResourceLoader struct {
-	Catalog   *InMem
 	Namespace string
 	CMClient  client.ConfigMapClient
 }
 
-func (d *ConfigMapCatalogResourceLoader) LoadCatalogResources(configMapName string) error {
+func (d *ConfigMapCatalogResourceLoader) LoadCatalogResources(catalog *InMem, configMapName string) error {
 	log.Debugf("Load ConfigMap     -- BEGIN %s", configMapName)
 
 	cm, err := d.CMClient.GetConfigMap(d.Namespace, configMapName)
 	if err != nil {
 		log.Debugf("Load ConfigMap     -- ERROR %s : error=%s", configMapName, err)
 		return fmt.Errorf("error loading catalog from ConfigMap %s: %s", configMapName, err)
 	}
-	return d.LoadCatalogResourcesFromConfigMap(cm)
+	return d.LoadCatalogResourcesFromConfigMap(catalog, cm)
 }
 
-func (d *ConfigMapCatalogResourceLoader) LoadCatalogResourcesFromConfigMap(cm *v1.ConfigMap) error {
+func (d *ConfigMapCatalogResourceLoader) LoadCatalogResourcesFromConfigMap(catalog *InMem, cm *v1.ConfigMap) error {
 	configMapName := cm.GetName()
 	found := false
 	crdListYaml, ok := cm.Data[ConfigMapCRDName]
@@ -57,7 +56,7 @@ func (d *ConfigMapCatalogResourceLoader) LoadCatalogResourcesFromConfigMap(cm *v
 
 		for _, crd := range parsedCRDList {
 			found = true
-			d.Catalog.SetCRDDefinition(crd)
+			catalog.SetCRDDefinition(crd)
 		}
 	}
 
@@ -78,7 +77,7 @@ func (d *ConfigMapCatalogResourceLoader) LoadCatalogResourcesFromConfigMap(cm *v
 
 		for _, csv := range parsedCSVList {
 			found = true
-			d.Catalog.setCSVDefinition(csv)
+			catalog.setCSVDefinition(csv)
 		}
 	}
 
@@ -99,11 +98,11 @@ func (d *ConfigMapCatalogResourceLoader) LoadCatalogResourcesFromConfigMap(cm *v
 		}
 		for _, packageManifest := range parsedPackageManifests {
 			found = true
-			if err := d.Catalog.addPackageManifest(packageManifest); err != nil {
+			if err := catalog.addPackageManifest(packageManifest); err != nil {
 				return err
 			}
 		}
-		log.Debugf("Load ConfigMap      -- Found packages: %v", d.Catalog.packages)
+		log.Debugf("Load ConfigMap      -- Found packages: %v", catalog.packages)
 	}
 
 	if !found {

pkg/controller/registry/mem.go

@@ -51,11 +51,12 @@ func NewInMemoryFromDirectory(directory string) (*InMem, error) {
 
 func NewInMemoryFromConfigMap(cmClient client.ConfigMapClient, namespace, cmName string) (*InMem, error) {
 	log.Infof("loading ui catalog entries from a configmap: %s", cmName)
-	loader := ConfigMapCatalogResourceLoader{NewInMem(), namespace, cmClient}
-	if err := loader.LoadCatalogResources(cmName); err != nil {
+	loader := ConfigMapCatalogResourceLoader{namespace, cmClient}
+	catalog := NewInMem()
+	if err := loader.LoadCatalogResources(catalog, cmName); err != nil {
 		return nil, err
 	}
-	return loader.Catalog, nil
+	return catalog, nil
 }
 
 // NewInMem returns a ptr to a new InMem instance

pkg/server/servicebroker/apiserver.go

@@ -97,15 +97,16 @@ func (c *inClusterCatalog) Load(namespace string) (registry.Source, error) {
 	}
 
 	// load service definitions from configmaps into temp in memory service registry
-	loader := registry.ConfigMapCatalogResourceLoader{registry.NewInMem(), namespace, c.opClient}
+	catalog := registry.NewInMem()
+	loader := registry.ConfigMapCatalogResourceLoader{namespace, c.opClient}
 	for _, cs := range csList.Items {
 		loader.Namespace = cs.GetNamespace()
-		if err := loader.LoadCatalogResources(cs.Spec.ConfigMap); err != nil {
+		if err := loader.LoadCatalogResources(catalog, cs.Spec.ConfigMap); err != nil {
 			log.Errorf("Component=ServiceBroker Endpoint=GetCatalog Error=%s", err)
 			return nil, err
 		}
 	}
-	return loader.Catalog, nil
+	return catalog, nil
 }
 
 func (a *ALMBroker) ValidateBrokerAPIVersion(version string) error {

pkg/server/servicebroker/apiserver_test.go

@@ -27,18 +27,18 @@ type mockCatalogLoader struct {
 
 func (m *mockCatalogLoader) Load(namespace string) (registry.Source, error) {
 	loader := registry.ConfigMapCatalogResourceLoader{
-		Catalog:   registry.NewInMem(),
 		Namespace: namespace,
 	}
+	catalog := registry.NewInMem()
 	for _, cm := range m.configMaps {
 		if namespace != "" && cm.GetNamespace() != namespace {
 			continue
 		}
-		if err := loader.LoadCatalogResourcesFromConfigMap(&cm); err != nil {
+		if err := loader.LoadCatalogResourcesFromConfigMap(catalog, &cm); err != nil {
 			return nil, err
 		}
 	}
-	return loader.Catalog, nil
+	return catalog, nil
 }
 
 func mockALMBroker(ctrl *gomock.Controller, namespace string, configMaps []v1.ConfigMap, objects []runtime.Object) *ALMBroker {

test/schema/catalog_versions_test.go

@@ -41,9 +41,7 @@ type LoadedCatalog struct {
 
 // loadCatalogFromFile loads an in memory catalog from a file path. Only used for testing.
 func loadCatalogFromFile(path string) (*LoadedCatalog, error) {
-	loader := registry.ConfigMapCatalogResourceLoader{
-		Catalog: registry.NewInMem(),
-	}
+	loader := registry.ConfigMapCatalogResourceLoader{}
 	currentBytes, err := ioutil.ReadFile(path)
 	if err != nil {
 		return nil, err
@@ -57,12 +55,13 @@ func loadCatalogFromFile(path string) (*LoadedCatalog, error) {
 	if err != nil {
 		return nil, err
 	}
-	err = loader.LoadCatalogResourcesFromConfigMap(&currentConfigMap)
+	catalog := registry.NewInMem()
+	err = loader.LoadCatalogResourcesFromConfigMap(catalog, &currentConfigMap)
 	if err != nil {
 		return nil, err
 	}
 	return &LoadedCatalog{
-		Registry: loader.Catalog,
+		Registry: catalog,
 		Name:     currentConfigMap.Name,
 	}, nil
 }