cache mutators at runtime to handle expensive to compile transformers

On-behalf-of: @SAP [email protected]

Author: xrstf

internal/controller/sync/controller.go

@@ -103,12 +103,7 @@ func Create(
 	remoteDummy.SetGroupVersionKind(remoteGVK)
 
 	// create the syncer that holds the meat&potatoes of the synchronization logic
-	mutator, err := mutation.NewMutator(pubRes.Spec.Mutation)
-	if err != nil {
-		return nil, fmt.Errorf("failed to create mutator: %w", err)
-	}
-
-	syncer, err := sync.NewResourceSyncer(log, localManager.GetClient(), virtualWorkspaceCluster.GetClient(), pubRes, localCRD, mutator, stateNamespace, agentName)
+	syncer, err := sync.NewResourceSyncer(log, localManager.GetClient(), virtualWorkspaceCluster.GetClient(), pubRes, localCRD, mutation.NewMutator, stateNamespace, agentName)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create syncer: %w", err)
 	}

internal/mutation/mutator.go

@@ -47,22 +47,18 @@ var _ Mutator = &mutator{}
 // NewMutator creates a new mutator, which will apply the mutation rules to a synced object, in
 // both directions. A nil spec is supported and will simply make the mutator not do anything.
 func NewMutator(spec *syncagentv1alpha1.ResourceMutationSpec) (Mutator, error) {
-	var (
-		specAgg   *transformer.AggregateTransformer
-		statusAgg *transformer.AggregateTransformer
-		err       error
-	)
-
-	if spec != nil {
-		specAgg, err = createAggregatedTransformer(spec.Spec)
-		if err != nil {
-			return nil, fmt.Errorf("cannot create transformer for spec: %w", err)
-		}
+	if spec == nil {
+		return nil, nil
+	}
 
-		statusAgg, err = createAggregatedTransformer(spec.Status)
-		if err != nil {
-			return nil, fmt.Errorf("cannot create transformer for status: %w", err)
-		}
+	specAgg, err := createAggregatedTransformer(spec.Spec)
+	if err != nil {
+		return nil, fmt.Errorf("cannot create transformer for spec: %w", err)
+	}
+
+	statusAgg, err := createAggregatedTransformer(spec.Status)
+	if err != nil {
+		return nil, fmt.Errorf("cannot create transformer for status: %w", err)
 	}
 
 	return &mutator{
@@ -72,10 +68,18 @@ func NewMutator(spec *syncagentv1alpha1.ResourceMutationSpec) (Mutator, error) {
 }
 
 func (m *mutator) MutateSpec(toMutate *unstructured.Unstructured, otherObj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	if m == nil {
+		return toMutate, nil
+	}
+
 	return m.spec.Apply(toMutate, otherObj)
 }
 
 func (m *mutator) MutateStatus(toMutate *unstructured.Unstructured, otherObj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	if m == nil {
+		return toMutate, nil
+	}
+
 	return m.status.Apply(toMutate, otherObj)
 }
 

internal/mutation/transformer/template_test.go

@@ -22,6 +22,7 @@ import (
 	"github.com/kcp-dev/api-syncagent/internal/test/diff"
 	syncagentv1alpha1 "github.com/kcp-dev/api-syncagent/sdk/apis/syncagent/v1alpha1"
 	"github.com/kcp-dev/api-syncagent/test/utils"
+
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 )
 

internal/sync/syncer.go

@@ -45,21 +45,25 @@ type ResourceSyncer struct {
 
 	destDummy *unstructured.Unstructured
 
-	mutator mutation.Mutator
+	// cached mutators (for those transformers that are expensive to compile, like CEL)
+	primaryMutator  mutation.Mutator
+	relatedMutators map[string]mutation.Mutator
 
 	agentName string
 
 	// newObjectStateStore is used for testing purposes
 	newObjectStateStore newObjectStateStoreFunc
 }
 
+type MutatorCreatorFunc func(*syncagentv1alpha1.ResourceMutationSpec) (mutation.Mutator, error)
+
 func NewResourceSyncer(
 	log *zap.SugaredLogger,
 	localClient ctrlruntimeclient.Client,
 	remoteClient ctrlruntimeclient.Client,
 	pubRes *syncagentv1alpha1.PublishedResource,
 	localCRD *apiextensionsv1.CustomResourceDefinition,
-	mutator mutation.Mutator,
+	mutatorCreator MutatorCreatorFunc,
 	stateNamespace string,
 	agentName string,
 ) (*ResourceSyncer, error) {
@@ -94,6 +98,21 @@ func NewResourceSyncer(
 		}
 	}
 
+	primaryMutator, err := mutatorCreator(pubRes.Spec.Mutation)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create primary object mutator: %w", err)
+	}
+
+	relatedMutators := map[string]mutation.Mutator{}
+	for _, rr := range pubRes.Spec.Related {
+		mutator, err := mutatorCreator(rr.Mutation)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create related object %q mutator: %w", rr.Identifier, err)
+		}
+
+		relatedMutators[rr.Identifier] = mutator
+	}
+
 	return &ResourceSyncer{
 		log:                 log.With("local-gvk", localGVK, "remote-gvk", remoteGVK),
 		localClient:         localClient,
@@ -102,7 +121,8 @@ func NewResourceSyncer(
 		localCRD:            localCRD,
 		subresources:        subresources,
 		destDummy:           localDummy,
-		mutator:             mutator,
+		primaryMutator:      primaryMutator,
+		relatedMutators:     relatedMutators,
 		agentName:           agentName,
 		newObjectStateStore: newKubernetesStateStoreCreator(stateNamespace),
 	}, nil
@@ -162,7 +182,7 @@ func (s *ResourceSyncer) Process(ctx Context, remoteObj *unstructured.Unstructur
 		// in kcp is deleted
 		blockSourceDeletion: true,
 		// use the configured mutations from the PublishedResource
-		mutator: s.mutator,
+		mutator: s.primaryMutator,
 		// make sure the syncer can remember the current state of any object
 		stateStore: stateStore,
 		// For the main resource, we need to store metadata on the destination copy

internal/sync/syncer_related.go

@@ -27,7 +27,6 @@ import (
 	"github.com/tidwall/gjson"
 	"go.uber.org/zap"
 
-	"github.com/kcp-dev/api-syncagent/internal/mutation"
 	"github.com/kcp-dev/api-syncagent/internal/sync/templating"
 	syncagentv1alpha1 "github.com/kcp-dev/api-syncagent/sdk/apis/syncagent/v1alpha1"
 
@@ -118,11 +117,6 @@ func (s *ResourceSyncer) processRelatedResource(log *zap.SugaredLogger, stateSto
 			object:      destObject,
 		}
 
-		mutator, err := mutation.NewMutator(relRes.Mutation)
-		if err != nil {
-			return false, fmt.Errorf("failed to create mutator: %w", err)
-		}
-
 		syncer := objectSyncer{
 			// Related objects within kcp are not labelled with the agent name because it's unnecessary.
 			// agentName: "",
@@ -147,7 +141,7 @@ func (s *ResourceSyncer) processRelatedResource(log *zap.SugaredLogger, stateSto
 			// sure we can clean up properly
 			blockSourceDeletion: relRes.Origin == "kcp",
 			// apply mutation rules configured for the related resource
-			mutator: mutator,
+			mutator: s.relatedMutators[relRes.Identifier],
 			// we never want to store sync-related metadata inside kcp
 			metadataOnDestination: false,
 		}

internal/sync/syncer_test.go

@@ -26,6 +26,7 @@ import (
 	"github.com/kcp-dev/logicalcluster/v3"
 	"go.uber.org/zap"
 
+	"github.com/kcp-dev/api-syncagent/internal/mutation"
 	dummyv1alpha1 "github.com/kcp-dev/api-syncagent/internal/sync/apis/dummy/v1alpha1"
 	"github.com/kcp-dev/api-syncagent/internal/test/diff"
 	syncagentv1alpha1 "github.com/kcp-dev/api-syncagent/sdk/apis/syncagent/v1alpha1"
@@ -899,7 +900,9 @@ func TestSyncerProcessingSingleResourceWithoutStatus(t *testing.T) {
 				remoteClient,
 				testcase.pubRes,
 				testcase.localCRD,
-				nil,
+				func(rms *syncagentv1alpha1.ResourceMutationSpec) (mutation.Mutator, error) {
+					return nil, nil
+				},
 				stateNamespace,
 				"textor-the-doctor",
 			)
@@ -1205,7 +1208,9 @@ func TestSyncerProcessingSingleResourceWithStatus(t *testing.T) {
 				remoteClient,
 				testcase.pubRes,
 				testcase.localCRD,
-				nil,
+				func(rms *syncagentv1alpha1.ResourceMutationSpec) (mutation.Mutator, error) {
+					return nil, nil
+				},
 				stateNamespace,
 				"textor-the-doctor",
 			)