CachedResource replication reconciler relies on supplied GVR

The reconciler should rely on the GVR passed from the controller
instead of having to reach out to REST mapper.

On-behalf-of: @SAP [email protected]
Signed-off-by: Robert Vasek <[email protected]>

Author: gman0

pkg/reconciler/cache/cachedresources/cachedresources_reconcile_replication.go

@@ -106,7 +106,6 @@ func (r *replication) reconcile(ctx context.Context, cachedResource *cachev1alph
 			r.shardName,
 			r.dynamicCacheClient,
 			r.kcpCacheClient,
-			r.dynRESTMapper,
 			gvr,
 			replicated,
 			callback,

pkg/reconciler/cache/cachedresources/replication/replication_controller.go

@@ -37,7 +37,6 @@ import (
 	"github.com/kcp-dev/kcp/pkg/cache/client/shard"
 	"github.com/kcp-dev/kcp/pkg/indexers"
 	"github.com/kcp-dev/kcp/pkg/logging"
-	"github.com/kcp-dev/kcp/pkg/reconciler/dynamicrestmapper"
 	cachev1alpha1 "github.com/kcp-dev/kcp/sdk/apis/cache/v1alpha1"
 	kcpclientset "github.com/kcp-dev/kcp/sdk/client/clientset/versioned/cluster"
 )
@@ -56,7 +55,6 @@ func NewController(
 	shardName string,
 	dynamicCacheClient kcpdynamic.ClusterInterface,
 	kcpCacheClient kcpclientset.ClusterInterface,
-	dynRESTMapper *dynamicrestmapper.DynamicRESTMapper,
 	gvr schema.GroupVersionResource,
 	replicated *ReplicatedGVR,
 	callback func(),
@@ -73,17 +71,15 @@ func NewController(
 		dynamicCacheClient: dynamicCacheClient,
 		kcpCacheClient:     kcpCacheClient,
 		replicated:         replicated,
-		gvr:                gvr,
 		callback:           callback,
-		dynRESTMapper:      dynRESTMapper,
 		cleanupFuncs:       make([]func(), 0),
 		localLabelSelector: localLabelSelector,
 	}
 
 	localHandler, err := c.replicated.Local.AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    func(obj interface{}) { c.enqueueObject(obj, c.gvr) },
-		UpdateFunc: func(_, obj interface{}) { c.enqueueObject(obj, c.gvr) },
-		DeleteFunc: func(obj interface{}) { c.enqueueObject(obj, c.gvr) },
+		AddFunc:    func(obj interface{}) { c.enqueueObject(obj, gvr) },
+		UpdateFunc: func(_, obj interface{}) { c.enqueueObject(obj, gvr) },
+		DeleteFunc: func(obj interface{}) { c.enqueueObject(obj, gvr) },
 	})
 	if err != nil {
 		return nil, err
@@ -211,9 +207,6 @@ type Controller struct {
 	kcpCacheClient     kcpclientset.ClusterInterface
 
 	replicated *ReplicatedGVR
-	gvr        schema.GroupVersionResource
-
-	dynRESTMapper *dynamicrestmapper.DynamicRESTMapper
 
 	// callback is called when we want to trigger parent object reconciliation.
 	// Cache state is being managed by child controller, so we need to trigger parent object reconciliation

pkg/reconciler/cache/cachedresources/replication/replication_reconcile.go

@@ -59,7 +59,7 @@ func (c *Controller) reconcile(ctx context.Context, gvrKey string) error {
 		return fmt.Errorf("incorrect key: %v, expected group.version.resource::key", gvrKey)
 	}
 	gvrParts := strings.SplitN(keyParts[0], ".", 3)
-	gvr := schema.GroupVersionResource{Version: gvrParts[0], Resource: gvrParts[1], Group: gvrParts[2]}
+	gvrFromKey := schema.GroupVersionResource{Version: gvrParts[0], Resource: gvrParts[1], Group: gvrParts[2]}
 
 	// Key will present in the form of namespace/name in the current logical cluster.
 	key := keyParts[1]
@@ -68,6 +68,7 @@ func (c *Controller) reconcile(ctx context.Context, gvrKey string) error {
 		shardName:          c.shardName,
 		localLabelSelector: c.localLabelSelector,
 		getLocalCopy: func(cluster logicalcluster.Name, namespace, name string) (*unstructured.Unstructured, error) {
+			gvr := gvrFromKey
 			key := kcpcache.ToClusterAwareKey(cluster.String(), namespace, name)
 			obj, exists, err := c.replicated.Local.GetIndexer().GetByKey(key)
 			if !exists {
@@ -92,7 +93,12 @@ func (c *Controller) reconcile(ctx context.Context, gvrKey string) error {
 			u.SetAPIVersion(gvr.GroupVersion().String())
 			return u, nil
 		},
-		getGlobalCopy: func(cluster logicalcluster.Name, namespace, name string, gvr schema.GroupVersionResource) (*unstructured.Unstructured, error) {
+		getGlobalCopy: func(cluster logicalcluster.Name, namespace, name string) (*unstructured.Unstructured, error) {
+			gvr := gvrFromKey
+			if gvr.Group == "" {
+				gvr.Group = "core"
+			}
+
 			key := GVRAndShardAndLogicalClusterAndNamespaceKey(gvr, c.shardName, cluster, namespace, name)
 			objs, err := c.replicated.Global.GetIndexer().ByIndex(ByGVRAndShardAndLogicalClusterAndNamespaceAndName, key)
 			if err != nil {
@@ -121,26 +127,22 @@ func (c *Controller) reconcile(ctx context.Context, gvrKey string) error {
 			return u, nil
 		},
 		createObject: func(ctx context.Context, cluster logicalcluster.Name, obj *unstructured.Unstructured) (*cachev1alpha1.CachedObject, error) {
-			gvk := obj.GroupVersionKind()
-			if gvk.Group == "" {
-				gvk.Group = "core"
+			gvr := gvrFromKey
+			if gvr.Group == "" {
+				gvr.Group = "core"
 			}
 
 			objBytes, err := json.Marshal(obj)
 			if err != nil {
 				return nil, err
 			}
-			mapper, err := c.dynRESTMapper.ForCluster(cluster).RESTMapping(gvk.GroupKind(), gvk.Version)
-			if err != nil {
-				return nil, err
-			}
 			cacheObj := &cachev1alpha1.CachedObject{
 				TypeMeta: metav1.TypeMeta{
 					Kind:       cache.CachedObjectKind,
 					APIVersion: cachev1alpha1.SchemeGroupVersion.String(),
 				},
 				ObjectMeta: metav1.ObjectMeta{
-					Name:              gvr.Version + "." + mapper.Resource.Resource + "." + gvr.Group + "." + obj.GetName(), // TODO: handle namespace
+					Name:              gvr.Version + "." + gvr.Resource + "." + gvr.Group + "." + obj.GetName(), // TODO: handle namespace
 					Labels:            obj.GetLabels(),
 					Annotations:       obj.GetAnnotations(),
 					CreationTimestamp: metav1.NewTime(time.Now()),
@@ -153,38 +155,34 @@ func (c *Controller) reconcile(ctx context.Context, gvrKey string) error {
 				cacheObj.Labels = map[string]string{}
 			}
 			// Append schema label to the object.
-			cacheObj.Labels[LabelKeyObjectSchema] = gvr.Version + "." + mapper.Resource.Resource + "." + gvr.Group
+			cacheObj.Labels[LabelKeyObjectSchema] = gvr.Version + "." + gvr.Resource + "." + gvr.Group
 			cacheObj.Labels[LabelKeyObjectGroup] = gvr.Group
 			cacheObj.Labels[LabelKeyObjectVersion] = gvr.Version
-			cacheObj.Labels[LabelKeyObjectResource] = mapper.Resource.Resource
+			cacheObj.Labels[LabelKeyObjectResource] = gvr.Resource
 			cacheObj.Labels[LabelKeyObjectOriginalName] = obj.GetName()
 			cacheObj.Labels[LabelKeyObjectOriginalNamespace] = obj.GetNamespace()
 
 			u, err := c.kcpCacheClient.Cluster(cluster.Path()).CacheV1alpha1().CachedObjects().Create(ctx, cacheObj, metav1.CreateOptions{})
 			return u, err
 		},
 		updateObject: func(ctx context.Context, cluster logicalcluster.Name, obj *unstructured.Unstructured) (*cachev1alpha1.CachedObject, error) {
-			gvk := obj.GroupVersionKind()
-			if gvk.Group == "" {
-				gvk.Group = "core"
+			gvr := gvrFromKey
+			if gvr.Group == "" {
+				gvr.Group = "core"
 			}
 
 			objBytes, err := json.Marshal(obj)
 			if err != nil {
 				return nil, err
 			}
 
-			mapper, err := c.dynRESTMapper.ForCluster(cluster).RESTMapping(gvk.GroupKind(), gvk.Version)
-			if err != nil {
-				return nil, err
-			}
 			cacheObj := &cachev1alpha1.CachedObject{
 				TypeMeta: metav1.TypeMeta{
 					Kind:       cache.CachedObjectKind,
 					APIVersion: cachev1alpha1.SchemeGroupVersion.String(),
 				},
 				ObjectMeta: metav1.ObjectMeta{
-					Name:            gvr.Version + "." + mapper.Resource.Resource + "." + gvr.Group + "." + obj.GetName(),
+					Name:            gvr.Version + "." + gvr.Resource + "." + gvr.Group + "." + obj.GetName(),
 					Labels:          obj.GetLabels(),
 					Annotations:     obj.GetAnnotations(),
 					ResourceVersion: obj.GetResourceVersion(),
@@ -197,23 +195,29 @@ func (c *Controller) reconcile(ctx context.Context, gvrKey string) error {
 				cacheObj.Labels = map[string]string{}
 			}
 			// Append schema label to the object.
-			cacheObj.Labels[LabelKeyObjectSchema] = gvr.Version + "." + mapper.Resource.Resource + "." + gvr.Group
+			cacheObj.Labels[LabelKeyObjectSchema] = gvr.Version + "." + gvr.Resource + "." + gvr.Group
 			cacheObj.Labels[LabelKeyObjectGroup] = gvr.Group
 			cacheObj.Labels[LabelKeyObjectVersion] = gvr.Version
-			cacheObj.Labels[LabelKeyObjectResource] = mapper.Resource.Resource
+			cacheObj.Labels[LabelKeyObjectResource] = gvr.Resource
 			cacheObj.Labels[LabelKeyObjectOriginalName] = obj.GetName()
 			cacheObj.Labels[LabelKeyObjectOriginalNamespace] = obj.GetNamespace()
 
 			return c.kcpCacheClient.Cluster(cluster.Path()).CacheV1alpha1().CachedObjects().Update(ctx, cacheObj, metav1.UpdateOptions{})
 		},
-		deleteObject: func(ctx context.Context, cluster logicalcluster.Name, ns, name string, gvr schema.GroupVersionResource) error {
+		deleteObject: func(ctx context.Context, cluster logicalcluster.Name, ns, name string) error {
 			// deleting from cache - means we delete the wrapper object
+
+			gvr := gvrFromKey
+			if gvr.Group == "" {
+				gvr.Group = "core"
+			}
+
 			nameCache := gvr.Version + "." + gvr.Resource + "." + gvr.Group + "." + name // TODO: handle namespace
 			return c.kcpCacheClient.Cluster(cluster.Path()).CacheV1alpha1().CachedObjects().Delete(ctx, nameCache, metav1.DeleteOptions{})
 		},
 	}
 	defer c.callback()
-	return r.reconcile(ctx, gvr, key)
+	return r.reconcile(ctx, key)
 }
 
 type replicationReconciler struct {
@@ -222,19 +226,19 @@ type replicationReconciler struct {
 	localLabelSelector labels.Selector
 
 	getLocalCopy  func(cluster logicalcluster.Name, namespace, name string) (*unstructured.Unstructured, error)
-	getGlobalCopy func(cluster logicalcluster.Name, namespace, name string, gvr schema.GroupVersionResource) (*unstructured.Unstructured, error)
+	getGlobalCopy func(cluster logicalcluster.Name, namespace, name string) (*unstructured.Unstructured, error)
 
 	createObject func(ctx context.Context, cluster logicalcluster.Name, obj *unstructured.Unstructured) (*cachev1alpha1.CachedObject, error)
 	updateObject func(ctx context.Context, cluster logicalcluster.Name, obj *unstructured.Unstructured) (*cachev1alpha1.CachedObject, error)
-	deleteObject func(ctx context.Context, cluster logicalcluster.Name, ns, name string, gvr schema.GroupVersionResource) error
+	deleteObject func(ctx context.Context, cluster logicalcluster.Name, ns, name string) error
 }
 
 // reconcile makes sure that the object under the given key from the local shard is replicated to the cache server.
 // the replication function handles the following cases:
 //  1. creation of the object in the cache server when the cached object is not found by getGlobalCopy
 //  2. deletion of the object from the cache server when the original/local object was removed OR was not found by getLocalCopy
 //  3. modification of the cached object to match the original one when meta.annotations, meta.labels, spec or status are different
-func (r *replicationReconciler) reconcile(ctx context.Context, gvr schema.GroupVersionResource, key string) error {
+func (r *replicationReconciler) reconcile(ctx context.Context, key string) error {
 	if r.deleted {
 		return nil
 	}
@@ -259,7 +263,7 @@ func (r *replicationReconciler) reconcile(ctx context.Context, gvr schema.GroupV
 		return nil
 	}
 
-	globalCopy, err := r.getGlobalCopy(clusterName, ns, name, gvr)
+	globalCopy, err := r.getGlobalCopy(clusterName, ns, name)
 	if err != nil && !apierrors.IsNotFound(err) {
 		utilruntime.HandleError(err)
 		return nil
@@ -274,7 +278,7 @@ func (r *replicationReconciler) reconcile(ctx context.Context, gvr schema.GroupV
 
 		// Object doesn't exist anymore, delete it from the global cache.
 		logger.V(2).WithValues("cluster", clusterName, "namespace", ns, "name", name).Info("Deleting object from global cache")
-		if err := r.deleteObject(ctx, clusterName, ns, name, gvr); err != nil && !apierrors.IsNotFound(err) {
+		if err := r.deleteObject(ctx, clusterName, ns, name); err != nil && !apierrors.IsNotFound(err) {
 			return err
 		}