Add ClusterFilter to ClusterCache Options

This allows filtering the Clusters that are handled by the cache.
It can be used for example by providers that only want to cache Clusters
of the relevant type to them.

Signed-off-by: Lennart Jern <[email protected]>

Author: lentzi90

controllers/clustercache/cluster_cache.go

@@ -60,13 +60,22 @@ type Options struct {
 	// will never be created.
 	WatchFilterValue string
 
+	// ClusterFilter is a function that can be used to filter which clusters should be handled
+	// by the ClusterCache. If nil, all clusters will be handled. If set, only clusters for which
+	// the filter returns true will be handled.
+	ClusterFilter ClusterFilter
+
 	// Cache are the cache options for the caches that are created per cluster.
 	Cache CacheOptions
 
 	// Client are the client options for the clients that are created per cluster.
 	Client ClientOptions
 }
 
+// ClusterFilter is a function that filters which clusters should be handled by the ClusterCache.
+// It returns true if the cluster should be handled, false otherwise.
+type ClusterFilter func(cluster *clusterv1.Cluster) bool
+
 // CacheOptions are the cache options for the caches that are created per cluster.
 type CacheOptions struct {
 	// SyncPeriod is the sync period of the cache.
@@ -364,6 +373,11 @@ type clusterCache struct {
 
 	// cacheCtxCancel is used during Shutdown to stop caches.
 	cacheCtxCancel context.CancelCauseFunc
+
+	// ClusterFilter is a function that can be used to filter which clusters should be handled
+	// by the ClusterCache. If nil, all clusters will be handled. If set, only clusters for which
+	// the filter returns true will be handled.
+	clusterFilter ClusterFilter
 }
 
 // clusterSource stores the necessary information so we can enqueue reconcile.Requests for reconcilers that
@@ -451,13 +465,14 @@ func (cc *clusterCache) Reconcile(ctx context.Context, req reconcile.Request) (r
 	log := ctrl.LoggerFrom(ctx)
 	clusterKey := client.ObjectKey{Namespace: req.Namespace, Name: req.Name}
 
-	accessor := cc.getOrCreateClusterAccessor(clusterKey)
-
 	cluster := &clusterv1.Cluster{}
 	if err := cc.client.Get(ctx, req.NamespacedName, cluster); err != nil {
 		if apierrors.IsNotFound(err) {
 			log.Info("Cluster has been deleted, disconnecting")
-			accessor.Disconnect(ctx)
+			accessor := cc.getClusterAccessor(clusterKey)
+			if accessor != nil {
+				accessor.Disconnect(ctx)
+			}
 			cc.deleteClusterAccessor(clusterKey)
 			cc.cleanupClusterSourcesForCluster(clusterKey)
 			return ctrl.Result{}, nil
@@ -468,6 +483,20 @@ func (cc *clusterCache) Reconcile(ctx context.Context, req reconcile.Request) (r
 		return ctrl.Result{RequeueAfter: defaultRequeueAfter}, nil
 	}
 
+	// Apply cluster filter if set
+	if cc.clusterFilter != nil && !cc.clusterFilter(cluster) {
+		log.V(6).Info("Cluster filtered out by ClusterFilter, not connecting")
+		accessor := cc.getClusterAccessor(clusterKey)
+		if accessor != nil {
+			accessor.Disconnect(ctx)
+		}
+		cc.deleteClusterAccessor(clusterKey)
+		cc.cleanupClusterSourcesForCluster(clusterKey)
+		return ctrl.Result{}, nil
+	}
+
+	accessor := cc.getOrCreateClusterAccessor(clusterKey)
+
 	// Return if infrastructure is not ready yet to avoid trying to open a connection when it cannot succeed.
 	// Requeue is not needed as there will be a new reconcile.Request when Cluster.status.initialization.infrastructureProvisioned is set.
 	if !ptr.Deref(cluster.Status.Initialization.InfrastructureProvisioned, false) {

controllers/clustercache/cluster_cache_test.go

@@ -57,6 +57,9 @@ func TestReconcile(t *testing.T) {
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "test-cluster",
 			Namespace: metav1.NamespaceDefault,
+			Labels: map[string]string{
+				"cluster.x-k8s.io/included-in-clustercache-tests": "true",
+			},
 		},
 		Spec: clusterv1.ClusterSpec{
 			ControlPlaneRef: clusterv1.ContractVersionedObjectReference{
@@ -87,6 +90,9 @@ func TestReconcile(t *testing.T) {
 		clusterAccessorConfig: accessorConfig,
 		clusterAccessors:      make(map[client.ObjectKey]*clusterAccessor),
 		cacheCtx:              context.Background(),
+		clusterFilter: func(cluster *clusterv1.Cluster) bool {
+			return (cluster.ObjectMeta.Labels["cluster.x-k8s.io/included-in-clustercache-tests"] == "true")
+		},
 	}
 
 	// Add a Cluster source and start it (queue will be later used to verify the source works correctly)
@@ -110,6 +116,31 @@ func TestReconcile(t *testing.T) {
 	testCluster.Status.Initialization.InfrastructureProvisioned = ptr.To(true)
 	g.Expect(env.Status().Patch(ctx, testCluster, patch)).To(Succeed())
 
+	// Exclude from clustercache by changing the label
+	patch = client.MergeFrom(testCluster.DeepCopy())
+	testCluster.ObjectMeta.Labels = map[string]string{
+		"cluster.x-k8s.io/included-in-clustercache-tests": "false",
+	}
+	g.Expect(env.Patch(ctx, testCluster, patch)).To(Succeed())
+	// Sanity check that the clusterFIlter does not include the cluster now
+	g.Expect(cc.clusterFilter(testCluster)).To((BeFalse()))
+
+	// Reconcile, cluster should be ignored now
+	// => no requeue, no cluster accessor created
+	res, err = cc.Reconcile(ctx, reconcile.Request{NamespacedName: clusterKey})
+	g.Expect(err).ToNot(HaveOccurred())
+	g.Expect(res).To(Equal(ctrl.Result{}))
+	g.Expect(res.IsZero()).To(BeTrue())
+
+	// Put the label back
+	patch = client.MergeFrom(testCluster.DeepCopy())
+	testCluster.ObjectMeta.Labels = map[string]string{
+		"cluster.x-k8s.io/included-in-clustercache-tests": "true",
+	}
+	g.Expect(env.Patch(ctx, testCluster, patch)).To(Succeed())
+	// Sanity check that the clusterFIlter does include the cluster now
+	g.Expect(cc.clusterFilter(testCluster)).To((BeTrue()))
+
 	// Reconcile, kubeconfig Secret doesn't exist
 	// => accessor.Connect will fail so we expect a retry with ConnectionCreationRetryInterval.
 	res, err = cc.Reconcile(ctx, reconcile.Request{NamespacedName: clusterKey})