feat: implement non-blocking CRD reconciliation with requeue support

Implement non blocking CRD reconciliation to improve controller throughput
when managing large amount of RGDs. The main idea is to free the workers to do
other tasks when we're waiting for the establishement.

Previously, the controller would block waiting for CRDs to become established,
preventing other ResourceGraphDefinitions from being processed. This change
introduces a requeue based approach where:

- CRD creation happens immediately without blocking
- controller checks CRD establishment status on each reconciliation
- if CRD exists but isn't established, reconciler requeues after 500ms
- other RGDs can be processed while waiting for CRD establishment

this patch also gets rid of the CRDWraper abstraction layer as it's no longer
needed.

Author: a-hilaly

pkg/client/crd.go

@@ -66,9 +66,8 @@ func (f *FakeSet) RESTConfig() *rest.Config {
 }
 
 // CRD returns a new CRD interface instance
-func (f *FakeSet) CRD(cfg client.CRDWrapperConfig) client.CRDInterface {
-	// Return a fake CRD implementation for testing
-	return &FakeCRD{}
+func (f *FakeSet) CRD() apiextensionsv1.CustomResourceDefinitionInterface {
+	return f.ApiExtensionsClient.CustomResourceDefinitions()
 }
 
 // WithImpersonation returns a new client that impersonates the given user
@@ -88,10 +87,8 @@ func (f *FakeSet) SetRESTMapper(restMapper meta.RESTMapper) {
 // FakeCRD is a fake implementation of CRDInterface for testing
 type FakeCRD struct{}
 
-var _ client.CRDInterface = (*FakeCRD)(nil)
-
-// Ensure ensures a CRD exists, up-to-date, and is ready
-func (f *FakeCRD) Ensure(ctx context.Context, crd v1.CustomResourceDefinition) error {
+// Create ensures a CRD exists, up-to-date, and is ready
+func (f *FakeCRD) Create(ctx context.Context, crd v1.CustomResourceDefinition) error {
 	// For testing, just return success
 	return nil
 }

pkg/client/fake/fake.go

@@ -41,8 +41,8 @@ type SetInterface interface {
 	// RESTConfig returns a copy of the underlying REST config
 	RESTConfig() *rest.Config
 
-	// CRD returns a new CRDInterface instance
-	CRD(cfg CRDWrapperConfig) CRDInterface
+	// CRD returns a new CustomResourceDefinitionInterface instance
+	CRD() apiextensionsv1.CustomResourceDefinitionInterface
 
 	// WithImpersonation returns a new client that impersonates the given user
 	WithImpersonation(user string) (SetInterface, error)
@@ -159,13 +159,9 @@ func (c *Set) RESTMapper() meta.RESTMapper {
 	return c.restMapper
 }
 
-// CRD returns a new CRDInterface instance
-func (c *Set) CRD(cfg CRDWrapperConfig) CRDInterface {
-	if cfg.Client == nil {
-		cfg.Client = c.apiExtensionsV1
-	}
-
-	return newCRDWrapper(cfg)
+// CRD returns a new CustomResourceDefinitionInterface instance
+func (c *Set) CRD() apiextensionsv1.CustomResourceDefinitionInterface {
+	return c.apiExtensionsV1.CustomResourceDefinitions()
 }
 
 // WithImpersonation returns a new client that impersonates the given user

pkg/client/set.go

@@ -35,6 +35,7 @@ import (
 	"github.com/kubernetes-sigs/kro/pkg/dynamiccontroller"
 	"github.com/kubernetes-sigs/kro/pkg/graph"
 	"github.com/kubernetes-sigs/kro/pkg/metadata"
+	"github.com/kubernetes-sigs/kro/pkg/requeue"
 )
 
 //+kubebuilder:rbac:groups=kro.run,resources=resourcegraphdefinitions,verbs=get;list;watch;create;update;patch;delete
@@ -51,8 +52,7 @@ type ResourceGraphDefinitionReconciler struct {
 
 	instanceLogger logr.Logger
 
-	clientSet  kroclient.SetInterface
-	crdManager kroclient.CRDClient
+	clientSet kroclient.SetInterface
 
 	metadataLabeler         metadata.Labeler
 	rgBuilder               *graph.Builder
@@ -67,12 +67,10 @@ func NewResourceGraphDefinitionReconciler(
 	builder *graph.Builder,
 	maxConcurrentReconciles int,
 ) *ResourceGraphDefinitionReconciler {
-	crdWrapper := clientSet.CRD(kroclient.CRDWrapperConfig{})
 
 	return &ResourceGraphDefinitionReconciler{
 		clientSet:               clientSet,
 		allowCRDDeletion:        allowCRDDeletion,
-		crdManager:              crdWrapper,
 		dynamicController:       dynamicController,
 		metadataLabeler:         metadata.NewKROMetaLabeler(),
 		rgBuilder:               builder,
@@ -171,9 +169,15 @@ func (r *ResourceGraphDefinitionReconciler) Reconcile(ctx context.Context, o *v1
 
 	topologicalOrder, resourcesInformation, reconcileErr := r.reconcileResourceGraphDefinition(ctx, o)
 
+	// Always update status
 	if err := r.updateStatus(ctx, o, topologicalOrder, resourcesInformation); err != nil {
 		reconcileErr = errors.Join(reconcileErr, err)
 	}
 
+	// Check if this is a requeue error
+	if needRequeue, duration := requeue.Check(reconcileErr); needRequeue {
+		return ctrl.Result{RequeueAfter: duration}, nil
+	}
+
 	return ctrl.Result{}, reconcileErr
 }

pkg/controller/resourcegraphdefinition/controller.go

@@ -70,7 +70,7 @@ func (r *ResourceGraphDefinitionReconciler) cleanupResourceGraphDefinitionCRD(ct
 		return nil
 	}
 
-	if err := r.crdManager.Delete(ctx, crdName); err != nil {
+	if err := r.deleteCRD(ctx, crdName); err != nil {
 		return fmt.Errorf("error deleting CRD: %w", err)
 	}
 	return nil