CachedResources: replicate full objects with

DDSIF, the informer for retrieving local objects, yields only PartialMetadataObjects.
This commit adds a live GET call to retrieve the full object for replication.

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

Author: gman0

pkg/reconciler/cache/cachedresources/cachedresources_controller.go

@@ -68,7 +68,6 @@ func NewController(
 	kcpClusterClient kcpclientset.ClusterInterface,
 	kcpCacheClient kcpclientset.ClusterInterface,
 	dynamicClient kcpdynamic.ClusterInterface,
-	cacheDynamicClient kcpdynamic.ClusterInterface,
 
 	kubeClusterClient kcpkubernetesclientset.ClusterInterface,
 	namespaceInformer kcpcorev1informers.NamespaceClusterInformer,
@@ -93,8 +92,7 @@ func NewController(
 		kcpClient:      kcpClusterClient,
 		kcpCacheClient: kcpCacheClient,
 
-		dynamicClient:      dynamicClient,
-		cacheDynamicClient: cacheDynamicClient,
+		dynamicClient: dynamicClient,
 
 		dynRESTMapper: dynRESTMapper,
 
@@ -165,8 +163,7 @@ type Controller struct {
 	kcpClient      kcpclientset.ClusterInterface
 	kcpCacheClient kcpclientset.ClusterInterface
 
-	dynamicClient      kcpdynamic.ClusterInterface
-	cacheDynamicClient kcpdynamic.ClusterInterface
+	dynamicClient kcpdynamic.ClusterInterface
 
 	dynRESTMapper *dynamicrestmapper.DynamicRESTMapper
 

pkg/reconciler/cache/cachedresources/cachedresources_reconcile.go

@@ -83,7 +83,7 @@ func (c *Controller) reconcile(ctx context.Context, cluster logicalcluster.Name,
 		},
 		&replication{
 			shardName:                      c.shardName,
-			dynamicCacheClient:             c.dynamicClient,
+			dynamicClusterClient:           c.dynamicClient,
 			kcpCacheClient:                 c.kcpCacheClient,
 			dynRESTMapper:                  c.dynRESTMapper,
 			cacheKcpInformers:              c.cacheKcpInformers,

pkg/reconciler/cache/cachedresources/cachedresources_reconcile_replication.go

@@ -41,7 +41,7 @@ import (
 // Or deletes the replication controller if the published resource is being deleted.
 type replication struct {
 	shardName                      string
-	dynamicCacheClient             kcpdynamic.ClusterInterface
+	dynamicClusterClient           kcpdynamic.ClusterInterface
 	kcpCacheClient                 kcpclientset.ClusterInterface
 	dynRESTMapper                  *dynamicrestmapper.DynamicRESTMapper
 	cacheKcpInformers              kcpinformers.SharedInformerFactory
@@ -103,7 +103,7 @@ func (r *replication) reconcile(ctx context.Context, cachedResource *cachev1alph
 
 		c, err := replicationcontroller.NewController(
 			r.shardName,
-			r.dynamicCacheClient,
+			r.dynamicClusterClient,
 			r.kcpCacheClient,
 			gvr,
 			replicated,

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

@@ -53,7 +53,7 @@ const (
 // NewController returns a new replication controller.
 func NewController(
 	shardName string,
-	dynamicCacheClient kcpdynamic.ClusterInterface,
+	dynamicClusterClient kcpdynamic.ClusterInterface,
 	kcpCacheClient kcpclientset.ClusterInterface,
 	gvr schema.GroupVersionResource,
 	replicated *ReplicatedGVR,
@@ -68,12 +68,12 @@ func NewController(
 				Name: ControllerName,
 			},
 		),
-		dynamicCacheClient: dynamicCacheClient,
-		kcpCacheClient:     kcpCacheClient,
-		replicated:         replicated,
-		callback:           callback,
-		cleanupFuncs:       make([]func(), 0),
-		localLabelSelector: localLabelSelector,
+		dynamicClusterClient: dynamicClusterClient,
+		kcpCacheClient:       kcpCacheClient,
+		replicated:           replicated,
+		callback:             callback,
+		cleanupFuncs:         make([]func(), 0),
+		localLabelSelector:   localLabelSelector,
 	}
 
 	localHandler, err := c.replicated.Local.AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -203,8 +203,8 @@ type Controller struct {
 	shardName string
 	queue     workqueue.TypedRateLimitingInterface[string]
 
-	dynamicCacheClient kcpdynamic.ClusterInterface
-	kcpCacheClient     kcpclientset.ClusterInterface
+	dynamicClusterClient kcpdynamic.ClusterInterface
+	kcpCacheClient       kcpclientset.ClusterInterface
 
 	replicated *ReplicatedGVR
 

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

@@ -85,32 +85,6 @@ func (c *Controller) reconcile(ctx context.Context, gvrKey string) error {
 	r := &replicationReconciler{
 		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 {
-				return nil, apierrors.NewNotFound(gvr.GroupResource(), name)
-			} else if err != nil {
-				return nil, err // necessary to avoid non-zero nil interface
-			}
-
-			u, err := toUnstructured(obj)
-			if err != nil {
-				return nil, err
-			}
-
-			if c.replicated.Filter != nil && !c.replicated.Filter(u) {
-				return nil, apierrors.NewNotFound(gvr.GroupResource(), name)
-			}
-
-			if _, ok := obj.(*unstructured.Unstructured); ok {
-				u = u.DeepCopy()
-			}
-			u.SetKind(c.replicated.Kind)
-			u.SetAPIVersion(gvr.GroupVersion().String())
-			return u, nil
-		},
 		getGlobalCopy: func(cluster logicalcluster.Name, namespace, name string) (*unstructured.Unstructured, error) {
 			gvr := gvrFromKey
 			if gvr.Group == "" {
@@ -237,6 +211,54 @@ func (c *Controller) reconcile(ctx context.Context, gvrKey string) error {
 			return c.kcpCacheClient.Cluster(cluster.Path()).CacheV1alpha1().CachedObjects().Delete(ctx, cachedObjName, metav1.DeleteOptions{})
 		},
 	}
+	// r.getLocalCopy is defined separately because it calls r.getGlobalCopy internally.
+	r.getLocalCopy = func(cluster logicalcluster.Name, namespace, name string) (*unstructured.Unstructured, error) {
+		gvr := gvrFromKey
+		key := kcpcache.ToClusterAwareKey(cluster.String(), namespace, name)
+		partialMetadataObj, exists, err := c.replicated.Local.GetIndexer().GetByKey(key)
+		if !exists {
+			return nil, apierrors.NewNotFound(gvr.GroupResource(), name)
+		} else if err != nil {
+			return nil, err // necessary to avoid non-zero nil interface
+		}
+
+		// The local informer only yields a PartialMetadataObject, but we need a full object for replication.
+		// We'll need to do a live GET to retrieve it, but before doing so we check RV against the cached copy
+		// to be sure we actually need do the GET call.
+
+		var globalCopyRV string
+		globalCopy, err := r.getGlobalCopy(cluster, namespace, name)
+		if err != nil {
+			if !apierrors.IsNotFound(err) {
+				return nil, err
+			}
+		} else {
+			ann := globalCopy.GetAnnotations()
+			if ann != nil {
+				globalCopyRV = ann[AnnotationKeyOriginalResourceVersion]
+			}
+		}
+
+		uPartialMetadataObj, err := toUnstructured(partialMetadataObj)
+		if err != nil {
+			return nil, err
+		}
+
+		if c.replicated.Filter != nil && !c.replicated.Filter(uPartialMetadataObj) {
+			return nil, apierrors.NewNotFound(gvr.GroupResource(), name)
+		}
+
+		if uPartialMetadataObj.GetResourceVersion() == globalCopyRV {
+			return globalCopy, nil
+		}
+
+		// The RV doesn't match either because it's different, or the globalCopy doesn't exist. Fallback to GET.
+		return c.dynamicClusterClient.
+			Cluster(cluster.Path()).
+			Resource(gvr).
+			Namespace(namespace).
+			Get(ctx, name, metav1.GetOptions{})
+	}
 	defer c.callback()
 	return r.reconcile(ctx, key)
 }

pkg/server/controllers.go

@@ -1742,7 +1742,6 @@ func (s *Server) installCacheController(ctx context.Context, config *rest.Config
 		kcpClusterClient,
 		s.KcpCacheClusterClient,
 		dynamicClient,
-		s.CacheDynamicClient,
 		s.KubeClusterClient,
 		s.KubeSharedInformerFactory.Core().V1().Namespaces(),
 		s.KubeSharedInformerFactory.Core().V1().Secrets(),