DynamicRESTMapper: add mapping for core k8s and kcp types

The mapper is now split into two controllers:
* DynamicTypesController: manages dynamic, cluster-local types added through bindings or local CRDs
* BuiltinTypesController: manages core k8s and kcp types, common to all clusters

On-behalf-of: @SAP [email protected]
Signed-off-by: Robert Vasek <[email protected]>

Author: gman0

pkg/reconciler/dynamicrestmapper/defaultrestmapper_mutable.go

@@ -17,6 +17,8 @@ limitations under the License.
 package dynamicrestmapper
 
 import (
+	"slices"
+
 	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
@@ -41,6 +43,24 @@ func (m *DefaultRESTMapper) add(typeMeta typeMeta) {
 
 	m.kindToPluralResource[kind] = plural
 	m.kindToScope[kind] = meta.RESTScopeRoot
+
+	foundDefaultVersion := false
+	for i := range m.defaultGroupVersions {
+		if m.defaultGroupVersions[i].Group == typeMeta.Group {
+			if typeMeta.Version > m.defaultGroupVersions[i].Version {
+				m.defaultGroupVersions[i].Version = typeMeta.Version
+			}
+			foundDefaultVersion = true
+			break
+		}
+	}
+
+	if !foundDefaultVersion {
+		m.defaultGroupVersions = append(m.defaultGroupVersions, schema.GroupVersion{
+			Group:   typeMeta.Group,
+			Version: typeMeta.Version,
+		})
+	}
 }
 
 func (m *DefaultRESTMapper) remove(typeMeta typeMeta) {
@@ -56,6 +76,37 @@ func (m *DefaultRESTMapper) remove(typeMeta typeMeta) {
 
 	delete(m.kindToPluralResource, kind)
 	delete(m.kindToScope, kind)
+
+	versionIdx := slices.IndexFunc(m.defaultGroupVersions, func(gv schema.GroupVersion) bool {
+		return gv.Group == typeMeta.Group
+	})
+	if versionIdx < 0 {
+		return
+	}
+
+	// Fixup the default version too.
+
+	// First, try to find resources with an older version in this group.
+	// In case there are any, we'll need to "downgrade", i.e. set the default
+	// group version to that.
+
+	latestGroupVersion := ""
+	for gvr := range m.pluralToSingular {
+		if gvr.Group != typeMeta.Group {
+			continue
+		}
+		if gvr.Version > latestGroupVersion {
+			latestGroupVersion = gvr.Version
+		}
+	}
+
+	if latestGroupVersion != "" {
+		m.defaultGroupVersions[versionIdx].Version = latestGroupVersion
+	} else {
+		// There are no more resources in this group.
+		// Delete the default version too.
+		m.defaultGroupVersions = slices.Delete(m.defaultGroupVersions, versionIdx, versionIdx+1)
+	}
 }
 
 func (m *DefaultRESTMapper) getGVKR(gvr schema.GroupVersionResource) typeMeta {

pkg/reconciler/dynamicrestmapper/dynamicrestmapper.go

@@ -19,38 +19,39 @@ package dynamicrestmapper
 import (
 	"sync"
 
-	"k8s.io/apimachinery/pkg/runtime/schema"
-
 	"github.com/kcp-dev/logicalcluster/v3"
 )
 
 // DynamicRESTMapper is a thread-safe RESTMapper with per-cluster GVK/GVR mappings.
 // The mapping data is fed from its associated Controller (in this package), triggered
-// on CRDs and APIBindings (and their respective APIResourceSchemas). Additionally,
-// mappings for built-in types (pkg/virtual/apiexport/schemas/builtin/builtin.go) are
-// added for each LogicalCluster by default.
+// on CRDs and APIBindings (and their respective APIResourceSchemas).
 type DynamicRESTMapper struct {
-	lock      sync.RWMutex
-	byCluster map[logicalcluster.Name]*DefaultRESTMapper
+	lock sync.RWMutex
+	// Built-in types consist of core k8s types and system CRDs.
+	// They are present in all clusters.
+	builtin *DefaultRESTMapper
+	// Dynamic types consist of bound types and CRDs local to the cluster.
+	dynamic map[logicalcluster.Name]*DefaultRESTMapper
 }
 
-func NewDynamicRESTMapper(defaultGroupVersions []schema.GroupVersion) *DynamicRESTMapper {
+func NewDynamicRESTMapper() *DynamicRESTMapper {
 	return &DynamicRESTMapper{
-		byCluster: make(map[logicalcluster.Name]*DefaultRESTMapper),
+		builtin: NewDefaultRESTMapper(nil),
+		dynamic: make(map[logicalcluster.Name]*DefaultRESTMapper),
 	}
 }
 
 func (d *DynamicRESTMapper) deleteMappingsForCluster(clusterName logicalcluster.Name) {
 	d.lock.Lock()
 	defer d.lock.Unlock()
-	delete(d.byCluster, clusterName)
+	delete(d.dynamic, clusterName)
 }
 
 // ForCluster returns a RESTMapper for the specified cluster name.
 // The method never returns nil. If the cluster doesn't exist at the time
 // of calling ForCluster, or if it is deleted while holding the returned
-// RESTMapper instance, all RESTMapper's methods will empty matches and
-// NoResourceMatchError error.
+// RESTMapper instance, all RESTMapper's methods will return empty matches
+// and a NoResourceMatchError error.
 func (d *DynamicRESTMapper) ForCluster(clusterName logicalcluster.Name) *ForCluster {
 	return newForCluster(clusterName, d)
 }

pkg/reconciler/dynamicrestmapper/dynamicrestmapper_builtin_controller.go

@@ -0,0 +1,235 @@
+/*
+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 dynamicrestmapper
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/go-logr/logr"
+
+	apiextensionshelpers "k8s.io/apiextensions-apiserver/pkg/apihelpers"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/klog/v2"
+
+	kcpapiextensionsv1informers "github.com/kcp-dev/client-go/apiextensions/informers/apiextensions/v1"
+	"github.com/kcp-dev/logicalcluster/v3"
+
+	"github.com/kcp-dev/kcp/pkg/logging"
+	"github.com/kcp-dev/kcp/pkg/tombstone"
+	builtinschemas "github.com/kcp-dev/kcp/pkg/virtual/apiexport/schemas/builtin"
+)
+
+const (
+	BuiltinTypesControllerName = "kcp-dynamicrestmapper-builtin"
+)
+
+var systemCRDClusterName = logicalcluster.Name("system:system-crds")
+
+type BuiltinTypesController struct {
+	queue workqueue.TypedRateLimitingInterface[string]
+
+	state         *DynamicRESTMapper
+	groupVersions map[string]string
+
+	getCRD func(clusterName logicalcluster.Name, name string) (*apiextensionsv1.CustomResourceDefinition, error)
+}
+
+func NewBuiltinTypesController(
+	ctx context.Context,
+	state *DynamicRESTMapper,
+	crdInformer kcpapiextensionsv1informers.CustomResourceDefinitionClusterInformer,
+) (*BuiltinTypesController, error) {
+	c := &BuiltinTypesController{
+		state: state,
+		queue: workqueue.NewTypedRateLimitingQueueWithConfig(
+			workqueue.DefaultTypedControllerRateLimiter[string](),
+			workqueue.TypedRateLimitingQueueConfig[string]{
+				Name: BuiltinTypesControllerName,
+			},
+		),
+		groupVersions: make(map[string]string),
+		getCRD: func(clusterName logicalcluster.Name, name string) (*apiextensionsv1.CustomResourceDefinition, error) {
+			return crdInformer.Lister().Cluster(clusterName).Get(name)
+		},
+	}
+
+	// Populate the builtin RESTMapper with core types.
+	// We assume those never change, so we add mapping for them only during initialization.
+
+	for i := range builtinschemas.BuiltInAPIs {
+		group := builtinschemas.BuiltInAPIs[i].GroupVersion.Group
+		version := builtinschemas.BuiltInAPIs[i].GroupVersion.Version
+		if version > c.groupVersions[group] {
+			c.groupVersions[group] = version
+		}
+		c.state.builtin.add(newTypeMeta(
+			builtinschemas.BuiltInAPIs[i].GroupVersion.Group,
+			builtinschemas.BuiltInAPIs[i].GroupVersion.Version,
+			builtinschemas.BuiltInAPIs[i].Names.Kind,
+			builtinschemas.BuiltInAPIs[i].Names.Singular,
+			builtinschemas.BuiltInAPIs[i].Names.Plural,
+			resourceScopeToRESTScope(builtinschemas.BuiltInAPIs[i].ResourceScope)),
+		)
+	}
+
+	logger := logging.WithReconciler(klog.Background(), BuiltinTypesControllerName)
+
+	// System CRDs could change over time, and so we build that part of the mapper dynamically.
+
+	// We are only interested in system CRDs.
+	_, _ = crdInformer.Informer().Cluster("system:system-crds").AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			c.enqueueCRD(tombstone.Obj[*apiextensionsv1.CustomResourceDefinition](obj), logger)
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			c.enqueueCRD(tombstone.Obj[*apiextensionsv1.CustomResourceDefinition](newObj), logger)
+		},
+		DeleteFunc: func(obj interface{}) {
+			c.enqueueCRD(tombstone.Obj[*apiextensionsv1.CustomResourceDefinition](obj), logger)
+		},
+	})
+
+	return c, nil
+}
+
+func (c *BuiltinTypesController) enqueueCRD(crd *apiextensionsv1.CustomResourceDefinition, logger logr.Logger) {
+	if !apiextensionshelpers.IsCRDConditionTrue(crd, apiextensionsv1.Established) {
+		// The CRD is not ready yet. Nothing to do, we'll get notified on the next update event.
+		return
+	}
+
+	key := fmt.Sprintf("%s.%s/%s", crd.Spec.Group, crd.Spec.Names.Plural, crd.Name)
+
+	logger.V(4).Info("queueing system CRD")
+	c.queue.Add(key)
+}
+
+func (c *BuiltinTypesController) Start(ctx context.Context, numThreads int) {
+	defer utilruntime.HandleCrash()
+	defer c.queue.ShutDown()
+
+	logger := logging.WithReconciler(klog.FromContext(ctx), BuiltinTypesControllerName)
+	ctx = klog.NewContext(ctx, logger)
+	logger.Info("Starting controller")
+	defer logger.Info("Shutting down controller")
+
+	for range numThreads {
+		go wait.UntilWithContext(ctx, c.startWorker, time.Second)
+	}
+
+	<-ctx.Done()
+}
+
+func (c *BuiltinTypesController) startWorker(ctx context.Context) {
+	for c.processNextWorkItem(ctx) {
+	}
+}
+
+func (c *BuiltinTypesController) processNextWorkItem(ctx context.Context) bool {
+	// Wait until there is a new item in the working queue
+	key, quit := c.queue.Get()
+	if quit {
+		return false
+	}
+
+	logger := logging.WithQueueKey(klog.FromContext(ctx), key)
+	ctx = klog.NewContext(ctx, logger)
+	logger.Info("processing key")
+
+	// No matter what, tell the queue we're done with this key, to unblock
+	// other workers.
+	defer c.queue.Done(key)
+
+	if err := c.process(ctx, key); err != nil {
+		utilruntime.HandleError(fmt.Errorf("%q controller failed to sync %q, err: %w", DynamicTypesControllerName, key, err))
+		c.queue.AddRateLimited(key)
+		return true
+	}
+	c.queue.Forget(key)
+	return true
+}
+
+func (c *BuiltinTypesController) gatherGVKRsForCRD(crd *apiextensionsv1.CustomResourceDefinition) []typeMeta {
+	if crd == nil {
+		return nil
+	}
+	gvkrs := make([]typeMeta, 0, len(crd.Spec.Versions))
+	for _, version := range crd.Spec.Versions {
+		if !version.Served {
+			continue
+		}
+
+		gvkrs = append(gvkrs, newTypeMeta(
+			crd.Spec.Group,
+			version.Name,
+			crd.Status.AcceptedNames.Kind,
+			crd.Status.AcceptedNames.Singular,
+			crd.Status.AcceptedNames.Plural,
+			resourceScopeToRESTScope(crd.Spec.Scope),
+		))
+	}
+	return gvkrs
+}
+
+func (c *BuiltinTypesController) gatherGVKRsForMappedGroupResource(gr schema.GroupResource) ([]typeMeta, error) {
+	gvkrs, err := c.state.builtin.getGVKRs(gr)
+	if err != nil {
+		if meta.IsNoMatchError(err) {
+			return nil, nil
+		}
+		return nil, err
+	}
+	return gvkrs, nil
+}
+
+func (c *BuiltinTypesController) process(ctx context.Context, key string) error {
+	logger := logging.WithQueueKey(klog.FromContext(ctx), key)
+
+	parts := strings.Split(key, "/")
+	gr := schema.ParseGroupResource(parts[0])
+	crdName := parts[1]
+
+	crd, err := c.getCRD(systemCRDClusterName, crdName)
+	if err != nil && apierrors.IsNotFound(err) {
+		return err
+	}
+
+	// Remove and add the mapping -- this way we can refresh any existing mappings.
+
+	typeMetaToRemove, err := c.gatherGVKRsForMappedGroupResource(gr)
+	if err != nil {
+		return err
+	}
+	typeMetaToAdd := c.gatherGVKRsForCRD(crd)
+
+	logger.V(4).Info("applying mappings")
+
+	c.state.lock.Lock()
+	defer c.state.lock.Unlock()
+	c.state.builtin.apply(typeMetaToRemove, typeMetaToAdd)
+	return nil
+}