UPSTREAM: <carry>: make RESTMapper in client cluster-aware

Signed-off-by: Dr. Stefan Schimanski <stefan.schimanski@gmail.com>

Author: sttts

examples/kcp/go.mod

@@ -53,6 +53,7 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect

examples/kcp/go.sum

@@ -89,6 +89,8 @@ github.com/grpc-ecosystem/grpc-gateway v1.16.0 h1:gmcG1KaJ57LophUzW0Hy8NmPhnMZb4
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 h1:bkypFPDjIYGfCYD5mRBvpqxfYX1YCS1PXdKYWi8FsN0=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0/go.mod h1:P+Lt/0by1T8bfcF3z737NnSbmxQAppXMRziHUxPOC8k=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
+github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/jonboulle/clockwork v0.4.0 h1:p4Cf1aMWXnXAUh8lVfewRBx1zaTSYKrKMF2g3ST4RZ4=

go.mod

@@ -32,6 +32,7 @@ require (
 )
 
 require (
+	github.com/hashicorp/golang-lru/v2 v2.0.7
 	github.com/kcp-dev/apimachinery/v2 v2.0.0
 	github.com/kcp-dev/logicalcluster/v3 v3.0.5
 )

go.sum

@@ -69,6 +69,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
 github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0 h1:bkypFPDjIYGfCYD5mRBvpqxfYX1YCS1PXdKYWi8FsN0=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.20.0/go.mod h1:P+Lt/0by1T8bfcF3z737NnSbmxQAppXMRziHUxPOC8k=
+github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
+github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=

pkg/client/client.go

@@ -23,6 +23,8 @@ import (
 	"net/http"
 	"strings"
 
+	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/kcp-dev/logicalcluster/v3"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -47,6 +49,10 @@ type Options struct {
 	// Mapper, if provided, will be used to map GroupVersionKinds to Resources
 	Mapper meta.RESTMapper
 
+	// MapperWithContext, if provided, will be used to map GroupVersionKinds to Resources.
+	// This overrides Mapper if set.
+	MapperWithContext func(context.Context) (meta.RESTMapper, error)
+
 	// Cache, if provided, is used to read objects from the cache.
 	Cache *CacheOptions
 
@@ -153,16 +159,23 @@ func newClient(config *rest.Config, options Options) (*client, error) {
 		}
 	}
 
+	// Init a MapperWithContext if none provided
+	if options.MapperWithContext == nil {
+		options.MapperWithContext = func(context.Context) (meta.RESTMapper, error) { return options.Mapper, nil }
+	}
+
 	resources := &clientRestResources{
 		httpClient: options.HTTPClient,
 		config:     config,
 		scheme:     options.Scheme,
-		mapper:     options.Mapper,
+		mapper:     options.MapperWithContext,
 		codecs:     serializer.NewCodecFactory(options.Scheme),
-
-		structuredResourceByType:   make(map[schema.GroupVersionKind]*resourceMeta),
-		unstructuredResourceByType: make(map[schema.GroupVersionKind]*resourceMeta),
 	}
+	cr, err := lru.New[logicalcluster.Path, clusterResources](1000)
+	if err != nil {
+		return nil, err
+	}
+	resources.clusterResources = cr
 
 	rawMetaClient, err := metadata.NewForConfigAndClient(metadata.ConfigFor(config), options.HTTPClient)
 	if err != nil {
@@ -180,11 +193,16 @@ func newClient(config *rest.Config, options Options) (*client, error) {
 		},
 		metadataClient: metadataClient{
 			client:     rawMetaClient,
-			restMapper: options.Mapper,
+			restMapper: options.MapperWithContext,
 		},
 		scheme: options.Scheme,
 		mapper: options.Mapper,
 	}
+	mapperCache, err := lru.New[logicalcluster.Name, meta.RESTMapper](1000)
+	if err != nil {
+		return nil, err
+	}
+	c.metadataClient.mapperCache = mapperCache
 	if options.Cache == nil || options.Cache.Reader == nil {
 		return c, nil
 	}

pkg/client/client_rest_resources.go

@@ -17,19 +17,32 @@ limitations under the License.
 package client
 
 import (
+	"context"
 	"net/http"
 	"strings"
 	"sync"
 
+	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/kcp-dev/logicalcluster/v3"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+	"sigs.k8s.io/controller-runtime/pkg/kontext"
 )
 
+type clusterResources struct {
+	mapper meta.RESTMapper
+
+	// structuredResourceByType stores structured type metadata
+	structuredResourceByType map[schema.GroupVersionKind]*resourceMeta
+	// unstructuredResourceByType stores unstructured type metadata
+	unstructuredResourceByType map[schema.GroupVersionKind]*resourceMeta
+}
+
 // clientRestResources creates and stores rest clients and metadata for Kubernetes types.
 type clientRestResources struct {
 	// httpClient is the http client to use for requests
@@ -42,21 +55,18 @@ type clientRestResources struct {
 	scheme *runtime.Scheme
 
 	// mapper maps GroupVersionKinds to Resources
-	mapper meta.RESTMapper
+	mapper func(ctx context.Context) (meta.RESTMapper, error)
 
 	// codecs are used to create a REST client for a gvk
 	codecs serializer.CodecFactory
 
-	// structuredResourceByType stores structured type metadata
-	structuredResourceByType map[schema.GroupVersionKind]*resourceMeta
-	// unstructuredResourceByType stores unstructured type metadata
-	unstructuredResourceByType map[schema.GroupVersionKind]*resourceMeta
-	mu                         sync.RWMutex
+	clusterResources *lru.Cache[logicalcluster.Path, clusterResources]
+	mu               sync.RWMutex
 }
 
 // newResource maps obj to a Kubernetes Resource and constructs a client for that Resource.
 // If the object is a list, the resource represents the item's type instead.
-func (c *clientRestResources) newResource(gvk schema.GroupVersionKind, isList, isUnstructured bool) (*resourceMeta, error) {
+func (c *clientRestResources) newResource(gvk schema.GroupVersionKind, isList, isUnstructured bool, mapper meta.RESTMapper) (*resourceMeta, error) {
 	if strings.HasSuffix(gvk.Kind, "List") && isList {
 		// if this was a list, treat it as a request for the item's resource
 		gvk.Kind = gvk.Kind[:len(gvk.Kind)-4]
@@ -66,7 +76,7 @@ func (c *clientRestResources) newResource(gvk schema.GroupVersionKind, isList, i
 	if err != nil {
 		return nil, err
 	}
-	mapping, err := c.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
+	mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
 	if err != nil {
 		return nil, err
 	}
@@ -75,7 +85,7 @@ func (c *clientRestResources) newResource(gvk schema.GroupVersionKind, isList, i
 
 // getResource returns the resource meta information for the given type of object.
 // If the object is a list, the resource represents the item's type instead.
-func (c *clientRestResources) getResource(obj runtime.Object) (*resourceMeta, error) {
+func (c *clientRestResources) getResource(ctx context.Context, obj runtime.Object) (*resourceMeta, error) {
 	gvk, err := apiutil.GVKForObject(obj, c.scheme)
 	if err != nil {
 		return nil, err
@@ -86,9 +96,25 @@ func (c *clientRestResources) getResource(obj runtime.Object) (*resourceMeta, er
 	// It's better to do creation work twice than to not let multiple
 	// people make requests at once
 	c.mu.RLock()
-	resourceByType := c.structuredResourceByType
+	cluster, _ := kontext.ClusterFrom(ctx)
+	cr, found := c.clusterResources.Get(cluster.Path())
+	if !found {
+		m, err := c.mapper(ctx)
+		if err != nil {
+			c.mu.RUnlock()
+			return nil, err
+		}
+		cr = clusterResources{
+			mapper:                     m,
+			structuredResourceByType:   make(map[schema.GroupVersionKind]*resourceMeta),
+			unstructuredResourceByType: make(map[schema.GroupVersionKind]*resourceMeta),
+		}
+		c.clusterResources.Purge()
+		c.clusterResources.Add(cluster.Path(), cr)
+	}
+	resourceByType := cr.structuredResourceByType
 	if isUnstructured {
-		resourceByType = c.unstructuredResourceByType
+		resourceByType = cr.unstructuredResourceByType
 	}
 	r, known := resourceByType[gvk]
 	c.mu.RUnlock()
@@ -100,7 +126,7 @@ func (c *clientRestResources) getResource(obj runtime.Object) (*resourceMeta, er
 	// Initialize a new Client
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	r, err = c.newResource(gvk, meta.IsListType(obj), isUnstructured)
+	r, err = c.newResource(gvk, meta.IsListType(obj), isUnstructured, cr.mapper)
 	if err != nil {
 		return nil, err
 	}
@@ -109,8 +135,8 @@ func (c *clientRestResources) getResource(obj runtime.Object) (*resourceMeta, er
 }
 
 // getObjMeta returns objMeta containing both type and object metadata and state.
-func (c *clientRestResources) getObjMeta(obj runtime.Object) (*objMeta, error) {
-	r, err := c.getResource(obj)
+func (c *clientRestResources) getObjMeta(ctx context.Context, obj runtime.Object) (*objMeta, error) {
+	r, err := c.getResource(ctx, obj)
 	if err != nil {
 		return nil, err
 	}

pkg/client/metadata_client.go

@@ -20,11 +20,15 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"sync"
 
+	lru "github.com/hashicorp/golang-lru/v2"
+	"github.com/kcp-dev/logicalcluster/v3"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/metadata"
+	"sigs.k8s.io/controller-runtime/pkg/kontext"
 )
 
 // TODO(directxman12): we could rewrite this on top of the low-level REST
@@ -34,12 +38,28 @@ import (
 
 // metadataClient is a client that reads & writes metadata-only requests to/from the API server.
 type metadataClient struct {
-	client     metadata.Interface
-	restMapper meta.RESTMapper
+	client      metadata.Interface
+	restMapper  func(ctx context.Context) (meta.RESTMapper, error)
+	mu          sync.Mutex
+	mapperCache *lru.Cache[logicalcluster.Name, meta.RESTMapper]
 }
 
-func (mc *metadataClient) getResourceInterface(gvk schema.GroupVersionKind, ns string) (metadata.ResourceInterface, error) {
-	mapping, err := mc.restMapper.RESTMapping(gvk.GroupKind(), gvk.Version)
+func (mc *metadataClient) getResourceInterface(ctx context.Context, gvk schema.GroupVersionKind, ns string) (metadata.ResourceInterface, error) {
+	cluster, _ := kontext.ClusterFrom(ctx)
+	mc.mu.Lock()
+	mapper, _ := mc.mapperCache.Get(cluster)
+	if mapper == nil {
+		var err error
+		mapper, err = mc.restMapper(ctx)
+		if err != nil {
+			mc.mu.Unlock()
+			return nil, err
+		}
+		mc.mapperCache.Add(cluster, mapper)
+	}
+	mc.mu.Unlock()
+
+	mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
 	if err != nil {
 		return nil, err
 	}
@@ -56,7 +76,7 @@ func (mc *metadataClient) Delete(ctx context.Context, obj Object, opts ...Delete
 		return fmt.Errorf("metadata client did not understand object: %T", obj)
 	}
 
-	resInt, err := mc.getResourceInterface(metadata.GroupVersionKind(), metadata.Namespace)
+	resInt, err := mc.getResourceInterface(ctx, metadata.GroupVersionKind(), metadata.Namespace)
 	if err != nil {
 		return err
 	}
@@ -77,7 +97,7 @@ func (mc *metadataClient) DeleteAllOf(ctx context.Context, obj Object, opts ...D
 	deleteAllOfOpts := DeleteAllOfOptions{}
 	deleteAllOfOpts.ApplyOptions(opts)
 
-	resInt, err := mc.getResourceInterface(metadata.GroupVersionKind(), deleteAllOfOpts.ListOptions.Namespace)
+	resInt, err := mc.getResourceInterface(ctx, metadata.GroupVersionKind(), deleteAllOfOpts.ListOptions.Namespace)
 	if err != nil {
 		return err
 	}
@@ -93,7 +113,7 @@ func (mc *metadataClient) Patch(ctx context.Context, obj Object, patch Patch, op
 	}
 
 	gvk := metadata.GroupVersionKind()
-	resInt, err := mc.getResourceInterface(gvk, metadata.Namespace)
+	resInt, err := mc.getResourceInterface(ctx, gvk, metadata.Namespace)
 	if err != nil {
 		return err
 	}
@@ -127,7 +147,7 @@ func (mc *metadataClient) Get(ctx context.Context, key ObjectKey, obj Object, op
 	getOpts := GetOptions{}
 	getOpts.ApplyOptions(opts)
 
-	resInt, err := mc.getResourceInterface(gvk, key.Namespace)
+	resInt, err := mc.getResourceInterface(ctx, gvk, key.Namespace)
 	if err != nil {
 		return err
 	}
@@ -154,7 +174,7 @@ func (mc *metadataClient) List(ctx context.Context, obj ObjectList, opts ...List
 	listOpts := ListOptions{}
 	listOpts.ApplyOptions(opts)
 
-	resInt, err := mc.getResourceInterface(gvk, listOpts.Namespace)
+	resInt, err := mc.getResourceInterface(ctx, gvk, listOpts.Namespace)
 	if err != nil {
 		return err
 	}
@@ -175,7 +195,7 @@ func (mc *metadataClient) PatchSubResource(ctx context.Context, obj Object, subR
 	}
 
 	gvk := metadata.GroupVersionKind()
-	resInt, err := mc.getResourceInterface(gvk, metadata.Namespace)
+	resInt, err := mc.getResourceInterface(ctx, gvk, metadata.Namespace)
 	if err != nil {
 		return err
 	}