[FEAT] Operator: Concurrent Reconciles supported

Enable concurrent reconciles for resources with the following defaults:
```
{
	CAPApplication:        1,
	CAPApplicationVersion: 3,
	CAPTenant:             10,
	CAPTenantOperation:    10,
	Domain:                1,
	ClusterDomain:         1,
}
```
This can be overridden using the corresponding env:
```
  "MAX_CONCURRENT_RECONCILES_CAP_APPLICATION",
  "MAX_CONCURRENT_RECONCILES_CAP_APPLICATION_VERSION",
  "MAX_CONCURRENT_RECONCILES_CAP_TENANT",
  "MAX_CONCURRENT_RECONCILES_CAP_TENANT_OPERATION",
  "MAX_CONCURRENT_RECONCILES_DOMAIN",
  "MAX_CONCURRENT_RECONCILES_CLUSTER_DOMAIN",
```
on the CAP Operator controller.

Apart from the above, the following changes were made:
 - custom ratelimit workqueue configuration for tenants and tenant
operations to avoid long delays in processing of these resources.
 - default resync period for operator resources set to 5 minutes.

Author: Pavan-SAP

cmd/controller/main.go

@@ -7,8 +7,10 @@ package main
 
 import (
 	"context"
+	"maps"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 	"time"
 
@@ -31,7 +33,8 @@ import (
 )
 
 const (
-	LeaseLockName = "capoperator-lease-lock"
+	LeaseLockName                    = "capoperator-lease-lock"
+	MaxConcurrentReconcilesEnvPrefix = "MAX_CONCURRENT_RECONCILES_"
 )
 
 func main() {
@@ -82,6 +85,9 @@ func main() {
 	// Initialize/start metrics server
 	util.InitMetricsServer()
 
+	// Get concurrency config for each resource kind from environment variables or use defaults
+	concurrencyConfig := getDefaultConcurrencyConfig()
+
 	// context for the reconciliation controller
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -133,6 +139,8 @@ func main() {
 				klog.InfoS("check & update of subscriptionGUID label done")
 
 				c := controller.NewController(coreClient, crdClient, istioClient, certClient, certManagerClient, dnsClient, promClient)
+				// Update the controller's concurrency config before starting the controller
+				maps.Copy(controller.DefaultConcurrentReconciles, concurrencyConfig)
 				go c.Start(ctx)
 			},
 			OnStoppedLeading: func() {
@@ -148,3 +156,28 @@ func main() {
 		},
 	})
 }
+
+func getDefaultConcurrencyConfig() map[int]int {
+	// inline function to get concurrency config for each resource kind from environment variables or use defaults
+	getConcurrencyConfigForResource := func(resourceEnvSuffix string, defaultVal int) int {
+		reconcileEnv := MaxConcurrentReconcilesEnvPrefix + resourceEnvSuffix
+		if val := os.Getenv(reconcileEnv); val != "" {
+			if intVal, err := strconv.Atoi(val); err == nil {
+				return intVal
+			}
+		}
+		return defaultVal
+	}
+
+	// Configure default concurrency for each resource kind, can be overridden by environment variables
+	concurrencyConfig := make(map[int]int)
+
+	for resourceKey, resourceEnvSuffix := range controller.ResourceEnvSuffixMap {
+		defaultReconcileForResource, ok := controller.DefaultConcurrentReconciles[resourceKey]
+		if !ok {
+			defaultReconcileForResource = controller.DefaultReconcile
+		}
+		concurrencyConfig[resourceKey] = getConcurrencyConfigForResource(resourceEnvSuffix, defaultReconcileForResource)
+	}
+	return concurrencyConfig
+}

internal/controller/controller.go

@@ -23,6 +23,7 @@ import (
 	"github.com/sap/cap-operator/pkg/client/clientset/versioned"
 	v1alpha1scheme "github.com/sap/cap-operator/pkg/client/clientset/versioned/scheme"
 	crdInformers "github.com/sap/cap-operator/pkg/client/informers/externalversions"
+	"golang.org/x/time/rate"
 	istio "istio.io/client-go/pkg/clientset/versioned"
 	istioscheme "istio.io/client-go/pkg/clientset/versioned/scheme"
 	istioInformers "istio.io/client-go/pkg/informers/externalversions"
@@ -56,17 +57,35 @@ type Controller struct {
 	eventRecorder                events.EventRecorder
 }
 
+var (
+	// Application and Domain resources are less frequently updated, so assume a default concurrency of 1.
+	DefaultReconcile            = 1
+	DefaultConcurrentReconciles = map[int]int{
+		ResourceCAPApplicationVersion: 3,  // Moderate concurrency to handle multiple versions efficiently
+		ResourceCAPTenant:             10, // High concurrency to handle multiple tenants efficiently
+		ResourceCAPTenantOperation:    10, // High concurrency to handle multiple tenant operations efficiently
+	}
+	ResourceEnvSuffixMap = map[int]string{
+		ResourceCAPApplication:        "CAP_APPLICATION",
+		ResourceCAPApplicationVersion: "CAP_APPLICATION_VERSION",
+		ResourceCAPTenant:             "CAP_TENANT",
+		ResourceCAPTenantOperation:    "CAP_TENANT_OPERATION",
+		ResourceDomain:                "DOMAIN",
+		ResourceClusterDomain:         "CLUSTER_DOMAIN",
+	}
+)
+
 func NewController(client kubernetes.Interface, crdClient versioned.Interface, istioClient istio.Interface, gardenerCertificateClient gardenerCert.Interface, certManagerCertificateClient certManager.Interface, gardenerDNSClient gardenerDNS.Interface, promClient promop.Interface) *Controller {
 	// Register metrics provider on the workqueue
 	initializeMetrics()
 
 	queues := map[int]workqueue.TypedRateLimitingInterface[QueueItem]{
 		ResourceCAPApplication:        workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[QueueItem](), workqueue.TypedRateLimitingQueueConfig[QueueItem]{Name: KindMap[ResourceCAPApplication]}),
 		ResourceCAPApplicationVersion: workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[QueueItem](), workqueue.TypedRateLimitingQueueConfig[QueueItem]{Name: KindMap[ResourceCAPApplicationVersion]}),
-		ResourceCAPTenant:             workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[QueueItem](), workqueue.TypedRateLimitingQueueConfig[QueueItem]{Name: KindMap[ResourceCAPTenant]}),
-		ResourceCAPTenantOperation:    workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[QueueItem](), workqueue.TypedRateLimitingQueueConfig[QueueItem]{Name: KindMap[ResourceCAPTenantOperation]}),
-		ResourceClusterDomain:         workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[QueueItem](), workqueue.TypedRateLimitingQueueConfig[QueueItem]{Name: KindMap[ResourceClusterDomain]}),
+		ResourceCAPTenant:             workqueue.NewTypedRateLimitingQueueWithConfig(customRateLimiter(), workqueue.TypedRateLimitingQueueConfig[QueueItem]{Name: KindMap[ResourceCAPTenant]}),
+		ResourceCAPTenantOperation:    workqueue.NewTypedRateLimitingQueueWithConfig(customRateLimiter(), workqueue.TypedRateLimitingQueueConfig[QueueItem]{Name: KindMap[ResourceCAPTenantOperation]}),
 		ResourceDomain:                workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[QueueItem](), workqueue.TypedRateLimitingQueueConfig[QueueItem]{Name: KindMap[ResourceDomain]}),
+		ResourceClusterDomain:         workqueue.NewTypedRateLimitingQueueWithConfig(workqueue.DefaultTypedControllerRateLimiter[QueueItem](), workqueue.TypedRateLimitingQueueConfig[QueueItem]{Name: KindMap[ResourceClusterDomain]}),
 	}
 
 	// Use 30mins as the default Resync interval for kube / proprietary  resources
@@ -89,8 +108,8 @@ func NewController(client kubernetes.Interface, crdClient versioned.Interface, i
 		// no activity needed on our side so far
 	}
 
-	// Use 60 as the default Resync interval for our custom resources (CAP CROs)
-	crdInformerFactory := crdInformers.NewSharedInformerFactory(crdClient, 60*time.Second)
+	// Use 5 mins as the default Resync interval for our custom resources (CAP CROs)
+	crdInformerFactory := crdInformers.NewSharedInformerFactory(crdClient, 5*time.Minute)
 
 	// initialize event recorder
 	scheme := runtime.NewScheme()
@@ -122,6 +141,21 @@ func NewController(client kubernetes.Interface, crdClient versioned.Interface, i
 	return c
 }
 
+// Custom Rate limiter for Tenant and TenantOperation queues to allow faster retries and higher throughput.
+func customRateLimiter() workqueue.TypedRateLimiter[QueueItem] {
+	return workqueue.NewTypedMaxOfRateLimiter(
+		// Faster exponential backoff for transient errors
+		workqueue.NewTypedItemExponentialFailureRateLimiter[QueueItem](
+			10*time.Millisecond, // base delay (was 5ms)
+			300*time.Second,     // max delay (was ~1000s)
+		),
+		// Higher QPS for bulk processing
+		&workqueue.TypedBucketRateLimiter[QueueItem]{
+			Limiter: rate.NewLimiter(rate.Limit(50), 200), // 50 QPS, 200 burst (was 10/100)
+		},
+	)
+}
+
 func throwInformerStartError(resources map[reflect.Type]bool) {
 	for resource, ok := range resources {
 		if !ok {
@@ -179,15 +213,19 @@ func (c *Controller) Start(ctx context.Context) {
 
 	var wg sync.WaitGroup
 	for k := range c.queues {
-		wg.Add(1)
-		go func(key int) {
-			defer wg.Done()
-			err := c.processQueue(qCxt, key)
-			if err != nil {
-				klog.ErrorS(err, "worker queue ended with error", "key", key)
-			}
-			qCancel() // cancel context to inform other workers
-		}(k)
+		concurrency := getConcurrencyForResource(k)
+		klog.InfoS("starting worker queue", "resource", getResourceKindFromKey(k), "concurrency", concurrency)
+		for i := range concurrency {
+			wg.Add(1)
+			go func(key, workerId int) {
+				defer wg.Done()
+				err := c.processQueue(qCxt, key, workerId)
+				if err != nil {
+					klog.ErrorS(err, "worker queue ended with error", "key", key)
+				}
+				qCancel() // cancel context to inform other workers
+			}(k, i)
+		}
 	}
 
 	// start version cleanup routines
@@ -199,33 +237,41 @@ func (c *Controller) Start(ctx context.Context) {
 	wg.Wait()
 }
 
-func (c *Controller) processQueue(ctx context.Context, key int) error {
-	klog.InfoS("starting to process queue", "resource", getResourceKindFromKey(key))
+func getConcurrencyForResource(key int) int {
+	concurrency, ok := DefaultConcurrentReconciles[key]
+	if !ok {
+		concurrency = DefaultReconcile // default concurrency
+	}
+	return concurrency
+}
+
+func (c *Controller) processQueue(ctx context.Context, key, workerId int) error {
+	klog.InfoS("starting to process queue", "resource", getResourceKindFromKey(key), "workerId", workerId)
 	for {
 		select {
 		case <-ctx.Done():
-			klog.InfoS("context done; ending processing of queue", "resource", getResourceKindFromKey(key))
+			klog.InfoS("context done; ending processing of queue", "resource", getResourceKindFromKey(key), "workerId", workerId)
 			return nil
 		default: // fall through - to avoid blocking
-			err := c.processQueueItem(ctx, key)
+			err := c.processQueueItem(ctx, key, workerId)
 			if err != nil {
 				return err
 			}
 		}
 	}
 }
 
-func (c *Controller) processQueueItem(ctx context.Context, key int) error {
+func (c *Controller) processQueueItem(ctx context.Context, key, workerId int) error {
 	q, ok := c.queues[key]
 	if !ok {
 		return fmt.Errorf("unknown queue; ending worker %d", key)
 	}
 
-	klog.V(2).InfoS("Processing queue item in work queue", "resource", getResourceKindFromKey(key), "queue length", q.Len())
+	klog.V(2).InfoS("Processing queue item in work queue", "resource", getResourceKindFromKey(key), "queue length", q.Len(), "workerId", workerId)
 
 	item, shutdown := q.Get()
 	if shutdown {
-		return fmt.Errorf("queue (%d) shutdown", key) // stop processing when the queue has been shutdown
+		return fmt.Errorf("queue (%d, %d) shutdown", key, workerId) // stop processing when the queue has been shutdown
 	}
 
 	// [IMPORTANT] always mark the item as done (after processing it)
@@ -242,7 +288,7 @@ func (c *Controller) processQueueItem(ctx context.Context, key int) error {
 	// Attempt to recover panics during reconciliation.
 	defer c.recoverFromPanic(ctx, item, q)
 
-	klog.InfoS("Processing Resource", "namespace", item.ResourceKey.Namespace, "name", item.ResourceKey.Name, "kind", getResourceKindFromKey(key), "attempt", attempts)
+	klog.InfoS("Processing Resource", "namespace", item.ResourceKey.Namespace, "name", item.ResourceKey.Name, "kind", getResourceKindFromKey(key), "attempt", attempts, "workerId", workerId)
 
 	switch item.Key {
 	case ResourceCAPApplication:
@@ -263,7 +309,7 @@ func (c *Controller) processQueueItem(ctx context.Context, key int) error {
 	}
 	// Handle reconcile errors
 	if err != nil {
-		klog.ErrorS(err, "queue processing error", "resource", getResourceKindFromKey(key))
+		klog.ErrorS(err, "queue processing error", "resource", getResourceKindFromKey(key), "workerId", workerId)
 		ReconcileErrors.WithLabelValues(getResourceKindFromKey(item.Key), item.ResourceKey.Namespace, item.ResourceKey.Name).Inc()
 		if !skipItem {
 			// add back to queue for re-processing

internal/controller/controller_test.go

@@ -108,9 +108,9 @@ func TestController_processQueue(t *testing.T) {
 			// Manual API checks
 			var expectedRes error
 			if tt.earlyShutDown {
-				expectedRes = testC.processQueue(ctx, tt.resource)
+				expectedRes = testC.processQueue(ctx, tt.resource, 0)
 			} else {
-				expectedRes = testC.processQueue(context.TODO(), tt.resource)
+				expectedRes = testC.processQueue(context.TODO(), tt.resource, 0)
 			}
 
 			if !tt.expectError && expectedRes != nil {
@@ -263,12 +263,12 @@ func TestController_processQueueItem(t *testing.T) {
 			if tt.earlyShutDown {
 				q.ShutDown()
 				cancel()
-				expectedRes = testC.processQueueItem(ctx, tt.resource)
+				expectedRes = testC.processQueueItem(ctx, tt.resource, 0)
 			} else {
 				if tt.resource < 4 || tt.resource == 9 || tt.resource == 99 {
 					q.Add(item)
 				}
-				expectedRes = testC.processQueueItem(context.TODO(), tt.resource)
+				expectedRes = testC.processQueueItem(context.TODO(), tt.resource, 0)
 			}
 
 			if !tt.expectError && expectedRes != nil {