pkg/sdk: Make number of workers configurable

pkg/sdk/api.go

@@ -40,7 +40,7 @@ var (
 // Consult the API reference for the Group, Version and Kind of a resource: https://kubernetes.io/docs/reference/
 // namespace is the Namespace to watch for the resource
 // TODO: support opts for specifying label selector
-func Watch(apiVersion, kind, namespace string, resyncPeriod int) {
+func Watch(apiVersion, kind, namespace string, resyncPeriod int, opts ...WatchOption) {
 	resourceClient, resourcePluralName, err := k8sclient.GetResourceClient(apiVersion, kind, namespace)
 	// TODO: Better error handling, e.g retry
 	if err != nil {
@@ -51,7 +51,9 @@ func Watch(apiVersion, kind, namespace string, resyncPeriod int) {
 		collector = metrics.New()
 		metrics.RegisterCollector(collector)
 	}
-	informer := NewInformer(resourcePluralName, namespace, resourceClient, resyncPeriod, collector)
+	o := NewWatchOp()
+	o.applyOpts(opts)
+	informer := NewInformer(resourcePluralName, namespace, resourceClient, resyncPeriod, collector, o.NumWorkers)
 	informers = append(informers, informer)
 }
 

pkg/sdk/informer.go

@@ -43,15 +43,17 @@ type informer struct {
 	context             context.Context
 	deletedObjects      map[string]interface{}
 	collector           *metrics.Collector
+	numWorkers          int
 }
 
-func NewInformer(resourcePluralName, namespace string, resourceClient dynamic.ResourceInterface, resyncPeriod int, c *metrics.Collector) Informer {
+func NewInformer(resourcePluralName, namespace string, resourceClient dynamic.ResourceInterface, resyncPeriod int, c *metrics.Collector, n int) Informer {
 	i := &informer{
 		resourcePluralName: resourcePluralName,
 		queue:              workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), resourcePluralName),
 		namespace:          namespace,
 		deletedObjects:     map[string]interface{}{},
 		collector:          c,
+		numWorkers:         n,
 	}
 
 	resyncDuration := time.Duration(resyncPeriod) * time.Second
@@ -87,8 +89,7 @@ func (i *informer) Run(ctx context.Context) {
 		panic("Timed out waiting for caches to sync")
 	}
 
-	const numWorkers = 1
-	for n := 0; n < numWorkers; n++ {
+	for n := 0; n < i.numWorkers; n++ {
 		go wait.Until(i.runWorker, time.Second, ctx.Done())
 	}
 	<-ctx.Done()

pkg/sdk/watch-opt.go

@@ -0,0 +1,49 @@
+// Copyright 2018 The Operator-SDK Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package sdk
+
+// WatchOp wraps all the options for Watch().
+type WatchOp struct {
+	NumWorkers int
+}
+
+// NewWatchOp create a new deafult WatchOp
+func NewWatchOp() *WatchOp {
+	op := &WatchOp{}
+	op.setDefaults()
+	return op
+}
+
+func (op *WatchOp) applyOpts(opts []WatchOption) {
+	for _, opt := range opts {
+		opt(op)
+	}
+}
+
+func (op *WatchOp) setDefaults() {
+	if op.NumWorkers == 0 {
+		op.NumWorkers = 1
+	}
+}
+
+// WatchOption configures WatchOp.
+type WatchOption func(*WatchOp)
+
+// WithWatchOptions sets the number of workers for the Watch() operation.
+func WithWatchOptions(numWorkers int) WatchOption {
+	return func(op *WatchOp) {
+		op.NumWorkers = numWorkers
+	}
+}