Merge pull request #125 from kcp-dev/cel-mutations

Add support for CEL-based mutation rules

Author: kcp-ci-bot

deploy/crd/kcp.io/syncagent.kcp.io_publishedresources.yaml

@@ -157,6 +157,16 @@ spec:
                     spec:
                       items:
                         properties:
+                          cel:
+                            properties:
+                              expression:
+                                type: string
+                              path:
+                                type: string
+                            required:
+                              - expression
+                              - path
+                            type: object
                           delete:
                             properties:
                               path:
@@ -193,6 +203,16 @@ spec:
                     status:
                       items:
                         properties:
+                          cel:
+                            properties:
+                              expression:
+                                type: string
+                              path:
+                                type: string
+                            required:
+                              - expression
+                              - path
+                            type: object
                           delete:
                             properties:
                               path:
@@ -396,6 +416,16 @@ spec:
                           spec:
                             items:
                               properties:
+                                cel:
+                                  properties:
+                                    expression:
+                                      type: string
+                                    path:
+                                      type: string
+                                  required:
+                                    - expression
+                                    - path
+                                  type: object
                                 delete:
                                   properties:
                                     path:
@@ -432,6 +462,16 @@ spec:
                           status:
                             items:
                               properties:
+                                cel:
+                                  properties:
+                                    expression:
+                                      type: string
+                                    path:
+                                      type: string
+                                  required:
+                                    - expression
+                                    - path
+                                  type: object
                                 delete:
                                   properties:
                                     path:

docs/content/publish-resources/index.md

@@ -236,6 +236,7 @@ spec:
       - regex: ...
         template: ...
         delete: ...
+        cel: ...
 ```
 
 #### Regex
@@ -273,6 +274,25 @@ delete:
 This mutation simply removes the value at the given path from the document. JSON path is the
 usual path, without a leading dot.
 
+#### CEL Expressions
+
+```yaml
+cel:
+  path: "metadata.resourceVersion"
+  expression: "value + 42"
+```
+
+This mutation applies a [CEL expression](https://cel.dev/) to a selected value (via `path`) in the
+source object. For this mutation the syncagent will first get the current value at the `path` from
+the Kubernetes object, then applies the CEl expression to it and updates the document with the
+resulting value.
+
+Inside the CEL expression, the following variables are available:
+
+* `value` is the value selected by the `path`
+* `self` is the object to modify
+* `other` is the copy of this object on the other side of the sync
+
 ### Related Resources
 
 The processing of resources on the service cluster often leads to additional resources being

go.mod

@@ -14,6 +14,7 @@ require (
 	github.com/evanphx/json-patch/v5 v5.9.11
 	github.com/go-logr/logr v1.4.3
 	github.com/go-logr/zapr v1.3.0
+	github.com/google/cel-go v0.23.2
 	github.com/google/go-cmp v0.7.0
 	github.com/kcp-dev/api-syncagent/sdk v0.0.0-00010101000000-000000000000
 	github.com/kcp-dev/kcp v0.28.1
@@ -81,7 +82,6 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/google/btree v1.1.3 // indirect
-	github.com/google/cel-go v0.23.2 // indirect
 	github.com/google/gnostic-models v0.6.9 // indirect
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect

internal/controller/syncmanager/controller.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"sync"
 
 	"go.uber.org/zap"
@@ -233,15 +234,23 @@ func (r *Reconciler) reconcile(ctx context.Context, log *zap.SugaredLogger, vwUR
 	}
 
 	// find all PublishedResources
-	pubResources := &syncagentv1alpha1.PublishedResourceList{}
-	if err := r.localManager.GetClient().List(ctx, pubResources, &ctrlruntimeclient.ListOptions{
+	pubResList := &syncagentv1alpha1.PublishedResourceList{}
+	if err := r.localManager.GetClient().List(ctx, pubResList, &ctrlruntimeclient.ListOptions{
 		LabelSelector: r.prFilter,
 	}); err != nil {
 		return fmt.Errorf("failed to list PublishedResources: %w", err)
 	}
 
+	// Filter out those that have not been processed into APIResourceSchemas yet; starting
+	// sync controllers too early, before the schemes are available, will make the watches
+	// not work properly.
+	// Also remove those PRs that have sync disabled.
+	pubResources := slices.DeleteFunc(pubResList.Items, func(pr syncagentv1alpha1.PublishedResource) bool {
+		return pr.Status.ResourceSchemaName == "" || !isSyncEnabled(&pr)
+	})
+
 	// make sure that for every PublishedResource, a matching sync controller exists
-	if err := r.ensureSyncControllers(ctx, log, pubResources.Items); err != nil {
+	if err := r.ensureSyncControllers(ctx, log, pubResources); err != nil {
 		return fmt.Errorf("failed to ensure sync controllers: %w", err)
 	}
 
@@ -415,9 +424,7 @@ func isSyncEnabled(pr *syncagentv1alpha1.PublishedResource) bool {
 func (r *Reconciler) ensureSyncControllers(ctx context.Context, log *zap.SugaredLogger, publishedResources []syncagentv1alpha1.PublishedResource) error {
 	requiredWorkers := sets.New[string]()
 	for _, pr := range publishedResources {
-		if isSyncEnabled(&pr) {
-			requiredWorkers.Insert(getPublishedResourceKey(&pr))
-		}
+		requiredWorkers.Insert(getPublishedResourceKey(&pr))
 	}
 
 	// stop controllers that are no longer needed
@@ -437,10 +444,6 @@ func (r *Reconciler) ensureSyncControllers(ctx context.Context, log *zap.Sugared
 
 	// start missing controllers
 	for _, pubRes := range publishedResources {
-		if !isSyncEnabled(&pubRes) {
-			continue
-		}
-
 		key := getPublishedResourceKey(&pubRes)
 
 		// controller already exists

internal/mutation/mutator.go

@@ -111,6 +111,12 @@ func createAggregatedTransformer(mutations []syncagentv1alpha1.ResourceMutation)
 				return nil, err
 			}
 
+		case mut.CEL != nil:
+			trans, err = transformer.NewCEL(mut.CEL)
+			if err != nil {
+				return nil, err
+			}
+
 		default:
 			return nil, errors.New("no valid mutation mechanism provided")
 		}

internal/mutation/transformer/cel.go

@@ -0,0 +1,94 @@
+/*
+Copyright 2025 The KCP Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package transformer
+
+import (
+	"fmt"
+
+	"github.com/google/cel-go/cel"
+	"github.com/google/cel-go/checker/decls"
+	"github.com/tidwall/gjson"
+	"github.com/tidwall/sjson"
+
+	syncagentv1alpha1 "github.com/kcp-dev/api-syncagent/sdk/apis/syncagent/v1alpha1"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+)
+
+type celTransformer struct {
+	path string
+	prg  cel.Program
+}
+
+func NewCEL(mut *syncagentv1alpha1.ResourceCELMutation) (*celTransformer, error) {
+	env, err := cel.NewEnv(cel.Declarations(
+		decls.NewVar("self", decls.Dyn),
+		decls.NewVar("other", decls.Dyn),
+		decls.NewVar("value", decls.Dyn),
+	))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create CEL env: %w", err)
+	}
+
+	expr, issues := env.Compile(mut.Expression)
+	if issues != nil && issues.Err() != nil {
+		return nil, fmt.Errorf("failed to compile CEL expression: %w", issues.Err())
+	}
+
+	prg, err := env.Program(expr)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create CEL program: %w", err)
+	}
+
+	return &celTransformer{
+		path: mut.Path,
+		prg:  prg,
+	}, nil
+}
+
+func (m *celTransformer) Apply(toMutate *unstructured.Unstructured, otherObj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
+	encoded, err := EncodeObject(toMutate)
+	if err != nil {
+		return nil, fmt.Errorf("failed to JSON encode object: %w", err)
+	}
+
+	// get the current value at the path
+	current := gjson.Get(encoded, m.path)
+
+	input := map[string]any{
+		"value": current.Value(),
+		"self":  toMutate.Object,
+		"other": nil,
+	}
+	if otherObj != nil {
+		input["other"] = otherObj.Object
+	}
+
+	// evaluate the expression
+	out, _, err := m.prg.Eval(input)
+	if err != nil {
+		return nil, fmt.Errorf("failed to evaluate CEL expression: %w", err)
+	}
+
+	// update the object
+	updated, err := sjson.Set(encoded, m.path, out)
+	if err != nil {
+		return nil, fmt.Errorf("failed to set updated value: %w", err)
+	}
+
+	return DecodeObject(updated)
+}