clustercache: add optional uncached client support for workload clusters

Signed-off-by: Bharath Nallapeta <bnallapeta@mirantis.com>

Author: bnallapeta

controllers/clustercache/cluster_accessor.go

@@ -191,6 +191,10 @@ type clusterAccessorLockedConnectionState struct {
 	// all typed objects except the ones for which caching has been disabled via DisableFor.
 	cachedClient client.Client
 
+	// uncachedClient to communicate with the workload cluster.
+	// It performs live GET/LIST calls directly against the API server with no caching.
+	uncachedClient client.Client
+
 	// cache is the cache used by the client.
 	// It manages informers that have been created e.g. by adding indexes to the cache,
 	// Get & List calls from the client or via the Watch method of the clusterAccessor.
@@ -297,11 +301,12 @@ func (ca *clusterAccessor) Connect(ctx context.Context) (retErr error) {
 		consecutiveFailures:  0,
 	}
 	ca.lockedState.connection = &clusterAccessorLockedConnectionState{
-		restConfig:   connection.RESTConfig,
-		restClient:   connection.RESTClient,
-		cachedClient: connection.CachedClient,
-		cache:        connection.Cache,
-		watches:      sets.Set[string]{},
+		restConfig:     connection.RESTConfig,
+		restClient:     connection.RESTClient,
+		cachedClient:   connection.CachedClient,
+		uncachedClient: connection.UncachedClient,
+		cache:          connection.Cache,
+		watches:        sets.Set[string]{},
 	}
 
 	return nil
@@ -407,6 +412,18 @@ func (ca *clusterAccessor) GetReader(ctx context.Context) (client.Reader, error)
 	return ca.lockedState.connection.cachedClient, nil
 }
 
+// GetUncachedClient returns a live (uncached) client for the given cluster.
+func (ca *clusterAccessor) GetUncachedClient(ctx context.Context) (client.Client, error) {
+	ca.rLock(ctx)
+	defer ca.rUnlock(ctx)
+
+	if ca.lockedState.connection == nil {
+		return nil, errors.Wrapf(ErrClusterNotConnected, "error getting uncached client")
+	}
+
+	return ca.lockedState.connection.uncachedClient, nil
+}
+
 func (ca *clusterAccessor) GetRESTConfig(ctx context.Context) (*rest.Config, error) {
 	ca.rLock(ctx)
 	defer ca.rUnlock(ctx)

controllers/clustercache/cluster_accessor_client.go

@@ -42,10 +42,11 @@ import (
 )
 
 type createConnectionResult struct {
-	RESTConfig   *rest.Config
-	RESTClient   *rest.RESTClient
-	CachedClient client.Client
-	Cache        *stoppableCache
+	RESTConfig     *rest.Config
+	RESTClient     *rest.RESTClient
+	CachedClient   client.Client
+	UncachedClient client.Client
+	Cache          *stoppableCache
 }
 
 func (ca *clusterAccessor) createConnection(ctx context.Context) (*createConnectionResult, error) {
@@ -99,17 +100,27 @@ func (ca *clusterAccessor) createConnection(ctx context.Context) (*createConnect
 		}
 	}
 
+	log.V(6).Info("Creating uncached client (finalized)")
+	// Create the production uncached client using the finalized rest.Config/httpClient/mapper.
+	// Wrap it so GET/LIST calls respect the configured timeout similar to the cached client path.
+	finalizedUncachedClient, err := createUncachedClient(ca.config.Scheme, restConfig, httpClient, mapper)
+	if err != nil {
+		return nil, err
+	}
+	finalizedUncachedClient = newClientWithTimeout(finalizedUncachedClient, restConfig.Timeout)
+
 	log.V(6).Info("Creating cached client and cache")
 	cachedClient, cache, err := createCachedClient(ctx, ca.cacheCtx, ca.config, restConfig, httpClient, mapper)
 	if err != nil {
 		return nil, err
 	}
 
 	return &createConnectionResult{
-		RESTConfig:   restConfig,
-		RESTClient:   restClient,
-		CachedClient: cachedClient,
-		Cache:        cache,
+		RESTConfig:     restConfig,
+		RESTClient:     restClient,
+		CachedClient:   cachedClient,
+		UncachedClient: finalizedUncachedClient,
+		Cache:          cache,
 	}, nil
 }
 

controllers/clustercache/cluster_accessor_test.go

@@ -210,6 +210,69 @@ func TestDisconnect(t *testing.T) {
 	g.Expect(accessor.lockedState.healthChecking.lastProbeSuccessTime.IsZero()).To(BeFalse())
 }
 
+func TestGetUncachedClientLifecycle(t *testing.T) {
+	g := NewWithT(t)
+
+	testCluster := &clusterv1.Cluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-cluster",
+			Namespace: metav1.NamespaceDefault,
+		},
+		Spec: clusterv1.ClusterSpec{
+			ControlPlaneRef: clusterv1.ContractVersionedObjectReference{
+				APIGroup: builder.ControlPlaneGroupVersion.Group,
+				Kind:     builder.GenericControlPlaneKind,
+				Name:     "cp1",
+			},
+		},
+	}
+	clusterKey := client.ObjectKeyFromObject(testCluster)
+	g.Expect(env.CreateAndWait(ctx, testCluster)).To(Succeed())
+	defer func() { g.Expect(env.CleanupAndWait(ctx, testCluster)).To(Succeed()) }()
+
+	// Create kubeconfig Secret
+	kubeconfigSecret := kubeconfig.GenerateSecret(testCluster, kubeconfig.FromEnvTestConfig(env.Config, testCluster))
+	g.Expect(env.CreateAndWait(ctx, kubeconfigSecret)).To(Succeed())
+	defer func() { g.Expect(env.CleanupAndWait(ctx, kubeconfigSecret)).To(Succeed()) }()
+
+	config := buildClusterAccessorConfig(env.GetScheme(), Options{
+		SecretClient: env.GetClient(),
+		Client: ClientOptions{
+			UserAgent: remote.DefaultClusterAPIUserAgent("test-controller-manager"),
+			Timeout:   10 * time.Second,
+		},
+	}, nil)
+	accessor := newClusterAccessor(context.Background(), clusterKey, config)
+
+	// Before connect, getting the uncached client should fail with ErrClusterNotConnected
+	_, err := accessor.GetUncachedClient(ctx)
+	g.Expect(err).To(HaveOccurred())
+	g.Expect(errors.Is(err, ErrClusterNotConnected)).To(BeTrue())
+
+	// Connect
+	g.Expect(accessor.Connect(ctx)).To(Succeed())
+	g.Expect(accessor.Connected(ctx)).To(BeTrue())
+
+	// After connect, getting the uncached client should succeed
+	r, err := accessor.GetUncachedClient(ctx)
+	g.Expect(err).ToNot(HaveOccurred())
+	g.Expect(r).ToNot(BeNil())
+
+	// List Nodes via the uncached client
+	nodeList := &corev1.NodeList{}
+	g.Expect(r.List(ctx, nodeList)).To(Succeed())
+	g.Expect(nodeList.Items).To(BeEmpty())
+
+	// Disconnect
+	accessor.Disconnect(ctx)
+	g.Expect(accessor.Connected(ctx)).To(BeFalse())
+
+	// After disconnect, getting the uncached client should fail with ErrClusterNotConnected
+	_, err = accessor.GetUncachedClient(ctx)
+	g.Expect(err).To(HaveOccurred())
+	g.Expect(errors.Is(err, ErrClusterNotConnected)).To(BeTrue())
+}
+
 func TestHealthCheck(t *testing.T) {
 	testCluster := &clusterv1.Cluster{
 		ObjectMeta: metav1.ObjectMeta{

controllers/clustercache/cluster_cache.go

@@ -166,6 +166,12 @@ type ClusterCache interface {
 	GetClusterSource(controllerName string, mapFunc func(ctx context.Context, cluster client.Object) []ctrl.Request, opts ...GetClusterSourceOption) source.Source
 }
 
+// UncachedClientProvider provides live (uncached) access to the workload cluster API server.
+// This is an optional extension interface; callers can type assert a ClusterCache to this interface to retrieve an uncached client for a given workload cluster.
+type UncachedClientProvider interface {
+	GetUncachedClient(ctx context.Context, cluster client.ObjectKey) (client.Client, error)
+}
+
 // HealthCheckingState holds the health checking state for a Cluster.
 type HealthCheckingState struct {
 	// LastProbeTime is the time when a health probe was executed last.
@@ -392,6 +398,16 @@ func (cc *clusterCache) GetReader(ctx context.Context, cluster client.ObjectKey)
 	return accessor.GetReader(ctx)
 }
 
+// GetUncachedClient returns a live (uncached) client for the given cluster.
+// If there is no connection to the workload cluster ErrClusterNotConnected will be returned.
+func (cc *clusterCache) GetUncachedClient(ctx context.Context, cluster client.ObjectKey) (client.Client, error) {
+	accessor := cc.getClusterAccessor(cluster)
+	if accessor == nil {
+		return nil, errors.Wrapf(ErrClusterNotConnected, "error getting uncached client")
+	}
+	return accessor.GetUncachedClient(ctx)
+}
+
 func (cc *clusterCache) GetRESTConfig(ctx context.Context, cluster client.ObjectKey) (*rest.Config, error) {
 	accessor := cc.getClusterAccessor(cluster)
 	if accessor == nil {

controllers/clustercache/cluster_cache_fake.go

@@ -32,8 +32,9 @@ func NewFakeClusterCache(workloadClient client.Client, clusterKey client.ObjectK
 	testCacheTracker.clusterAccessors[clusterKey] = &clusterAccessor{
 		lockedState: clusterAccessorLockedState{
 			connection: &clusterAccessorLockedConnectionState{
-				cachedClient: workloadClient,
-				watches:      sets.Set[string]{}.Insert(watchObjects...),
+				cachedClient:   workloadClient,
+				uncachedClient: workloadClient,
+				watches:        sets.Set[string]{}.Insert(watchObjects...),
 			},
 			healthChecking: clusterAccessorLockedHealthCheckingState{
 				lastProbeTime:        time.Now(),