fix(queueInformers): use separate queue for each namespace

- Uses a separate queue for each namespaced QueueInformer
- Extends `make e2e-local` test timeout to 20 minutes
- Checks for olm.operatorNamespace difference to determine copied CSVs

Author: njhale

go.mod

@@ -66,7 +66,7 @@ require (
 	google.golang.org/grpc v1.16.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0-20170531160350-a96e63847dc3 // indirect
-	k8s.io/api v0.0.0-20180904230853-4e7be11eab3f
+	k8s.io/api v0.0.0-20181108095152-eee84a6322ca
 	k8s.io/apiextensions-apiserver v0.0.0-20180905004947-16750353bf97
 	k8s.io/apimachinery v0.0.0-20181026144827-8ee1a638bafa
 	k8s.io/apiserver v0.0.0-20181026151315-13cfe3978170

go.sum

@@ -210,8 +210,8 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWD
 gopkg.in/yaml.v2 v2.2.1 h1:mUhvW9EsL+naU5Q3cakzfE91YhliOondGd6ZrsDBHQE=
 gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 honnef.co/go/tools v0.0.0-20180728063816-88497007e858/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
-k8s.io/api v0.0.0-20180904230853-4e7be11eab3f h1:DLRkv8Ps4Sdx8Srj+UtGisj4whV7v/HezlHx6QqiZqE=
-k8s.io/api v0.0.0-20180904230853-4e7be11eab3f/go.mod h1:iuAfoD4hCxJ8Onx9kaTIt30j7jUFS00AXQi6QMi99vA=
+k8s.io/api v0.0.0-20181108095152-eee84a6322ca h1:0gIeW03B5m7yni69Y95oPgDXv7ow7puUMt2WqhJIKY8=
+k8s.io/api v0.0.0-20181108095152-eee84a6322ca/go.mod h1:iuAfoD4hCxJ8Onx9kaTIt30j7jUFS00AXQi6QMi99vA=
 k8s.io/apiextensions-apiserver v0.0.0-20180905004947-16750353bf97 h1:s4lWWs6JN5kWVzk5bztddkr5kgO/cGIbqTDP+QttUeQ=
 k8s.io/apiextensions-apiserver v0.0.0-20180905004947-16750353bf97/go.mod h1:IxkesAMoaCRoLrPJdZNZUQp9NfZnzqaVzLhb2VEQzXE=
 k8s.io/apimachinery v0.0.0-20181026144827-8ee1a638bafa h1:i0EOpPFWExNx7efINILpw8LJeah7gakRl1zjvwVfjiI=

pkg/controller/operators/olm/operator.go

@@ -50,10 +50,10 @@ const (
 
 type Operator struct {
 	*queueinformer.Operator
-	csvQueue workqueue.RateLimitingInterface
-	client   versioned.Interface
-	resolver install.StrategyResolverInterface
-	lister   operatorlister.OperatorLister
+	csvQueues   map[string]workqueue.RateLimitingInterface
+	client      versioned.Interface
+	resolver    install.StrategyResolverInterface
+	lister      operatorlister.OperatorLister
 	recorder    record.EventRecorder
 }
 
@@ -75,11 +75,12 @@ func NewOperator(crClient versioned.Interface, opClient operatorclient.ClientInt
 	}
 
 	op := &Operator{
-		Operator: queueOperator,
-		client:   crClient,
-		lister:   operatorlister.NewLister(),
-		resolver: resolver,
-		recorder: eventRecorder,
+		Operator:  queueOperator,
+		csvQueues: make(map[string]workqueue.RateLimitingInterface),
+		client:    crClient,
+		lister:    operatorlister.NewLister(),
+		resolver:  resolver,
+		recorder:  eventRecorder,
 	}
 
 	// Set up RBAC informers
@@ -96,7 +97,7 @@ func NewOperator(crClient versioned.Interface, opClient operatorclient.ClientInt
 		namespaceInformer.Informer(),
 		op.syncObject,
 		nil,
-		"namespace",
+		"namespaces",
 		metrics.NewMetricsNil(),
 	)
 	op.RegisterQueueInformer(queueInformer)
@@ -117,7 +118,7 @@ func NewOperator(crClient versioned.Interface, opClient operatorclient.ClientInt
 		&cache.ResourceEventHandlerFuncs{
 			DeleteFunc: op.handleDeletion,
 		},
-		"namespace",
+		"rbac",
 		metrics.NewMetricsNil(),
 	)
 	for _, informer := range rbacQueueInformers {
@@ -181,95 +182,74 @@ func NewOperator(crClient versioned.Interface, opClient operatorclient.ClientInt
 		&cache.ResourceEventHandlerFuncs{
 			DeleteFunc: op.handleDeletion,
 		},
-		"services",
+		"serviceaccounts",
 		metrics.NewMetricsNil(),
 	))
 	op.lister.CoreV1().RegisterServiceAccountLister(metav1.NamespaceAll, serviceAccountInformer.Lister())
 
-	// Set up watch on CSVs
-	csvInformers := []cache.SharedIndexInformer{}
+	// csvInformers for each namespace all use the same backing queue keys are namespaced
+	csvHandlers := &cache.ResourceEventHandlerFuncs{
+		DeleteFunc: op.deleteClusterServiceVersion,
+	}
 	for _, namespace := range namespaces {
-		log.Debugf("watching for CSVs in namespace %s", namespace)
+		log.WithField("namespace", namespace).Infof("watching CSVs")
 		sharedInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(crClient, wakeupInterval, externalversions.WithNamespace(namespace))
-		informer := sharedInformerFactory.Operators().V1alpha1().ClusterServiceVersions()
-		csvInformers = append(csvInformers, informer.Informer())
-		op.lister.OperatorsV1alpha1().RegisterClusterServiceVersionLister(namespace, informer.Lister())
-	}
+		csvInformer := sharedInformerFactory.Operators().V1alpha1().ClusterServiceVersions()
+		op.lister.OperatorsV1alpha1().RegisterClusterServiceVersionLister(namespace, csvInformer.Lister())
 
-	// csvInformers for each namespace all use the same backing queue
-	// queue keys are namespaced
-	csvQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "clusterserviceversions")
-	csvHandlers := cache.ResourceEventHandlerFuncs{
-		DeleteFunc: op.deleteClusterServiceVersion,
-	}
-	queueInformers := queueinformer.New(
-		csvQueue,
-		csvInformers,
-		op.syncClusterServiceVersion,
-		&csvHandlers,
-		"csv",
-		metrics.NewMetricsCSV(op.client),
-	)
-	for _, informer := range queueInformers {
-		op.RegisterQueueInformer(informer)
+		// Register queue and QueueInformer
+		queueName := fmt.Sprintf("%s/clusterserviceversions", namespace)
+		csvQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), queueName)
+		csvQueueInformer := queueinformer.NewInformer(csvQueue, csvInformer.Informer(), op.syncClusterServiceVersion, csvHandlers, queueName, metrics.NewMetricsCSV(op.client))
+		op.RegisterQueueInformer(csvQueueInformer)
+		op.csvQueues[namespace] = csvQueue
 	}
-	op.csvQueue = csvQueue
 
-	// set up watch on deployments
-	depInformers := []cache.SharedIndexInformer{}
-	for _, namespace := range namespaces {
-		log.Debugf("watching deployments in namespace %s", namespace)
-		informer := informers.NewSharedInformerFactoryWithOptions(opClient.KubernetesInterface(), wakeupInterval, informers.WithNamespace(namespace)).Apps().V1().Deployments()
-		depInformers = append(depInformers, informer.Informer())
-		op.lister.AppsV1().RegisterDeploymentLister(namespace, informer.Lister())
+	// Set up watch on deployments
+	depHandlers := &cache.ResourceEventHandlerFuncs{
+		DeleteFunc: op.handleDeletion,
 	}
+	for _, namespace := range namespaces {
+		log.WithField("namespace", namespace).Infof("watching deployments")
+		depInformer := informers.NewSharedInformerFactoryWithOptions(opClient.KubernetesInterface(), wakeupInterval, informers.WithNamespace(namespace)).Apps().V1().Deployments()
+		op.lister.AppsV1().RegisterDeploymentLister(namespace, depInformer.Lister())
 
-	depQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "csv-deployments")
-	depQueueInformers := queueinformer.New(
-		depQueue,
-		depInformers,
-		op.syncDeployment,
-		&cache.ResourceEventHandlerFuncs{
-			DeleteFunc: op.handleDeletion,
-		},
-		"deployment",
-		metrics.NewMetricsNil(),
-	)
-	for _, informer := range depQueueInformers {
-		op.RegisterQueueInformer(informer)
+		// Register queue and QueueInformer
+		queueName := fmt.Sprintf("%s/csv-deployments", namespace)
+		depQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), queueName)
+		depQueueInformer := queueinformer.NewInformer(depQueue, depInformer.Informer(), op.syncDeployment, depHandlers, queueName, metrics.NewMetricsNil())
+		op.RegisterQueueInformer(depQueueInformer)
 	}
 
 	// Create an informer for the operator group
-	operatorGroupInformers := []cache.SharedIndexInformer{}
 	for _, namespace := range namespaces {
-		informerFactory := externalversions.NewSharedInformerFactoryWithOptions(crClient, wakeupInterval, externalversions.WithNamespace(namespace))
-		informer := informerFactory.Operators().V1alpha2().OperatorGroups()
-		operatorGroupInformers = append(operatorGroupInformers, informer.Informer())
-		op.lister.OperatorsV1alpha2().RegisterOperatorGroupLister(namespace, informer.Lister())
-	}
-
-	// Register OperatorGroup informers.
-	operatorGroupQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "operatorgroups")
-	operatorGroupQueueInformer := queueinformer.New(
-		operatorGroupQueue,
-		operatorGroupInformers,
-		op.syncOperatorGroups,
-		nil,
-		"operatorgroups",
-		metrics.NewMetricsNil(),
-	)
-	for _, informer := range operatorGroupQueueInformer {
-		op.RegisterQueueInformer(informer)
+		log.WithField("namespace", namespace).Infof("watching OperatorGroups")
+		sharedInformerFactory := externalversions.NewSharedInformerFactoryWithOptions(crClient, wakeupInterval, externalversions.WithNamespace(namespace))
+		operatorGroupInformer := sharedInformerFactory.Operators().V1alpha2().OperatorGroups()
+		op.lister.OperatorsV1alpha2().RegisterOperatorGroupLister(namespace, operatorGroupInformer.Lister())
+
+		// Register queue and QueueInformer
+		queueName := fmt.Sprintf("%s/operatorgroups", namespace)
+		operatorGroupQueue := workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), queueName)
+		operatorGroupQueueInformer := queueinformer.NewInformer(operatorGroupQueue, operatorGroupInformer.Informer(), op.syncOperatorGroups, nil, queueName, metrics.NewMetricsNil())
+		op.RegisterQueueInformer(operatorGroupQueueInformer)
 	}
 
 	return op, nil
 }
 
 func (a *Operator) requeueCSV(name, namespace string) {
-	// we can build the key directly, will need to change if queue uses different key scheme
+	// We can build the key directly, will need to change if queue uses different key scheme
 	key := fmt.Sprintf("%s/%s", namespace, name)
-	log.Debugf("requeueing CSV %s", key)
-	a.csvQueue.AddRateLimited(key)
+	logger := log.WithField("key", key)
+	logger.Debugf("requeueing CSV")
+
+	if queue, ok := a.csvQueues[namespace]; ok {
+		queue.AddRateLimited(key)
+		return
+	}
+
+	logger.Debugf("couldn't find queue for CSV")
 }
 
 func (a *Operator) syncDeployment(obj interface{}) (syncError error) {
@@ -398,7 +378,9 @@ func (a *Operator) syncClusterServiceVersion(obj interface{}) (syncError error)
 		"phase":     clusterServiceVersion.Status.Phase,
 	})
 
-	if clusterServiceVersion.Status.Reason == v1alpha1.CSVReasonCopied {
+	operatorNamespace, ok := clusterServiceVersion.GetAnnotations()["olm.operatorNamespace"]
+	if clusterServiceVersion.Status.Reason == v1alpha1.CSVReasonCopied || 
+		ok && clusterServiceVersion.GetNamespace() != operatorNamespace {
 		logger.Info("skip sync of dummy CSV")
 		return a.removeDanglingChildCSVs(clusterServiceVersion)
 	}

pkg/controller/operators/olm/operator_test.go

@@ -133,11 +133,11 @@ func NewFakeOperator(clientObjs []runtime.Object, k8sObjs []runtime.Object, extO
 	// Create the new operator
 	queueOperator, err := queueinformer.NewOperatorFromClient(opClientFake)
 	op := &Operator{
-		Operator:  queueOperator,
-		client:    clientFake,
-		lister:    operatorlister.NewLister(),
-		resolver:  resolver,
-		csvQueue:  workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "clusterserviceversions"),
+		Operator: queueOperator,
+		client:   clientFake,
+		lister:   operatorlister.NewLister(),
+		resolver: resolver,
+		csvQueues: make(map[string]workqueue.RateLimitingInterface),
 		recorder:  eventRecorder,
 	}
 

scripts/run_e2e_local.sh

@@ -42,4 +42,4 @@ trap cleanupAndExit SIGINT SIGTERM EXIT
 
 # run tests
 e2e_kubeconfig=${KUBECONFIG:-~/.kube/config}
-KUBECONFIG=${e2e_kubeconfig} NAMESPACE=${namespace} go test -v ./test/e2e/... ${1/[[:alnum:]-]*/-run ${1}}
+KUBECONFIG=${e2e_kubeconfig} NAMESPACE=${namespace} go test -v -timeout 20m ./test/e2e/... ${1/[[:alnum:]-]*/-run ${1}}