Merge pull request #3465 from gman0/cachedresource-repl-fixes

CachedResource replication reconciler fixes

Author: kcp-ci-bot

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

@@ -17,12 +17,16 @@ limitations under the License.
 package replication
 
 import (
+	"bytes"
 	"context"
+	"crypto/sha256"
 	"encoding/json"
 	"fmt"
 	"strings"
 	"time"
 
+	"github.com/martinlindhe/base36"
+
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
@@ -49,6 +53,18 @@ const (
 	LabelKeyObjectOriginalNamespace = "cache.kcp.io/object-original-namespace"
 )
 
+func GenCachedObjectName(gvr schema.GroupVersionResource, namespace, name string) string {
+	buf := bytes.Buffer{}
+	buf.WriteString(gvr.String())
+	buf.WriteString(namespace)
+	buf.WriteString(name)
+
+	hash := sha256.Sum256([]byte(name))
+	base36hash := strings.ToLower(base36.EncodeBytes(hash[:]))
+
+	return base36hash
+}
+
 func (c *Controller) reconcile(ctx context.Context, gvrKey string) error {
 	if c.deleted {
 		return nil
@@ -59,7 +75,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 +84,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 +109,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 +143,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:              GenCachedObjectName(gvr, obj.GetNamespace(), obj.GetName()),
 					Labels:            obj.GetLabels(),
 					Annotations:       obj.GetAnnotations(),
 					CreationTimestamp: metav1.NewTime(time.Now()),
@@ -153,38 +171,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:            GenCachedObjectName(gvr, obj.GetNamespace(), obj.GetName()),
 					Labels:          obj.GetLabels(),
 					Annotations:     obj.GetAnnotations(),
 					ResourceVersion: obj.GetResourceVersion(),
@@ -197,23 +211,32 @@ 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
-			nameCache := gvr.Version + "." + gvr.Resource + "." + gvr.Group + "." + name // TODO: handle namespace
-			return c.kcpCacheClient.Cluster(cluster.Path()).CacheV1alpha1().CachedObjects().Delete(ctx, nameCache, metav1.DeleteOptions{})
+
+			gvr := gvrFromKey
+			if gvr.Group == "" {
+				gvr.Group = "core"
+			}
+
+			cachedObjName := GenCachedObjectName(gvr, ns, name)
+			if ns != "" {
+				cachedObjName += "." + ns
+			}
+			return c.kcpCacheClient.Cluster(cluster.Path()).CacheV1alpha1().CachedObjects().Delete(ctx, cachedObjName, metav1.DeleteOptions{})
 		},
 	}
 	defer c.callback()
-	return r.reconcile(ctx, gvr, key)
+	return r.reconcile(ctx, key)
 }
 
 type replicationReconciler struct {
@@ -222,19 +245,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 +282,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 +297,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
 		}
 
@@ -295,17 +318,7 @@ func (r *replicationReconciler) reconcile(ctx context.Context, gvr schema.GroupV
 
 		logger.V(2).Info("Creating object in global cache")
 		_, err := r.createObject(ctx, clusterName, localCopy)
-		if err != nil {
-			if apierrors.IsAlreadyExists(err) {
-				// If the object already exists, try to update it instead
-				logger.V(2).Info("Object already exists in global cache, updating instead")
-				_, err = r.updateObject(ctx, clusterName, localCopy)
-			}
-			if err != nil {
-				return err
-			}
-		}
-		return nil
+		return err
 	}
 
 	// update global copy and compare