Merge pull request #3573 from olamilekan000/fix-workspace-extend-with-bug

fix workspace type installation with extend.with without path bug

Author: kcp-ci-bot

pkg/admission/workspacetypeexists/admission.go

@@ -182,27 +182,32 @@ func (o *workspacetypeExists) Admit(ctx context.Context, a admission.Attributes,
 }
 
 func (o *workspacetypeExists) resolveTypeRef(workspacePath logicalcluster.Path, ref tenancyv1alpha1.WorkspaceTypeReference) (*tenancyv1alpha1.WorkspaceType, error) {
+	return resolveWorkspaceTypeReference(workspacePath, ref, o.getType)
+}
+
+func resolveWorkspaceTypeReference(workspacePath logicalcluster.Path, ref tenancyv1alpha1.WorkspaceTypeReference, getter func(logicalcluster.Path, string) (*tenancyv1alpha1.WorkspaceType, error)) (*tenancyv1alpha1.WorkspaceType, error) {
 	if ref.Path != "" {
-		wt, err := o.getType(logicalcluster.NewPath(ref.Path), string(ref.Name))
+		wt, err := getter(logicalcluster.NewPath(ref.Path), string(ref.Name))
 		if err != nil {
 			return nil, apierrors.NewInternalError(err)
 		}
 
 		return wt, err
 	}
 
+	currentPath := workspacePath
 	for {
-		wt, err := o.getType(workspacePath, string(ref.Name))
+		wt, err := getter(currentPath, string(ref.Name))
 		if err != nil {
 			if apierrors.IsNotFound(err) {
-				parent, hasParent := workspacePath.Parent()
-				if !hasParent && workspacePath != core.RootCluster.Path() {
+				parent, hasParent := currentPath.Parent()
+				if !hasParent && currentPath != core.RootCluster.Path() {
 					// fall through with root cluster. We always check types in there as last chance.
 					parent = core.RootCluster.Path()
 				} else if !hasParent {
 					return nil, fmt.Errorf("workspace type %q cannot be resolved", ref.String())
 				}
-				workspacePath = parent
+				currentPath = parent
 				continue
 			}
 			return nil, apierrors.NewInternalError(err)
@@ -439,7 +444,12 @@ func (r *transitiveTypeResolver) resolve(wt *tenancyv1alpha1.WorkspaceType, seen
 
 	ret := make([]*tenancyv1alpha1.WorkspaceType, 0, len(wt.Spec.Extend.With))
 	for _, baseTypeRef := range wt.Spec.Extend.With {
-		qualifiedName := logicalcluster.NewPath(baseTypeRef.Path).Join(tenancyv1alpha1.ObjectName(baseTypeRef.Name)).String()
+		baseType, err := resolveWorkspaceTypeReference(canonicalPathFrom(wt), baseTypeRef, r.getter)
+		if err != nil {
+			return nil, fmt.Errorf("unable to find inherited workspace type %s", baseTypeRef.Name)
+		}
+
+		qualifiedName := canonicalPathFrom(baseType).Join(baseType.Name).String()
 		if pathSeen[qualifiedName] {
 			// seen in this path already. That's a cycle.
 			for i, t := range path {
@@ -455,10 +465,6 @@ func (r *transitiveTypeResolver) resolve(wt *tenancyv1alpha1.WorkspaceType, seen
 			continue // already seen trunk
 		}
 
-		baseType, err := r.getter(logicalcluster.NewPath(baseTypeRef.Path), tenancyv1alpha1.ObjectName(baseTypeRef.Name))
-		if err != nil {
-			return nil, fmt.Errorf("unable to find inherited workspace type %s", qualifiedName)
-		}
 		ret = append(ret, baseType)
 
 		parents, err := r.resolve(baseType, seen, pathSeen, path)

pkg/admission/workspacetypeexists/admission_test.go

@@ -622,6 +622,23 @@ func TestTransitiveTypeResolverResolve(t *testing.T) {
 			want:    sets.New[string]("root:universal", "root:organization", "root:org:type"),
 			wantErr: true,
 		},
+		{
+			name: "extending types without explicit path",
+			types: map[string]*tenancyv1alpha1.WorkspaceType{
+				"root:org:base":      newType("root:org:base").WorkspaceType,
+				"root:org:universal": newType("root:org:universal").WorkspaceType,
+			},
+			input: func() *tenancyv1alpha1.WorkspaceType {
+				wt := newType("root:org:extended").WorkspaceType
+				wt.Spec.Extend.With = []tenancyv1alpha1.WorkspaceTypeReference{
+					{Name: "base"},
+					{Name: "universal", Path: "root:org"},
+				}
+				return wt
+			}(),
+			want:    sets.New[string]("root:org:base", "root:org:universal", "root:org:extended"),
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

pkg/reconciler/tenancy/workspace/workspace_controller.go

@@ -19,6 +19,7 @@ package workspace
 import (
 	"context"
 	"fmt"
+	"sync"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -50,6 +51,77 @@ const (
 	ControllerName = "kcp-workspace"
 )
 
+type clientPool struct {
+	mu          sync.RWMutex
+	kcpClients  map[string]kcpclientset.ClusterInterface
+	kubeClients map[string]kubernetes.ClusterInterface
+	adminConfig *rest.Config
+}
+
+func newClientPool(adminConfig *rest.Config) *clientPool {
+	return &clientPool{
+		kcpClients:  make(map[string]kcpclientset.ClusterInterface),
+		kubeClients: make(map[string]kubernetes.ClusterInterface),
+		adminConfig: adminConfig,
+	}
+}
+
+// getKcpClient returns a kcp client (i.e. a client that implements kcpclient.ClusterInterface) for the given shard.
+// the returned client establishes a direct connection with the shard with credentials stored in adminConfig.
+// clients are cached per shard to prevent connection leaks.
+func (p *clientPool) getKcpClient(shardName, baseURL string) (kcpclientset.ClusterInterface, error) {
+	p.mu.RLock()
+	if client, exists := p.kcpClients[shardName]; exists {
+		p.mu.RUnlock()
+		return client, nil
+	}
+	p.mu.RUnlock()
+
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if client, exists := p.kcpClients[shardName]; exists {
+		return client, nil
+	}
+
+	shardConfig := rest.CopyConfig(p.adminConfig)
+	shardConfig.Host = baseURL
+	client, err := kcpclientset.NewForConfig(shardConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create shard %q kcp client: %w", shardName, err)
+	}
+	p.kcpClients[shardName] = client
+	return client, nil
+}
+
+// getKubeClient returns a kube client (i.e. a client that implements kubernetes.ClusterInterface) for the given shard.
+// the returned client establishes a direct connection with the shard with credentials stored in adminConfig.
+// clients are cached per shard to prevent connection leaks.
+func (p *clientPool) getKubeClient(shardName, baseURL string) (kubernetes.ClusterInterface, error) {
+	p.mu.RLock()
+	if client, exists := p.kubeClients[shardName]; exists {
+		p.mu.RUnlock()
+		return client, nil
+	}
+	p.mu.RUnlock()
+
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	if client, exists := p.kubeClients[shardName]; exists {
+		return client, nil
+	}
+
+	shardConfig := rest.CopyConfig(p.adminConfig)
+	shardConfig.Host = baseURL
+	client, err := kubernetes.NewForConfig(shardConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create shard %q kube client: %w", shardName, err)
+	}
+	p.kubeClients[shardName] = client
+	return client, nil
+}
+
 func NewController(
 	shardName string,
 	kcpClusterClient kcpclientset.ClusterInterface,
@@ -88,6 +160,8 @@ func NewController(
 		logicalClusterIndexer: logicalClusterInformer.Informer().GetIndexer(),
 		logicalClusterLister:  logicalClusterInformer.Lister(),
 
+		clientPool: newClientPool(logicalClusterAdminConfig),
+
 		commit: committer.NewCommitter[*tenancyv1alpha1.Workspace, tenancyv1alpha1client.WorkspaceInterface, *tenancyv1alpha1.WorkspaceSpec, *tenancyv1alpha1.WorkspaceStatus](kcpClusterClient.TenancyV1alpha1().Workspaces()),
 	}
 
@@ -132,6 +206,9 @@ type Controller struct {
 	logicalClusterIndexer cache.Indexer
 	logicalClusterLister  corev1alpha1listers.LogicalClusterClusterLister
 
+	// clientPool manages reusable clients to prevent connection leaks
+	clientPool *clientPool
+
 	// commit creates a patch and submits it, if needed.
 	commit func(ctx context.Context, old, new *workspaceResource) error
 }

pkg/reconciler/tenancy/workspace/workspace_reconcile.go

@@ -18,13 +18,11 @@ package workspace
 
 import (
 	"context"
-	"fmt"
 	"time"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilserrors "k8s.io/apimachinery/pkg/util/errors"
-	restclient "k8s.io/client-go/rest"
 
 	kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache"
 	"github.com/kcp-dev/client-go/kubernetes"
@@ -61,36 +59,18 @@ func (c *Controller) reconcile(ctx context.Context, ws *tenancyv1alpha1.Workspac
 		return shards[0].(*corev1alpha1.Shard), nil
 	}
 
-	// kcpLogicalClusterAdminClientFor returns a kcp client (i.e. a client that implements kcpclient.ClusterInterface) for the given shard.
-	// the returned client establishes a direct connection with the shard with credentials stored in r.logicalClusterAdminConfig.
-	// TODO:(p0lyn0mial): make it more efficient, maybe we need a per shard client pool or we could use an HTTPRoundTripper
 	kcpDirectClientFor := func(shard *corev1alpha1.Shard) (kcpclientset.ClusterInterface, error) {
 		if shard.Name == c.shardName {
 			return c.kcpClusterClient, nil
 		}
-		shardConfig := restclient.CopyConfig(c.logicalClusterAdminConfig)
-		shardConfig.Host = shard.Spec.BaseURL
-		shardClient, err := kcpclientset.NewForConfig(shardConfig)
-		if err != nil {
-			return nil, fmt.Errorf("failed to create shard %q kube client: %w", shard.Name, err)
-		}
-		return shardClient, nil
+		return c.clientPool.getKcpClient(shard.Name, shard.Spec.BaseURL)
 	}
 
-	// kubeLogicalClusterAdminClientFor returns a kube client (i.e. a client that implements kubernetes.ClusterInterface) for the given shard.
-	// the returned client establishes a direct connection with the shard with credentials stored in r.logicalClusterAdminConfig.
-	// TODO:(p0lyn0mial): make it more efficient, maybe we need a per shard client pool or we could use an HTTPRoundTripper
 	kubeDirectClientFor := func(shard *corev1alpha1.Shard) (kubernetes.ClusterInterface, error) {
 		if shard.Name == c.shardName {
 			return c.kubeClusterClient, nil
 		}
-		shardConfig := restclient.CopyConfig(c.logicalClusterAdminConfig)
-		shardConfig.Host = shard.Spec.BaseURL
-		shardClient, err := kubernetes.NewForConfig(shardConfig)
-		if err != nil {
-			return nil, fmt.Errorf("failed to create shard %q kube client: %w", shard.Name, err)
-		}
-		return shardClient, nil
+		return c.clientPool.getKubeClient(shard.Name, shard.Spec.BaseURL)
 	}
 
 	getType := func(path logicalcluster.Path, name string) (*tenancyv1alpha1.WorkspaceType, error) {