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

Signed-off-by: Dr. Stefan Schimanski <[email protected]>

Author: sttts

go.mod

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

go.sum

@@ -65,6 +65,8 @@ github.com/google/uuid v1.3.1 h1:KjJaJ9iWZ3jOFZIf1Lqf4laDRCasjl0BCmnEGxkdLb4=
 github.com/google/uuid v1.3.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 h1:YBftPWNWd4WwGqtY2yeZL2ef8rHAxPBD8KFhJpmcqms=
 github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0/go.mod h1:YN5jB8ie0yfIUg6VvR9Kz84aCaG7AsGZnLjhHbUqwPg=
+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/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/imdario/mergo v0.3.13 h1:lFzP57bqS/wsqKssCGmtLAb8A0wKjLGrve2q3PPVcBk=
 github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg=

pkg/client/client.go

@@ -22,7 +22,10 @@ import (
 	"fmt"
 	"net/http"
 	"strings"
+	"time"
 
+	"github.com/hashicorp/golang-lru/v2/expirable"
+	"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 +50,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
 
@@ -170,15 +177,19 @@ 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),
+		clusterResources: expirable.NewLRU[logicalcluster.Path, clusterResources](1000, nil, 10*time.Minute),
 	}
 
 	rawMetaClient, err := metadata.NewForConfigAndClient(metadata.ConfigFor(config), options.HTTPClient)
@@ -197,7 +208,7 @@ 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,

pkg/client/client_rest_resources.go

@@ -17,19 +17,32 @@ limitations under the License.
 package client
 
 import (
+	"context"
 	"net/http"
 	"strings"
 	"sync"
 
+	"github.com/hashicorp/golang-lru/v2/expirable"
+	"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 *expirable.LRU[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(ctx context.Context, gvk schema.GroupVersionKind, isList, isUnstructured bool) (*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,11 @@ func (c *clientRestResources) newResource(gvk schema.GroupVersionKind, isList, i
 	if err != nil {
 		return nil, err
 	}
-	mapping, err := c.mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
+	mapper, err := c.mapper(ctx)
+	if err != nil {
+		return nil, err
+	}
+	mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
 	if err != nil {
 		return nil, err
 	}
@@ -75,7 +89,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 +100,24 @@ 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.Add(cluster.Path(), cr)
+	}
+	resourceByType := cr.structuredResourceByType
 	if isUnstructured {
-		resourceByType = c.unstructuredResourceByType
+		resourceByType = cr.unstructuredResourceByType
 	}
 	r, known := resourceByType[gvk]
 	c.mu.RUnlock()
@@ -100,7 +129,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(ctx, gvk, meta.IsListType(obj), isUnstructured)
 	if err != nil {
 		return nil, err
 	}
@@ -109,8 +138,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

@@ -35,11 +35,15 @@ 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
+	restMapper func(ctx context.Context) (meta.RESTMapper, error)
 }
 
-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) {
+	mapper, err := mc.restMapper(ctx)
+	if err != nil {
+		return nil, err
+	}
+	mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
 	if err != nil {
 		return nil, err
 	}
@@ -56,7 +60,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 +81,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 +97,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 +131,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 +158,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 +179,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
 	}

pkg/client/typed_client.go

@@ -32,7 +32,7 @@ type typedClient struct {
 
 // Create implements client.Client.
 func (c *typedClient) Create(ctx context.Context, obj Object, opts ...CreateOption) error {
-	o, err := c.resources.getObjMeta(obj)
+	o, err := c.resources.getObjMeta(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -51,7 +51,7 @@ func (c *typedClient) Create(ctx context.Context, obj Object, opts ...CreateOpti
 
 // Update implements client.Client.
 func (c *typedClient) Update(ctx context.Context, obj Object, opts ...UpdateOption) error {
-	o, err := c.resources.getObjMeta(obj)
+	o, err := c.resources.getObjMeta(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -71,7 +71,7 @@ func (c *typedClient) Update(ctx context.Context, obj Object, opts ...UpdateOpti
 
 // Delete implements client.Client.
 func (c *typedClient) Delete(ctx context.Context, obj Object, opts ...DeleteOption) error {
-	o, err := c.resources.getObjMeta(obj)
+	o, err := c.resources.getObjMeta(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -90,7 +90,7 @@ func (c *typedClient) Delete(ctx context.Context, obj Object, opts ...DeleteOpti
 
 // DeleteAllOf implements client.Client.
 func (c *typedClient) DeleteAllOf(ctx context.Context, obj Object, opts ...DeleteAllOfOption) error {
-	o, err := c.resources.getObjMeta(obj)
+	o, err := c.resources.getObjMeta(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -109,7 +109,7 @@ func (c *typedClient) DeleteAllOf(ctx context.Context, obj Object, opts ...Delet
 
 // Patch implements client.Client.
 func (c *typedClient) Patch(ctx context.Context, obj Object, patch Patch, opts ...PatchOption) error {
-	o, err := c.resources.getObjMeta(obj)
+	o, err := c.resources.getObjMeta(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -134,7 +134,7 @@ func (c *typedClient) Patch(ctx context.Context, obj Object, patch Patch, opts .
 
 // Get implements client.Client.
 func (c *typedClient) Get(ctx context.Context, key ObjectKey, obj Object, opts ...GetOption) error {
-	r, err := c.resources.getResource(obj)
+	r, err := c.resources.getResource(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -149,7 +149,7 @@ func (c *typedClient) Get(ctx context.Context, key ObjectKey, obj Object, opts .
 
 // List implements client.Client.
 func (c *typedClient) List(ctx context.Context, obj ObjectList, opts ...ListOption) error {
-	r, err := c.resources.getResource(obj)
+	r, err := c.resources.getResource(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -166,7 +166,7 @@ func (c *typedClient) List(ctx context.Context, obj ObjectList, opts ...ListOpti
 }
 
 func (c *typedClient) GetSubResource(ctx context.Context, obj, subResourceObj Object, subResource string, opts ...SubResourceGetOption) error {
-	o, err := c.resources.getObjMeta(obj)
+	o, err := c.resources.getObjMeta(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -189,7 +189,7 @@ func (c *typedClient) GetSubResource(ctx context.Context, obj, subResourceObj Ob
 }
 
 func (c *typedClient) CreateSubResource(ctx context.Context, obj Object, subResourceObj Object, subResource string, opts ...SubResourceCreateOption) error {
-	o, err := c.resources.getObjMeta(obj)
+	o, err := c.resources.getObjMeta(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -214,7 +214,7 @@ func (c *typedClient) CreateSubResource(ctx context.Context, obj Object, subReso
 
 // UpdateSubResource used by SubResourceWriter to write status.
 func (c *typedClient) UpdateSubResource(ctx context.Context, obj Object, subResource string, opts ...SubResourceUpdateOption) error {
-	o, err := c.resources.getObjMeta(obj)
+	o, err := c.resources.getObjMeta(ctx, obj)
 	if err != nil {
 		return err
 	}
@@ -249,7 +249,7 @@ func (c *typedClient) UpdateSubResource(ctx context.Context, obj Object, subReso
 
 // PatchSubResource used by SubResourceWriter to write subresource.
 func (c *typedClient) PatchSubResource(ctx context.Context, obj Object, subResource string, patch Patch, opts ...SubResourcePatchOption) error {
-	o, err := c.resources.getObjMeta(obj)
+	o, err := c.resources.getObjMeta(ctx, obj)
 	if err != nil {
 		return err
 	}