wip: add handling for APIBindings

Author: gman0

pkg/dynamicrestmapper/dynamicrestmapper_controller.go

@@ -20,14 +20,8 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"sync"
 	"time"
 
-	kcpapiextensionsclientset "github.com/kcp-dev/client-go/apiextensions/client"
-	kcpapiextensionsv1informers "github.com/kcp-dev/client-go/apiextensions/informers/apiextensions/v1"
-
-	kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache"
-
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/meta"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -36,7 +30,15 @@ import (
 	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
+	kcpcache "github.com/kcp-dev/apimachinery/v2/pkg/cache"
+	kcpapiextensionsv1informers "github.com/kcp-dev/client-go/apiextensions/informers/apiextensions/v1"
+	"github.com/kcp-dev/logicalcluster/v3"
+
+	"github.com/kcp-dev/kcp/pkg/indexers"
+	"github.com/kcp-dev/kcp/pkg/informer"
 	"github.com/kcp-dev/kcp/pkg/logging"
+	apisv1alpha1 "github.com/kcp-dev/kcp/sdk/apis/apis/v1alpha1"
+	apisv1alpha1informers "github.com/kcp-dev/kcp/sdk/client/informers/externalversions/apis/v1alpha1"
 )
 
 const (
@@ -46,8 +48,12 @@ const (
 func NewController(
 	ctx context.Context,
 	state *DynamicRESTMapper,
-	crdClusterClient *kcpapiextensionsclientset.ClusterClientset,
 	crdInformer kcpapiextensionsv1informers.CustomResourceDefinitionClusterInformer,
+	apiBindingInformer apisv1alpha1informers.APIBindingClusterInformer,
+	apiExportInformer apisv1alpha1informers.APIExportClusterInformer,
+	apiResourceSchemaInformer apisv1alpha1informers.APIResourceSchemaClusterInformer,
+	globalAPIExportInformer apisv1alpha1informers.APIExportClusterInformer,
+	globalAPIResourceSchemaInformer apisv1alpha1informers.APIResourceSchemaClusterInformer,
 ) (*Controller, error) {
 	queue := workqueue.NewTypedRateLimitingQueueWithConfig(
 		workqueue.DefaultTypedControllerRateLimiter[string](),
@@ -58,10 +64,19 @@ func NewController(
 
 	c := &Controller{
 		queue: queue,
+		state: state,
 
-		crdInformer: crdInformer,
+		getAPIExportByPath: func(path logicalcluster.Path, name string) (*apisv1alpha1.APIExport, error) {
+			return indexers.ByPathAndNameWithFallback[*apisv1alpha1.APIExport](
+				apisv1alpha1.Resource("apiexports"),
+				apiExportInformer.Informer().GetIndexer(),
+				globalAPIExportInformer.Informer().GetIndexer(),
+				path,
+				name,
+			)
+		},
 
-		state: state,
+		getAPIResourceSchema: informer.NewScopedGetterWithFallback(apiResourceSchemaInformer.Lister(), globalAPIResourceSchemaInformer.Lister()),
 	}
 
 	_, _ = crdInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
@@ -76,6 +91,18 @@ func NewController(
 		},
 	})
 
+	_, _ = apiBindingInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			c.enqueueAPIBinding(obj.(*apisv1alpha1.APIBinding), false)
+		},
+		DeleteFunc: func(obj interface{}) {
+			if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok {
+				obj = tombstone.Obj
+			}
+			c.enqueueAPIBinding(obj.(*apisv1alpha1.APIBinding), true)
+		},
+	})
+
 	return c, nil
 }
 
@@ -84,12 +111,11 @@ func NewController(
 // updates the workspace index, which maps logical clusters to shard URLs.
 type Controller struct {
 	queue workqueue.TypedRateLimitingInterface[string]
+	state *DynamicRESTMapper
 
-	crdInformer kcpapiextensionsv1informers.CustomResourceDefinitionClusterInformer
-
-	lock sync.RWMutex
+	getAPIExportByPath func(path logicalcluster.Path, name string) (*apisv1alpha1.APIExport, error)
 
-	state *DynamicRESTMapper
+	getAPIResourceSchema func(clusterName logicalcluster.Name, name string) (*apisv1alpha1.APIResourceSchema, error)
 }
 
 type gvkrKey struct {
@@ -138,6 +164,56 @@ func (c *Controller) enqueueCRD(crd *apiextensionsv1.CustomResourceDefinition, d
 	}
 }
 
+func (c *Controller) enqueueAPIBinding(apiBinding *apisv1alpha1.APIBinding, deleted bool) {
+	key, err := kcpcache.DeletionHandlingMetaClusterNamespaceKeyFunc(apiBinding)
+	if err != nil {
+		utilruntime.HandleError(err)
+		return
+	}
+
+	apiExportPath := logicalcluster.NewPath(apiBinding.Spec.Reference.Export.Path)
+	if apiExportPath.Empty() {
+		apiExportPath = logicalcluster.From(apiBinding).Path()
+	}
+
+	apiExport, err := c.getAPIExportByPath(apiExportPath, apiBinding.Spec.Reference.Export.Name)
+	if err != nil {
+		utilruntime.HandleError(err)
+		return
+	}
+
+	for _, schemaName := range apiExport.Spec.LatestResourceSchemas {
+		sch, err := c.getAPIResourceSchema(logicalcluster.From(apiExport), schemaName)
+		if err != nil {
+			fmt.Printf("\n\n!!! X key=%s,apiExportPath=%s,logicalcluster.From(apiExport)=%s\n\n", key, apiExportPath, logicalcluster.From(apiExport))
+			utilruntime.HandleError(err)
+			return
+		}
+
+		for _, schVersion := range sch.Spec.Versions {
+			encodedGvkKey, err := encodeGvkrKey(
+				gvkrKey{
+					Gvkr: gvkr{
+						Group:            sch.Spec.Group,
+						Version:          schVersion.Name,
+						Kind:             sch.Spec.Names.Kind,
+						ResourcePlural:   sch.Spec.Names.Plural,
+						ResourceSingular: sch.Spec.Names.Singular,
+					},
+					Key:     key,
+					Deleted: deleted,
+				},
+			)
+			if err != nil {
+				utilruntime.HandleError(fmt.Errorf("%q controller failed encode APIResourceSchema object with key %q, err: %w", ControllerName, key, err))
+			}
+			logging.WithQueueKey(logging.WithReconciler(klog.Background(), ControllerName), encodedGvkKey).V(4).
+				Info("queueing APIResourceSchema because of APIBinding")
+			c.queue.Add(encodedGvkKey)
+		}
+	}
+}
+
 func (c *Controller) Start(ctx context.Context, numThreads int) {
 	defer utilruntime.HandleCrash()
 	defer c.queue.ShutDown()

pkg/server/controllers.go

@@ -1599,13 +1599,14 @@ func (s *Server) installDynamicRESTMapper(ctx context.Context, config *rest.Conf
 	config = rest.CopyConfig(config)
 	config = rest.AddUserAgent(config, dynamicrestmapper.ControllerName)
 
-	crdClusterClient, err := kcpapiextensionsclientset.NewForConfig(config)
-	if err != nil {
-		return err
-	}
-
-	c, err := dynamicrestmapper.NewController(ctx, s.DynRESTMapper, crdClusterClient,
-		s.ApiExtensionsSharedInformerFactory.Apiextensions().V1().CustomResourceDefinitions())
+	c, err := dynamicrestmapper.NewController(ctx, s.DynRESTMapper,
+		s.ApiExtensionsSharedInformerFactory.Apiextensions().V1().CustomResourceDefinitions(),
+		s.KcpSharedInformerFactory.Apis().V1alpha1().APIBindings(),
+		s.KcpSharedInformerFactory.Apis().V1alpha1().APIExports(),
+		s.KcpSharedInformerFactory.Apis().V1alpha1().APIResourceSchemas(),
+		s.CacheKcpSharedInformerFactory.Apis().V1alpha1().APIExports(),
+		s.CacheKcpSharedInformerFactory.Apis().V1alpha1().APIResourceSchemas(),
+	)
 	if err != nil {
 		return err
 	}
@@ -1614,7 +1615,15 @@ func (s *Server) installDynamicRESTMapper(ctx context.Context, config *rest.Conf
 		Name: dynamicrestmapper.ControllerName,
 		Wait: func(ctx context.Context, s *Server) error {
 			return wait.PollUntilContextCancel(ctx, waitPollInterval, true, func(ctx context.Context) (bool, error) {
-				return s.ApiExtensionsSharedInformerFactory.Apiextensions().V1().CustomResourceDefinitions().Informer().HasSynced(), nil
+				return s.ApiExtensionsSharedInformerFactory.Apiextensions().V1().CustomResourceDefinitions().Informer().HasSynced() &&
+					// APIExport
+					s.KcpSharedInformerFactory.Apis().V1alpha1().APIExports().Informer().HasSynced() &&
+					s.CacheKcpSharedInformerFactory.Apis().V1alpha1().APIExports().Informer().HasSynced() &&
+					// APIResourceSchema
+					s.KcpSharedInformerFactory.Apis().V1alpha1().APIResourceSchemas().Informer().HasSynced() &&
+					s.CacheKcpSharedInformerFactory.Apis().V1alpha1().APIResourceSchemas().Informer().HasSynced() &&
+					// APIBinding
+					s.KcpSharedInformerFactory.Apis().V1alpha1().APIBindings().Informer().HasSynced(), nil
 			})
 		},
 		Runner: func(ctx context.Context) {