fix cached resource replication after leader election loss

pkg/reconciler/cache/clustercachedresources/clustercachedresources_controller.go

@@ -25,6 +25,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilerrors "k8s.io/apimachinery/pkg/util/errors"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -201,6 +203,28 @@ func (c *Controller) enqueue(obj interface{}) {
 	c.queue.Add(key)
 }
 
+// enqueueForGVR enqueues all ClusterCachedResources for gvr when its CRD comes back.
+// They might be in a long backoff from failing while the CRD was gone, so we reset it.
+func (c *Controller) enqueueForGVR(gvr schema.GroupVersionResource) {
+	clusterCachedResources, err := c.ClusterCachedResourceLister.List(labels.Everything())
+	if err != nil {
+		utilruntime.HandleError(err)
+		return
+	}
+	for _, clusterCachedResource := range clusterCachedResources {
+		if schema.GroupVersionResource(clusterCachedResource.Spec.GroupVersionResource) != gvr {
+			continue
+		}
+		key, err := kcpcache.MetaClusterNamespaceKeyFunc(clusterCachedResource)
+		if err != nil {
+			utilruntime.HandleError(err)
+			continue
+		}
+		c.queue.Forget(key)
+		c.queue.Add(key)
+	}
+}
+
 // Start starts the controller, which stops when ctx.Done() is closed.
 func (c *Controller) Start(ctx context.Context, numThreads int) {
 	defer utilruntime.HandleCrash()
@@ -211,6 +235,10 @@ func (c *Controller) Start(ctx context.Context, numThreads int) {
 	logger.Info("Starting controller")
 	defer logger.Info("Shutting down controller")
 
+	c.localDiscoveringDynamicKcpInformers.AddGVRLifecycleHandler(ctx, informer.GVRLifecycleHandlerFuncs{
+		AddedFunc: c.enqueueForGVR,
+	})
+
 	for range numThreads {
 		go wait.Until(func() { c.startWorker(ctx) }, time.Second, ctx.Done())
 	}
@@ -331,3 +359,19 @@ func (c *controllerRegistry) unregister(name string) {
 	delete(c.controllers, name)
 	delete(c.cancels, name)
 }
+
+// unregisterIf unregisters the controller only if it is still registered under name.
+// Returns true if it was removed.
+func (c *controllerRegistry) unregisterIf(name string, controller *replicationcontroller.Controller) bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.controllers[name] != controller {
+		return false
+	}
+	if c.cancels[name] != nil {
+		c.cancels[name]()
+	}
+	delete(c.controllers, name)
+	delete(c.cancels, name)
+	return true
+}

pkg/reconciler/cache/clustercachedresources/clustercachedresources_reconcile.go

@@ -21,6 +21,7 @@ import (
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -33,6 +34,8 @@ import (
 	"github.com/kcp-dev/logicalcluster/v3"
 	apisv1alpha1 "github.com/kcp-dev/sdk/apis/apis/v1alpha1"
 	cachev1alpha1 "github.com/kcp-dev/sdk/apis/cache/v1alpha1"
+	conditionsv1alpha1 "github.com/kcp-dev/sdk/apis/third_party/conditions/apis/conditions/v1alpha1"
+	"github.com/kcp-dev/sdk/apis/third_party/conditions/util/conditions"
 
 	cacheclient "github.com/kcp-dev/kcp/pkg/cache/client"
 	"github.com/kcp-dev/kcp/pkg/cache/client/shard"
@@ -108,11 +111,13 @@ func (c *Controller) reconcile(ctx context.Context, cluster logicalcluster.Name,
 			},
 		},
 		&replication{
-			shardName:                            c.shardName,
-			localDynamicClusterClient:            c.localDynamicClient,
-			globalDynamicClusterClient:           c.globalDynamicClient,
-			kcpCacheClient:                       c.kcpCacheClient,
-			dynRESTMapper:                        c.dynRESTMapper,
+			shardName:                  c.shardName,
+			localDynamicClusterClient:  c.localDynamicClient,
+			globalDynamicClusterClient: c.globalDynamicClient,
+			kcpCacheClient:             c.kcpCacheClient,
+			getKind: func(cluster logicalcluster.Name, gvr schema.GroupVersionResource) (schema.GroupVersionKind, error) {
+				return c.dynRESTMapper.ForCluster(cluster).KindFor(gvr)
+			},
 			localDiscoveringDynamicKcpInformers:  c.localDiscoveringDynamicKcpInformers,
 			globalDiscoveringDynamicKcpInformers: c.globalDiscoveringDynamicKcpInformers,
 			requeueSelf:                          c.enqueue,
@@ -122,6 +127,7 @@ func (c *Controller) reconcile(ctx context.Context, cluster logicalcluster.Name,
 
 	var errs []error
 
+	before := clusterCachedResource.DeepCopy()
 	requeue := false
 	for _, r := range reconcilers {
 		var err error
@@ -139,9 +145,38 @@ func (c *Controller) reconcile(ctx context.Context, cluster logicalcluster.Name,
 		}
 	}
 
+	// We can't update status together with spec or metadata (committer panics),
+	// so leave it for the next round.
+	if equality.Semantic.DeepEqual(before.ObjectMeta, clusterCachedResource.ObjectMeta) &&
+		equality.Semantic.DeepEqual(before.Spec, clusterCachedResource.Spec) {
+		c.updateReplicationStatus(clusterCachedResource)
+	}
+
 	return requeue, utilerrors.NewAggregate(errs)
 }
 
+// updateReplicationStatus is done here and not in the replication reconciler, as
+// reconcilers before it can stop (e.g. CRD was deleted) and status would stay Ready.
+func (c *Controller) updateReplicationStatus(clusterCachedResource *cachev1alpha1.ClusterCachedResource) {
+	switch clusterCachedResource.Status.Phase {
+	case cachev1alpha1.ClusterCachedResourcePhaseDeleting, cachev1alpha1.ClusterCachedResourcePhaseDeleted:
+		return
+	}
+
+	controller := c.controllerRegistry.get(controllerNameFor(clusterCachedResource))
+	switch {
+	case controller != nil && controller.Started():
+		conditions.MarkTrue(clusterCachedResource, cachev1alpha1.ReplicationStarted)
+		clusterCachedResource.Status.Phase = cachev1alpha1.ClusterCachedResourcePhaseReady
+		return
+	case controller != nil:
+		conditions.MarkFalse(clusterCachedResource, cachev1alpha1.ReplicationStarted, "Syncing", conditionsv1alpha1.ConditionSeverityInfo, "Waiting for informers to sync")
+	default:
+		conditions.MarkFalse(clusterCachedResource, cachev1alpha1.ReplicationStarted, "NotRunning", conditionsv1alpha1.ConditionSeverityInfo, "Replication controller is not running")
+	}
+	clusterCachedResource.Status.Phase = cachev1alpha1.ClusterCachedResourcePhaseInitializing
+}
+
 func (c *Controller) listSelectedLocalResources(ctx context.Context, cluster logicalcluster.Name, clusterCachedResource *cachev1alpha1.ClusterCachedResource) (*unstructured.UnstructuredList, error) {
 	gvr := schema.GroupVersionResource{
 		Group:    clusterCachedResource.Spec.Group,

pkg/reconciler/cache/clustercachedresources/clustercachedresources_reconcile_replication.go

@@ -21,19 +21,16 @@ import (
 	"fmt"
 
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
 	kcpapiextensionsclientset "github.com/kcp-dev/client-go/apiextensions/client"
 	kcpdynamic "github.com/kcp-dev/client-go/dynamic"
 	"github.com/kcp-dev/logicalcluster/v3"
 	cachev1alpha1 "github.com/kcp-dev/sdk/apis/cache/v1alpha1"
-	"github.com/kcp-dev/sdk/apis/third_party/conditions/util/conditions"
 	kcpclientset "github.com/kcp-dev/sdk/client/clientset/versioned/cluster"
 
 	"github.com/kcp-dev/kcp/pkg/informer"
 	replicationcontroller "github.com/kcp-dev/kcp/pkg/reconciler/cache/clustercachedresources/replication"
-	"github.com/kcp-dev/kcp/pkg/reconciler/dynamicrestmapper"
 )
 
 // replication starts the replication machinery for a published resource.
@@ -43,7 +40,7 @@ type replication struct {
 	localDynamicClusterClient            kcpdynamic.ClusterInterface
 	globalDynamicClusterClient           kcpdynamic.ClusterInterface
 	kcpCacheClient                       kcpclientset.ClusterInterface
-	dynRESTMapper                        *dynamicrestmapper.DynamicRESTMapper
+	getKind                              func(cluster logicalcluster.Name, gvr schema.GroupVersionResource) (schema.GroupVersionKind, error)
 	localDiscoveringDynamicKcpInformers  *informer.DiscoveringDynamicSharedInformerFactory
 	globalDiscoveringDynamicKcpInformers *informer.DiscoveringDynamicSharedInformerFactory
 	cacheApiExtensionsClusterClient      kcpapiextensionsclientset.ClusterInterface
@@ -65,7 +62,7 @@ func (r *replication) reconcile(ctx context.Context, clusterCachedResource *cach
 	selection := replicationcontroller.SelectionFor(clusterCachedResource)
 
 	clusterName := logicalcluster.From(clusterCachedResource)
-	controllerName := fmt.Sprintf("%s.%s.%s.%s.%s", clusterName, gvr.Version, gvr.Resource, gvr.Group, clusterCachedResource.Name)
+	controllerName := controllerNameFor(clusterCachedResource)
 	// TODO: Add locking here when multiple workers are supported.
 	controller := r.controllerRegistry.get(controllerName)
 	// We setup controller even if we are deleting. This is to ensure that we can purge the cache.
@@ -77,26 +74,28 @@ func (r *replication) reconcile(ctx context.Context, clusterCachedResource *cach
 			return reconcileStatusStopAndRequeue, nil
 		}
 
-		controllerCtx, cancel := context.WithCancel(ctx)
+		// Get kind first, so we don't create informers for resources which are not served.
+		replicatedKind, err := r.getKind(clusterName, gvr)
+		if err != nil {
+			logger.Error(err, "Failed to get Kind for resource", "resource", gvr)
+			return reconcileStatusStopAndRequeue, err
+		}
+		// The REST mapper can still have a CRD which the local factory already removed.
+		// ForResource would then create an informer the factory never stops.
+		if !r.servedLocally(gvr) {
+			return reconcileStatusStopAndRequeue, fmt.Errorf("resource %s is not served by local informers", gvr)
+		}
 
 		global, err := r.globalDiscoveringDynamicKcpInformers.ForResource(gvr)
 		if err != nil {
 			logger.Error(err, "Failed to get global informer for resource", "resource", gvr)
-			cancel()
 			return reconcileStatusStopAndRequeue, err
 		}
 
 		// Local informer is based on the specific types we want to replicate.
 		local, err := r.localDiscoveringDynamicKcpInformers.ForResource(gvr)
 		if err != nil {
 			logger.Error(err, "Failed to get local informer for resource", "resource", gvr)
-			cancel()
-			return reconcileStatusStopAndRequeue, err
-		}
-		replicatedKind, err := r.dynRESTMapper.ForCluster(clusterName).KindFor(gvr)
-		if err != nil {
-			logger.Error(err, "Failed to get Kind for resource", "resource", gvr)
-			cancel()
 			return reconcileStatusStopAndRequeue, err
 		}
 		replicated := &replicationcontroller.ReplicatedGVR{
@@ -105,12 +104,22 @@ func (r *replication) reconcile(ctx context.Context, clusterCachedResource *cach
 			Local:    local.Informer(),
 			Global:   global.Informer(),
 		}
+		if replicated.Local.IsStopped() || replicated.Global.IsStopped() {
+			return reconcileStatusStopAndRequeue, fmt.Errorf("informer for %s is stopped", gvr)
+		}
 		replicationcontroller.InstallIndexers(replicated)
+
+		// Run informers with the factories' stop channels, not ctx. ctx is cancelled when we
+		// lose the leader lease and a stopped informer can't be reused. The global factory
+		// has no worker, so global informers run until the process exits.
+		r.localDiscoveringDynamicKcpInformers.Start(nil)
+		r.globalDiscoveringDynamicKcpInformers.Start(nil)
+
 		requeueSelf := func() {
 			r.requeueSelf(clusterCachedResource)
 		}
 
-		c, err := replicationcontroller.NewController(
+		c := replicationcontroller.NewController(
 			r.shardName,
 			r.localDynamicClusterClient,
 			r.globalDynamicClusterClient,
@@ -122,31 +131,19 @@ func (r *replication) reconcile(ctx context.Context, clusterCachedResource *cach
 			requeueSelf,
 			selection,
 		)
-		if err != nil {
-			cancel()
-			return reconcileStatusContinue, err
-		}
-
-		go replicated.Local.Run(ctx.Done())
-		go replicated.Global.Run(ctx.Done())
 
+		controllerCtx, cancel := context.WithCancel(ctx)
 		r.controllerRegistry.register(controllerName, c, cancel)
-		if clusterCachedResource.Status.Phase != cachev1alpha1.ClusterCachedResourcePhaseDeleting {
-			conditions.MarkTrue(clusterCachedResource, cachev1alpha1.ReplicationStarted)
-			clusterCachedResource.Status.Phase = cachev1alpha1.ClusterCachedResourcePhaseReady
-		}
 
 		go func() {
-			defer cancel()
-
-			if !cache.WaitForCacheSync(controllerCtx.Done(), replicated.Local.HasSynced, replicated.Global.HasSynced) {
-				logger.Error(nil, "Informers failed to sync, removing controller", "controller", controllerName)
-				r.controllerRegistry.unregister(controllerName)
+			c.Start(controllerCtx, 1)
+			removed := r.controllerRegistry.unregisterIf(controllerName, c)
+			cancel()
+			// controller stopped by itself (e.g. CRD was deleted), recreate it
+			if removed && ctx.Err() == nil {
+				logger.Info("Replication controller stopped, recreating", "controller", controllerName)
 				requeueSelf()
-				return
 			}
-
-			c.Start(controllerCtx, 1)
 		}()
 
 		return reconcileStatusStopAndRequeue, nil // Once controller is started, we requeue to check if we need to delete it.
@@ -169,3 +166,20 @@ func (r *replication) reconcile(ctx context.Context, clusterCachedResource *cach
 		return reconcileStatusContinue, nil
 	}
 }
+
+func (r *replication) servedLocally(gvr schema.GroupVersionResource) bool {
+	resources, err := r.localDiscoveringDynamicKcpInformers.ServerResourcesForGroupVersion(gvr.GroupVersion().String())
+	if err != nil {
+		return false
+	}
+	for _, resource := range resources.APIResources {
+		if resource.Name == gvr.Resource {
+			return true
+		}
+	}
+	return false
+}
+
+func controllerNameFor(ccr *cachev1alpha1.ClusterCachedResource) string {
+	return fmt.Sprintf("%s.%s.%s.%s.%s", logicalcluster.From(ccr), ccr.Spec.Version, ccr.Spec.Resource, ccr.Spec.Group, ccr.Name)
+}