Implement Receiver resource filtering with CEL

Signed-off-by: Kevin McDermott <[email protected]>
Co-authored-by: Matheus Pimenta <[email protected]>

api/v1/receiver_types.go

@@ -67,6 +67,16 @@ type ReceiverSpec struct {
 	// +required
 	Resources []CrossNamespaceObjectReference `json:"resources"`
 
+	// ResourceFilter is a CEL expression expected to return a boolean that is
+	// evaluated for each resource referenced in the Resources field when a
+	// webhook is received. If the expression returns false then the controller
+	// will not request a reconciliation for the resource.
+	// When the expression is specified the controller will parse it and mark
+	// the object as terminally failed if the expression is invalid or does not
+	// return a boolean.
+	// +optional
+	ResourceFilter string `json:"resourceFilter,omitempty"`
+
 	// SecretRef specifies the Secret containing the token used
 	// to validate the payload authenticity.
 	// +required

config/crd/bases/notification.toolkit.fluxcd.io_receivers.yaml

@@ -62,6 +62,16 @@ spec:
                   Secret references.
                 pattern: ^([0-9]+(\.[0-9]+)?(ms|s|m|h))+$
                 type: string
+              resourceFilter:
+                description: |-
+                  ResourceFilter is a CEL expression expected to return a boolean that is
+                  evaluated for each resource referenced in the Resources field when a
+                  webhook is received. If the expression returns false then the controller
+                  will not request a reconciliation for the resource.
+                  When the expression is specified the controller will parse it and mark
+                  the object as terminally failed if the expression is invalid or does not
+                  return a boolean.
+                type: string
               resources:
                 description: A list of resources to be notified about changes.
                 items:

docs/api/v1/notification.md

@@ -122,6 +122,24 @@ e.g. &lsquo;push&rsquo; for GitHub or &lsquo;Push Hook&rsquo; for GitLab.</p>
 </tr>
 <tr>
 <td>
+<code>resourceFilter</code><br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>ResourceFilter is a CEL expression expected to return a boolean that is
+evaluated for each resource referenced in the Resources field when a
+webhook is received. If the expression returns false then the controller
+will not request a reconciliation for the resource.
+When the expression is specified the controller will parse it and mark
+the object as terminally failed if the expression is invalid or does not
+return a boolean.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>secretRef</code><br>
 <em>
 <a href="https://pkg.go.dev/github.com/fluxcd/pkg/apis/meta#LocalObjectReference">
@@ -321,6 +339,24 @@ e.g. &lsquo;push&rsquo; for GitHub or &lsquo;Push Hook&rsquo; for GitLab.</p>
 </tr>
 <tr>
 <td>
+<code>resourceFilter</code><br>
+<em>
+string
+</em>
+</td>
+<td>
+<em>(Optional)</em>
+<p>ResourceFilter is a CEL expression expected to return a boolean that is
+evaluated for each resource referenced in the Resources field when a
+webhook is received. If the expression returns false then the controller
+will not request a reconciliation for the resource.
+When the expression is specified the controller will parse it and mark
+the object as terminally failed if the expression is invalid or does not
+return a boolean.</p>
+</td>
+</tr>
+<tr>
+<td>
 <code>secretRef</code><br>
 <em>
 <a href="https://pkg.go.dev/github.com/fluxcd/pkg/apis/meta#LocalObjectReference">

docs/spec/v1/receivers.md

@@ -700,6 +700,75 @@ resources:
 **Note:** Cross-namespace references [can be disabled for security
 reasons](#disabling-cross-namespace-selectors).
 
+#### Filtering reconciled objects with CEL
+
+To filter the resources that are reconciled you can use [Common Expression Language (CEL)](https://cel.dev/).
+
+For example, to trigger `ImageRepositories` on notifications from [Google Artifact Registry](https://cloud.google.com/artifact-registry/docs/configure-notifications#examples) you can define the following receiver:
+
+```yaml
+apiVersion: notification.toolkit.fluxcd.io/v1
+kind: Receiver
+metadata:
+  name: gar-receiver
+  namespace: apps
+spec:
+  type: gcr
+  secretRef:
+    name: flux-gar-token
+  resources:
+    - apiVersion: image.toolkit.fluxcd.io/v1beta2
+      kind: ImageRepository
+      name: "*"
+      matchLabels:
+        registry: gar
+```
+
+This will trigger the reconciliation of all `ImageRepositories` with the label `registry: gar`.
+
+But if you want to only notify `ImageRepository` resources that are referenced from the incoming hook you can use CEL to filter the resources.
+
+```yaml
+apiVersion: notification.toolkit.fluxcd.io/v1
+kind: Receiver
+metadata:
+  name: gar-receiver
+  namespace: apps
+spec:
+  type: gcr
+  secretRef:
+    name: flux-gar-token
+  resources:
+    - apiVersion: image.toolkit.fluxcd.io/v1beta2
+      kind: ImageRepository
+      name: "*"
+      matchLabels:
+        registry: gar
+  resourceFilter: 'req.tag.contains(res.metadata.name)'
+```
+
+If the body of the incoming hook looks like this:
+
+```json
+{
+  "action":"INSERT",
+  "digest":"us-east1-docker.pkg.dev/my-project/my-repo/hello-world@sha256:6ec128e26cd5...",
+  "tag":"us-east1-docker.pkg.dev/my-project/my-repo/hello-world:1.1"
+}
+```
+
+This simple example would match `ImageRepositories` containing the name `hello-world`.
+
+If you want to do more complex processing:
+
+```yaml
+  resourceFilter: has(res.metadata.annotations) && req.tag.split('/').last().value().split(":").first().value() == res.metadata.annotations['update-image']
+```
+
+This would look for an annotation "update-image" on the resource, and match it to the `hello-world` part of the tag name.
+
+**Note:** Currently the `resource` value in the CEL expression only provides the object metadata, this means you can access things like `res.metadata.labels`, `res.metadata.annotations` and `res.metadata.name`.
+
 ### Secret reference
 
 `.spec.secretRef.name` is a required field to specify a name reference to a

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/fluxcd/pkg/ssa v0.44.0
 	github.com/getsentry/sentry-go v0.31.1
 	github.com/go-logr/logr v1.4.2
+	github.com/google/cel-go v0.23.1
 	github.com/google/go-github/v64 v64.0.0
 	github.com/hashicorp/go-retryablehttp v0.7.7
 	github.com/ktrysmt/go-bitbucket v0.9.81
@@ -51,6 +52,7 @@ require (
 replace gopkg.in/yaml.v3 => gopkg.in/yaml.v3 v3.0.1
 
 require (
+	cel.dev/expr v0.19.1 // indirect
 	cloud.google.com/go v0.116.0 // indirect
 	cloud.google.com/go/auth v0.12.1 // indirect
 	cloud.google.com/go/auth/oauth2adapt v0.2.6 // indirect
@@ -75,6 +77,7 @@ require (
 	github.com/DataDog/zstd v1.5.2 // indirect
 	github.com/MakeNowJust/heredoc v1.0.0 // indirect
 	github.com/ProtonMail/go-crypto v1.1.5 // indirect
+	github.com/antlr4-go/antlr/v4 v4.13.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/bradleyfalzon/ghinstallation/v2 v2.13.0 // indirect
@@ -92,6 +95,7 @@ require (
 	github.com/fatih/color v1.16.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/fluxcd/pkg/apis/acl v0.6.0 // indirect
+	github.com/fluxcd/pkg/apis/kustomize v1.9.0 // indirect
 	github.com/fluxcd/pkg/auth v0.3.0 // indirect
 	github.com/fsnotify/fsnotify v1.8.0 // indirect
 	github.com/fxamacker/cbor/v2 v2.7.0 // indirect
@@ -160,6 +164,7 @@ require (
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/santhosh-tekuri/jsonschema/v6 v6.0.1 // indirect
 	github.com/spf13/cobra v1.8.1 // indirect
+	github.com/stoewer/go-strcase v1.3.0 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/xlab/treeprint v1.2.0 // indirect
 	go.opencensus.io v0.24.0 // indirect
@@ -172,6 +177,7 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.32.0 // indirect
+	golang.org/x/exp v0.0.0-20240904232852-e7e105dedf7e // indirect
 	golang.org/x/mod v0.22.0 // indirect
 	golang.org/x/net v0.34.0 // indirect
 	golang.org/x/sync v0.10.0 // indirect

go.sum

@@ -1,3 +1,5 @@
+cel.dev/expr v0.19.1 h1:NciYrtDRIR0lNCnH1LFJegdjspNx9fI59O7TWcua/W4=
+cel.dev/expr v0.19.1/go.mod h1:MrpN08Q+lEBs+bGYdLxxHkZoUSsCp0nSKTs0nTymJgw=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.116.0 h1:B3fRrSDkLRt5qSHWe40ERJvhvnQwdZiHu0bJOpldweE=
 cloud.google.com/go v0.116.0/go.mod h1:cEPSRWPzZEswwdr9BxE6ChEn01dWlTaF05LiC2Xs70U=
@@ -79,6 +81,8 @@ github.com/PagerDuty/go-pagerduty v1.8.0 h1:MTFqTffIcAervB83U7Bx6HERzLbyaSPL/+ox
 github.com/PagerDuty/go-pagerduty v1.8.0/go.mod h1:nzIeAqyFSJAFkjWKvMzug0JtwDg+V+UoCWjFrfFH5mI=
 github.com/ProtonMail/go-crypto v1.1.5 h1:eoAQfK2dwL+tFSFpr7TbOaPNUbPiJj4fLYwwGE1FQO4=
 github.com/ProtonMail/go-crypto v1.1.5/go.mod h1:rA3QumHc/FZ8pAHreoekgiAbzpNsfQAosU5td4SnOrE=
+github.com/antlr4-go/antlr/v4 v4.13.0 h1:lxCg3LAv+EUK6t1i0y1V6/SLeUi0eKEKdhQAlS8TVTI=
+github.com/antlr4-go/antlr/v4 v4.13.0/go.mod h1:pfChB/xh/Unjila75QW7+VU4TSnWnnk9UTnmpPaOR2g=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 h1:0CwZNZbxp69SHPdPJAN/hZIm0C4OItdklCFmMRWYpio=
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
@@ -146,6 +150,8 @@ github.com/fluxcd/pkg/apis/acl v0.6.0 h1:rllf5uQLzTow81ZCslkQ6LPpDNqVQr6/fWaNksd
 github.com/fluxcd/pkg/apis/acl v0.6.0/go.mod h1:IVDZx3MAoDWjlLrJHMF9Z27huFuXAEQlnbWw0M6EcTs=
 github.com/fluxcd/pkg/apis/event v0.16.0 h1:ffKc/3erowPnh72lFszz7sPQhLZ7bhqNrq+pu1Pb+JE=
 github.com/fluxcd/pkg/apis/event v0.16.0/go.mod h1:D/QQi5lHT9/Ur3OMFLJO71D4KDQHbJ5s8dQV3h1ZAT0=
+github.com/fluxcd/pkg/apis/kustomize v1.9.0 h1:SJpT1CK58AnTvCpDKeGfMNA0Xud/4VReZNvPe8XkTxo=
+github.com/fluxcd/pkg/apis/kustomize v1.9.0/go.mod h1:AZl2GU03oPVue6SUivdiIYd/3mvF94j7t1G2JO26d4s=
 github.com/fluxcd/pkg/apis/meta v1.10.0 h1:rqbAuyl5ug7A5jjRf/rNwBXmNl6tJ9wG2iIsriwnQUk=
 github.com/fluxcd/pkg/apis/meta v1.10.0/go.mod h1:n7NstXHDaleAUMajcXTVkhz0MYkvEXy1C/eLI/t1xoI=
 github.com/fluxcd/pkg/auth v0.3.0 h1:I1A3e81O+bpAgEcJ3e+rXqObKPjzBu6FLYXQTSxXLOs=
@@ -226,6 +232,8 @@ github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek
 github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg=
 github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4=
+github.com/google/cel-go v0.23.1 h1:91ThhEZlBcE5rB2adBVXqvDoqdL8BG2oyhd0bK1I/r4=
+github.com/google/cel-go v0.23.1/go.mod h1:52Pb6QsDbC5kvgxvZhiL9QX1oZEkcUF/ZqaPx1J5Wwo=
 github.com/google/gnostic-models v0.6.9 h1:MU/8wDLif2qCXZmzncUQ/BOfxWfthHi63KqpoNbWqVw=
 github.com/google/gnostic-models v0.6.9/go.mod h1:CiWsm0s6BSQd1hRn8/QmxqB6BesYcbSZxsz9b0KuDBw=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
@@ -396,6 +404,8 @@ github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3k
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o=
 github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stoewer/go-strcase v1.3.0 h1:g0eASXYtp+yvN9fK8sH94oCIk0fau9uV1/ZdJ0AVEzs=
+github.com/stoewer/go-strcase v1.3.0/go.mod h1:fAH5hQ5pehh+j3nZfvwdk2RgEgQjAoM8wodgtPmh1xo=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
@@ -459,6 +469,8 @@ golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v
 golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
 golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
 golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
+golang.org/x/exp v0.0.0-20240904232852-e7e105dedf7e h1:I88y4caeGeuDQxgdoFPUq097j7kNfw6uvuiNxUBfcBk=
+golang.org/x/exp v0.0.0-20240904232852-e7e105dedf7e/go.mod h1:akd2r19cwCdwSwWeIdzYQGa/EZZyqcOdwWiwj5L5eKQ=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=

internal/controller/receiver_controller.go

@@ -156,14 +156,29 @@ func (r *ReceiverReconciler) Reconcile(ctx context.Context, req ctrl.Request) (r
 // reconcile steps through the actual reconciliation tasks for the object, it returns early on the first step that
 // produces an error.
 func (r *ReceiverReconciler) reconcile(ctx context.Context, obj *apiv1.Receiver) (ctrl.Result, error) {
+	log := ctrl.LoggerFrom(ctx)
+
+	if filter := obj.Spec.ResourceFilter; filter != "" {
+		if err := server.ValidateResourceFilter(filter); err != nil {
+			const msg = "Reconciliation failed terminally due to configuration error"
+			errMsg := fmt.Sprintf("%s: %v", msg, err)
+			conditions.MarkFalse(obj, meta.ReadyCondition, meta.InvalidCELExpressionReason, "%s", errMsg)
+			conditions.MarkStalled(obj, meta.InvalidCELExpressionReason, "%s", errMsg)
+			obj.Status.ObservedGeneration = obj.Generation
+			log.Error(err, msg)
+			r.Event(obj, corev1.EventTypeWarning, meta.InvalidCELExpressionReason, errMsg)
+			return ctrl.Result{}, nil
+		}
+	}
+
 	// Mark the resource as under reconciliation.
 	conditions.MarkReconciling(obj, meta.ProgressingReason, "Reconciliation in progress")
 
 	token, err := r.token(ctx, obj)
 	if err != nil {
 		conditions.MarkFalse(obj, meta.ReadyCondition, apiv1.TokenNotFoundReason, "%s", err)
 		obj.Status.WebhookPath = ""
-		return ctrl.Result{Requeue: true}, err
+		return ctrl.Result{}, err
 	}
 
 	webhookPath := obj.GetWebhookPath(token)
@@ -174,7 +189,7 @@ func (r *ReceiverReconciler) reconcile(ctx context.Context, obj *apiv1.Receiver)
 
 	if obj.Status.WebhookPath != webhookPath {
 		obj.Status.WebhookPath = webhookPath
-		ctrl.LoggerFrom(ctx).Info(msg)
+		log.Info(msg)
 	}
 
 	return ctrl.Result{RequeueAfter: obj.GetInterval()}, nil

internal/controller/receiver_controller_test.go

@@ -144,6 +144,45 @@ func TestReceiverReconciler_Reconcile(t *testing.T) {
 		g.Expect(resultR.Spec.Interval.Duration).To(BeIdenticalTo(10 * time.Minute))
 	})
 
+	t.Run("fails with invalid CEL resource filter", func(t *testing.T) {
+		g := NewWithT(t)
+		g.Expect(k8sClient.Get(context.Background(), client.ObjectKeyFromObject(receiver), resultR)).To(Succeed())
+
+		// Incomplete CEL expression
+		patch := []byte(`{"spec":{"resourceFilter":"has(res.metadata.annotations"}}`)
+		g.Expect(k8sClient.Patch(context.Background(), resultR, client.RawPatch(types.MergePatchType, patch))).To(Succeed())
+
+		g.Eventually(func() bool {
+			_ = k8sClient.Get(context.Background(), client.ObjectKeyFromObject(receiver), resultR)
+			return !conditions.IsReady(resultR)
+		}, timeout, time.Second).Should(BeTrue())
+
+		g.Expect(resultR.Status.ObservedGeneration).To(Equal(resultR.Generation))
+
+		g.Expect(conditions.GetReason(resultR, meta.ReadyCondition)).To(BeIdenticalTo(meta.InvalidCELExpressionReason))
+		g.Expect(conditions.GetMessage(resultR, meta.ReadyCondition)).To(ContainSubstring("annotations"))
+
+		g.Expect(conditions.Has(resultR, meta.StalledCondition)).To(BeTrue())
+		g.Expect(conditions.GetReason(resultR, meta.StalledCondition)).To(BeIdenticalTo(meta.InvalidCELExpressionReason))
+		g.Expect(conditions.GetObservedGeneration(resultR, meta.StalledCondition)).To(BeIdenticalTo(resultR.Generation))
+	})
+
+	t.Run("recovers when the CEL expression is valid", func(t *testing.T) {
+		g := NewWithT(t)
+		// Incomplete CEL expression
+		patch := []byte(`{"spec":{"resourceFilter":"has(res.metadata.annotations)"}}`)
+		g.Expect(k8sClient.Patch(context.Background(), resultR, client.RawPatch(types.MergePatchType, patch))).To(Succeed())
+
+		g.Eventually(func() bool {
+			_ = k8sClient.Get(context.Background(), client.ObjectKeyFromObject(receiver), resultR)
+			return conditions.IsReady(resultR)
+		}, timeout, time.Second).Should(BeTrue())
+
+		g.Expect(conditions.GetObservedGeneration(resultR, meta.ReadyCondition)).To(BeIdenticalTo(resultR.Generation))
+		g.Expect(resultR.Status.ObservedGeneration).To(BeIdenticalTo(resultR.Generation))
+		g.Expect(conditions.Has(resultR, meta.ReconcilingCondition)).To(BeFalse())
+	})
+
 	t.Run("fails with secret not found error", func(t *testing.T) {
 		g := NewWithT(t)