Adding ability to use dynamic list for unstructured list watcher.

pkg/cache/internal/deleg_map.go

@@ -0,0 +1,95 @@
+/*
+Copyright 2018 The Kubernetes 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 internal
+
+import (
+	"time"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
+)
+
+// InformersMap create and caches Informers for (runtime.Object, schema.GroupVersionKind) pairs.
+// It uses a standard parameter codec constructed based on the given generated Scheme.
+type InformersMap struct {
+	// we abstract over the details of structured vs unstructured with the specificInformerMaps
+
+	structured   *specificInformersMap
+	unstructured *specificInformersMap
+
+	// Scheme maps runtime.Objects to GroupVersionKinds
+	Scheme *runtime.Scheme
+}
+
+// NewInformersMap creates a new InformersMap that can create informers for
+// both structured and unstructured objects.
+func NewInformersMap(config *rest.Config,
+	scheme *runtime.Scheme,
+	mapper meta.RESTMapper,
+	resync time.Duration) *InformersMap {
+
+	return &InformersMap{
+		structured:   newStructuredInformersMap(config, scheme, mapper, resync),
+		unstructured: newUnstructuredInformersMap(config, scheme, mapper, resync),
+
+		Scheme: scheme,
+	}
+}
+
+// Start calls Run on each of the informers and sets started to true.  Blocks on the stop channel.
+func (m *InformersMap) Start(stop <-chan struct{}) error {
+	go m.structured.Start(stop)
+	go m.unstructured.Start(stop)
+	<-stop
+	return nil
+}
+
+// WaitForCacheSync waits until all the caches have been synced.
+func (m *InformersMap) WaitForCacheSync(stop <-chan struct{}) bool {
+	syncedFuncs := append([]cache.InformerSynced(nil), m.structured.HasSyncedFuncs()...)
+	syncedFuncs = append(syncedFuncs, m.unstructured.HasSyncedFuncs()...)
+
+	return cache.WaitForCacheSync(stop, syncedFuncs...)
+}
+
+// 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) (*MapEntry, error) {
+	_, isUnstructured := obj.(*unstructured.Unstructured)
+	_, isUnstructuredList := obj.(*unstructured.UnstructuredList)
+	isUnstructured = isUnstructured || isUnstructuredList
+
+	if isUnstructured {
+		return m.unstructured.Get(gvk, obj)
+	}
+
+	return m.structured.Get(gvk, obj)
+}
+
+// newStructuredInformersMap creates a new InformersMap for structured objects.
+func newStructuredInformersMap(config *rest.Config, scheme *runtime.Scheme, mapper meta.RESTMapper, resync time.Duration) *specificInformersMap {
+	return newSpecificInformersMap(config, scheme, mapper, resync, createStructuredListWatch)
+}
+
+// newUnstructuredInformersMap creates a new InformersMap for unstructured objects.
+func newUnstructuredInformersMap(config *rest.Config, scheme *runtime.Scheme, mapper meta.RESTMapper, resync time.Duration) *specificInformersMap {
+	return newSpecificInformersMap(config, scheme, mapper, resync, createUnstructuredListWatch)
+}

pkg/cache/internal/informers_map.go

@@ -23,30 +23,35 @@ import (
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/apimachinery/pkg/watch"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
+
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
 )
 
-// NewInformersMap returns a new InformersMap
-func NewInformersMap(config *rest.Config,
+// clientListWatcherFunc knows how to create a ListWatcher
+type createListWatcherFunc func(gvk schema.GroupVersionKind, ip *specificInformersMap) (*cache.ListWatch, error)
+
+// newSpecificInformersMap returns a new specificInformersMap (like
+// the generical InformersMap, except that it doesn't implement WaitForCacheSync).
+func newSpecificInformersMap(config *rest.Config,
 	scheme *runtime.Scheme,
 	mapper meta.RESTMapper,
-	resync time.Duration) *InformersMap {
-	ip := &InformersMap{
-		config:                    config,
-		Scheme:                    scheme,
-		mapper:                    mapper,
-		informersByGVK:            make(map[schema.GroupVersionKind]*MapEntry),
-		unstructuredInformerByGVK: make(map[schema.GroupVersionKind]*MapEntry),
-		codecs:     serializer.NewCodecFactory(scheme),
-		paramCodec: runtime.NewParameterCodec(scheme),
-		resync:     resync,
+	resync time.Duration, createListWatcher createListWatcherFunc) *specificInformersMap {
+	ip := &specificInformersMap{
+		config:            config,
+		Scheme:            scheme,
+		mapper:            mapper,
+		informersByGVK:    make(map[schema.GroupVersionKind]*MapEntry),
+		codecs:            serializer.NewCodecFactory(scheme),
+		paramCodec:        runtime.NewParameterCodec(scheme),
+		resync:            resync,
+		createListWatcher: createListWatcher,
 	}
 	return ip
 }
@@ -60,9 +65,9 @@ type MapEntry struct {
 	Reader CacheReader
 }
 
-// InformersMap create and caches Informers for (runtime.Object, schema.GroupVersionKind) pairs.
-//It uses a standard parameter codec constructed based on the given generated Scheme.
-type InformersMap struct {
+// specificInformersMap create and caches Informers for (runtime.Object, schema.GroupVersionKind) pairs.
+// It uses a standard parameter codec constructed based on the given generated Scheme.
+type specificInformersMap struct {
 	// Scheme maps runtime.Objects to GroupVersionKinds
 	Scheme *runtime.Scheme
 
@@ -75,10 +80,6 @@ type InformersMap struct {
 	// informersByGVK is the cache of informers keyed by groupVersionKind
 	informersByGVK map[schema.GroupVersionKind]*MapEntry
 
-	// unstructuredInformerByGVK is a cache of informers for unstructured types
-	// keyed by groupVersionKind
-	unstructuredInformerByGVK map[schema.GroupVersionKind]*MapEntry
-
 	// codecs is used to create a new REST client
 	codecs serializer.CodecFactory
 
@@ -93,20 +94,22 @@ type InformersMap struct {
 
 	// mu guards access to the map
 	mu sync.RWMutex
-	// mu guards access to the unstructured map
-	unstructuredMu sync.RWMutex
 
 	// start is true if the informers have been started
 	started bool
+
+	// createClient knows how to create a client and a list object,
+	// and allows for abstracting over the particulars of structured vs
+	// unstructured objects.
+	createListWatcher createListWatcherFunc
 }
 
 // Start calls Run on each of the informers and sets started to true.  Blocks on the stop channel.
-func (ip *InformersMap) Start(stop <-chan struct{}) error {
+// It doesn't return start because it can't return an error, and it's not a runnable directly.
+func (ip *specificInformersMap) Start(stop <-chan struct{}) {
 	func() {
 		ip.mu.Lock()
-		ip.unstructuredMu.Lock()
 		defer ip.mu.Unlock()
-		defer ip.unstructuredMu.Unlock()
 
 		// Set the stop channel so it can be passed to informers that are added later
 		ip.stop = stop
@@ -116,52 +119,31 @@ func (ip *InformersMap) Start(stop <-chan struct{}) error {
 			go informer.Informer.Run(stop)
 		}
 
-		// Start each unstructured informer
-		for _, informer := range ip.unstructuredInformerByGVK {
-			go informer.Informer.Run(stop)
-		}
-
 		// Set started to true so we immediately start any informers added later.
 		ip.started = true
 	}()
 	<-stop
-	return nil
 }
 
-// WaitForCacheSync waits until all the caches have been synced
-func (ip *InformersMap) WaitForCacheSync(stop <-chan struct{}) bool {
-	syncedFuncs := make([]cache.InformerSynced, 0, len(ip.informersByGVK)+len(ip.unstructuredInformerByGVK))
+// HasSyncedFuncs returns all the HasSynced functions for the informers in this map.
+func (ip *specificInformersMap) HasSyncedFuncs() []cache.InformerSynced {
+	syncedFuncs := make([]cache.InformerSynced, 0, len(ip.informersByGVK))
 	for _, informer := range ip.informersByGVK {
 		syncedFuncs = append(syncedFuncs, informer.Informer.HasSynced)
 	}
-	for _, informer := range ip.unstructuredInformerByGVK {
-		syncedFuncs = append(syncedFuncs, informer.Informer.HasSynced)
-	}
-	return cache.WaitForCacheSync(stop, syncedFuncs...)
-}
-
-func (ip *InformersMap) getMapEntry(gvk schema.GroupVersionKind, isUnstructured bool) (*MapEntry, bool) {
-	if isUnstructured {
-		ip.unstructuredMu.RLock()
-		defer ip.unstructuredMu.RUnlock()
-		i, ok := ip.unstructuredInformerByGVK[gvk]
-		return i, ok
-	}
-	ip.mu.RLock()
-	defer ip.mu.RUnlock()
-	i, ok := ip.informersByGVK[gvk]
-	return i, ok
-
+	return syncedFuncs
 }
 
-// Get will create a new Informer and add it to the map of InformersMap if none exists.  Returns
+// 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 *InformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object) (*MapEntry, error) {
-	_, isUnstructured := obj.(*unstructured.Unstructured)
-	_, isUnstructuredList := obj.(*unstructured.UnstructuredList)
-	isUnstructured = isUnstructured || isUnstructuredList
+func (ip *specificInformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object) (*MapEntry, error) {
 	// Return the informer if it is found
-	i, ok := ip.getMapEntry(gvk, isUnstructured)
+	i, ok := func() (*MapEntry, bool) {
+		ip.mu.RLock()
+		defer ip.mu.RUnlock()
+		i, ok := ip.informersByGVK[gvk]
+		return i, ok
+	}()
 	if ok {
 		return i, nil
 	}
@@ -170,27 +152,21 @@ func (ip *InformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object) (*M
 	// need to be locked
 	var sync bool
 	i, err := func() (*MapEntry, error) {
-		var ok bool
-		var i *MapEntry
-		// Check the caches to see if we already have an Informer.  If we do, return the Informer.
+		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 isUnstructured {
-			ip.unstructuredMu.Lock()
-			defer ip.unstructuredMu.Unlock()
-			i, ok = ip.unstructuredInformerByGVK[gvk]
-		} else {
-			ip.mu.Lock()
-			defer ip.mu.Unlock()
-			i, ok = ip.informersByGVK[gvk]
-		}
+		var ok bool
+		i, ok := ip.informersByGVK[gvk]
 		if ok {
 			return i, nil
 		}
 
 		// Create a NewSharedIndexInformer and add it to the map.
 		var lw *cache.ListWatch
-		lw, err := ip.newListWatch(gvk, isUnstructured)
+		lw, err := ip.createListWatcher(gvk, ip)
 		if err != nil {
 			return nil, err
 		}
@@ -201,7 +177,7 @@ func (ip *InformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object) (*M
 			Informer: ni,
 			Reader:   CacheReader{indexer: ni.GetIndexer(), groupVersionKind: gvk},
 		}
-		ip.setMap(i, gvk, isUnstructured)
+		ip.informersByGVK[gvk] = i
 
 		// Start the Informer if need by
 		// TODO(seans): write thorough tests and document what happens here - can you add indexers?
@@ -226,45 +202,23 @@ func (ip *InformersMap) Get(gvk schema.GroupVersionKind, obj runtime.Object) (*M
 	return i, err
 }
 
-// setMap - helper function to decide which map to add to.
-func (ip *InformersMap) setMap(i *MapEntry, gvk schema.GroupVersionKind, isUnstructured bool) {
-	if isUnstructured {
-		ip.unstructuredInformerByGVK[gvk] = i
-	} else {
-
-		ip.informersByGVK[gvk] = i
-	}
-}
-
 // newListWatch returns a new ListWatch object that can be used to create a SharedIndexInformer.
-func (ip *InformersMap) newListWatch(gvk schema.GroupVersionKind, isUnstructured bool) (*cache.ListWatch, error) {
+func createStructuredListWatch(gvk schema.GroupVersionKind, 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)
 	if err != nil {
 		return nil, err
 	}
 
-	// Construct a RESTClient for the groupVersionKind that we will use to
-	// talk to the apiserver.
-	var client rest.Interface
-	var listObj runtime.Object
-	if isUnstructured {
-		listObj = &unstructured.UnstructuredList{}
-		client, err = apiutil.RESTUnstructuredClientForGVK(gvk, ip.config)
-		if err != nil {
-			return nil, err
-		}
-	} else {
-		client, err = apiutil.RESTClientForGVK(gvk, ip.config, ip.codecs)
-		if err != nil {
-			return nil, err
-		}
-		listGVK := gvk.GroupVersion().WithKind(gvk.Kind + "List")
-		listObj, err = ip.Scheme.New(listGVK)
-		if err != nil {
-			return nil, err
-		}
+	client, err := apiutil.RESTClientForGVK(gvk, ip.config, ip.codecs)
+	if err != nil {
+		return nil, err
+	}
+	listGVK := gvk.GroupVersion().WithKind(gvk.Kind + "List")
+	listObj, err := ip.Scheme.New(listGVK)
+	if err != nil {
+		return nil, err
 	}
 
 	// Create a new ListWatch for the obj
@@ -282,3 +236,29 @@ func (ip *InformersMap) newListWatch(gvk schema.GroupVersionKind, isUnstructured
 		},
 	}, nil
 }
+
+func createUnstructuredListWatch(gvk schema.GroupVersionKind, 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)
+	if err != nil {
+		return nil, err
+	}
+	dynamicClient, err := dynamic.NewForConfig(ip.config)
+	if err != nil {
+		return nil, err
+	}
+
+	// Create a new ListWatch for the obj
+	return &cache.ListWatch{
+		ListFunc: func(opts metav1.ListOptions) (runtime.Object, error) {
+			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
+			return dynamicClient.Resource(mapping.Resource).Watch(opts)
+		},
+	}, nil
+}