modify cache to accept clusterName

Author: varshaprasad96

pkg/cache/cache.go

@@ -56,7 +56,7 @@ type Informers interface {
 
 	// GetInformerForKind is similar to GetInformer, except that it takes a group-version-kind, instead
 	// of the underlying object.
-	GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind, clusterName string) (Informer, error)
+	GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind) (Informer, error)
 
 	// Start runs all the informers known to this cache until the context is closed.
 	// It blocks.

pkg/cache/informer_cache.go

@@ -22,7 +22,6 @@ import (
 	"reflect"
 	"strings"
 
-	"k8s.io/apimachinery/pkg/api/meta"
 	apimeta "k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -59,7 +58,7 @@ func (ip *informerCache) Get(ctx context.Context, key client.ObjectKey, out clie
 		return err
 	}
 
-	started, cache, err := ip.InformersMap.Get(ctx, gvk, out, out.GetClusterName())
+	started, cache, err := ip.InformersMap.Get(ctx, gvk, out)
 	if err != nil {
 		return err
 	}
@@ -76,11 +75,11 @@ func (ip *informerCache) List(ctx context.Context, out client.ObjectList, opts .
 	if err != nil {
 		return err
 	}
-	metadata, err := meta.Accessor(cacheTypeObj)
-	if err != nil {
-		return err
-	}
-	started, cache, err := ip.InformersMap.Get(ctx, *gvk, cacheTypeObj, metadata.GetClusterName())
+	// metadata, err := meta.Accessor(cacheTypeObj)
+	// if err != nil {
+	// 	return err
+	// }
+	started, cache, err := ip.InformersMap.Get(ctx, *gvk, cacheTypeObj)
 	if err != nil {
 		return err
 	}
@@ -135,14 +134,14 @@ func (ip *informerCache) objectTypeForListObject(list client.ObjectList) (*schem
 }
 
 // GetInformerForKind returns the informer for the GroupVersionKind.
-func (ip *informerCache) GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind, clusterName string) (Informer, error) {
+func (ip *informerCache) GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind) (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(ctx, gvk, obj, clusterName)
+	_, i, err := ip.InformersMap.Get(ctx, gvk, obj)
 	if err != nil {
 		return nil, err
 	}
@@ -157,7 +156,7 @@ func (ip *informerCache) GetInformer(ctx context.Context, obj client.Object) (In
 	}
 
 	fmt.Println("getting informer from indexer")
-	_, i, err := ip.InformersMap.Get(ctx, gvk, obj, obj.GetClusterName())
+	_, i, err := ip.InformersMap.Get(ctx, gvk, obj)
 	if err != nil {
 		return nil, err
 	}

pkg/cache/informertest/fake_cache.go

@@ -39,7 +39,7 @@ type FakeInformers struct {
 }
 
 // GetInformerForKind implements Informers.
-func (c *FakeInformers) GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind, clusterName string) (cache.Informer, error) {
+func (c *FakeInformers) GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind) (cache.Informer, error) {
 	if c.Scheme == nil {
 		c.Scheme = scheme.Scheme
 	}

pkg/cache/internal/clusterscoped_cache.go

@@ -0,0 +1,299 @@
+/*
+Copyright 2019 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 cache
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/prometheus/common/log"
+	apimeta "k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/rest"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// NewCacheFunc - Function for creating a new cache from the options and a rest config.
+type NewCacheFunc func(config *rest.Config, opts Options) (Cache, error)
+
+// a new global namespaced cache to handle cluster scoped resources.
+const globalClusterCache = "_cluster"
+
+// [_cluster -> cache ("*"), clusterNames -> caches ]
+
+// MultiNamespacedCacheBuilder - Builder function to create a new multi-namespaced cache.
+// This will scope the cache to a list of namespaces. Listing for all namespaces
+// will list for all the namespaces that this knows about. By default this will create
+// a global cache for cluster scoped resource. Note that this is not intended
+// to be used for excluding namespaces, this is better done via a Predicate. Also note that
+// you may face performance issues when using this with a high number of namespaces.
+func MultiClusterCacheBuilder(clusterNames []string) NewCacheFunc {
+	return func(config *rest.Config, opts Options) (Cache, error) {
+		opts, err := defaultOpts(config, opts)
+		if err != nil {
+			return nil, err
+		}
+
+		caches := map[string]Cache{}
+
+		// create aglobal cache for * scope
+		gCache, err := New(config, opts)
+		if err != nil {
+			return nil, fmt.Errorf("error creating global cache %v", err)
+		}
+
+		for _, cs := range clusterNames {
+			opts.ClusterName = cs
+			c, err := New(config, opts)
+			if err != nil {
+				return nil, err
+			}
+			caches[cs] = c
+		}
+		return &multiNamespaceCache{namespaceToCache: caches, Scheme: opts.Scheme, RESTMapper: opts.Mapper, clusterCache: gCache}, nil
+	}
+}
+
+// multiNamespaceCache knows how to handle multiple namespaced caches
+// Use this feature when scoping permissions for your
+// operator to a list of namespaces instead of watching every namespace
+// in the cluster.
+type multiClusterCache struct {
+	clusterToCache map[string]Cache
+	Scheme         *runtime.Scheme
+	RESTMapper     apimeta.RESTMapper
+	gClusterCache  Cache // Point to "*"
+}
+
+var _ Cache = &multiClusterCache{}
+
+// Methods for multiNamespaceCache to conform to the Informers interface.
+func (c *multiClusterCache) GetInformer(ctx context.Context, obj client.Object) (Informer, error) {
+	informers := map[string]Informer{}
+
+	//get CLusterName
+	clusterName, err := getClusterName(obj)
+	if err != nil {
+		return nil, fmt.Errorf("error getting clustername %q", err)
+	}
+
+	if len(clusterName) == "*" {
+		globalInformer, err := c.gClusterCache.GetInformer(ctx, obj)
+		if err != nil {
+			return err
+		}
+		informers[globalClusterCache] = globalInformer
+	}
+
+	for cs, cache := range c.clusterToCache {
+		informer, err := cache.GetInformer(ctx, obj)
+		if err != nil {
+			return nil, err
+		}
+		informers[cs] = informer
+	}
+
+	return &multiClusterCache{clusterToCache: informers}, nil
+
+}
+
+func getClusterName(obj client.Object) (string, error) {
+	if obj == nil {
+		return nil, fmt.Errorf("object cannot be empty %v", obj)
+	}
+	if obj.GetClusterName() != "" {
+		return "*", nil
+	}
+
+	return obj.GetClusterName(), nil
+}
+
+func (c *multiNamespaceCache) GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind, clusterName string) (Informer, error) {
+	informers := map[string]Informer{} // clusterName -> informer
+
+	clusterName, err := getCLusterName(obj)
+	if err != nil {
+		return err
+	}
+	// globalCLusterCache
+	if len(clusterName) == "*" {
+		clusterCacheInf, err := c.gClusterCache.GetInformerForKind(ctx, gvk)
+		if err != nil {
+			return err
+		}
+
+		informers[globalClusterCache] = clusterCacheInf
+		return multiClusterCache{clusterToCache: informers}, nil
+	}
+
+	for cs, cache := range c.clusterToCache {
+		informer, err := cache.GetInformerForKind(ctx, gvk)
+		if err != nil {
+			return err
+		}
+		informer[cs] = informer
+	}
+
+	return multiClusterCache{clusterToCache: informer}, nil
+}
+
+func (c *multiClusterCache) Start(ctx context.Context) error {
+	// start global cache
+	go func() {
+		err := c.glusterCache.Start(ctx)
+		if err != nil {
+			log.Error(err, "cluster scoped cache failed to start")
+		}
+	}()
+
+	// start namespaced caches
+	for cs, cache := range c.clusterToCache {
+		go func(cs string, cache Cache) {
+			err := cache.Start(ctx)
+			if err != nil {
+				log.Error(err, "multiClusterCache cache failed to start cluster informer", "cluster", cs)
+			}
+		}(cs, cache)
+	}
+
+	<-ctx.Done()
+	return nil
+}
+
+func (c *multiClusterCache) WaitForCacheSync(ctx context.Context) bool {
+	synced := true
+	for _, cache := range c.clusterToCache {
+		if s := cache.WaitForCacheSync(ctx); !s {
+			synced = s
+		}
+	}
+
+	// check if global cluster cache has synced
+	if !c.gClusterCache.WaitForCacheSync(ctx) {
+		synced = false
+	}
+	return synced
+}
+
+func (c *multiNamespaceCache) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error {
+
+	clusterName, err := getCLusterName(obj)
+	if err != nil {
+		return err
+	}
+
+	if clusterName == "*" {
+		return c.gClusterCache.IndexField(ctx, obj, field, extractValue)
+	}
+
+	for _, cache := range c.clusterToCache {
+		if err := cache.IndexField(ctx, obj, field, extractValue); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (c *multiClusterCache) Get(ctx context.Context, key client.ObjectKey, obj client.Object) error {
+	clusterName, err := getClusterName(obj)
+	if err != nil {
+		return err
+	}
+
+	if clusterName == "*" {
+		// Look into the global cache to fetch the object
+		return c.gClusterCache.Get(ctx, key, obj)
+	}
+
+	cache, ok := c.clusterToCache[clusterName]
+	if !ok {
+		return fmt.Errorf("unable to get: %v because of unknown clusterName for the cache", key)
+	}
+	return cache.Get(ctx, key, obj)
+}
+
+// List
+// ClusterName is not passed => error
+// ClusterName is passed => getCache
+// ListAll clusters => clusterName is "*"
+
+func (c *multiClusterCache) List(ctx context.Context, list client.ObjectList, opts ...client.ListOptions) {
+	listOpts := client.ListOptions{}
+	listOpts.ApplyOptions(opts)
+
+	clusterName := opts.ClusterName
+	if clusterName == "" {
+		// initial stab - error out
+		fmt.Errorf("cluster Name is empty in listOpts %v", listOpts)
+	}
+
+	if clusterName == "*" {
+		// Look at gloabal cluster cache
+		return c.gClusterCache.List(ctx, list, opts...)
+	}
+
+	// look at individual caches
+	cache, ok := c.clusterToCache[clusterName]
+	if !ok { // cache is not found to the particular cluster
+		return fmt.Errorf("unable to get cache because clusterName %v is not known", clusterName)
+	}
+	return cache.List(ctx, list, opts...)
+}
+
+// informer maps
+type multiClusterInformer struct {
+	clusterNameToInformer map[string]Informer
+}
+
+var _Informer = &multiClusterInformer
+
+// AddEventHandler adds the handler to each namespaced informer.
+func (i *multiClusterInformer) AddEventHandler(handler toolscache.ResourceEventHandler) {
+	for _, informer := range i.clusterNameToInformer {
+		informer.AddEventHandler(handler)
+	}
+}
+
+// AddEventHandlerWithResyncPeriod adds the handler with a resync period to each namespaced informer.
+func (i *multiClusterInformer) AddEventHandlerWithResyncPeriod(handler toolscache.ResourceEventHandler, resyncPeriod time.Duration) {
+	for _, informer := range i.clusterNameToInformer {
+		informer.AddEventHandlerWithResyncPeriod(handler, resyncPeriod)
+	}
+}
+
+// AddIndexers adds the indexer for each namespaced informer.
+func (i *multiClusterInformer) AddIndexers(indexers toolscache.Indexers) error {
+	for _, informer := range i.clusterNameToInformer {
+		err := informer.AddIndexers(indexers)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// HasSynced checks if each namespaced informer has synced.
+func (i *multiClusterInformer) HasSynced() bool {
+	for _, informer := range i.clusterNameToInformer {
+		if ok := informer.HasSynced(); !ok {
+			return ok
+		}
+	}
+	return true
+}