Merge pull request #428 from CSdread/master

pkg/sdk: Add label selector for sdk.Watch()

Author: fanminshi

doc/user-guide.md

@@ -65,11 +65,22 @@ func main() {
 }
 ```
 
-**Note:** The number of concurrent informer workers can be configured with an additional Watch option. The default value is 1 if an argument is not given.
+#### Options
+**Worker Count**
+The number of concurrent informer workers can be configured with an additional Watch option. The default value is 1 if an argument is not given.
 ```Go
 sdk.Watch("cache.example.com/v1alpha1", "Memcached", "default", 5, sdk.WithNumWorkers(n))
 ```
 
+**Label Selector**
+Label selectors allow the watch to filter resources by kubernetes labels. It can be specified using the standard kubernetes label selector format:
+
+https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#label-selectors
+
+```Go
+sdk.Watch("cache.example.com/v1alpha1", "Memcached", "default", 5, sdk.WithLabelSelector("app=myapp"))
+```
+
 ### Define the Memcached spec and status
 
 Modify the spec and status of the `Memcached` CR at `pkg/apis/cache/v1alpha1/types.go`:

pkg/sdk/api.go

@@ -54,7 +54,7 @@ func Watch(apiVersion, kind, namespace string, resyncPeriod time.Duration, opts
 	}
 	o := newWatchOp()
 	o.applyOpts(opts)
-	informer := NewInformer(resourcePluralName, namespace, resourceClient, resyncPeriod, collector, o.numWorkers)
+	informer := NewInformer(resourcePluralName, namespace, resourceClient, resyncPeriod, collector, o.numWorkers, o.labelSelector)
 	informers = append(informers, informer)
 }
 

pkg/sdk/informer.go

@@ -46,7 +46,7 @@ type informer struct {
 	numWorkers          int
 }
 
-func NewInformer(resourcePluralName, namespace string, resourceClient dynamic.ResourceInterface, resyncPeriod time.Duration, c *metrics.Collector, n int) Informer {
+func NewInformer(resourcePluralName, namespace string, resourceClient dynamic.ResourceInterface, resyncPeriod time.Duration, c *metrics.Collector, n int, labelSelector string) Informer {
 	i := &informer{
 		resourcePluralName: resourcePluralName,
 		queue:              workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), resourcePluralName),
@@ -57,7 +57,7 @@ func NewInformer(resourcePluralName, namespace string, resourceClient dynamic.Re
 	}
 
 	i.sharedIndexInformer = cache.NewSharedIndexInformer(
-		newListWatcherFromResourceClient(resourceClient), &unstructured.Unstructured{}, resyncPeriod, cache.Indexers{},
+		newListWatcherFromResourceClient(resourceClient, labelSelector), &unstructured.Unstructured{}, resyncPeriod, cache.Indexers{},
 	)
 	i.sharedIndexInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    i.handleAddResourceEvent,
@@ -67,11 +67,17 @@ func NewInformer(resourcePluralName, namespace string, resourceClient dynamic.Re
 	return i
 }
 
-func newListWatcherFromResourceClient(resourceClient dynamic.ResourceInterface) *cache.ListWatch {
+func newListWatcherFromResourceClient(resourceClient dynamic.ResourceInterface, labelSelector string) *cache.ListWatch {
 	listFunc := func(options metav1.ListOptions) (runtime.Object, error) {
+		if labelSelector != "" {
+			options.LabelSelector = labelSelector
+		}
 		return resourceClient.List(options)
 	}
 	watchFunc := func(options metav1.ListOptions) (watch.Interface, error) {
+		if labelSelector != "" {
+			options.LabelSelector = labelSelector
+		}
 		return resourceClient.Watch(options)
 	}
 	return &cache.ListWatch{ListFunc: listFunc, WatchFunc: watchFunc}

pkg/sdk/watch-opt.go

@@ -16,7 +16,8 @@ package sdk
 
 // WatchOp wraps all the options for Watch().
 type watchOp struct {
-	numWorkers int
+	numWorkers    int
+	labelSelector string
 }
 
 // NewWatchOp create a new deafult WatchOp
@@ -47,3 +48,9 @@ func WithNumWorkers(numWorkers int) watchOption {
 		op.numWorkers = numWorkers
 	}
 }
+
+func WithLabelSelector(labelSelector string) watchOption {
+	return func(op *watchOp) {
+		op.labelSelector = labelSelector
+	}
+}