Merge pull request #57 from sttts/sttts-cluster-aware-rest-mapper

Make RESTMapper in client cluster-aware

Author: kcp-ci-bot

.golangci.yml

@@ -13,7 +13,6 @@ linters:
     - exhaustive
     - exportloopref
     - ginkgolinter
-    - goconst
     - gocritic
     - gocyclo
     - gofmt

examples/kcp/go.mod

@@ -50,6 +50,7 @@ require (
 	github.com/google/uuid v1.3.1 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect
+	github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
 	github.com/imdario/mergo v0.3.13 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect

examples/kcp/go.sum

@@ -96,6 +96,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.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/imdario/mergo v0.3.13 h1:lFzP57bqS/wsqKssCGmtLAb8A0wKjLGrve2q3PPVcBk=
 github.com/imdario/mergo v0.3.13/go.mod h1:4lJ1jqUDcsbIECGy0RUJAXNIhg+6ocWgb1ALK2O4oXg=
 github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8=

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

@@ -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
 
@@ -56,6 +62,10 @@ type Options struct {
 
 	// DryRun instructs the client to only perform dry run requests.
 	DryRun *bool
+
+	// KcpClusterDiscoveryCacheSize is the size of the cache for cluster discovery
+	// information backing the client's REST mapper.
+	KcpClusterDiscoveryCacheSize int
 }
 
 // WarningHandlerOptions are options for configuring a
@@ -170,16 +180,27 @@ func newClient(config *rest.Config, options Options) (*client, error) {
 		}
 	}
 
+	if options.KcpClusterDiscoveryCacheSize == 0 {
+		options.KcpClusterDiscoveryCacheSize = 1000
+	}
+
+	// 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](options.KcpClusterDiscoveryCacheSize)
+	if err != nil {
+		return nil, err
+	}
+	resources.clusterResources = cr
 
 	rawMetaClient, err := metadata.NewForConfigAndClient(metadata.ConfigFor(config), options.HTTPClient)
 	if err != nil {
@@ -197,11 +218,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](options.KcpClusterDiscoveryCacheSize)
+	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
 	}