feat: add concurrency support in applyset operations (#712)

* feat: add concurrency support in applyset operations

Introduce configurable concurrency limit for applyset operations using `sync/errgroup`. This improves performance by enabling concurrent processing of apply api calls. Defaults to the number of objects in the applyset if unspecified.

Signed-off-by: Jakob Möller <[email protected]>

* feat: remove hardcoded prune concurrency limit in favor of concurrency spec

Replaced the fixed `PruneGVKParallelizationLimit` constant with the existing configurable `concurrency` field to allow dynamic adjustment of prune operation parallelism. Updated comments for clarity.

Signed-off-by: Jakob Möller <[email protected]>

---------

Signed-off-by: Jakob Möller <[email protected]>

Author: jakobmoellerdev

pkg/applyset/applyset.go

@@ -29,8 +29,10 @@ import (
 	"reflect"
 	"sort"
 	"strings"
+	"sync"
 
 	"github.com/go-logr/logr"
+	"golang.org/x/sync/errgroup"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -116,6 +118,10 @@ type Config struct {
 
 	// Log is used to inject the calling reconciler's logger
 	Log logr.Logger
+
+	// Concurrency is the maximum number of concurrent apply and prune operations in a single applyset.
+	// If not provided, the default value is the number of objects in the applyset.
+	Concurrency int
 }
 
 /*
@@ -186,9 +192,10 @@ func New(
 				FieldManager: config.FieldManager,
 				Force:        true,
 			},
-			//deleteOptions: metav1.DeleteOptions{},
+			// deleteOptions: metav1.DeleteOptions{},
 		},
-		log: config.Log,
+		log:         config.Log,
+		concurrency: config.Concurrency,
 	}
 
 	gvk := parent.GroupVersionKind()
@@ -259,6 +266,10 @@ type applySet struct {
 	serverOptions
 
 	log logr.Logger
+
+	// concurrency is the maximum number of concurrent apply and prune operations in a single applyset.
+	// If not provided, the default value is the number of objects in the applyset.
+	concurrency int
 }
 
 func (a *applySet) getAndRecordNamespace(obj ApplyableObject, restMapping *meta.RESTMapping) error {
@@ -544,19 +555,35 @@ func (a *applySet) apply(ctx context.Context, dryRun bool) (*ApplyResult, error)
 	if dryRun {
 		options.DryRun = []string{"All"}
 	}
-	for _, obj := range a.desired.objects {
 
+	concurrency := a.concurrency
+	if concurrency <= 0 {
+		concurrency = len(a.desired.objects)
+	}
+
+	eg, egctx := errgroup.WithContext(ctx)
+	eg.SetLimit(concurrency)
+
+	// protect concurrent access to write the apply results
+	var mu sync.Mutex
+
+	for _, obj := range a.desired.objects {
 		dynResource, err := a.resourceClient(obj)
 		if err != nil {
 			return results, err
 		}
-		lastApplied, err := dynResource.Apply(ctx, obj.GetName(), obj.Unstructured, options)
-		results.recordApplied(obj, lastApplied, err)
-		a.log.V(2).Info("applied object", "object", obj.String(), "applied-revision", lastApplied.GetResourceVersion(),
-			"error", err)
-	}
-
-	return results, nil
+		eg.Go(func() error {
+			lastApplied, err := dynResource.Apply(egctx, obj.GetName(), obj.Unstructured, options)
+			mu.Lock()
+			defer mu.Unlock()
+			results.recordApplied(obj, lastApplied, err)
+			a.log.V(2).Info("applied object", "object", obj.String(), "applied-revision", lastApplied.GetResourceVersion(),
+				"error", err)
+			return nil
+		})
+	}
+
+	return results, eg.Wait()
 }
 
 func (a *applySet) prune(ctx context.Context, results *ApplyResult, dryRun bool) (*ApplyResult, error) {

pkg/applyset/prune.go

@@ -36,14 +36,6 @@ import (
 	"k8s.io/client-go/dynamic"
 )
 
-const (
-	// This is set to an arbitrary number here for now.
-	// This ensures we are no unbounded when pruning many GVKs.
-	// Could be parameterized later on
-	// TODO (barney-s): Possible parameterization target
-	PruneGVKParallelizationLimit = 1
-)
-
 // PruneObject is an apiserver object that should be deleted as part of prune.
 type PruneObject struct {
 	*unstructured.Unstructured
@@ -124,7 +116,7 @@ func (a *applySet) findAllObjectsToPrune(
 		}
 	}
 
-	if PruneGVKParallelizationLimit <= 1 {
+	if a.concurrency <= 1 {
 		for i := range tasks {
 			task := tasks[i]
 			results, err := a.findObjectsToPrune(ctx, dynamicClient, visitedUIDs, task.namespace, task.restMapping)
@@ -135,7 +127,7 @@ func (a *applySet) findAllObjectsToPrune(
 		}
 	} else {
 		group, ctx := errgroup.WithContext(ctx)
-		group.SetLimit(PruneGVKParallelizationLimit)
+		group.SetLimit(a.concurrency)
 		for i := range tasks {
 			task := tasks[i]
 			group.Go(func() error {