Move serviceexport to controller-runtime

Author: mjudeikis

cmd/example-backend/main.go

@@ -80,7 +80,7 @@ func run(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	server, err := backend.NewServer(config)
+	server, err := backend.NewServer(ctx, config)
 	if err != nil {
 		return err
 	}

contrib/example-backend/controllers/serviceexport/serviceexport_controller.go

@@ -19,43 +19,47 @@ package serviceexport
 import (
 	"context"
 	"fmt"
-	"time"
+	"reflect"
 
 	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
-	apiextensionsinformers "k8s.io/apiextensions-apiserver/pkg/client/informers/externalversions/apiextensions/v1"
-	apiextensionslisters "k8s.io/apiextensions-apiserver/pkg/client/listers/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	utilerrors "k8s.io/apimachinery/pkg/util/errors"
-	"k8s.io/apimachinery/pkg/util/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/cache"
-	"k8s.io/client-go/util/workqueue"
-	"k8s.io/klog/v2"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
-	"github.com/kube-bind/kube-bind/pkg/committer"
 	"github.com/kube-bind/kube-bind/pkg/indexers"
 	kubebindv1alpha2 "github.com/kube-bind/kube-bind/sdk/apis/kubebind/v1alpha2"
 	bindclient "github.com/kube-bind/kube-bind/sdk/client/clientset/versioned"
-	bindinformers "github.com/kube-bind/kube-bind/sdk/client/informers/externalversions/kubebind/v1alpha2"
-	bindlisters "github.com/kube-bind/kube-bind/sdk/client/listers/kubebind/v1alpha2"
 )
 
 const (
 	controllerName = "kube-bind-example-backend-serviceexport"
 )
 
-// NewController returns a new controller to reconcile ServiceExports.
-func NewController(
-	config *rest.Config,
-	serviceExportInformer bindinformers.APIServiceExportInformer,
-	crdInformer apiextensionsinformers.CustomResourceDefinitionInformer,
-) (*Controller, error) {
-	queue := workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[string](), workqueue.TypedRateLimitingQueueConfig[string]{Name: controllerName})
+// APIServiceExportReconciler reconciles a APIServiceExport object.
+type APIServiceExportReconciler struct {
+	client.Client
+	Scheme *runtime.Scheme
 
-	logger := klog.Background().WithValues("controller", controllerName)
+	bindClient bindclient.Interface
+	reconciler reconciler
+}
 
+// NewAPIServiceExportReconciler returns a new APIServiceExportReconciler to reconcile APIServiceExports.
+func NewAPIServiceExportReconciler(
+	ctx context.Context,
+	client client.Client,
+	scheme *runtime.Scheme,
+	config *rest.Config,
+	cache cache.Cache,
+) (*APIServiceExportReconciler, error) {
 	config = rest.CopyConfig(config)
 	config = rest.AddUserAgent(config, controllerName)
 
@@ -64,219 +68,119 @@ func NewController(
 		return nil, err
 	}
 
-	c := &Controller{
-		queue: queue,
+	// Set up field indexer for APIServiceExports by CustomResourceDefinition name
+	// This replicates the behavior of the original indexer setup
+	if err := cache.IndexField(ctx, &kubebindv1alpha2.APIServiceExport{}, indexers.ServiceExportByCustomResourceDefinition,
+		indexers.IndexServiceExportByCustomResourceDefinitionControllerRuntime); err != nil {
+		return nil, fmt.Errorf("failed to setup ServiceExportByCustomResourceDefinition indexer: %w", err)
+	}
 
+	r := &APIServiceExportReconciler{
+		Client:     client,
+		Scheme:     scheme,
 		bindClient: bindClient,
-
-		serviceExportLister:  serviceExportInformer.Lister(),
-		serviceExportIndexer: serviceExportInformer.Informer().GetIndexer(),
-
-		crdLister:  crdInformer.Lister(),
-		crdIndexer: crdInformer.Informer().GetIndexer(),
-
 		reconciler: reconciler{
-			getCRD: func(name string) (*apiextensionsv1.CustomResourceDefinition, error) {
-				return crdInformer.Lister().Get(name)
+			getCRD: func(ctx context.Context, name string) (*apiextensionsv1.CustomResourceDefinition, error) {
+				var crd apiextensionsv1.CustomResourceDefinition
+				key := types.NamespacedName{Name: name}
+				if err := cache.Get(ctx, key, &crd); err != nil {
+					return nil, err
+				}
+				return &crd, nil
 			},
 			getAPIResourceSchema: func(ctx context.Context, name string) (*kubebindv1alpha2.APIResourceSchema, error) {
 				return bindClient.KubeBindV1alpha2().APIResourceSchemas().Get(ctx, name, metav1.GetOptions{})
 			},
 			deleteServiceExport: func(ctx context.Context, ns, name string) error {
 				return bindClient.KubeBindV1alpha2().APIServiceExports(ns).Delete(ctx, name, metav1.DeleteOptions{})
 			},
-			requeue: func(export *kubebindv1alpha2.APIServiceExport) {
-				key, err := cache.MetaNamespaceKeyFunc(export)
-				if err != nil {
-					runtime.HandleError(err)
-					return
-				}
-				queue.Add(key)
-			},
-		},
-
-		commit: committer.NewCommitter[*kubebindv1alpha2.APIServiceExport, *kubebindv1alpha2.APIServiceExportSpec, *kubebindv1alpha2.APIServiceExportStatus](
-			func(ns string) committer.Patcher[*kubebindv1alpha2.APIServiceExport] {
-				return bindClient.KubeBindV1alpha2().APIServiceExports(ns)
-			},
-		),
-	}
-
-	indexers.AddIfNotPresentOrDie(serviceExportInformer.Informer().GetIndexer(), cache.Indexers{
-		indexers.ServiceExportByCustomResourceDefinition: indexers.IndexServiceExportByCustomResourceDefinition,
-	})
-
-	if _, err := serviceExportInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj any) {
-			c.enqueueServiceExport(logger, obj)
 		},
-		UpdateFunc: func(old, newObj any) {
-			c.enqueueServiceExport(logger, newObj)
-		},
-		DeleteFunc: func(obj any) {
-			c.enqueueServiceExport(logger, obj)
-		},
-	}); err != nil {
-		return nil, err
-	}
-
-	if _, err := crdInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc: func(obj any) {
-			c.enqueueCRD(logger, obj)
-		},
-		UpdateFunc: func(old, newObj any) {
-			c.enqueueCRD(logger, newObj)
-		},
-		DeleteFunc: func(obj any) {
-			c.enqueueCRD(logger, obj)
-		},
-	}); err != nil {
-		return nil, err
 	}
 
-	return c, nil
+	return r, nil
 }
 
-type Resource = committer.Resource[*kubebindv1alpha2.APIServiceExportSpec, *kubebindv1alpha2.APIServiceExportStatus]
-type CommitFunc = func(context.Context, *Resource, *Resource) error
+// createCRDMapper creates a mapping function that can access the client for looking up related APIServiceExports.
+func (r *APIServiceExportReconciler) createCRDMapper() handler.MapFunc {
+	return func(ctx context.Context, obj client.Object) []reconcile.Request {
+		crd := obj.(*apiextensionsv1.CustomResourceDefinition)
 
-// Controller reconciles ServiceNamespaces by creating a Namespace for each, and deleting it if
-// the APIServiceNamespace is deleted.
-type Controller struct {
-	queue workqueue.TypedRateLimitingInterface[string]
+		// Use the field indexer to find APIServiceExports related to this CRD
+		// The indexer key should match the CRD name (namespace/name format for namespaced objects)
+		crdKey := crd.Name // CRDs are cluster-scoped, so just the name
 
-	bindClient bindclient.Interface
-
-	serviceExportLister  bindlisters.APIServiceExportLister
-	serviceExportIndexer cache.Indexer
-
-	crdLister  apiextensionslisters.CustomResourceDefinitionLister
-	crdIndexer cache.Indexer
-
-	reconciler
+		var exports kubebindv1alpha2.APIServiceExportList
+		if err := r.List(ctx, &exports, client.MatchingFields{indexers.ServiceExportByCustomResourceDefinition: crdKey}); err != nil {
+			return []reconcile.Request{}
+		}
 
-	commit CommitFunc
-}
+		var requests []reconcile.Request
+		for _, export := range exports.Items {
+			requests = append(requests, reconcile.Request{
+				NamespacedName: types.NamespacedName{
+					Namespace: export.Namespace,
+					Name:      export.Name,
+				},
+			})
+		}
 
-func (c *Controller) enqueueServiceExport(logger klog.Logger, obj any) {
-	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-	if err != nil {
-		runtime.HandleError(err)
-		return
+		return requests
 	}
-
-	logger.V(2).Info("queueing APIServiceExport", "key", key)
-	c.queue.Add(key)
 }
 
-func (c *Controller) enqueueCRD(logger klog.Logger, obj any) {
-	crdKey, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-	if err != nil {
-		runtime.HandleError(err)
-		return
-	}
-
-	exports, err := c.serviceExportIndexer.ByIndex(indexers.ServiceExportByCustomResourceDefinition, crdKey)
-	if err != nil {
-		runtime.HandleError(err)
-		return
-	}
-
-	for _, obj := range exports {
-		export, ok := obj.(*kubebindv1alpha2.APIServiceExport)
-		if !ok {
-			runtime.HandleError(fmt.Errorf("unexpected type %T", obj))
-			return
-		}
-		key, err := cache.MetaNamespaceKeyFunc(export)
-		if err != nil {
-			runtime.HandleError(err)
-			continue
+//+kubebuilder:rbac:groups=kubebind.k8s.io,resources=apiserviceexports,verbs=get;list;watch;create;update;patch;delete
+//+kubebuilder:rbac:groups=kubebind.k8s.io,resources=apiserviceexports/status,verbs=get;update;patch
+//+kubebuilder:rbac:groups=kubebind.k8s.io,resources=apiserviceexports/finalizers,verbs=update
+//+kubebuilder:rbac:groups=apiextensions.k8s.io,resources=customresourcedefinitions,verbs=get;list;watch
+
+// Reconcile is part of the main kubernetes reconciliation loop which aims to
+// move the current state of the cluster closer to the desired state.
+func (r *APIServiceExportReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	logger := log.FromContext(ctx)
+	logger.Info("Reconciling APIServiceExport", "namespace", req.Namespace, "name", req.Name)
+
+	// Fetch the APIServiceExport instance
+	apiServiceExport := &kubebindv1alpha2.APIServiceExport{}
+	if err := r.Get(ctx, req.NamespacedName, apiServiceExport); err != nil {
+		if errors.IsNotFound(err) {
+			// Request object not found, could have been deleted after reconcile request.
+			logger.Info("APIServiceExport not found, ignoring")
+			return ctrl.Result{}, nil
 		}
-		logger.V(2).Info("queueing APIServiceExport", "key", key, "reason", "CustomResourceDefinition", "CustomResourceDefinitionKey", crdKey)
-		c.queue.Add(key)
+		// Error reading the object - requeue the request.
+		return ctrl.Result{}, fmt.Errorf("failed to get APIServiceExport: %w", err)
 	}
-}
-
-// Start starts the controller, which stops when ctx.Done() is closed.
-func (c *Controller) Start(ctx context.Context, numThreads int) {
-	defer runtime.HandleCrash()
-	defer c.queue.ShutDown()
-
-	logger := klog.FromContext(ctx).WithValues("controller", controllerName)
-
-	logger.Info("Starting controller")
-	defer logger.Info("Shutting down controller")
-
-	for i := 0; i < numThreads; i++ {
-		go wait.UntilWithContext(ctx, c.startWorker, time.Second)
-	}
-
-	<-ctx.Done()
-}
 
-func (c *Controller) startWorker(ctx context.Context) {
-	defer runtime.HandleCrash()
+	// Create a copy to modify
+	original := apiServiceExport.DeepCopy()
 
-	for c.processNextWorkItem(ctx) {
+	// Run the reconciliation logic
+	if err := r.reconciler.reconcile(ctx, apiServiceExport); err != nil {
+		logger.Error(err, "Failed to reconcile APIServiceExport")
+		return ctrl.Result{}, err
 	}
-}
 
-func (c *Controller) 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
+	// Update status if it has changed
+	if !reflect.DeepEqual(original, apiServiceExport) {
+		err := r.Update(ctx, apiServiceExport)
+		if err != nil {
+			logger.Error(err, "Failed to update APIServiceExport status")
+			return ctrl.Result{}, fmt.Errorf("failed to update APIServiceExport status: %w", err)
+		}
+		logger.Info("APIServiceExport status updated", "namespace", apiServiceExport.Namespace, "name", apiServiceExport.Name)
 	}
 
-	logger := klog.FromContext(ctx).WithValues("key", key)
-	ctx = klog.NewContext(ctx, logger)
-	logger.V(2).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 {
-		runtime.HandleError(fmt.Errorf("%q controller failed to sync %q, err: %w", controllerName, key, err))
-		c.queue.AddRateLimited(key)
-		return true
-	}
-	c.queue.Forget(key)
-	return true
+	return ctrl.Result{}, nil
 }
 
-func (c *Controller) process(ctx context.Context, key string) error {
-	ns, name, err := cache.SplitMetaNamespaceKey(key)
-	if err != nil {
-		runtime.HandleError(err)
-		return nil // we cannot do anything
-	}
-
-	obj, err := c.serviceExportLister.APIServiceExports(ns).Get(name)
-	if err != nil && !errors.IsNotFound(err) {
-		return err
-	} else if errors.IsNotFound(err) {
-		return nil // nothing we can do
-	}
-
-	old := obj
-	obj = obj.DeepCopy()
-
-	var errs []error
-	if err := c.reconcile(ctx, obj); err != nil {
-		errs = append(errs, err)
-	}
-
-	// Regardless of whether reconcile returned an error or not, always try to patch status if needed. Return the
-	// reconciliation error at the end.
-
-	// If the object being reconciled changed as a result, update it.
-	oldResource := &Resource{ObjectMeta: old.ObjectMeta, Spec: &old.Spec, Status: &old.Status}
-	newResource := &Resource{ObjectMeta: obj.ObjectMeta, Spec: &obj.Spec, Status: &obj.Status}
-	if err := c.commit(ctx, oldResource, newResource); err != nil {
-		errs = append(errs, err)
-	}
-
-	return utilerrors.NewAggregate(errs)
+// SetupWithManager sets up the controller with the Manager.
+func (r *APIServiceExportReconciler) SetupWithManager(mgr ctrl.Manager) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&kubebindv1alpha2.APIServiceExport{}).
+		Watches(
+			&apiextensionsv1.CustomResourceDefinition{},
+			handler.EnqueueRequestsFromMapFunc(r.createCRDMapper()),
+		).
+		Named(controllerName).
+		Complete(r)
 }

contrib/example-backend/controllers/serviceexport/serviceexport_reconcile.go

@@ -34,11 +34,9 @@ import (
 )
 
 type reconciler struct {
-	getCRD               func(name string) (*apiextensionsv1.CustomResourceDefinition, error)
+	getCRD               func(ctx context.Context, name string) (*apiextensionsv1.CustomResourceDefinition, error)
 	getAPIResourceSchema func(ctx context.Context, name string) (*kubebindv1alpha2.APIResourceSchema, error)
 	deleteServiceExport  func(ctx context.Context, namespace, name string) error
-
-	requeue func(export *kubebindv1alpha2.APIServiceExport)
 }
 
 func (r *reconciler) reconcile(ctx context.Context, export *kubebindv1alpha2.APIServiceExport) error {