add specific cluster and start individual cache

Signed-off-by: Karol Szwaj <[email protected]>

On-behalf-of: @SAP [email protected]

Author: cnvergence

initializingworkspaces/cluster.go

@@ -0,0 +1,149 @@
+/*
+Copyright 2025 The KCP Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package initializingworkspaces
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"time"
+
+	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/cluster"
+
+	"github.com/kcp-dev/logicalcluster/v3"
+)
+
+var _ cluster.Cluster = &specificCluster{}
+
+type specificCluster struct {
+	clusterName logicalcluster.Name
+
+	scheme     *runtime.Scheme
+	config     *rest.Config
+	httpClient *http.Client
+	client     client.Client
+	mapper     meta.RESTMapper
+	cache      cache.Cache
+}
+
+// New method to create a cluster with specific URL
+func (p *Provider) createSpecificCluster(clusterName logicalcluster.Name, scheme *runtime.Scheme) (cluster.Cluster, error) {
+	specificConfig := rest.CopyConfig(p.config)
+	specificConfig.Host = fmt.Sprintf("%s/clusters/%s", specificConfig.Host, clusterName)
+
+	cli, err := client.New(specificConfig, client.Options{Scheme: scheme})
+	if err != nil {
+		return nil, err
+	}
+
+	httpClient, err := rest.HTTPClientFor(specificConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	mapper, err := apiutil.NewDynamicRESTMapper(specificConfig, httpClient)
+	if err != nil {
+		return nil, err
+	}
+
+	specificCache, err := cache.New(specificConfig, cache.Options{
+		Scheme: scheme,
+		Mapper: mapper,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create cache: %w", err)
+	}
+
+	// Return the custom cluster implementation
+	return &specificCluster{
+		clusterName: clusterName,
+		config:      specificConfig,
+		scheme:      scheme,
+		client:      cli, // Direct client
+		httpClient:  httpClient,
+		mapper:      mapper, // Dedicated mapper
+		cache:       specificCache,
+	}, nil
+}
+
+// GetHTTPClient returns the HTTP client scoped to the cluster.
+func (c *specificCluster) GetHTTPClient() *http.Client {
+	return c.httpClient
+}
+
+// GetConfig returns the rest.Config scoped to the cluster.
+func (c *specificCluster) GetConfig() *rest.Config {
+	return c.config
+}
+
+// GetScheme returns the scheme scoped to the cluster.
+func (c *specificCluster) GetScheme() *runtime.Scheme {
+	return c.scheme
+}
+
+// GetFieldIndexer returns a FieldIndexer scoped to the cluster.
+func (c *specificCluster) GetFieldIndexer() client.FieldIndexer {
+	return c.cache
+}
+
+// GetRESTMapper returns a RESTMapper scoped to the cluster.
+func (c *specificCluster) GetRESTMapper() meta.RESTMapper {
+	return c.mapper
+}
+
+// GetCache returns a cache.Cache.
+func (c *specificCluster) GetCache() cache.Cache {
+	return c.cache
+}
+
+// GetClient returns a client scoped to the namespace.
+func (c *specificCluster) GetClient() client.Client {
+	return c.client
+}
+
+// GetEventRecorderFor returns a new EventRecorder for the provided name.
+func (c *specificCluster) GetEventRecorderFor(name string) record.EventRecorder {
+	panic("implement me")
+}
+
+// GetAPIReader returns a reader against the cluster.
+func (c *specificCluster) GetAPIReader() client.Reader {
+	return c.cache
+}
+
+// Start starts the cluster.
+func (c *specificCluster) Start(ctx context.Context) error {
+	go func() {
+		if err := c.cache.Start(ctx); err != nil {
+			klog.Errorf("Error starting cache for cluster %s: %v", c.clusterName, err)
+		}
+	}()
+
+	_, cancel := context.WithTimeout(ctx, 30*time.Second)
+	defer cancel()
+
+	return nil
+}

initializingworkspaces/provider.go

@@ -243,15 +243,24 @@ func (p *Provider) handleLogicalClusterEvent(ctx context.Context, mgr mcmanager.
 		return
 	}
 
-	// create new scoped cluster.
+	// create new specific cluster.
 	clusterCtx, cancel := context.WithCancel(ctx)
-	cl, err := mcpcache.NewScopedCluster(p.config, clusterName, p.cache, p.scheme)
+	cl, err := p.createSpecificCluster(clusterName, p.scheme)
 	if err != nil {
 		p.log.Error(err, "failed to create cluster for initializing workspace", "cluster", clusterName)
 		cancel()
 		p.lock.Unlock()
 		return
 	}
+
+	// Start the cluster's cache and wait for it to sync BEFORE engaging
+	p.log.Info("starting cluster cache", "cluster", clusterName)
+	if err := cl.Start(clusterCtx); err != nil {
+		p.log.Error(err, "failed to start cluster cache", "cluster", clusterName)
+		cancel()
+		p.lock.Unlock()
+		return
+	}
 	p.clusters[clusterName] = cl
 	p.cancelFns[clusterName] = cancel
 	p.lock.Unlock()