fix errors - clusterCaching works

Author: varshaprasad96

examples/kcp/main.go

@@ -25,6 +25,7 @@ import (
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	ctrl "sigs.k8s.io/controller-runtime"
 	api "sigs.k8s.io/controller-runtime/examples/crd/pkg"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
@@ -57,7 +58,10 @@ func (r *reconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu
 func main() {
 	ctrl.SetLogger(zap.New())
 
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{})
+	cfg := ctrl.GetConfigOrDie()
+	mgr, err := ctrl.NewManager(cfg, ctrl.Options{
+		NewCache: cache.MultiClusterCacheBuilder([]string{"*"}),
+	})
 	if err != nil {
 		setupLog.Error(err, "unable to start manager")
 		os.Exit(1)

pkg/builder/controller.go

@@ -229,7 +229,11 @@ func (blder *Builder) doWatch() error {
 	if err != nil {
 		return err
 	}
-	src := &source.Kind{Type: typeForSrc, ClusterName: blder.cluster}
+
+	// watch the cluster which is being passed to the controller - initial watch
+	typeForSrc.SetClusterName(blder.cluster)
+
+	src := &source.Kind{Type: typeForSrc}
 	hdler := &handler.EnqueueRequestForObject{}
 	allPredicates := append(blder.globalPredicates, blder.forInput.predicates...)
 	if err := blder.ctrl.Watch(src, hdler, allPredicates...); err != nil {
@@ -242,7 +246,7 @@ func (blder *Builder) doWatch() error {
 		if err != nil {
 			return err
 		}
-		src := &source.Kind{Type: typeForSrc, ClusterName: blder.cluster}
+		src := &source.Kind{Type: typeForSrc}
 		hdler := &handler.EnqueueRequestForOwner{
 			OwnerType:    blder.forInput.object,
 			IsController: true,

pkg/cache/internal/clusterscoped_cache.go renamed to pkg/cache/clusterscoped_cache.go

@@ -21,17 +21,14 @@ import (
 	"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"
+	toolscache "k8s.io/client-go/tools/cache"
 	"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"
 
@@ -66,7 +63,7 @@ func MultiClusterCacheBuilder(clusterNames []string) NewCacheFunc {
 			}
 			caches[cs] = c
 		}
-		return &multiNamespaceCache{namespaceToCache: caches, Scheme: opts.Scheme, RESTMapper: opts.Mapper, clusterCache: gCache}, nil
+		return &multiClusterCache{clusterToCache: caches, Scheme: opts.Scheme, RESTMapper: opts.Mapper, gClusterCache: gCache}, nil
 	}
 }
 
@@ -93,10 +90,10 @@ func (c *multiClusterCache) GetInformer(ctx context.Context, obj client.Object)
 		return nil, fmt.Errorf("error getting clustername %q", err)
 	}
 
-	if len(clusterName) == "*" {
+	if (clusterName) == "*" {
 		globalInformer, err := c.gClusterCache.GetInformer(ctx, obj)
 		if err != nil {
-			return err
+			return nil, err
 		}
 		informers[globalClusterCache] = globalInformer
 	}
@@ -109,13 +106,13 @@ func (c *multiClusterCache) GetInformer(ctx context.Context, obj client.Object)
 		informers[cs] = informer
 	}
 
-	return &multiClusterCache{clusterToCache: informers}, nil
+	return &multiClusterInformer{clusterNameToInformer: informers}, nil
 
 }
 
 func getClusterName(obj client.Object) (string, error) {
 	if obj == nil {
-		return nil, fmt.Errorf("object cannot be empty %v", obj)
+		return "", fmt.Errorf("object cannot be empty %v", obj)
 	}
 	if obj.GetClusterName() != "" {
 		return "*", nil
@@ -124,39 +121,14 @@ func getClusterName(obj client.Object) (string, error) {
 	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) GetInformerForKind(ctx context.Context, gvk schema.GroupVersionKind) (Informer, error) {
+	return nil, fmt.Errorf("not supported in multiClustercache")
 }
 
 func (c *multiClusterCache) Start(ctx context.Context) error {
 	// start global cache
 	go func() {
-		err := c.glusterCache.Start(ctx)
+		err := c.gClusterCache.Start(ctx)
 		if err != nil {
 			log.Error(err, "cluster scoped cache failed to start")
 		}
@@ -191,9 +163,9 @@ func (c *multiClusterCache) WaitForCacheSync(ctx context.Context) bool {
 	return synced
 }
 
-func (c *multiNamespaceCache) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error {
+func (c *multiClusterCache) IndexField(ctx context.Context, obj client.Object, field string, extractValue client.IndexerFunc) error {
 
-	clusterName, err := getCLusterName(obj)
+	clusterName, err := getClusterName(obj)
 	if err != nil {
 		return err
 	}
@@ -233,16 +205,17 @@ func (c *multiClusterCache) Get(ctx context.Context, key client.ObjectKey, obj c
 // ClusterName is passed => getCache
 // ListAll clusters => clusterName is "*"
 
-func (c *multiClusterCache) List(ctx context.Context, list client.ObjectList, opts ...client.ListOptions) {
+func (c *multiClusterCache) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
 	listOpts := client.ListOptions{}
-	listOpts.ApplyOptions(opts)
 
-	clusterName := opts.ClusterName
+	clusterName := listOpts.ClusterName
 	if clusterName == "" {
 		// initial stab - error out
 		fmt.Errorf("cluster Name is empty in listOpts %v", listOpts)
 	}
 
+	listOpts.ApplyOptions(opts)
+
 	if clusterName == "*" {
 		// Look at gloabal cluster cache
 		return c.gClusterCache.List(ctx, list, opts...)
@@ -261,7 +234,7 @@ type multiClusterInformer struct {
 	clusterNameToInformer map[string]Informer
 }
 
-var _Informer = &multiClusterInformer
+var _Informer = &multiClusterInformer{}
 
 // AddEventHandler adds the handler to each namespaced informer.
 func (i *multiClusterInformer) AddEventHandler(handler toolscache.ResourceEventHandler) {

pkg/cache/internal/deleg_map.go

@@ -56,9 +56,9 @@ func NewInformersMap(config *rest.Config,
 	disableDeepCopy DisableDeepCopyByGVK,
 ) *InformersMap {
 	return &InformersMap{
-		structured:   newStructuredInformersMap(config, scheme, mapper, resync, namespace, clusterName, selectors, disableDeepCopy),
-		unstructured: newUnstructuredInformersMap(config, scheme, mapper, resync, namespace, clusterName, selectors, disableDeepCopy),
-		metadata:     newMetadataInformersMap(config, scheme, mapper, resync, namespace, clusterName, selectors, disableDeepCopy),
+		structured:   newStructuredInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy),
+		unstructured: newUnstructuredInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy),
+		metadata:     newMetadataInformersMap(config, scheme, mapper, resync, namespace, selectors, disableDeepCopy),
 
 		Scheme: scheme,
 	}

pkg/source/source.go

@@ -94,8 +94,6 @@ type Kind struct {
 	// cache used to watch APIs
 	cache cache.Cache
 
-	ClusterName string
-
 	// started may contain an error if one was encountered during startup. If its closed and does not
 	// contain an error, startup and syncing finished.
 	started     chan error