CachedResources: replicate full objects

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

@@ -46,8 +46,7 @@ import (
 	cachev1alpha1 "github.com/kcp-dev/kcp/sdk/apis/cache/v1alpha1"
 	kcpclientset "github.com/kcp-dev/kcp/sdk/client/clientset/versioned/cluster"
 	cachev1alpha1client "github.com/kcp-dev/kcp/sdk/client/clientset/versioned/typed/cache/v1alpha1"
-	kcpinformers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions"
-	cacheinformers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/cache/v1alpha1"
+	cachev1alpha1informers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/cache/v1alpha1"
 	cachev1alpha1listers "github.com/kcp-dev/kcp/sdk/client/listers/cache/v1alpha1"
 )
 
@@ -68,7 +67,6 @@ func NewController(
 	kcpClusterClient kcpclientset.ClusterInterface,
 	kcpCacheClient kcpclientset.ClusterInterface,
 	dynamicClient kcpdynamic.ClusterInterface,
-	cacheDynamicClient kcpdynamic.ClusterInterface,
 
 	kubeClusterClient kcpkubernetesclientset.ClusterInterface,
 	namespaceInformer kcpcorev1informers.NamespaceClusterInformer,
@@ -77,10 +75,10 @@ func NewController(
 	dynRESTMapper *dynamicrestmapper.DynamicRESTMapper,
 
 	discoveringDynamicKcpInformers *informer.DiscoveringDynamicSharedInformerFactory,
-	cacheKcpInformers kcpinformers.SharedInformerFactory,
 
-	cachedResourceInformer cacheinformers.CachedResourceClusterInformer,
-	cachedResourceEndpointSliceInformer cacheinformers.CachedResourceEndpointSliceClusterInformer,
+	cachedObjectsInformer cachev1alpha1informers.CachedObjectClusterInformer,
+	cachedResourceInformer cachev1alpha1informers.CachedResourceClusterInformer,
+	cachedResourceEndpointSliceInformer cachev1alpha1informers.CachedResourceEndpointSliceClusterInformer,
 ) (*Controller, error) {
 	c := &Controller{
 		shardName: shardName,
@@ -93,13 +91,12 @@ func NewController(
 		kcpClient:      kcpClusterClient,
 		kcpCacheClient: kcpCacheClient,
 
-		dynamicClient:      dynamicClient,
-		cacheDynamicClient: cacheDynamicClient,
+		dynamicClient: dynamicClient,
 
 		dynRESTMapper: dynRESTMapper,
 
 		discoveringDynamicKcpInformers: discoveringDynamicKcpInformers,
-		cacheKcpInformers:              cacheKcpInformers,
+		cachedObjectsInformer:          cachedObjectsInformer,
 
 		CachedResourceLister:  cachedResourceInformer.Lister(),
 		CachedResourceIndexer: cachedResourceInformer.Informer().GetIndexer(),
@@ -165,18 +162,17 @@ type Controller struct {
 	kcpClient      kcpclientset.ClusterInterface
 	kcpCacheClient kcpclientset.ClusterInterface
 
-	dynamicClient      kcpdynamic.ClusterInterface
-	cacheDynamicClient kcpdynamic.ClusterInterface
+	dynamicClient kcpdynamic.ClusterInterface
 
 	dynRESTMapper *dynamicrestmapper.DynamicRESTMapper
 
 	discoveringDynamicKcpInformers *informer.DiscoveringDynamicSharedInformerFactory
-	cacheKcpInformers              kcpinformers.SharedInformerFactory
+	cachedObjectsInformer          cachev1alpha1informers.CachedObjectClusterInformer
 
 	CachedResourceIndexer cache.Indexer
 	CachedResourceLister  cachev1alpha1listers.CachedResourceClusterLister
 
-	CachedResourceEndpointSliceInformer cacheinformers.CachedResourceEndpointSliceClusterInformer
+	CachedResourceEndpointSliceInformer cachev1alpha1informers.CachedResourceEndpointSliceClusterInformer
 
 	commit func(ctx context.Context, new, old *CachedResourceResource) error
 

pkg/reconciler/cache/cachedresources/cachedresources_reconcile.go

@@ -83,10 +83,10 @@ 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,
+			cachedObjectsInformer:          c.cachedObjectsInformer,
 			discoveringDynamicKcpInformers: c.discoveringDynamicKcpInformers,
 			callback:                       c.enqueue,
 			controllerRegistry:             c.controllerRegistry,

pkg/reconciler/cache/cachedresources/cachedresources_reconcile_replication.go

@@ -34,17 +34,17 @@ import (
 	cachev1alpha1 "github.com/kcp-dev/kcp/sdk/apis/cache/v1alpha1"
 	"github.com/kcp-dev/kcp/sdk/apis/third_party/conditions/util/conditions"
 	kcpclientset "github.com/kcp-dev/kcp/sdk/client/clientset/versioned/cluster"
-	kcpinformers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions"
+	cachev1alpha1informers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/cache/v1alpha1"
 )
 
 // replication starts the replication machinery for a published resource.
 // 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
+	cachedObjectsInformer          cachev1alpha1informers.CachedObjectClusterInformer
 	discoveringDynamicKcpInformers *informer.DiscoveringDynamicSharedInformerFactory
 	callback                       func(obj interface{})
 	controllerRegistry             *controllerRegistry
@@ -53,7 +53,6 @@ type replication struct {
 func (r *replication) reconcile(ctx context.Context, cachedResource *cachev1alpha1.CachedResource) (reconcileStatus, error) {
 	logger := klog.FromContext(ctx)
 	logger.Info("reconciling published resource", "CachedResource", cachedResource.Name)
-
 	gvr := schema.GroupVersionResource{
 		Group:    cachedResource.Spec.Group,
 		Version:  cachedResource.Spec.Version,
@@ -76,8 +75,6 @@ func (r *replication) reconcile(ctx context.Context, cachedResource *cachev1alph
 		// Global informer is based on the CachedResource type and we construct index based on the schema labels.
 		controllerCtx, cancel := context.WithCancel(ctx)
 
-		global := r.cacheKcpInformers.Cache().V1alpha1().CachedObjects()
-
 		// Local informer is based on the specific types we want to replicate.
 		local, err := r.discoveringDynamicKcpInformers.ClusterWithContext(ctx, cluster).ForResource(gvr)
 		if err != nil {
@@ -92,19 +89,18 @@ func (r *replication) reconcile(ctx context.Context, cachedResource *cachev1alph
 			return reconcileStatusStopAndRequeue, err
 		}
 		replicated := &replicationcontroller.ReplicatedGVR{
-			Kind:   replicatedKind.Kind,
-			Local:  local.Informer(),
-			Global: global.Informer(),
+			Kind:  replicatedKind.Kind,
+			Local: local.Informer(),
 		}
-		replicationcontroller.InstallIndexers(replicated)
 		callback := func() {
 			r.callback(cachedResource)
 		}
 
 		c, err := replicationcontroller.NewController(
 			r.shardName,
-			r.dynamicCacheClient,
+			r.dynamicClusterClient,
 			r.kcpCacheClient,
+			r.cachedObjectsInformer,
 			gvr,
 			replicated,
 			callback,
@@ -116,9 +112,8 @@ func (r *replication) reconcile(ctx context.Context, cachedResource *cachev1alph
 		}
 
 		go replicated.Local.Run(ctx.Done())
-		go replicated.Global.Run(ctx.Done())
 
-		if !cache.WaitForCacheSync(ctx.Done(), replicated.Local.HasSynced, replicated.Global.HasSynced) {
+		if !cache.WaitForCacheSync(ctx.Done(), replicated.Local.HasSynced) {
 			cancel()
 			return reconcileStatusContinue, fmt.Errorf("failed to wait for informers to sync")
 		}

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

@@ -32,13 +32,14 @@ import (
 
 	kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache"
 	kcpdynamic "github.com/kcp-dev/client-go/dynamic"
+	"github.com/kcp-dev/logicalcluster/v3"
 
 	cacheclient "github.com/kcp-dev/kcp/pkg/cache/client"
 	"github.com/kcp-dev/kcp/pkg/cache/client/shard"
-	"github.com/kcp-dev/kcp/pkg/indexers"
 	"github.com/kcp-dev/kcp/pkg/logging"
 	cachev1alpha1 "github.com/kcp-dev/kcp/sdk/apis/cache/v1alpha1"
 	kcpclientset "github.com/kcp-dev/kcp/sdk/client/clientset/versioned/cluster"
+	cachev1alpha1informers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/cache/v1alpha1"
 )
 
 // Locally we store object with the original form of the object.
@@ -53,8 +54,9 @@ const (
 // NewController returns a new replication controller.
 func NewController(
 	shardName string,
-	dynamicCacheClient kcpdynamic.ClusterInterface,
+	dynamicClusterClient kcpdynamic.ClusterInterface,
 	kcpCacheClient kcpclientset.ClusterInterface,
+	cachedObjectsInformer cachev1alpha1informers.CachedObjectClusterInformer,
 	gvr schema.GroupVersionResource,
 	replicated *ReplicatedGVR,
 	callback func(),
@@ -68,12 +70,13 @@ func NewController(
 				Name: ControllerName,
 			},
 		),
-		dynamicCacheClient: dynamicCacheClient,
-		kcpCacheClient:     kcpCacheClient,
-		replicated:         replicated,
-		callback:           callback,
-		cleanupFuncs:       make([]func(), 0),
-		localLabelSelector: localLabelSelector,
+		dynamicClusterClient:  dynamicClusterClient,
+		kcpCacheClient:        kcpCacheClient,
+		cachedObjectsInformer: cachedObjectsInformer,
+		replicated:            replicated,
+		callback:              callback,
+		cleanupFuncs:          make([]func(), 0),
+		localLabelSelector:    localLabelSelector,
 	}
 
 	localHandler, err := c.replicated.Local.AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -88,7 +91,7 @@ func NewController(
 		_ = c.replicated.Local.RemoveEventHandler(localHandler)
 	})
 
-	globalHandler, err := c.replicated.Global.AddEventHandler(cache.ResourceEventHandlerFuncs{
+	globalHandler, err := c.cachedObjectsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc:    func(obj interface{}) { c.enqueueCacheObject(obj) },
 		UpdateFunc: func(_, obj interface{}) { c.enqueueCacheObject(obj) },
 		DeleteFunc: func(obj interface{}) { c.enqueueCacheObject(obj) },
@@ -97,7 +100,7 @@ func NewController(
 		return nil, err
 	}
 	c.cleanupFuncs = append(c.cleanupFuncs, func() {
-		_ = c.replicated.Global.RemoveEventHandler(globalHandler)
+		_ = c.cachedObjectsInformer.Informer().RemoveEventHandler(globalHandler)
 	})
 
 	return c, nil
@@ -198,8 +201,9 @@ type Controller struct {
 	shardName string
 	queue     workqueue.TypedRateLimitingInterface[string]
 
-	dynamicCacheClient kcpdynamic.ClusterInterface
-	kcpCacheClient     kcpclientset.ClusterInterface
+	dynamicClusterClient  kcpdynamic.ClusterInterface
+	kcpCacheClient        kcpclientset.ClusterInterface
+	cachedObjectsInformer cachev1alpha1informers.CachedObjectClusterInformer
 
 	replicated *ReplicatedGVR
 
@@ -223,17 +227,7 @@ type Controller struct {
 }
 
 type ReplicatedGVR struct {
-	Kind          string
-	Filter        func(u *unstructured.Unstructured) bool
-	Global, Local cache.SharedIndexInformer
-}
-
-// InstallIndexers adds the additional indexers that this controller requires to the informers.
-func InstallIndexers(replicated *ReplicatedGVR) {
-	indexers.AddIfNotPresentOrDie(
-		replicated.Global.GetIndexer(),
-		cache.Indexers{
-			ByGVRAndShardAndLogicalClusterAndNamespaceAndName: IndexByGVRAndShardAndLogicalClusterAndNamespace,
-		},
-	)
+	Kind   string
+	Filter func(u *unstructured.Unstructured) bool
+	Local  cache.SharedIndexInformer
 }