Add support for creating namespaced watches

Author: Ulf Lilleengen

pkg/cache/cache.go

@@ -53,10 +53,18 @@ type Informers interface {
 	// API kind and resource.
 	GetInformer(obj runtime.Object) (Informer, error)
 
+	// GetInformer fetches or constructs an informer for the given object that corresponds to a single
+	// API kind and resource.
+	GetInformerInNamespace(obj runtime.Object, namespace *string) (Informer, error)
+
 	// GetInformerForKind is similar to GetInformer, except that it takes a group-version-kind, instead
 	// of the underlying object.
 	GetInformerForKind(gvk schema.GroupVersionKind) (Informer, error)
 
+	// GetInformerForKindInNamespace is similar to GetInformerInNamespace, except that it takes a group-version-kind, instead
+	// of the underlying object.
+	GetInformerForKindInNamespace(gvk schema.GroupVersionKind, namespace *string) (Informer, error)
+
 	// Start runs all the informers known to this cache until the given channel is closed.
 	// It blocks.
 	Start(stopCh <-chan struct{}) error

pkg/cache/informer_cache.go

@@ -57,7 +57,7 @@ func (ip *informerCache) Get(ctx context.Context, key client.ObjectKey, out runt
 		return err
 	}
 
-	started, cache, err := ip.InformersMap.Get(gvk, out)
+	started, cache, err := ip.InformersMap.Get(gvk, out, &key.Namespace)
 	if err != nil {
 		return err
 	}
@@ -101,7 +101,7 @@ func (ip *informerCache) List(ctx context.Context, out runtime.Object, opts ...c
 		}
 	}
 
-	started, cache, err := ip.InformersMap.Get(gvk, cacheTypeObj)
+	started, cache, err := ip.InformersMap.Get(gvk, cacheTypeObj, nil)
 	if err != nil {
 		return err
 	}
@@ -115,12 +115,16 @@ func (ip *informerCache) List(ctx context.Context, out runtime.Object, opts ...c
 
 // GetInformerForKind returns the informer for the GroupVersionKind
 func (ip *informerCache) GetInformerForKind(gvk schema.GroupVersionKind) (Informer, error) {
+	return ip.GetInformerForKindInNamespace(gvk, nil)
+}
+
+func (ip *informerCache) GetInformerForKindInNamespace(gvk schema.GroupVersionKind, namespace *string) (Informer, error) {
 	// Map the gvk to an object
 	obj, err := ip.Scheme.New(gvk)
 	if err != nil {
 		return nil, err
 	}
-	_, i, err := ip.InformersMap.Get(gvk, obj)
+	_, i, err := ip.InformersMap.Get(gvk, obj, namespace)
 	if err != nil {
 		return nil, err
 	}
@@ -129,11 +133,15 @@ func (ip *informerCache) GetInformerForKind(gvk schema.GroupVersionKind) (Inform
 
 // GetInformer returns the informer for the obj
 func (ip *informerCache) GetInformer(obj runtime.Object) (Informer, error) {
+	return ip.GetInformerInNamespace(obj, nil)
+}
+
+func (ip *informerCache) GetInformerInNamespace(obj runtime.Object, namespace *string) (Informer, error) {
 	gvk, err := apiutil.GVKForObject(obj, ip.Scheme)
 	if err != nil {
 		return nil, err
 	}
-	_, i, err := ip.InformersMap.Get(gvk, obj)
+	_, i, err := ip.InformersMap.Get(gvk, obj, namespace)
 	if err != nil {
 		return nil, err
 	}

pkg/cache/internal/deleg_map.go

@@ -73,16 +73,16 @@ func (m *InformersMap) WaitForCacheSync(stop <-chan struct{}) bool {
 
 // Get will create a new Informer and add it to the map of InformersMap if none exists.  Returns
 // the Informer from the map.
-func (m *InformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object) (bool, *MapEntry, error) {
+func (m *InformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object, namespace *string) (bool, *MapEntry, error) {
 	_, isUnstructured := obj.(*unstructured.Unstructured)
 	_, isUnstructuredList := obj.(*unstructured.UnstructuredList)
 	isUnstructured = isUnstructured || isUnstructuredList
 
 	if isUnstructured {
-		return m.unstructured.Get(gvk, obj)
+		return m.unstructured.Get(gvk, obj, namespace)
 	}
 
-	return m.structured.Get(gvk, obj)
+	return m.structured.Get(gvk, obj, namespace)
 }
 
 // newStructuredInformersMap creates a new InformersMap for structured objects.

pkg/cache/internal/informers_map.go

@@ -35,7 +35,7 @@ import (
 )
 
 // clientListWatcherFunc knows how to create a ListWatcher
-type createListWatcherFunc func(gvk schema.GroupVersionKind, ip *specificInformersMap) (*cache.ListWatch, error)
+type createListWatcherFunc func(gvk schema.GroupVersionKind, namespace *string, ip *specificInformersMap) (*cache.ListWatch, error)
 
 // newSpecificInformersMap returns a new specificInformersMap (like
 // the generical InformersMap, except that it doesn't implement WaitForCacheSync).
@@ -49,7 +49,7 @@ func newSpecificInformersMap(config *rest.Config,
 		config:            config,
 		Scheme:            scheme,
 		mapper:            mapper,
-		informersByGVK:    make(map[schema.GroupVersionKind]*MapEntry),
+		informers:         make(map[MapKey]*MapEntry),
 		codecs:            serializer.NewCodecFactory(scheme),
 		paramCodec:        runtime.NewParameterCodec(scheme),
 		resync:            resync,
@@ -59,6 +59,11 @@ func newSpecificInformersMap(config *rest.Config,
 	return ip
 }
 
+type MapKey struct {
+	gvk       schema.GroupVersionKind
+	namespace *string
+}
+
 // MapEntry contains the cached data for an Informer
 type MapEntry struct {
 	// Informer is the cached informer
@@ -80,8 +85,8 @@ type specificInformersMap struct {
 	// mapper maps GroupVersionKinds to Resources
 	mapper meta.RESTMapper
 
-	// informersByGVK is the cache of informers keyed by groupVersionKind
-	informersByGVK map[schema.GroupVersionKind]*MapEntry
+	// informers is the cache of informers keyed by groupVersionKind and namespace
+	informers map[MapKey]*MapEntry
 
 	// codecs is used to create a new REST client
 	codecs serializer.CodecFactory
@@ -122,7 +127,7 @@ func (ip *specificInformersMap) Start(stop <-chan struct{}) {
 		ip.stop = stop
 
 		// Start each informer
-		for _, informer := range ip.informersByGVK {
+		for _, informer := range ip.informers {
 			go informer.Informer.Run(stop)
 		}
 
@@ -136,27 +141,31 @@ func (ip *specificInformersMap) Start(stop <-chan struct{}) {
 func (ip *specificInformersMap) HasSyncedFuncs() []cache.InformerSynced {
 	ip.mu.RLock()
 	defer ip.mu.RUnlock()
-	syncedFuncs := make([]cache.InformerSynced, 0, len(ip.informersByGVK))
-	for _, informer := range ip.informersByGVK {
+	syncedFuncs := make([]cache.InformerSynced, 0, len(ip.informers))
+	for _, informer := range ip.informers {
 		syncedFuncs = append(syncedFuncs, informer.Informer.HasSynced)
 	}
 	return syncedFuncs
 }
 
 // Get will create a new Informer and add it to the map of specificInformersMap if none exists.  Returns
 // the Informer from the map.
-func (ip *specificInformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object) (bool, *MapEntry, error) {
+func (ip *specificInformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object, namespace *string) (bool, *MapEntry, error) {
+	key := MapKey{
+		gvk:       gvk,
+		namespace: namespace,
+	}
 	// Return the informer if it is found
 	i, started, ok := func() (*MapEntry, bool, bool) {
 		ip.mu.RLock()
 		defer ip.mu.RUnlock()
-		i, ok := ip.informersByGVK[gvk]
+		i, ok := ip.informers[key]
 		return i, ip.started, ok
 	}()
 
 	if !ok {
 		var err error
-		if i, started, err = ip.addInformerToMap(gvk, obj); err != nil {
+		if i, started, err = ip.addInformerToMap(key, obj); err != nil {
 			return started, nil, err
 		}
 	}
@@ -171,20 +180,20 @@ func (ip *specificInformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Obj
 	return started, i, nil
 }
 
-func (ip *specificInformersMap) addInformerToMap(gvk schema.GroupVersionKind, obj runtime.Object) (*MapEntry, bool, error) {
+func (ip *specificInformersMap) addInformerToMap(key MapKey, obj runtime.Object) (*MapEntry, bool, error) {
 	ip.mu.Lock()
 	defer ip.mu.Unlock()
 
 	// Check the cache to see if we already have an Informer.  If we do, return the Informer.
 	// This is for the case where 2 routines tried to get the informer when it wasn't in the map
 	// so neither returned early, but the first one created it.
-	if i, ok := ip.informersByGVK[gvk]; ok {
+	if i, ok := ip.informers[key]; ok {
 		return i, ip.started, nil
 	}
 
 	// Create a NewSharedIndexInformer and add it to the map.
 	var lw *cache.ListWatch
-	lw, err := ip.createListWatcher(gvk, ip)
+	lw, err := ip.createListWatcher(key.gvk, key.namespace, ip)
 	if err != nil {
 		return nil, false, err
 	}
@@ -193,9 +202,9 @@ func (ip *specificInformersMap) addInformerToMap(gvk schema.GroupVersionKind, ob
 	})
 	i := &MapEntry{
 		Informer: ni,
-		Reader:   CacheReader{indexer: ni.GetIndexer(), groupVersionKind: gvk},
+		Reader:   CacheReader{indexer: ni.GetIndexer(), groupVersionKind: key.gvk},
 	}
-	ip.informersByGVK[gvk] = i
+	ip.informers[key] = i
 
 	// Start the Informer if need by
 	// TODO(seans): write thorough tests and document what happens here - can you add indexers?
@@ -207,7 +216,7 @@ func (ip *specificInformersMap) addInformerToMap(gvk schema.GroupVersionKind, ob
 }
 
 // newListWatch returns a new ListWatch object that can be used to create a SharedIndexInformer.
-func createStructuredListWatch(gvk schema.GroupVersionKind, ip *specificInformersMap) (*cache.ListWatch, error) {
+func createStructuredListWatch(gvk schema.GroupVersionKind, namespace *string, ip *specificInformersMap) (*cache.ListWatch, error) {
 	// Kubernetes APIs work against Resources, not GroupVersionKinds.  Map the
 	// groupVersionKind to the Resource API we will use.
 	mapping, err := ip.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
@@ -225,25 +234,30 @@ func createStructuredListWatch(gvk schema.GroupVersionKind, ip *specificInformer
 		return nil, err
 	}
 
+	ns := ip.namespace
+	if namespace != nil {
+		ns = *namespace
+	}
+
 	// Create a new ListWatch for the obj
 	return &cache.ListWatch{
 		ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
 			res := listObj.DeepCopyObject()
-			isNamespaceScoped := ip.namespace != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot
-			err := client.Get().NamespaceIfScoped(ip.namespace, isNamespaceScoped).Resource(mapping.Resource.Resource).VersionedParams(&opts, ip.paramCodec).Do().Into(res)
+			isNamespaceScoped := ns != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot
+			err := client.Get().NamespaceIfScoped(ns, isNamespaceScoped).Resource(mapping.Resource.Resource).VersionedParams(&opts, ip.paramCodec).Do().Into(res)
 			return res, err
 		},
 		// Setup the watch function
 		WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
 			// Watch needs to be set to true separately
 			opts.Watch = true
-			isNamespaceScoped := ip.namespace != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot
-			return client.Get().NamespaceIfScoped(ip.namespace, isNamespaceScoped).Resource(mapping.Resource.Resource).VersionedParams(&opts, ip.paramCodec).Watch()
+			isNamespaceScoped := ns != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot
+			return client.Get().NamespaceIfScoped(ns, isNamespaceScoped).Resource(mapping.Resource.Resource).VersionedParams(&opts, ip.paramCodec).Watch()
 		},
 	}, nil
 }
 
-func createUnstructuredListWatch(gvk schema.GroupVersionKind, ip *specificInformersMap) (*cache.ListWatch, error) {
+func createUnstructuredListWatch(gvk schema.GroupVersionKind, namespace *string, ip *specificInformersMap) (*cache.ListWatch, error) {
 	// Kubernetes APIs work against Resources, not GroupVersionKinds.  Map the
 	// groupVersionKind to the Resource API we will use.
 	mapping, err := ip.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
@@ -255,20 +269,25 @@ func createUnstructuredListWatch(gvk schema.GroupVersionKind, ip *specificInform
 		return nil, err
 	}
 
+	ns := ip.namespace
+	if namespace != nil {
+		ns = *namespace
+	}
+
 	// Create a new ListWatch for the obj
 	return &cache.ListWatch{
 		ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
-			if ip.namespace != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot {
-				return dynamicClient.Resource(mapping.Resource).Namespace(ip.namespace).List(opts)
+			if ns != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot {
+				return dynamicClient.Resource(mapping.Resource).Namespace(ns).List(opts)
 			}
 			return dynamicClient.Resource(mapping.Resource).List(opts)
 		},
 		// Setup the watch function
 		WatchFunc: func(opts metav1.ListOptions) (watch.Interface, error) {
 			// Watch needs to be set to true separately
 			opts.Watch = true
-			if ip.namespace != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot {
-				return dynamicClient.Resource(mapping.Resource).Namespace(ip.namespace).Watch(opts)
+			if ns != "" && mapping.Scope.Name() != meta.RESTScopeNameRoot {
+				return dynamicClient.Resource(mapping.Resource).Namespace(ns).Watch(opts)
 			}
 			return dynamicClient.Resource(mapping.Resource).Watch(opts)
 		},

pkg/cache/multi_namespace_cache.go

@@ -69,9 +69,12 @@ var _ Cache = &multiNamespaceCache{}
 
 // Methods for multiNamespaceCache to conform to the Informers interface
 func (c *multiNamespaceCache) GetInformer(obj runtime.Object) (Informer, error) {
+	return c.GetInformerInNamespace(obj, nil)
+}
+func (c *multiNamespaceCache) GetInformerInNamespace(obj runtime.Object, namespace *string) (Informer, error) {
 	informers := map[string]Informer{}
 	for ns, cache := range c.namespaceToCache {
-		informer, err := cache.GetInformer(obj)
+		informer, err := cache.GetInformerInNamespace(obj, namespace)
 		if err != nil {
 			return nil, err
 		}
@@ -81,9 +84,13 @@ func (c *multiNamespaceCache) GetInformer(obj runtime.Object) (Informer, error)
 }
 
 func (c *multiNamespaceCache) GetInformerForKind(gvk schema.GroupVersionKind) (Informer, error) {
+	return c.GetInformerForKindInNamespace(gvk, nil)
+}
+
+func (c *multiNamespaceCache) GetInformerForKindInNamespace(gvk schema.GroupVersionKind, namespace *string) (Informer, error) {
 	informers := map[string]Informer{}
 	for ns, cache := range c.namespaceToCache {
-		informer, err := cache.GetInformerForKind(gvk)
+		informer, err := cache.GetInformerForKindInNamespace(gvk, namespace)
 		if err != nil {
 			return nil, err
 		}

pkg/source/source.go

@@ -60,6 +60,9 @@ type Kind struct {
 	// Type is the type of object to watch.  e.g. &v1.Pod{}
 	Type runtime.Object
 
+	// Namespace is the namespace for watching the objects
+	Namespace *string
+
 	// cache used to watch APIs
 	cache cache.Cache
 }
@@ -82,7 +85,7 @@ func (ks *Kind) Start(handler handler.EventHandler, queue workqueue.RateLimiting
 	}
 
 	// Lookup the Informer from the Cache and add an EventHandler which populates the Queue
-	i, err := ks.cache.GetInformer(ks.Type)
+	i, err := ks.cache.GetInformerInNamespace(ks.Type, ks.Namespace)
 	if err != nil {
 		if kindMatchErr, ok := err.(*meta.NoKindMatchError); ok {
 			log.Error(err, "if kind is a CRD, it should be installed before calling Start",