Add ttl input parameter

.golangci.yml

@@ -117,6 +117,7 @@ linters:
       check-type-assertions: false
       check-blank: false
     goconst:
+      ignore-tests: true
       min-len: 3
       min-occurrences: 5
     gocritic:

README.md

@@ -86,6 +86,25 @@ spec:
  See the [RunFunctionRequest][proto] protobuf message for schema details. The
  [introduction to CEL][cel-intro] documentation shows more example expressions.
 
+## Function Response Caching
+
+You can set the `ttl` input to control the Function response cache time-to-live.
+This is useful for tuning reconciliation behavior in large compositions.
+
+```yaml
+  - step: filter-composed-resources
+    functionRef:
+      name: function-cel-filter
+    input:
+      apiVersion: cel.fn.crossplane.io/v1beta1
+      kind: Filters
+      filters:
+      # Only create the bucket if the XR's spec.export field is set to "S3".
+       - name: bucket
+         expression: observed.composite.resource.spec.export == "S3"
+      ttl: "5m"
+```
+
 [functions]: https://docs.crossplane.io/latest/concepts/composition-functions
 [cel]: https://github.com/google/cel-spec
 [filter]: https://en.wikipedia.org/wiki/Filter_(higher-order_function)

fn.go

@@ -3,9 +3,11 @@ package main
 import (
 	"context"
 	"regexp"
+	"time"
 
 	"cel.dev/cel-go/cel"
 	"github.com/crossplane-contrib/function-cel-filter/input/v1beta1"
+	"google.golang.org/protobuf/types/known/durationpb"
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/crossplane/function-sdk-go/errors"
@@ -20,6 +22,7 @@ type Function struct {
 	fnv1.UnimplementedFunctionRunnerServiceServer
 
 	log logging.Logger
+	ttl time.Duration
 	env *cel.Env
 }
 
@@ -38,14 +41,21 @@ func NewFunction(log logging.Logger) (*Function, error) {
 func (f *Function) RunFunction(_ context.Context, req *fnv1.RunFunctionRequest) (*fnv1.RunFunctionResponse, error) {
 	f.log.Info("Running function", "tag", req.GetMeta().GetTag())
 
-	rsp := response.To(req, response.DefaultTTL)
+	rsp := response.To(req, f.ttl)
 
 	in := &v1beta1.Filters{}
 	if err := request.GetInput(req, in); err != nil {
 		response.Fatal(rsp, errors.Wrapf(err, "cannot get filters from %T", req))
 		return rsp, nil
 	}
-
+	if in.TTL != "" {
+		dur, err := time.ParseDuration(in.TTL)
+		if err != nil {
+			response.Fatal(rsp, errors.Wrapf(err, "cannot set ttl"))
+			return rsp, nil
+		}
+		rsp.Meta.Ttl = durationpb.New(dur)
+	}
 	regexps := make([]*regexp.Regexp, len(in.Filters))
 	celexps := make([]bool, len(in.Filters))
 

fn_test.go

@@ -102,11 +102,30 @@ func TestRunFunction(t *testing.T) {
 				},
 			},
 		},
+		"InputTTL": {
+			reason: "The function should use the TTL specified in the input",
+			args: args{
+				req: &fnv1.RunFunctionRequest{
+					Meta: &fnv1.RequestMeta{Tag: "hello"},
+					Input: resource.MustStructJSON(`{
+						"apiVersion": "filters.cel.crossplane.io/v1beta1",
+						"kind": "Filters",
+						"ttl": "5m0s"
+					}`),
+				},
+			},
+			want: want{
+				rsp: &fnv1.RunFunctionResponse{
+					Meta: &fnv1.ResponseMeta{Tag: "hello", Ttl: durationpb.New(5 * response.DefaultTTL)},
+				},
+			},
+		},
 	}
 
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
 			f, _ := NewFunction(logging.NewNopLogger())
+			f.ttl = response.DefaultTTL
 			rsp, err := f.RunFunction(tc.args.ctx, tc.args.req)
 
 			if diff := cmp.Diff(tc.want.rsp, rsp, protocmp.Transform()); diff != "" {

input/v1beta1/input.go

@@ -25,6 +25,10 @@ type Filters struct {
 	// resource will be included. Desired composed resources that don't match
 	// any filter are always included.
 	Filters []Filter `json:"filters"`
+
+	// TTL for which a response can be cached in time.Duration format
+	// +optional
+	TTL string `json:"ttl"`
 }
 
 // A Filter can be used to filter a desired composed resource produced by a

main.go

@@ -2,19 +2,25 @@
 package main
 
 import (
+	"time"
+
 	"github.com/alecthomas/kong"
 
 	"github.com/crossplane/function-sdk-go"
+	"github.com/crossplane/function-sdk-go/response"
 )
 
 // CLI of this Function.
 type CLI struct {
 	Debug bool `help:"Emit debug logs in addition to info logs." short:"d"`
 
-	Network     string `default:"tcp"                                                                                        help:"Network on which to listen for gRPC connections."`
-	Address     string `default:":9443"                                                                                      help:"Address at which to listen for gRPC connections."`
-	TLSCertsDir string `env:"TLS_SERVER_CERTS_DIR"                                                                           help:"Directory containing server certs (tls.key, tls.crt) and the CA used to verify client certificates (ca.crt)"`
-	Insecure    bool   `help:"Run without mTLS credentials. If you supply this flag --tls-server-certs-dir will be ignored."`
+	Network     string `default:"tcp"              help:"Network on which to listen for gRPC connections."`
+	Address     string `default:":9443"            help:"Address at which to listen for gRPC connections."`
+	TLSCertsDir string `env:"TLS_SERVER_CERTS_DIR" help:"Directory containing server certs (tls.key, tls.crt) and the CA used to verify client certificates (ca.crt)"`
+
+	Insecure           bool           `help:"Run without mTLS credentials. If you supply this flag --tls-server-certs-dir will be ignored."`
+	MaxRecvMessageSize int            `default:"4"                                                                                          help:"Maximum size of received messages in MB."`
+	TTL                *time.Duration `default:"1m"                                                                                         help:"Time to live for function response."`
 }
 
 // Run this Function.
@@ -24,15 +30,22 @@ func (c *CLI) Run() error {
 		return err
 	}
 
+	ttl := response.DefaultTTL
+	if c.TTL != nil {
+		ttl = *c.TTL
+	}
+
 	fn, err := NewFunction(log)
 	if err != nil {
 		return err
 	}
+	fn.ttl = ttl
 
 	return function.Serve(fn,
 		function.Listen(c.Network, c.Address),
 		function.MTLSCertificates(c.TLSCertsDir),
-		function.Insecure(c.Insecure))
+		function.Insecure(c.Insecure),
+		function.MaxRecvMessageSize(c.MaxRecvMessageSize*1024*1024))
 }
 
 func main() {

package/input/cel.fn.crossplane.io_filters.yaml

@@ -3,7 +3,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
-    controller-gen.kubebuilder.io/version: v0.20.1
+    controller-gen.kubebuilder.io/version: v0.22.0
   name: filters.cel.fn.crossplane.io
 spec:
   group: cel.fn.crossplane.io
@@ -81,6 +81,9 @@ spec:
             type: string
           metadata:
             type: object
+          ttl:
+            description: TTL for which a response can be cached in time.Duration format
+            type: string
         required:
         - filters
         type: object