From 03c814bfa60b3a65c41b14d38cb415a1bdbba6d9 Mon Sep 17 00:00:00 2001 From: Nikshay Date: Sun, 11 Oct 2026 22:10:22 +0530 Subject: [PATCH 1/6] Add --timeout flag to kpt fn render and kpt fn eval The container and exec function runtimes accept a timeout, but nothing set it, so every function ran with the default of 5 minutes. Add a --timeout flag to both commands and pass it to the runtimes. A function that times out is now reported with its stderr and a 'Timed out after' line, and kpt no longer waits past the timeout for a child process that keeps the output open. Part of #4473 Assisted-by: Claude:claude-opus-5-5 Assisted-by: Codex:gpt-6.1-sol Signed-off-by: Nikshay --- commands/fn/render/cmdrender.go | 5 ++ commands/fn/render/cmdrender_test.go | 46 +++++++++- .../en/reference/cli/fn/eval/_index.md | 5 ++ .../en/reference/cli/fn/render/_index.md | 5 ++ internal/docs/generated/fndocs/docs.go | 10 +++ pkg/fn/runtime/container.go | 29 ++++--- pkg/fn/runtime/exec.go | 23 ++++- pkg/fn/runtime/exec_test.go | 86 +++++++++++++++++++ pkg/fn/runtime/fnerrors.go | 8 ++ pkg/fn/runtime/fnerrors_test.go | 13 +++ pkg/fn/runtime/runner.go | 5 ++ pkg/fn/runtime/runner_test.go | 25 ++++++ pkg/lib/runneroptions/runneroptions.go | 14 +++ .../cmdconfig/commands/cmdeval/cmdeval.go | 5 ++ .../commands/cmdeval/cmdeval_test.go | 20 ++++- thirdparty/kyaml/runfn/runfn.go | 2 + thirdparty/kyaml/runfn/runfn_test.go | 29 +++++++ 17 files changed, 317 insertions(+), 13 deletions(-) create mode 100644 pkg/fn/runtime/exec_test.go diff --git a/commands/fn/render/cmdrender.go b/commands/fn/render/cmdrender.go index 37ac34b3b4..d5b27cdf2c 100644 --- a/commands/fn/render/cmdrender.go +++ b/commands/fn/render/cmdrender.go @@ -66,6 +66,8 @@ func NewRunner(ctx context.Context, parent string) *Runner { "allow functions to access network during pipeline execution.") c.Flags().BoolVar(&r.RunnerOptions.AllowWasm, "allow-alpha-wasm", r.RunnerOptions.AllowWasm, "allow wasm to be used during pipeline execution.") + c.Flags().DurationVar(&r.RunnerOptions.Timeout, "timeout", 0, + "maximum time each function is allowed to run, e.g. 30s or 10m (default 5m)") cmdutil.FixDocs("kpt", parent, c) r.Command = c return r @@ -97,6 +99,9 @@ func (r *Runner) InitDefaults() { } func (r *Runner) preRunE(_ *cobra.Command, args []string) error { + if err := r.RunnerOptions.ValidateTimeout(); err != nil { + return err + } if len(args) == 0 { // no pkg path specified, default to current working dir wd, err := os.Getwd() diff --git a/commands/fn/render/cmdrender_test.go b/commands/fn/render/cmdrender_test.go index c777855464..d947fa99dd 100644 --- a/commands/fn/render/cmdrender_test.go +++ b/commands/fn/render/cmdrender_test.go @@ -1,4 +1,4 @@ -// Copyright 2021 The kpt Authors +// Copyright 2021,2026 The kpt Authors // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -20,6 +20,7 @@ import ( "runtime" "strings" "testing" + "time" "github.com/kptdev/kpt/internal/testutil" "github.com/kptdev/kpt/pkg/printer/fake" @@ -49,5 +50,48 @@ func TestCmd_flagAndArgParsing_Symlink(t *testing.T) { assert.Equal(t, strings.ToLower(filepath.Join("path", "to", "pkg", "dir")), strings.ToLower(r.pkgPath)) } +func TestCmd_timeoutFlag(t *testing.T) { + testCases := map[string]struct { + args []string + expected time.Duration + expectedErr string + }{ + "not set uses the runtime default": { + args: []string{}, + expected: 0, + }, + "valid duration": { + args: []string{"--timeout", "30s"}, + expected: 30 * time.Second, + }, + "negative duration": { + args: []string{"--timeout", "-1s"}, + expectedErr: "invalid timeout -1s: must not be negative", + }, + "not a duration": { + args: []string{"--timeout", "soon"}, + expectedErr: "invalid argument \"soon\" for \"--timeout\" flag", + }, + } + + for tn, tc := range testCases { + t.Run(tn, func(t *testing.T) { + r := NewRunner(fake.CtxWithDefaultPrinter(), "kpt") + r.Command.RunE = NoOpRunE + r.Command.SilenceErrors = true + r.Command.SilenceUsage = true + r.Command.SetArgs(append([]string{t.TempDir()}, tc.args...)) + + err := r.Command.Execute() + if tc.expectedErr != "" { + assert.ErrorContains(t, err, tc.expectedErr) + return + } + assert.NoError(t, err) + assert.Equal(t, tc.expected, r.RunnerOptions.Timeout) + }) + } +} + // NoOpRunE is a noop function to replace the run function of a command. Useful for testing argument parsing. var NoOpRunE = func(_ *cobra.Command, _ []string) error { return nil } diff --git a/documentation/content/en/reference/cli/fn/eval/_index.md b/documentation/content/en/reference/cli/fn/eval/_index.md index a3a6588172..39025ed206 100644 --- a/documentation/content/en/reference/cli/fn/eval/_index.md +++ b/documentation/content/en/reference/cli/fn/eval/_index.md @@ -180,6 +180,11 @@ fn-args: b) the literal value of `--tag`, if it is not a semantic version constraint. See the constraints spec here: https://github.com/Masterminds/semver +--timeout: + Maximum time the function is allowed to run, e.g. `30s` or `10m`. + If the function has not finished by then, the command fails. Default: `5m`. + Note that this is applicable to container and exec based functions only. + --type, t; Specify the function type. Accept value `mutator` (default), `validator`. If used with `--save`, this flag will save the evaluated function to the corresponding diff --git a/documentation/content/en/reference/cli/fn/render/_index.md b/documentation/content/en/reference/cli/fn/render/_index.md index 3661834620..e1a845767d 100644 --- a/documentation/content/en/reference/cli/fn/render/_index.md +++ b/documentation/content/en/reference/cli/fn/render/_index.md @@ -80,6 +80,11 @@ PKG_PATH: it doesn't exist. Structured results emitted by the functions are aggregated and saved to `results.yaml` file in the specified directory. If not specified, no result files are written to the local filesystem. + +--timeout: + Maximum time each function in the pipeline is allowed to run, e.g. `30s` or `10m`. + If a function has not finished by then, the render fails. Default: `5m`. + Note that this is applicable to container and exec based functions only. ``` #### Kptfile Annotations diff --git a/internal/docs/generated/fndocs/docs.go b/internal/docs/generated/fndocs/docs.go index 0b2b24a16e..cf7a9b5e82 100644 --- a/internal/docs/generated/fndocs/docs.go +++ b/internal/docs/generated/fndocs/docs.go @@ -185,6 +185,11 @@ Flags: b) the literal value of ` + "`" + `--tag` + "`" + `, if it is not a semantic version constraint. See the constraints spec here: https://github.com/Masterminds/semver + --timeout: + Maximum time the function is allowed to run, e.g. ` + "`" + `30s` + "`" + ` or ` + "`" + `10m` + "`" + `. + If the function has not finished by then, the command fails. Default: ` + "`" + `5m` + "`" + `. + Note that this is applicable to container and exec based functions only. + --type, t; Specify the function type. Accept value ` + "`" + `mutator` + "`" + ` (default), ` + "`" + `validator` + "`" + `. If used with ` + "`" + `--save` + "`" + `, this flag will save the evaluated function to the corresponding @@ -338,6 +343,11 @@ Flags: it doesn't exist. Structured results emitted by the functions are aggregated and saved to ` + "`" + `results.yaml` + "`" + ` file in the specified directory. If not specified, no result files are written to the local filesystem. + + --timeout: + Maximum time each function in the pipeline is allowed to run, e.g. ` + "`" + `30s` + "`" + ` or ` + "`" + `10m` + "`" + `. + If a function has not finished by then, the render fails. Default: ` + "`" + `5m` + "`" + `. + Note that this is applicable to container and exec based functions only. Kptfile Annotations: diff --git a/pkg/fn/runtime/container.go b/pkg/fn/runtime/container.go index 06db5e732c..7d0f6433a0 100644 --- a/pkg/fn/runtime/container.go +++ b/pkg/fn/runtime/container.go @@ -82,7 +82,7 @@ type ContainerFn struct { Image string // ImagePullPolicy controls the image pulling behavior. ImagePullPolicy runneroptions.ImagePullPolicy - // Container function will be killed after this timeour. + // Container function will be killed after this timeout. // The default value is 5 minutes. Timeout time.Duration Perm ContainerFnPermission @@ -140,13 +140,28 @@ func (f *ContainerFn) Run(reader io.Reader, writer io.Writer) error { func (f *ContainerFn) runCLI(reader io.Reader, writer io.Writer, bin string, filterCLIOutputFn func(io.Reader) string) error { errSink := bytes.Buffer{} - cmd, cancel := f.getCmd(bin) + // setup container run timeout + timeout := defaultLongTimeout + if f.Timeout != 0 { + timeout = f.Timeout + } + ctx, cancel := context.WithTimeout(context.Background(), timeout) defer cancel() + cmd := f.getCmd(ctx, bin) cmd.Stdin = reader cmd.Stdout = writer cmd.Stderr = &errSink if err := cmd.Run(); err != nil { + if goerrors.Is(ctx.Err(), context.DeadlineExceeded) { + return &ExecError{ + OriginalErr: err, + ExitCode: timeoutExitCode, + Stderr: filterCLIOutputFn(&errSink), + Timeout: timeout, + TruncateOutput: printer.TruncateOutput, + } + } if exitErr, ok := goerrors.AsType[*exec.ExitError](err); ok { return &ExecError{ OriginalErr: exitErr, @@ -166,7 +181,7 @@ func (f *ContainerFn) runCLI(reader io.Reader, writer io.Writer, bin string, fil // getCmd assembles a command for docker, podman or nerdctl. The input binName // is expected to be one of "docker", "podman" and "nerdctl". -func (f *ContainerFn) getCmd(binName string) (*exec.Cmd, context.CancelFunc) { +func (f *ContainerFn) getCmd(ctx context.Context, binName string) *exec.Cmd { network := networkNameNone if f.Perm.AllowNetwork { network = networkNameHost @@ -198,13 +213,7 @@ func (f *ContainerFn) getCmd(binName string) (*exec.Cmd, context.CancelFunc) { } args = append(args, NewContainerEnvFromStringSlice(f.Env).GetDockerFlags()...) args = append(args, f.Image) - // setup container run timeout - timeout := defaultLongTimeout - if f.Timeout != 0 { - timeout = f.Timeout - } - ctx, cancel := context.WithTimeout(context.Background(), timeout) - return exec.CommandContext(ctx, binName, args...), cancel + return exec.CommandContext(ctx, binName, args...) } // NewContainerEnvFromStringSlice returns a new ContainerEnv pointer with parsing diff --git a/pkg/fn/runtime/exec.go b/pkg/fn/runtime/exec.go index e5bdf881a3..5ef7668fe0 100644 --- a/pkg/fn/runtime/exec.go +++ b/pkg/fn/runtime/exec.go @@ -27,6 +27,15 @@ import ( "github.com/kptdev/kpt/pkg/printer" ) +const ( + // execWaitDelay is how long kpt waits for the output of an exec function + // to be closed after the function has exited or has been killed. + execWaitDelay = time.Second + // timeoutExitCode is the exit code reported for a function that timed out. + // It is the exit code of a process that was killed by a signal. + timeoutExitCode = -1 +) + type ExecFn struct { // Path is the os specific path to the executable // file. It can be relative or absolute. @@ -35,7 +44,7 @@ type ExecFn struct { Args []string Env map[string]string - // Container function will be killed after this timeour. + // Container function will be killed after this timeout. // The default value is 5 minutes. Timeout time.Duration // FnResult is used to store the information about the result from @@ -55,6 +64,9 @@ func (f *ExecFn) Run(r io.Reader, w io.Writer) error { defer cancel() cmd := exec.CommandContext(ctx, f.Path, f.Args...) + // do not wait past the timeout for a child process of the function that + // still has the output open + cmd.WaitDelay = execWaitDelay errSink := bytes.Buffer{} cmd.Stdin = r @@ -66,6 +78,15 @@ func (f *ExecFn) Run(r io.Reader, w io.Writer) error { } if err := cmd.Run(); err != nil { + if goerrors.Is(ctx.Err(), context.DeadlineExceeded) { + return &ExecError{ + OriginalErr: err, + ExitCode: timeoutExitCode, + Stderr: errSink.String(), + Timeout: timeout, + TruncateOutput: printer.TruncateOutput, + } + } if exitErr, ok := goerrors.AsType[*exec.ExitError](err); ok { return &ExecError{ OriginalErr: exitErr, diff --git a/pkg/fn/runtime/exec_test.go b/pkg/fn/runtime/exec_test.go new file mode 100644 index 0000000000..e4d38ed827 --- /dev/null +++ b/pkg/fn/runtime/exec_test.go @@ -0,0 +1,86 @@ +// Copyright 2026 The kpt 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 runtime + +import ( + "bytes" + "os/exec" + "testing" + "time" + + fnresultv1 "github.com/kptdev/kpt/api/fnresult/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestExecFn_Timeout(t *testing.T) { + shBin, err := exec.LookPath("sh") + if err != nil { + t.Skip("skipping exec timeout test: no sh binary found") + } + + testCases := map[string]struct { + script string + timeout time.Duration + expectTimeout bool + expectedStderr string + }{ + "function that outlives the timeout fails": { + script: "echo stuck >&2; exec sleep 30", + timeout: 100 * time.Millisecond, + expectTimeout: true, + expectedStderr: "stuck\n", + }, + "child process of the function that keeps the output open is not waited for": { + script: "sleep 30; true", + timeout: 100 * time.Millisecond, + expectTimeout: true, + }, + "function that finishes within the timeout succeeds": { + script: "true", + timeout: 30 * time.Second, + }, + "zero timeout falls back to the default": { + script: "true", + timeout: 0, + }, + } + + for tn, tc := range testCases { + t.Run(tn, func(t *testing.T) { + fn := &ExecFn{ + Path: shBin, + Args: []string{"-c", tc.script}, + Timeout: tc.timeout, + FnResult: &fnresultv1.Result{}, + } + + start := time.Now() + err := fn.Run(&bytes.Buffer{}, &bytes.Buffer{}) + if !tc.expectTimeout { + assert.NoError(t, err) + return + } + + var execErr *ExecError + require.ErrorAs(t, err, &execErr) + assert.Equal(t, tc.timeout, execErr.Timeout) + assert.Equal(t, timeoutExitCode, execErr.ExitCode) + assert.Equal(t, tc.expectedStderr, execErr.Stderr) + // kpt must not have waited for the function to finish + assert.Less(t, time.Since(start), 10*time.Second) + }) + } +} diff --git a/pkg/fn/runtime/fnerrors.go b/pkg/fn/runtime/fnerrors.go index eaa69c5693..967885a8b2 100644 --- a/pkg/fn/runtime/fnerrors.go +++ b/pkg/fn/runtime/fnerrors.go @@ -17,6 +17,7 @@ package runtime import ( "fmt" "strings" + "time" "github.com/kptdev/kpt/pkg/lib/runneroptions" ) @@ -41,6 +42,10 @@ type ExecError struct { // ExitCode is the exit code returned from function ExitCode int `yaml:"exitCode,omitempty"` + + // Timeout is the time limit that the function exceeded. It is zero if the + // function did not time out. + Timeout time.Duration `yaml:"-"` } // String returns string representation of the failure. @@ -55,6 +60,9 @@ func (fe *ExecError) String() string { } b.WriteString(errLine.String()) fmt.Fprintf(&b, " Exit Code: %d\n", fe.ExitCode) + if fe.Timeout != 0 { + fmt.Fprintf(&b, " Timed Out After: %v\n", fe.Timeout) + } return b.String() } diff --git a/pkg/fn/runtime/fnerrors_test.go b/pkg/fn/runtime/fnerrors_test.go index 08dc4cd4ea..5d85e56263 100644 --- a/pkg/fn/runtime/fnerrors_test.go +++ b/pkg/fn/runtime/fnerrors_test.go @@ -17,6 +17,7 @@ package runtime import ( "bytes" "testing" + "time" "github.com/stretchr/testify/assert" ) @@ -44,6 +45,18 @@ error message2`, }, expected: `[Stderr]: error message1, error message2 Exit Code: 1 +`, + }, + { + name: "no truncate - timeout", + fnExecError: ExecError{ + Stderr: "error message", + ExitCode: -1, + Timeout: 30 * time.Second, + }, + expected: `[Stderr]: +error message Exit Code: -1 + Timed Out After: 30s `, }, { diff --git a/pkg/fn/runtime/runner.go b/pkg/fn/runtime/runner.go index 51629dd741..58a127c776 100644 --- a/pkg/fn/runtime/runner.go +++ b/pkg/fn/runtime/runner.go @@ -143,6 +143,7 @@ func NewRunner( cfn := &ContainerFn{ Image: f.Image, ImagePullPolicy: opts.ImagePullPolicy, + Timeout: opts.Timeout, Perm: ContainerFnPermission{ AllowNetwork: opts.AllowNetwork, // mounts are disabled for render operations (currently) @@ -179,6 +180,7 @@ func NewRunner( eFn := &ExecFn{ Path: execPath, Args: execArgs, + Timeout: opts.Timeout, FnResult: fnResult, } fltr.Run = eFn.Run @@ -567,6 +569,9 @@ func printFnExecErr(ctx context.Context, fnErr *ExecError) { pr := printer.FromContextOrDie(ctx) printFnStderr(ctx, fnErr.Stderr) pr.Printf(" Exit code: %d\n", fnErr.ExitCode) + if fnErr.Timeout != 0 { + pr.Printf(" Timed out after: %v\n", fnErr.Timeout) + } } // printFnStderr prints given stdErr in a user friendly format on kpt CLI. diff --git a/pkg/fn/runtime/runner_test.go b/pkg/fn/runtime/runner_test.go index fffcecb875..6097b8ce59 100644 --- a/pkg/fn/runtime/runner_test.go +++ b/pkg/fn/runtime/runner_test.go @@ -20,9 +20,11 @@ import ( "bytes" "context" "os" + "os/exec" "path/filepath" "strings" "testing" + "time" fnresultv1 "github.com/kptdev/kpt/api/fnresult/v1" kptfilev1 "github.com/kptdev/kpt/api/kptfile/v1" @@ -1120,3 +1122,26 @@ func TestPrintFnResult(t *testing.T) { }) } } + +func TestNewRunner_ExecTimeout(t *testing.T) { + if _, err := exec.LookPath("sleep"); err != nil { + t.Skip("skipping exec timeout test: no sleep binary found") + } + + out := &bytes.Buffer{} + ctx := printer.WithContext(context.Background(), printer.New(out, out)) + opts := runneroptions.RunnerOptions{} + opts.InitDefaults(runneroptions.GHCRImagePrefix) + opts.Timeout = 100 * time.Millisecond + + fnResults := fnresultv1.NewResultList() + fr, err := NewRunner(ctx, filesys.MakeFsInMemory(), &kptfilev1.Function{Exec: "sleep 30"}, + "", fnResults, opts, nil) + require.NoError(t, err) + + _, err = fr.Filter(nil) + assert.Error(t, err) + assert.Contains(t, out.String(), "Timed out after: 100ms") + require.Len(t, fnResults.Items, 1) + assert.Equal(t, timeoutExitCode, fnResults.Items[0].ExitCode) +} diff --git a/pkg/lib/runneroptions/runneroptions.go b/pkg/lib/runneroptions/runneroptions.go index 421250168d..1fd1497f35 100644 --- a/pkg/lib/runneroptions/runneroptions.go +++ b/pkg/lib/runneroptions/runneroptions.go @@ -20,6 +20,7 @@ import ( "net/url" "os" "strings" + "time" ) const ( @@ -85,6 +86,11 @@ type RunnerOptions struct { CelCostLimit uint64 ExtraTlsCerts [][]byte + + // Timeout is the maximum time a single container or exec function is + // allowed to run before it is killed. When it is zero, the default timeout + // of the function runtime (5 minutes) is used. + Timeout time.Duration } type PackageIDType int8 @@ -170,6 +176,14 @@ func (opts *RunnerOptions) ResolveToImage(image string) string { return image } +// ValidateTimeout checks that Timeout is not negative. +func (opts *RunnerOptions) ValidateTimeout() error { + if opts.Timeout < 0 { + return fmt.Errorf("invalid timeout %v: must not be negative", opts.Timeout) + } + return nil +} + // ValidatePrefix checks that ImagePrefix is a valid registry path. func (opts *RunnerOptions) ValidatePrefix() error { if opts.ImagePrefix == "" { diff --git a/thirdparty/cmdconfig/commands/cmdeval/cmdeval.go b/thirdparty/cmdconfig/commands/cmdeval/cmdeval.go index 8888368186..c21676a915 100644 --- a/thirdparty/cmdconfig/commands/cmdeval/cmdeval.go +++ b/thirdparty/cmdconfig/commands/cmdeval/cmdeval.go @@ -120,6 +120,8 @@ func GetEvalFnRunner(ctx context.Context, parent string) *EvalFnRunner { &r.excludeLabels, "exclude-labels", []string{}, "exclude resources matching the given labels") r.Command.Flags().StringVar( &r.CelCondition, "when", "", "CEL expression to determine if the function should be run") + r.Command.Flags().DurationVar( + &r.RunnerOptions.Timeout, "timeout", 0, "maximum time the function is allowed to run, e.g. 30s or 10m (default 5m)") if err := r.Command.Flags().MarkHidden("include-meta-resources"); err != nil { panic(err) @@ -465,6 +467,9 @@ func (r *EvalFnRunner) preRunE(c *cobra.Command, args []string) error { if err := r.RunnerOptions.ValidatePrefix(); err != nil { return err } + if err := r.RunnerOptions.ValidateTimeout(); err != nil { + return err + } if err := r.validateOptionalFlags(); err != nil { return err } diff --git a/thirdparty/cmdconfig/commands/cmdeval/cmdeval_test.go b/thirdparty/cmdconfig/commands/cmdeval/cmdeval_test.go index 7e03b56886..eb132d6af0 100644 --- a/thirdparty/cmdconfig/commands/cmdeval/cmdeval_test.go +++ b/thirdparty/cmdconfig/commands/cmdeval/cmdeval_test.go @@ -1,4 +1,4 @@ -// Copyright 2019 The Kubernetes Authors. +// Copyright 2019,2026 The Kubernetes Authors. // SPDX-License-Identifier: Apache-2.0 package cmdeval @@ -12,6 +12,7 @@ import ( "runtime" "strings" "testing" + "time" "github.com/kptdev/kpt/internal/testutil" "github.com/kptdev/kpt/pkg/lib/runneroptions" @@ -50,6 +51,7 @@ func TestRunFnCommand_preRunE(t *testing.T) { fnConfigPath string network bool mount []string + timeout time.Duration }{ { name: "config map", @@ -257,6 +259,17 @@ apiVersion: v1 args: []string{"eval", dir, "--image", "foo:bar", "--", "a=b", "c", "e=f"}, err: "must have keys and values separated by", }, + { + name: "timeout", + args: []string{"eval", dir, "--image", "foo:bar", "--timeout", "30s"}, + path: dir, + timeout: 30 * time.Second, + }, + { + name: "negative timeout", + args: []string{"eval", dir, "--image", "foo:bar", "--timeout", "-1s"}, + err: "invalid timeout -1s: must not be negative", + }, { name: "envs", args: []string{"eval", dir, "--env", "FOO=BAR", "-e", "BAR", "--image", "foo:bar"}, @@ -407,6 +420,11 @@ apiVersion: v1 t.FailNow() } + // check if the timeout was passed to the function runner + if !assert.Equal(t, tt.timeout, r.runFns.RunnerOptions.Timeout) { + t.FailNow() + } + if !assert.Equal(t, toStorageMounts(tt.mount), r.runFns.StorageMounts) { t.FailNow() } diff --git a/thirdparty/kyaml/runfn/runfn.go b/thirdparty/kyaml/runfn/runfn.go index 2bb88c22eb..a4b5901508 100644 --- a/thirdparty/kyaml/runfn/runfn.go +++ b/thirdparty/kyaml/runfn/runfn.go @@ -410,6 +410,7 @@ func (r *RunFns) defaultFnFilterProvider(spec runtimeutil.FunctionSpec, fnConfig c := &fnruntime.ContainerFn{ Image: resolvedImage, ImagePullPolicy: r.RunnerOptions.ImagePullPolicy, + Timeout: r.RunnerOptions.Timeout, UIDGID: uidgid, StorageMounts: r.StorageMounts, Env: spec.Container.Env, @@ -438,6 +439,7 @@ func (r *RunFns) defaultFnFilterProvider(spec runtimeutil.FunctionSpec, fnConfig e := &fnruntime.ExecFn{ Path: spec.Exec.Path, Args: r.ExecArgs, + Timeout: r.RunnerOptions.Timeout, FnResult: fnResult, } fltr.Run = e.Run diff --git a/thirdparty/kyaml/runfn/runfn_test.go b/thirdparty/kyaml/runfn/runfn_test.go index cf76b79a5c..3490b1d23f 100644 --- a/thirdparty/kyaml/runfn/runfn_test.go +++ b/thirdparty/kyaml/runfn/runfn_test.go @@ -5,13 +5,18 @@ package runfn import ( "bytes" + "context" "os" + "os/exec" "path/filepath" "runtime" "testing" + "time" fnresultv1 "github.com/kptdev/kpt/api/fnresult/v1" kptfilev1 "github.com/kptdev/kpt/api/kptfile/v1" + "github.com/kptdev/kpt/pkg/lib/runneroptions" + "github.com/kptdev/kpt/pkg/printer" "github.com/kptdev/kpt/pkg/printer/fake" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -671,3 +676,27 @@ func TestRunFns_mergeContainerEnv(t *testing.T) { }) } } + +func TestDefaultFnFilterProvider_execTimeout(t *testing.T) { + sleepBin, err := exec.LookPath("sleep") + if err != nil { + t.Skip("skipping exec timeout test: no sleep binary found") + } + + out := &bytes.Buffer{} + r := &RunFns{ + Ctx: printer.WithContext(context.Background(), printer.New(out, out)), + ExecArgs: []string{"30"}, + fnResults: fnresultv1.NewResultList(), + RunnerOptions: runneroptions.RunnerOptions{Timeout: 100 * time.Millisecond}, + } + fltr, err := r.defaultFnFilterProvider( + runtimeutil.FunctionSpec{Exec: runtimeutil.ExecSpec{Path: sleepBin}}, nil, nil) + require.NoError(t, err) + + _, err = fltr.Filter(nil) + assert.Error(t, err) + assert.Contains(t, out.String(), "Timed out after: 100ms") + require.Len(t, r.fnResults.Items, 1) + assert.Equal(t, -1, r.fnResults.Items[0].ExitCode) +} From 85507056cec0ab9cfc165ab1e1a72c840421348b Mon Sep 17 00:00:00 2001 From: Nikshay Date: Sun, 11 Oct 2026 22:22:57 +0530 Subject: [PATCH 2/6] Add a container timeout test and fix the ExecFn.Timeout comment Assisted-by: Claude:claude-opus-5-5 Signed-off-by: Nikshay --- pkg/fn/runtime/container_timeout_test.go | 62 ++++++++++++++++++++++++ pkg/fn/runtime/exec.go | 2 +- 2 files changed, 63 insertions(+), 1 deletion(-) create mode 100644 pkg/fn/runtime/container_timeout_test.go diff --git a/pkg/fn/runtime/container_timeout_test.go b/pkg/fn/runtime/container_timeout_test.go new file mode 100644 index 0000000000..280d6c2612 --- /dev/null +++ b/pkg/fn/runtime/container_timeout_test.go @@ -0,0 +1,62 @@ +// Copyright 2026 The kpt 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 runtime + +import ( + "bytes" + "os" + "path/filepath" + "runtime" + "testing" + "time" + + fnresultv1 "github.com/kptdev/kpt/api/fnresult/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestContainerFn_Timeout(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("skipping container timeout test on Windows") + } + + // A stand-in for the container CLI, so that the test does not need a + // container runtime. It prints a CLI message and a function message, and + // then does not finish. + fakeCLI := filepath.Join(t.TempDir(), "fake-container-cli") + script := `#!/bin/sh +echo "Unable to find image 'example.com/fn:v1' locally" >&2 +echo "function is stuck" >&2 +exec sleep 30 +` + require.NoError(t, os.WriteFile(fakeCLI, []byte(script), 0700)) + + fn := &ContainerFn{ + Image: "example.com/fn:v1", + Timeout: 100 * time.Millisecond, + FnResult: &fnresultv1.Result{}, + } + + start := time.Now() + err := fn.runCLI(&bytes.Buffer{}, &bytes.Buffer{}, fakeCLI, filterDockerCLIOutput) + + var execErr *ExecError + require.ErrorAs(t, err, &execErr) + assert.Equal(t, 100*time.Millisecond, execErr.Timeout) + assert.Equal(t, timeoutExitCode, execErr.ExitCode) + // the CLI message is filtered out, the message of the function is kept + assert.Equal(t, "function is stuck", execErr.Stderr) + assert.Less(t, time.Since(start), 10*time.Second) +} diff --git a/pkg/fn/runtime/exec.go b/pkg/fn/runtime/exec.go index 5ef7668fe0..e94c0001cf 100644 --- a/pkg/fn/runtime/exec.go +++ b/pkg/fn/runtime/exec.go @@ -44,7 +44,7 @@ type ExecFn struct { Args []string Env map[string]string - // Container function will be killed after this timeout. + // Exec function will be killed after this timeout. // The default value is 5 minutes. Timeout time.Duration // FnResult is used to store the information about the result from From 4894ab251c871e36da26a9e32ebdec5d1e9d263a Mon Sep 17 00:00:00 2001 From: Nikshay Date: Mon, 12 Oct 2026 20:41:06 +0530 Subject: [PATCH 3/6] Make --timeout limit the whole render pipeline Signed-off-by: Nikshay Assisted-by: Claude:claude-opus-5-5 --- commands/fn/render/cmdrender.go | 3 +- .../en/reference/cli/fn/render/_index.md | 7 +- internal/docs/generated/fndocs/docs.go | 7 +- pkg/fn/runtime/container.go | 16 +--- pkg/fn/runtime/exec.go | 31 +++---- pkg/fn/runtime/exec_test.go | 59 ++++++++++--- pkg/fn/runtime/fnerrors.go | 23 ++++- pkg/fn/runtime/fnerrors_test.go | 15 +++- pkg/fn/runtime/runner.go | 7 +- pkg/fn/runtime/runner_test.go | 9 +- pkg/fn/runtime/timeout.go | 85 +++++++++++++++++++ pkg/lib/kptops/render_executor.go | 5 +- pkg/lib/kptops/render_executor_test.go | 46 ++++++++++ pkg/lib/runneroptions/runneroptions.go | 9 +- thirdparty/kyaml/runfn/runfn_test.go | 2 +- 15 files changed, 255 insertions(+), 69 deletions(-) create mode 100644 pkg/fn/runtime/timeout.go diff --git a/commands/fn/render/cmdrender.go b/commands/fn/render/cmdrender.go index d5b27cdf2c..96bffda627 100644 --- a/commands/fn/render/cmdrender.go +++ b/commands/fn/render/cmdrender.go @@ -67,7 +67,8 @@ func NewRunner(ctx context.Context, parent string) *Runner { c.Flags().BoolVar(&r.RunnerOptions.AllowWasm, "allow-alpha-wasm", r.RunnerOptions.AllowWasm, "allow wasm to be used during pipeline execution.") c.Flags().DurationVar(&r.RunnerOptions.Timeout, "timeout", 0, - "maximum time each function is allowed to run, e.g. 30s or 10m (default 5m)") + "maximum time the whole pipeline is allowed to run, e.g. 30s or 10m. "+ + "When it is not set, each function is allowed to run for 5m.") cmdutil.FixDocs("kpt", parent, c) r.Command = c return r diff --git a/documentation/content/en/reference/cli/fn/render/_index.md b/documentation/content/en/reference/cli/fn/render/_index.md index e1a845767d..0bb2b93e82 100644 --- a/documentation/content/en/reference/cli/fn/render/_index.md +++ b/documentation/content/en/reference/cli/fn/render/_index.md @@ -82,8 +82,11 @@ PKG_PATH: If not specified, no result files are written to the local filesystem. --timeout: - Maximum time each function in the pipeline is allowed to run, e.g. `30s` or `10m`. - If a function has not finished by then, the render fails. Default: `5m`. + Maximum time the whole pipeline is allowed to run, e.g. `30s` or `10m`. + This includes the functions of all subpackages. If the pipeline has not + finished by then, the function that is running is stopped and the render fails. + When this flag is not set, there is no limit for the whole pipeline, and each + function is allowed to run for 5 minutes. Note that this is applicable to container and exec based functions only. ``` diff --git a/internal/docs/generated/fndocs/docs.go b/internal/docs/generated/fndocs/docs.go index cf7a9b5e82..1d78c33047 100644 --- a/internal/docs/generated/fndocs/docs.go +++ b/internal/docs/generated/fndocs/docs.go @@ -345,8 +345,11 @@ Flags: If not specified, no result files are written to the local filesystem. --timeout: - Maximum time each function in the pipeline is allowed to run, e.g. ` + "`" + `30s` + "`" + ` or ` + "`" + `10m` + "`" + `. - If a function has not finished by then, the render fails. Default: ` + "`" + `5m` + "`" + `. + Maximum time the whole pipeline is allowed to run, e.g. ` + "`" + `30s` + "`" + ` or ` + "`" + `10m` + "`" + `. + This includes the functions of all subpackages. If the pipeline has not + finished by then, the function that is running is stopped and the render fails. + When this flag is not set, there is no limit for the whole pipeline, and each + function is allowed to run for 5 minutes. Note that this is applicable to container and exec based functions only. Kptfile Annotations: diff --git a/pkg/fn/runtime/container.go b/pkg/fn/runtime/container.go index 7d0f6433a0..95cc9f274d 100644 --- a/pkg/fn/runtime/container.go +++ b/pkg/fn/runtime/container.go @@ -83,7 +83,7 @@ type ContainerFn struct { // ImagePullPolicy controls the image pulling behavior. ImagePullPolicy runneroptions.ImagePullPolicy // Container function will be killed after this timeout. - // The default value is 5 minutes. + // The default value is 5 minutes, or the time limit of Ctx if it has one. Timeout time.Duration Perm ContainerFnPermission // UIDGID is the os User ID and Group ID that will be @@ -141,11 +141,7 @@ func (f *ContainerFn) Run(reader io.Reader, writer io.Writer) error { func (f *ContainerFn) runCLI(reader io.Reader, writer io.Writer, bin string, filterCLIOutputFn func(io.Reader) string) error { errSink := bytes.Buffer{} // setup container run timeout - timeout := defaultLongTimeout - if f.Timeout != 0 { - timeout = f.Timeout - } - ctx, cancel := context.WithTimeout(context.Background(), timeout) + ctx, cancel := fnContext(f.Ctx, f.Timeout) defer cancel() cmd := f.getCmd(ctx, bin) cmd.Stdin = reader @@ -154,13 +150,7 @@ func (f *ContainerFn) runCLI(reader io.Reader, writer io.Writer, bin string, fil if err := cmd.Run(); err != nil { if goerrors.Is(ctx.Err(), context.DeadlineExceeded) { - return &ExecError{ - OriginalErr: err, - ExitCode: timeoutExitCode, - Stderr: filterCLIOutputFn(&errSink), - Timeout: timeout, - TruncateOutput: printer.TruncateOutput, - } + return newTimeoutExecError(ctx, err, filterCLIOutputFn(&errSink)) } if exitErr, ok := goerrors.AsType[*exec.ExitError](err); ok { return &ExecError{ diff --git a/pkg/fn/runtime/exec.go b/pkg/fn/runtime/exec.go index e94c0001cf..9ab045b363 100644 --- a/pkg/fn/runtime/exec.go +++ b/pkg/fn/runtime/exec.go @@ -27,16 +27,15 @@ import ( "github.com/kptdev/kpt/pkg/printer" ) -const ( - // execWaitDelay is how long kpt waits for the output of an exec function - // to be closed after the function has exited or has been killed. - execWaitDelay = time.Second - // timeoutExitCode is the exit code reported for a function that timed out. - // It is the exit code of a process that was killed by a signal. - timeoutExitCode = -1 -) +// execWaitDelay is how long kpt waits for the output of an exec function +// to be closed after the function has exited or has been killed. +const execWaitDelay = time.Second type ExecFn struct { + // Ctx is the context of the run. If it has a time limit, the function is + // killed when that limit has passed. + Ctx context.Context + // Path is the os specific path to the executable // file. It can be relative or absolute. Path string @@ -45,7 +44,7 @@ type ExecFn struct { Env map[string]string // Exec function will be killed after this timeout. - // The default value is 5 minutes. + // The default value is 5 minutes, or the time limit of Ctx if it has one. Timeout time.Duration // FnResult is used to store the information about the result from // the function. @@ -56,11 +55,7 @@ type ExecFn struct { // writes the output to w. func (f *ExecFn) Run(r io.Reader, w io.Writer) error { // setup exec run timeout - timeout := defaultLongTimeout - if f.Timeout != 0 { - timeout = f.Timeout - } - ctx, cancel := context.WithTimeout(context.Background(), timeout) + ctx, cancel := fnContext(f.Ctx, f.Timeout) defer cancel() cmd := exec.CommandContext(ctx, f.Path, f.Args...) @@ -79,13 +74,7 @@ func (f *ExecFn) Run(r io.Reader, w io.Writer) error { if err := cmd.Run(); err != nil { if goerrors.Is(ctx.Err(), context.DeadlineExceeded) { - return &ExecError{ - OriginalErr: err, - ExitCode: timeoutExitCode, - Stderr: errSink.String(), - Timeout: timeout, - TruncateOutput: printer.TruncateOutput, - } + return newTimeoutExecError(ctx, err, errSink.String()) } if exitErr, ok := goerrors.AsType[*exec.ExitError](err); ok { return &ExecError{ diff --git a/pkg/fn/runtime/exec_test.go b/pkg/fn/runtime/exec_test.go index e4d38ed827..5d5a985819 100644 --- a/pkg/fn/runtime/exec_test.go +++ b/pkg/fn/runtime/exec_test.go @@ -16,6 +16,7 @@ package runtime import ( "bytes" + "context" "os/exec" "testing" "time" @@ -32,21 +33,49 @@ func TestExecFn_Timeout(t *testing.T) { } testCases := map[string]struct { - script string - timeout time.Duration - expectTimeout bool - expectedStderr string + script string + // timeout is the time limit of the function + timeout time.Duration + // runTimeout is the time limit of the whole run + runTimeout time.Duration + expectedTimeout time.Duration + expectRunTimeout bool + expectedStderr string }{ "function that outlives the timeout fails": { - script: "echo stuck >&2; exec sleep 30", - timeout: 100 * time.Millisecond, - expectTimeout: true, - expectedStderr: "stuck\n", + script: "echo stuck >&2; exec sleep 30", + timeout: 100 * time.Millisecond, + expectedTimeout: 100 * time.Millisecond, + expectedStderr: "stuck\n", }, "child process of the function that keeps the output open is not waited for": { - script: "sleep 30; true", - timeout: 100 * time.Millisecond, - expectTimeout: true, + script: "sleep 30; true", + timeout: 100 * time.Millisecond, + expectedTimeout: 100 * time.Millisecond, + }, + "function that outlives the timeout of the whole run fails": { + script: "echo stuck >&2; exec sleep 30", + runTimeout: 100 * time.Millisecond, + expectedTimeout: 100 * time.Millisecond, + expectRunTimeout: true, + expectedStderr: "stuck\n", + }, + "shorter timeout of the function comes before the timeout of the whole run": { + script: "exec sleep 30", + timeout: 100 * time.Millisecond, + runTimeout: 30 * time.Second, + expectedTimeout: 100 * time.Millisecond, + }, + "shorter timeout of the whole run comes before the timeout of the function": { + script: "exec sleep 30", + timeout: 30 * time.Second, + runTimeout: 100 * time.Millisecond, + expectedTimeout: 100 * time.Millisecond, + expectRunTimeout: true, + }, + "function that finishes within the timeout of the whole run succeeds": { + script: "true", + runTimeout: 30 * time.Second, }, "function that finishes within the timeout succeeds": { script: "true", @@ -60,7 +89,10 @@ func TestExecFn_Timeout(t *testing.T) { for tn, tc := range testCases { t.Run(tn, func(t *testing.T) { + ctx, cancel := WithRunTimeout(context.Background(), tc.runTimeout) + defer cancel() fn := &ExecFn{ + Ctx: ctx, Path: shBin, Args: []string{"-c", tc.script}, Timeout: tc.timeout, @@ -69,14 +101,15 @@ func TestExecFn_Timeout(t *testing.T) { start := time.Now() err := fn.Run(&bytes.Buffer{}, &bytes.Buffer{}) - if !tc.expectTimeout { + if tc.expectedTimeout == 0 { assert.NoError(t, err) return } var execErr *ExecError require.ErrorAs(t, err, &execErr) - assert.Equal(t, tc.timeout, execErr.Timeout) + assert.Equal(t, tc.expectedTimeout, execErr.Timeout) + assert.Equal(t, tc.expectRunTimeout, execErr.RunTimeout) assert.Equal(t, timeoutExitCode, execErr.ExitCode) assert.Equal(t, tc.expectedStderr, execErr.Stderr) // kpt must not have waited for the function to finish diff --git a/pkg/fn/runtime/fnerrors.go b/pkg/fn/runtime/fnerrors.go index 967885a8b2..6c80cad589 100644 --- a/pkg/fn/runtime/fnerrors.go +++ b/pkg/fn/runtime/fnerrors.go @@ -43,9 +43,13 @@ type ExecError struct { // ExitCode is the exit code returned from function ExitCode int `yaml:"exitCode,omitempty"` - // Timeout is the time limit that the function exceeded. It is zero if the + // Timeout is the time limit that was exceeded. It is zero if the // function did not time out. Timeout time.Duration `yaml:"-"` + + // RunTimeout is true if Timeout is the time limit of the whole run, and + // false if it is the time limit of this function. + RunTimeout bool `yaml:"-"` } // String returns string representation of the failure. @@ -60,12 +64,25 @@ func (fe *ExecError) String() string { } b.WriteString(errLine.String()) fmt.Fprintf(&b, " Exit Code: %d\n", fe.ExitCode) - if fe.Timeout != 0 { - fmt.Fprintf(&b, " Timed Out After: %v\n", fe.Timeout) + if msg := fe.timeoutMessage(); msg != "" { + fmt.Fprintf(&b, " Timed Out: %s\n", msg) } return b.String() } +// timeoutMessage describes the time limit that was exceeded. It is empty if +// the function did not time out. +func (fe *ExecError) timeoutMessage() string { + switch { + case fe.Timeout == 0: + return "" + case fe.RunTimeout: + return fmt.Sprintf("the whole run took longer than %v", fe.Timeout) + default: + return fmt.Sprintf("the function took longer than %v", fe.Timeout) + } +} + func (fe *ExecError) Error() string { return fe.String() } diff --git a/pkg/fn/runtime/fnerrors_test.go b/pkg/fn/runtime/fnerrors_test.go index 5d85e56263..65dff0ccc8 100644 --- a/pkg/fn/runtime/fnerrors_test.go +++ b/pkg/fn/runtime/fnerrors_test.go @@ -56,7 +56,20 @@ error message1, error message2 Exit Code: 1 }, expected: `[Stderr]: error message Exit Code: -1 - Timed Out After: 30s + Timed Out: the function took longer than 30s +`, + }, + { + name: "no truncate - timeout of the whole run", + fnExecError: ExecError{ + Stderr: "error message", + ExitCode: -1, + Timeout: 30 * time.Second, + RunTimeout: true, + }, + expected: `[Stderr]: +error message Exit Code: -1 + Timed Out: the whole run took longer than 30s `, }, { diff --git a/pkg/fn/runtime/runner.go b/pkg/fn/runtime/runner.go index 58a127c776..cccab3d4f2 100644 --- a/pkg/fn/runtime/runner.go +++ b/pkg/fn/runtime/runner.go @@ -143,7 +143,6 @@ func NewRunner( cfn := &ContainerFn{ Image: f.Image, ImagePullPolicy: opts.ImagePullPolicy, - Timeout: opts.Timeout, Perm: ContainerFnPermission{ AllowNetwork: opts.AllowNetwork, // mounts are disabled for render operations (currently) @@ -178,9 +177,9 @@ func NewRunner( execArgs = s[1:] } eFn := &ExecFn{ + Ctx: ctx, Path: execPath, Args: execArgs, - Timeout: opts.Timeout, FnResult: fnResult, } fltr.Run = eFn.Run @@ -569,8 +568,8 @@ func printFnExecErr(ctx context.Context, fnErr *ExecError) { pr := printer.FromContextOrDie(ctx) printFnStderr(ctx, fnErr.Stderr) pr.Printf(" Exit code: %d\n", fnErr.ExitCode) - if fnErr.Timeout != 0 { - pr.Printf(" Timed out after: %v\n", fnErr.Timeout) + if msg := fnErr.timeoutMessage(); msg != "" { + pr.Printf(" Timed out: %s\n", msg) } } diff --git a/pkg/fn/runtime/runner_test.go b/pkg/fn/runtime/runner_test.go index 6097b8ce59..e00e75f61f 100644 --- a/pkg/fn/runtime/runner_test.go +++ b/pkg/fn/runtime/runner_test.go @@ -1123,16 +1123,17 @@ func TestPrintFnResult(t *testing.T) { } } -func TestNewRunner_ExecTimeout(t *testing.T) { +func TestNewRunner_RunTimeout(t *testing.T) { if _, err := exec.LookPath("sleep"); err != nil { t.Skip("skipping exec timeout test: no sleep binary found") } out := &bytes.Buffer{} - ctx := printer.WithContext(context.Background(), printer.New(out, out)) + ctx, cancel := WithRunTimeout( + printer.WithContext(context.Background(), printer.New(out, out)), 100*time.Millisecond) + defer cancel() opts := runneroptions.RunnerOptions{} opts.InitDefaults(runneroptions.GHCRImagePrefix) - opts.Timeout = 100 * time.Millisecond fnResults := fnresultv1.NewResultList() fr, err := NewRunner(ctx, filesys.MakeFsInMemory(), &kptfilev1.Function{Exec: "sleep 30"}, @@ -1141,7 +1142,7 @@ func TestNewRunner_ExecTimeout(t *testing.T) { _, err = fr.Filter(nil) assert.Error(t, err) - assert.Contains(t, out.String(), "Timed out after: 100ms") + assert.Contains(t, out.String(), "Timed out: the whole run took longer than 100ms") require.Len(t, fnResults.Items, 1) assert.Equal(t, timeoutExitCode, fnResults.Items[0].ExitCode) } diff --git a/pkg/fn/runtime/timeout.go b/pkg/fn/runtime/timeout.go new file mode 100644 index 0000000000..d62b152609 --- /dev/null +++ b/pkg/fn/runtime/timeout.go @@ -0,0 +1,85 @@ +// Copyright 2026 The kpt 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 runtime + +import ( + "context" + goerrors "errors" + "fmt" + "time" + + "github.com/kptdev/kpt/pkg/printer" +) + +// timeoutExitCode is the exit code reported for a function that timed out. +// It is the exit code of a process that was killed by a signal. +const timeoutExitCode = -1 + +// timeoutError is the cause of a context that ended because a time limit +// has passed. +type timeoutError struct { + timeout time.Duration + // wholeRun is true for the time limit of the whole run, and false for the + // time limit of a single function. + wholeRun bool +} + +func (e *timeoutError) Error() string { + return fmt.Sprintf("timed out after %v", e.timeout) +} + +// WithRunTimeout returns a copy of ctx that ends when timeout has passed. The +// functions that are run with the returned context share this time limit, so +// it limits the whole run and not each function. When timeout is zero, ctx is +// returned as it is. +func WithRunTimeout(ctx context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { + if timeout == 0 { + return ctx, func() {} + } + return context.WithTimeoutCause(ctx, timeout, &timeoutError{timeout: timeout, wholeRun: true}) +} + +// fnContext returns the context that a single function is run with. The +// function is limited by timeout. When timeout is zero, the function is +// limited by the time limit of the whole run if parent has one, and by the +// default of 5 minutes otherwise. +func fnContext(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { + if parent == nil { + parent = context.Background() + } + if timeout == 0 { + if _, ok := parent.Deadline(); ok { + return context.WithCancel(parent) + } + timeout = defaultLongTimeout + } + return context.WithTimeoutCause(parent, timeout, &timeoutError{timeout: timeout}) +} + +// newTimeoutExecError returns the error for a function that was killed +// because the time limit of ctx has passed. +func newTimeoutExecError(ctx context.Context, err error, stderr string) *ExecError { + execErr := &ExecError{ + OriginalErr: err, + ExitCode: timeoutExitCode, + Stderr: stderr, + TruncateOutput: printer.TruncateOutput, + } + if cause, ok := goerrors.AsType[*timeoutError](context.Cause(ctx)); ok { + execErr.Timeout = cause.timeout + execErr.RunTimeout = cause.wholeRun + } + return execErr +} diff --git a/pkg/lib/kptops/render_executor.go b/pkg/lib/kptops/render_executor.go index 3e532df408..f1a1eeae33 100644 --- a/pkg/lib/kptops/render_executor.go +++ b/pkg/lib/kptops/render_executor.go @@ -107,7 +107,10 @@ func (e *Renderer) Execute(ctx context.Context) (*fnresultv1.ResultList, error) hydrateFn = hydrateBfsOrder } - _, hydErr := hydrateFn(ctx, root, hctx) + // the timeout, if set, limits the run time of the whole pipeline + hydCtx, cancel := fnruntime.WithRunTimeout(ctx, e.RunnerOptions.Timeout) + _, hydErr := hydrateFn(hydCtx, root, hctx) + cancel() if hydErr != nil && !hctx.saveOnRenderFailure { if e.Output == nil { diff --git a/pkg/lib/kptops/render_executor_test.go b/pkg/lib/kptops/render_executor_test.go index b38968a4d3..7528121fd7 100644 --- a/pkg/lib/kptops/render_executor_test.go +++ b/pkg/lib/kptops/render_executor_test.go @@ -18,9 +18,11 @@ import ( "bytes" "context" "fmt" + "os/exec" "path/filepath" "strings" "testing" + "time" fnresultv1 "github.com/kptdev/kpt/api/fnresult/v1" kptfilev1 "github.com/kptdev/kpt/api/kptfile/v1" @@ -396,6 +398,50 @@ func TestRenderer_Execute_RenderOrder(t *testing.T) { }) } } +func TestRenderer_Execute_Timeout(t *testing.T) { + if _, err := exec.LookPath("sleep"); err != nil { + t.Skip("skipping render timeout test: no sleep binary found") + } + + var outputBuffer bytes.Buffer + ctx := printer.WithContext(t.Context(), printer.New(&outputBuffer, &outputBuffer)) + + mockFileSystem := filesys.MakeFsInMemory() + assert.NoError(t, mockFileSystem.Mkdir(rootString)) + assert.NoError(t, mockFileSystem.WriteFile(filepath.Join(rootString, "Kptfile"), []byte(`apiVersion: kpt.dev/v1 +kind: Kptfile +metadata: + name: root-package +pipeline: + mutators: + - exec: sleep 30 +`))) + assert.NoError(t, mockFileSystem.WriteFile(filepath.Join(rootString, "resources.yaml"), []byte(`apiVersion: v1 +kind: ConfigMap +metadata: + name: root-cm +`))) + + opts := runneroptions.RunnerOptions{} + opts.InitDefaults(runneroptions.GHCRImagePrefix) + opts.AllowExec = true + opts.Timeout = 100 * time.Millisecond + renderer := &Renderer{ + PkgPath: rootString, + FileSystem: mockFileSystem, + RunnerOptions: opts, + // the output is not written back to the package + Output: &bytes.Buffer{}, + } + + start := time.Now() + _, err := renderer.Execute(ctx) + assert.Error(t, err) + assert.Contains(t, outputBuffer.String(), "Timed out: the whole run took longer than 100ms") + // kpt must not have waited for the function to finish + assert.Less(t, time.Since(start), 10*time.Second) +} + func TestRunPipeline_SubpkgNameDisplay(t *testing.T) { tests := []struct { name string diff --git a/pkg/lib/runneroptions/runneroptions.go b/pkg/lib/runneroptions/runneroptions.go index 1fd1497f35..02f08f03d7 100644 --- a/pkg/lib/runneroptions/runneroptions.go +++ b/pkg/lib/runneroptions/runneroptions.go @@ -87,9 +87,12 @@ type RunnerOptions struct { ExtraTlsCerts [][]byte - // Timeout is the maximum time a single container or exec function is - // allowed to run before it is killed. When it is zero, the default timeout - // of the function runtime (5 minutes) is used. + // Timeout is the maximum time the whole run is allowed to take: all the + // functions of the pipeline for `kpt fn render`, and the function for + // `kpt fn eval`. A container or exec function that is still running when + // it has passed is killed. When it is zero, there is no limit for the + // whole run, and each function is limited by the default timeout of the + // function runtime (5 minutes). Timeout time.Duration } diff --git a/thirdparty/kyaml/runfn/runfn_test.go b/thirdparty/kyaml/runfn/runfn_test.go index 3490b1d23f..3538dc6f0a 100644 --- a/thirdparty/kyaml/runfn/runfn_test.go +++ b/thirdparty/kyaml/runfn/runfn_test.go @@ -696,7 +696,7 @@ func TestDefaultFnFilterProvider_execTimeout(t *testing.T) { _, err = fltr.Filter(nil) assert.Error(t, err) - assert.Contains(t, out.String(), "Timed out after: 100ms") + assert.Contains(t, out.String(), "Timed out: the function took longer than 100ms") require.Len(t, r.fnResults.Items, 1) assert.Equal(t, -1, r.fnResults.Items[0].ExitCode) } From 4e1534cefbde6d6ad8dc14d0bdb952a49aa9ff23 Mon Sep 17 00:00:00 2001 From: Nikshay Date: Thu, 15 Oct 2026 16:52:09 +0530 Subject: [PATCH 4/6] Pass the run context to ExecFn as a parameter Signed-off-by: Nikshay Assisted-by: Claude:claude-opus-5-5 --- pkg/fn/runtime/exec.go | 15 +++++++++------ pkg/fn/runtime/exec_test.go | 3 +-- pkg/fn/runtime/runner.go | 5 +++-- pkg/fn/runtime/timeout.go | 4 +++- 4 files changed, 16 insertions(+), 11 deletions(-) diff --git a/pkg/fn/runtime/exec.go b/pkg/fn/runtime/exec.go index 9ab045b363..036ceb2f2f 100644 --- a/pkg/fn/runtime/exec.go +++ b/pkg/fn/runtime/exec.go @@ -32,10 +32,6 @@ import ( const execWaitDelay = time.Second type ExecFn struct { - // Ctx is the context of the run. If it has a time limit, the function is - // killed when that limit has passed. - Ctx context.Context - // Path is the os specific path to the executable // file. It can be relative or absolute. Path string @@ -44,7 +40,8 @@ type ExecFn struct { Env map[string]string // Exec function will be killed after this timeout. - // The default value is 5 minutes, or the time limit of Ctx if it has one. + // The default value is 5 minutes, or the time limit of the context passed + // to RunContext if it has one. Timeout time.Duration // FnResult is used to store the information about the result from // the function. @@ -54,8 +51,14 @@ type ExecFn struct { // Run runs the executable file which reads the input from r and // writes the output to w. func (f *ExecFn) Run(r io.Reader, w io.Writer) error { + return f.RunContext(context.Background(), r, w) +} + +// RunContext is like Run, but the function is also killed when the time +// limit of ctx has passed. +func (f *ExecFn) RunContext(ctx context.Context, r io.Reader, w io.Writer) error { // setup exec run timeout - ctx, cancel := fnContext(f.Ctx, f.Timeout) + ctx, cancel := fnContext(ctx, f.Timeout) defer cancel() cmd := exec.CommandContext(ctx, f.Path, f.Args...) diff --git a/pkg/fn/runtime/exec_test.go b/pkg/fn/runtime/exec_test.go index 5d5a985819..9e8412e54b 100644 --- a/pkg/fn/runtime/exec_test.go +++ b/pkg/fn/runtime/exec_test.go @@ -92,7 +92,6 @@ func TestExecFn_Timeout(t *testing.T) { ctx, cancel := WithRunTimeout(context.Background(), tc.runTimeout) defer cancel() fn := &ExecFn{ - Ctx: ctx, Path: shBin, Args: []string{"-c", tc.script}, Timeout: tc.timeout, @@ -100,7 +99,7 @@ func TestExecFn_Timeout(t *testing.T) { } start := time.Now() - err := fn.Run(&bytes.Buffer{}, &bytes.Buffer{}) + err := fn.RunContext(ctx, &bytes.Buffer{}, &bytes.Buffer{}) if tc.expectedTimeout == 0 { assert.NoError(t, err) return diff --git a/pkg/fn/runtime/runner.go b/pkg/fn/runtime/runner.go index cccab3d4f2..41003acfef 100644 --- a/pkg/fn/runtime/runner.go +++ b/pkg/fn/runtime/runner.go @@ -177,12 +177,13 @@ func NewRunner( execArgs = s[1:] } eFn := &ExecFn{ - Ctx: ctx, Path: execPath, Args: execArgs, FnResult: fnResult, } - fltr.Run = eFn.Run + fltr.Run = func(r io.Reader, w io.Writer) error { + return eFn.RunContext(ctx, r, w) + } } default: return nil, fmt.Errorf("must specify `exec` or `image` to execute a function") diff --git a/pkg/fn/runtime/timeout.go b/pkg/fn/runtime/timeout.go index d62b152609..9158997c5d 100644 --- a/pkg/fn/runtime/timeout.go +++ b/pkg/fn/runtime/timeout.go @@ -46,7 +46,9 @@ func (e *timeoutError) Error() string { // returned as it is. func WithRunTimeout(ctx context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { if timeout == 0 { - return ctx, func() {} + return ctx, func() { + // Nothing to cancel: ctx is returned as it is. + } } return context.WithTimeoutCause(ctx, timeout, &timeoutError{timeout: timeout, wholeRun: true}) } From ba5dc1698b7b0928a40e085a4a0a931fa82bad36 Mon Sep 17 00:00:00 2001 From: Nikshay Date: Thu, 15 Oct 2026 23:12:32 +0530 Subject: [PATCH 5/6] Say which timeout was exceeded in timeout errors Signed-off-by: Nikshay --- pkg/fn/runtime/fnerrors.go | 4 ++-- pkg/fn/runtime/fnerrors_test.go | 4 ++-- pkg/fn/runtime/runner_test.go | 2 +- pkg/lib/kptops/render_executor_test.go | 2 +- thirdparty/kyaml/runfn/runfn_test.go | 2 +- 5 files changed, 7 insertions(+), 7 deletions(-) diff --git a/pkg/fn/runtime/fnerrors.go b/pkg/fn/runtime/fnerrors.go index 6c80cad589..8909611af0 100644 --- a/pkg/fn/runtime/fnerrors.go +++ b/pkg/fn/runtime/fnerrors.go @@ -77,9 +77,9 @@ func (fe *ExecError) timeoutMessage() string { case fe.Timeout == 0: return "" case fe.RunTimeout: - return fmt.Sprintf("the whole run took longer than %v", fe.Timeout) + return fmt.Sprintf("pipeline execution timeout of %v exceeded", fe.Timeout) default: - return fmt.Sprintf("the function took longer than %v", fe.Timeout) + return fmt.Sprintf("function execution timeout of %v exceeded", fe.Timeout) } } diff --git a/pkg/fn/runtime/fnerrors_test.go b/pkg/fn/runtime/fnerrors_test.go index 65dff0ccc8..ae7035cec4 100644 --- a/pkg/fn/runtime/fnerrors_test.go +++ b/pkg/fn/runtime/fnerrors_test.go @@ -56,7 +56,7 @@ error message1, error message2 Exit Code: 1 }, expected: `[Stderr]: error message Exit Code: -1 - Timed Out: the function took longer than 30s + Timed Out: function execution timeout of 30s exceeded `, }, { @@ -69,7 +69,7 @@ error message Exit Code: -1 }, expected: `[Stderr]: error message Exit Code: -1 - Timed Out: the whole run took longer than 30s + Timed Out: pipeline execution timeout of 30s exceeded `, }, { diff --git a/pkg/fn/runtime/runner_test.go b/pkg/fn/runtime/runner_test.go index e00e75f61f..73ed48fb9d 100644 --- a/pkg/fn/runtime/runner_test.go +++ b/pkg/fn/runtime/runner_test.go @@ -1142,7 +1142,7 @@ func TestNewRunner_RunTimeout(t *testing.T) { _, err = fr.Filter(nil) assert.Error(t, err) - assert.Contains(t, out.String(), "Timed out: the whole run took longer than 100ms") + assert.Contains(t, out.String(), "Timed out: pipeline execution timeout of 100ms exceeded") require.Len(t, fnResults.Items, 1) assert.Equal(t, timeoutExitCode, fnResults.Items[0].ExitCode) } diff --git a/pkg/lib/kptops/render_executor_test.go b/pkg/lib/kptops/render_executor_test.go index 7528121fd7..9531b36e69 100644 --- a/pkg/lib/kptops/render_executor_test.go +++ b/pkg/lib/kptops/render_executor_test.go @@ -437,7 +437,7 @@ metadata: start := time.Now() _, err := renderer.Execute(ctx) assert.Error(t, err) - assert.Contains(t, outputBuffer.String(), "Timed out: the whole run took longer than 100ms") + assert.Contains(t, outputBuffer.String(), "Timed out: pipeline execution timeout of 100ms exceeded") // kpt must not have waited for the function to finish assert.Less(t, time.Since(start), 10*time.Second) } diff --git a/thirdparty/kyaml/runfn/runfn_test.go b/thirdparty/kyaml/runfn/runfn_test.go index 3538dc6f0a..694f4f3044 100644 --- a/thirdparty/kyaml/runfn/runfn_test.go +++ b/thirdparty/kyaml/runfn/runfn_test.go @@ -696,7 +696,7 @@ func TestDefaultFnFilterProvider_execTimeout(t *testing.T) { _, err = fltr.Filter(nil) assert.Error(t, err) - assert.Contains(t, out.String(), "Timed out: the function took longer than 100ms") + assert.Contains(t, out.String(), "Timed out: function execution timeout of 100ms exceeded") require.Len(t, r.fnResults.Items, 1) assert.Equal(t, -1, r.fnResults.Items[0].ExitCode) } From 0d88a1054e14dee55a8380c7f26f6cc82cad19f0 Mon Sep 17 00:00:00 2001 From: SurbhiAgarwal Date: Sat, 17 Oct 2026 21:59:25 +0530 Subject: [PATCH 6/6] Support timeout field in Kptfile pipeline steps Add support for specifying a timeout field on individual mutator and validator pipeline steps in the Kptfile (e.g. timeout: 30s). - Add Timeout field to Function type in api/kptfile/v1/types.go and generated applyconfigurations. - Validate that the timeout field contains a valid, positive duration. - Parse step timeout and pass to ContainerFn and ExecFn runtimes. - Maintain precedence: the shorter limit wins (pipeline-level --timeout flag limits the whole run, step-level timeout limits the individual function, and step-level timeout overrides the default 5-minute timeout). - Update Kptfile schema definitions (JSON and YAML). - Add unit tests for validation, step execution timeout, and precedence. Part of #4473 Signed-off-by: SurbhiAgarwal --- .../kptfile/v1/function.go | 12 ++ api/kptfile/v1/types.go | 6 + api/kptfile/v1/validation.go | 52 +++++-- api/kptfile/v1/validation_test.go | 70 ++++++++++ commands/fn/render/cmdrender.go | 2 +- .../en/reference/cli/fn/render/_index.md | 3 +- .../en/reference/schema/kptfile/kptfile.json | 5 + .../en/reference/schema/kptfile/kptfile.yaml | 7 + go.mod | 2 +- internal/docs/generated/fndocs/docs.go | 3 +- pkg/fn/runtime/runner.go | 11 ++ pkg/fn/runtime/runner_test.go | 79 +++++++++++ pkg/lib/kptops/render_executor_test.go | 128 ++++++++++++++++++ 13 files changed, 366 insertions(+), 14 deletions(-) diff --git a/api/generated/applyconfigurations/kptfile/v1/function.go b/api/generated/applyconfigurations/kptfile/v1/function.go index a520062f03..0b1e7becb0 100644 --- a/api/generated/applyconfigurations/kptfile/v1/function.go +++ b/api/generated/applyconfigurations/kptfile/v1/function.go @@ -77,6 +77,10 @@ type FunctionApplyConfiguration struct { // - Check resource count among the selected resources: // `when: "resources.filter(r, r.kind == 'Deployment').size() > 0"` CelCondition *string `json:"when,omitempty"` + // Timeout specifies the maximum execution time for this function step + // (e.g. "30s", "1m", "5m"). If omitted, the pipeline timeout or default + // function timeout (5m) is used. + Timeout *string `json:"timeout,omitempty"` } // FunctionApplyConfiguration constructs a declarative configuration of the Function type for use with @@ -180,3 +184,11 @@ func (b *FunctionApplyConfiguration) WithCelCondition(value string) *FunctionApp b.CelCondition = &value return b } + +// WithTimeout sets the Timeout field in the declarative configuration to the given value +// and returns the receiver, so that objects can be built by chaining "With" function invocations. +// If called multiple times, the Timeout field is set to the value of the last call. +func (b *FunctionApplyConfiguration) WithTimeout(value string) *FunctionApplyConfiguration { + b.Timeout = &value + return b +} diff --git a/api/kptfile/v1/types.go b/api/kptfile/v1/types.go index e069defa1f..f75535590f 100644 --- a/api/kptfile/v1/types.go +++ b/api/kptfile/v1/types.go @@ -391,6 +391,12 @@ type Function struct { // - Check resource count among the selected resources: // `when: "resources.filter(r, r.kind == 'Deployment').size() > 0"` CelCondition string `yaml:"when,omitempty" json:"when,omitempty"` + + // Timeout specifies the maximum execution time for this function step + // (e.g. "30s", "1m", "5m"). If omitted, the pipeline timeout or default + // function timeout (5m) is used. Applicable to container and exec based + // functions only. + Timeout string `yaml:"timeout,omitempty" json:"timeout,omitempty"` } // ResourceReference identifies a resource within the package by its API identity. diff --git a/api/kptfile/v1/validation.go b/api/kptfile/v1/validation.go index dbbf0410af..770a388f26 100644 --- a/api/kptfile/v1/validation.go +++ b/api/kptfile/v1/validation.go @@ -21,6 +21,7 @@ import ( "regexp" "slices" "strings" + "time" "github.com/distribution/reference" "sigs.k8s.io/kustomize/kyaml/filesys" @@ -120,16 +121,21 @@ func (l *Locator) validate() error { if l.Type == GenericOrigin && l.Generic == nil { return &ValidateError{Field: "upstreamLock.generic", Reason: `must be set when type is "generic"`} } - if l.Git != nil { - if l.Git.Repo == "" { - return &ValidateError{Field: "upstreamLock.git.repo", Reason: mustNotBeEmpty} - } - if l.Git.Ref == "" { - return &ValidateError{Field: "upstreamLock.git.ref", Reason: mustNotBeEmpty} - } - if l.Git.Commit == "" { - return &ValidateError{Field: "upstreamLock.git.commit", Reason: mustNotBeEmpty} - } + return l.Git.validate() +} + +func (g *GitLock) validate() error { + if g == nil { + return nil + } + if g.Repo == "" { + return &ValidateError{Field: "upstreamLock.git.repo", Reason: mustNotBeEmpty} + } + if g.Ref == "" { + return &ValidateError{Field: "upstreamLock.git.ref", Reason: mustNotBeEmpty} + } + if g.Commit == "" { + return &ValidateError{Field: "upstreamLock.git.commit", Reason: mustNotBeEmpty} } return nil } @@ -235,6 +241,32 @@ func (f *Function) validate(fsys filesys.FileSystem, fnType string, idx int, pkg if err := f.validateExecutor(fnType, idx); err != nil { return err } + if err := f.validateTimeout(fnType, idx); err != nil { + return err + } + return nil +} + +// validateTimeout checks that the optional timeout string is a valid positive duration. +func (f *Function) validateTimeout(fnType string, idx int) error { + if f.Timeout == "" { + return nil + } + d, err := time.ParseDuration(f.Timeout) + if err != nil { + return &ValidateError{ + Field: fmt.Sprintf("pipeline.%s[%d].timeout", fnType, idx), + Value: f.Timeout, + Reason: fmt.Sprintf("invalid duration %q: %v", f.Timeout, err), + } + } + if d <= 0 { + return &ValidateError{ + Field: fmt.Sprintf("pipeline.%s[%d].timeout", fnType, idx), + Value: f.Timeout, + Reason: fmt.Sprintf("timeout %q must be greater than zero", f.Timeout), + } + } return nil } diff --git a/api/kptfile/v1/validation_test.go b/api/kptfile/v1/validation_test.go index c6856e141c..853d091cc5 100644 --- a/api/kptfile/v1/validation_test.go +++ b/api/kptfile/v1/validation_test.go @@ -148,6 +148,76 @@ func TestKptfileValidate(t *testing.T) { }, valid: false, }, + { + name: "pipeline: valid mutator step timeout", + kptfile: KptFile{ + Pipeline: &Pipeline{ + Mutators: []Function{ + { + Image: "image", + Timeout: "30s", + }, + }, + }, + }, + valid: true, + }, + { + name: "pipeline: valid validator step timeout", + kptfile: KptFile{ + Pipeline: &Pipeline{ + Validators: []Function{ + { + Image: "image", + Timeout: "1m", + }, + }, + }, + }, + valid: true, + }, + { + name: "pipeline: invalid step timeout syntax", + kptfile: KptFile{ + Pipeline: &Pipeline{ + Mutators: []Function{ + { + Image: "image", + Timeout: "not-a-duration", + }, + }, + }, + }, + valid: false, + }, + { + name: "pipeline: non-positive zero step timeout", + kptfile: KptFile{ + Pipeline: &Pipeline{ + Mutators: []Function{ + { + Image: "image", + Timeout: "0s", + }, + }, + }, + }, + valid: false, + }, + { + name: "pipeline: negative step timeout", + kptfile: KptFile{ + Pipeline: &Pipeline{ + Validators: []Function{ + { + Image: "image", + Timeout: "-10s", + }, + }, + }, + }, + valid: false, + }, } for _, c := range cases { diff --git a/commands/fn/render/cmdrender.go b/commands/fn/render/cmdrender.go index 96bffda627..f7c62680eb 100644 --- a/commands/fn/render/cmdrender.go +++ b/commands/fn/render/cmdrender.go @@ -68,7 +68,7 @@ func NewRunner(ctx context.Context, parent string) *Runner { "allow wasm to be used during pipeline execution.") c.Flags().DurationVar(&r.RunnerOptions.Timeout, "timeout", 0, "maximum time the whole pipeline is allowed to run, e.g. 30s or 10m. "+ - "When it is not set, each function is allowed to run for 5m.") + "When it is not set, each function is allowed to run for 5m, or for its timeout from the Kptfile if it sets one.") cmdutil.FixDocs("kpt", parent, c) r.Command = c return r diff --git a/documentation/content/en/reference/cli/fn/render/_index.md b/documentation/content/en/reference/cli/fn/render/_index.md index 0bb2b93e82..c8bf87993d 100644 --- a/documentation/content/en/reference/cli/fn/render/_index.md +++ b/documentation/content/en/reference/cli/fn/render/_index.md @@ -86,7 +86,8 @@ PKG_PATH: This includes the functions of all subpackages. If the pipeline has not finished by then, the function that is running is stopped and the render fails. When this flag is not set, there is no limit for the whole pipeline, and each - function is allowed to run for 5 minutes. + function is allowed to run for 5 minutes, or for its `timeout` from the + Kptfile if it sets one. Note that this is applicable to container and exec based functions only. ``` diff --git a/documentation/content/en/reference/schema/kptfile/kptfile.json b/documentation/content/en/reference/schema/kptfile/kptfile.json index 615eb26197..689c5ac2c6 100644 --- a/documentation/content/en/reference/schema/kptfile/kptfile.json +++ b/documentation/content/en/reference/schema/kptfile/kptfile.json @@ -129,6 +129,11 @@ "type": "string", "x-go-name": "Tag" }, + "timeout": { + "description": "`Timeout` is an optional field for specifying the maximum time the function is allowed to run,\ne.g. '30s' or '10m'. If omitted, the function is limited by the pipeline run timeout (if set)\nor defaults to 5 minutes. Applicable to container and exec based functions only.", + "type": "string", + "x-go-name": "Timeout" + }, "when": { "description": "CelCondition is an optional CEL expression (exposed as 'when' in YAML) that determines whether this\nfunction should be executed. The expression is evaluated against the list\nof KRM resources passed to this function step (after `Selectors` and\n`Exclude` have been applied) and should return a boolean value.\nIf omitted or evaluates to true, the function executes normally.\nIf evaluates to false, the function is skipped.\n\nExamples:\n\nCheck if a specific ConfigMap exists among the selected resources:\nwhen: `\"resources.exists(r, r.kind == 'ConfigMap' && r.metadata.name == 'my-config')\"`\n\nCheck resource count among the selected resources:\n`when: \"resources.filter(r, r.kind == 'Deployment').size() > 0\"`", "type": "string", diff --git a/documentation/content/en/reference/schema/kptfile/kptfile.yaml b/documentation/content/en/reference/schema/kptfile/kptfile.yaml index 7711310a16..f90c8afe0d 100644 --- a/documentation/content/en/reference/schema/kptfile/kptfile.yaml +++ b/documentation/content/en/reference/schema/kptfile/kptfile.yaml @@ -121,6 +121,13 @@ definitions: Can be a semver constraint satisfying this spec: https://github.com/Masterminds/semver?tab=readme-ov-file#checking-version-constraints type: string x-go-name: Tag + timeout: + description: |- + `Timeout` is an optional field for specifying the maximum time the function is allowed to run, + e.g. '30s' or '10m'. If omitted, the function is limited by the pipeline run timeout (if set) + or defaults to 5 minutes. Applicable to container and exec based functions only. + type: string + x-go-name: Timeout when: description: |- CelCondition is an optional CEL expression (exposed as 'when' in YAML) that determines whether this diff --git a/go.mod b/go.mod index 28d02213af..43c6f0f420 100644 --- a/go.mod +++ b/go.mod @@ -2,7 +2,7 @@ module github.com/kptdev/kpt go 1.26.5 -//replace github.com/kptdev/kpt/api => ./api +replace github.com/kptdev/kpt/api => ./api require ( github.com/Masterminds/semver/v3 v3.5.0 diff --git a/internal/docs/generated/fndocs/docs.go b/internal/docs/generated/fndocs/docs.go index 1d78c33047..69d2fddca0 100644 --- a/internal/docs/generated/fndocs/docs.go +++ b/internal/docs/generated/fndocs/docs.go @@ -349,7 +349,8 @@ Flags: This includes the functions of all subpackages. If the pipeline has not finished by then, the function that is running is stopped and the render fails. When this flag is not set, there is no limit for the whole pipeline, and each - function is allowed to run for 5 minutes. + function is allowed to run for 5 minutes, or for its ` + "`" + `timeout` + "`" + ` from the + Kptfile if it sets one. Note that this is applicable to container and exec based functions only. Kptfile Annotations: diff --git a/pkg/fn/runtime/runner.go b/pkg/fn/runtime/runner.go index 41003acfef..dac113420f 100644 --- a/pkg/fn/runtime/runner.go +++ b/pkg/fn/runtime/runner.go @@ -130,6 +130,15 @@ func NewRunner( pkgCtxGenerator := &builtins.PackageContextGenerator{} fltr.Run = pkgCtxGenerator.Run } else { + var fnTimeout time.Duration + if f.Timeout != "" { + var err error + fnTimeout, err = time.ParseDuration(f.Timeout) + if err != nil { + return nil, fmt.Errorf("invalid function timeout %q: %w", f.Timeout, err) + } + } + switch { case f.Image != "": // If allowWasm is true, we will use wasm runtime for image field. @@ -143,6 +152,7 @@ func NewRunner( cfn := &ContainerFn{ Image: f.Image, ImagePullPolicy: opts.ImagePullPolicy, + Timeout: fnTimeout, Perm: ContainerFnPermission{ AllowNetwork: opts.AllowNetwork, // mounts are disabled for render operations (currently) @@ -179,6 +189,7 @@ func NewRunner( eFn := &ExecFn{ Path: execPath, Args: execArgs, + Timeout: fnTimeout, FnResult: fnResult, } fltr.Run = func(r io.Reader, w io.Writer) error { diff --git a/pkg/fn/runtime/runner_test.go b/pkg/fn/runtime/runner_test.go index 73ed48fb9d..0a31a53131 100644 --- a/pkg/fn/runtime/runner_test.go +++ b/pkg/fn/runtime/runner_test.go @@ -1146,3 +1146,82 @@ func TestNewRunner_RunTimeout(t *testing.T) { require.Len(t, fnResults.Items, 1) assert.Equal(t, timeoutExitCode, fnResults.Items[0].ExitCode) } + +func TestNewRunner_StepTimeout(t *testing.T) { + if _, err := exec.LookPath("sleep"); err != nil { + t.Skip("skipping exec timeout test: no sleep binary found") + } + + out := &bytes.Buffer{} + ctx := printer.WithContext(context.Background(), printer.New(out, out)) + opts := runneroptions.RunnerOptions{} + opts.InitDefaults(runneroptions.GHCRImagePrefix) + + fnResults := fnresultv1.NewResultList() + fr, err := NewRunner(ctx, filesys.MakeFsInMemory(), &kptfilev1.Function{Exec: "sleep 30", Timeout: "100ms"}, + "", fnResults, opts, nil) + require.NoError(t, err) + + _, err = fr.Filter(nil) + assert.Error(t, err) + assert.Contains(t, out.String(), "Timed out: function execution timeout of 100ms exceeded") + require.Len(t, fnResults.Items, 1) + assert.Equal(t, timeoutExitCode, fnResults.Items[0].ExitCode) +} + +func TestNewRunner_InvalidTimeout(t *testing.T) { + opts := runneroptions.RunnerOptions{} + opts.InitDefaults(runneroptions.GHCRImagePrefix) + fnResults := fnresultv1.NewResultList() + _, err := NewRunner(context.Background(), filesys.MakeFsInMemory(), &kptfilev1.Function{Exec: "sleep 30", Timeout: "not-a-duration"}, + "", fnResults, opts, nil) + assert.ErrorContains(t, err, "invalid function timeout \"not-a-duration\"") +} + +func TestNewRunner_StepTimeout_Precedence_RunTimeoutFirst(t *testing.T) { + if _, err := exec.LookPath("sleep"); err != nil { + t.Skip("skipping exec timeout test: no sleep binary found") + } + + out := &bytes.Buffer{} + ctx, cancel := WithRunTimeout( + printer.WithContext(context.Background(), printer.New(out, out)), 100*time.Millisecond) + defer cancel() + opts := runneroptions.RunnerOptions{} + opts.InitDefaults(runneroptions.GHCRImagePrefix) + + fnResults := fnresultv1.NewResultList() + fr, err := NewRunner(ctx, filesys.MakeFsInMemory(), &kptfilev1.Function{Exec: "sleep 30", Timeout: "10s"}, + "", fnResults, opts, nil) + require.NoError(t, err) + + _, err = fr.Filter(nil) + assert.Error(t, err) + assert.Contains(t, out.String(), "Timed out: pipeline execution timeout of 100ms exceeded") + require.Len(t, fnResults.Items, 1) + assert.Equal(t, timeoutExitCode, fnResults.Items[0].ExitCode) +} + +func TestNewRunner_StepTimeout_Precedence_StepTimeoutFirst(t *testing.T) { + if _, err := exec.LookPath("sleep"); err != nil { + t.Skip("skipping exec timeout test: no sleep binary found") + } + + out := &bytes.Buffer{} + ctx, cancel := WithRunTimeout( + printer.WithContext(context.Background(), printer.New(out, out)), 10*time.Second) + defer cancel() + opts := runneroptions.RunnerOptions{} + opts.InitDefaults(runneroptions.GHCRImagePrefix) + + fnResults := fnresultv1.NewResultList() + fr, err := NewRunner(ctx, filesys.MakeFsInMemory(), &kptfilev1.Function{Exec: "sleep 30", Timeout: "100ms"}, + "", fnResults, opts, nil) + require.NoError(t, err) + + _, err = fr.Filter(nil) + assert.Error(t, err) + assert.Contains(t, out.String(), "Timed out: function execution timeout of 100ms exceeded") + require.Len(t, fnResults.Items, 1) + assert.Equal(t, timeoutExitCode, fnResults.Items[0].ExitCode) +} diff --git a/pkg/lib/kptops/render_executor_test.go b/pkg/lib/kptops/render_executor_test.go index 9531b36e69..d2846397f5 100644 --- a/pkg/lib/kptops/render_executor_test.go +++ b/pkg/lib/kptops/render_executor_test.go @@ -442,6 +442,134 @@ metadata: assert.Less(t, time.Since(start), 10*time.Second) } +func TestRenderer_Execute_StepTimeout(t *testing.T) { + if _, err := exec.LookPath("sleep"); err != nil { + t.Skip("skipping render timeout test: no sleep binary found") + } + + var outputBuffer bytes.Buffer + ctx := printer.WithContext(t.Context(), printer.New(&outputBuffer, &outputBuffer)) + + mockFileSystem := filesys.MakeFsInMemory() + assert.NoError(t, mockFileSystem.Mkdir(rootString)) + assert.NoError(t, mockFileSystem.WriteFile(filepath.Join(rootString, "Kptfile"), []byte(`apiVersion: kpt.dev/v1 +kind: Kptfile +metadata: + name: root-package +pipeline: + mutators: + - exec: sleep 30 + timeout: 100ms +`))) + assert.NoError(t, mockFileSystem.WriteFile(filepath.Join(rootString, "resources.yaml"), []byte(`apiVersion: v1 +kind: ConfigMap +metadata: + name: root-cm +`))) + + opts := runneroptions.RunnerOptions{} + opts.InitDefaults(runneroptions.GHCRImagePrefix) + opts.AllowExec = true + renderer := &Renderer{ + PkgPath: rootString, + FileSystem: mockFileSystem, + RunnerOptions: opts, + Output: &bytes.Buffer{}, + } + + start := time.Now() + _, err := renderer.Execute(ctx) + assert.Error(t, err) + assert.Contains(t, outputBuffer.String(), "Timed out: function execution timeout of 100ms exceeded") + assert.Less(t, time.Since(start), 10*time.Second) +} + +func TestRenderer_Execute_StepTimeout_Precedence_CLIFirst(t *testing.T) { + if _, err := exec.LookPath("sleep"); err != nil { + t.Skip("skipping render timeout test: no sleep binary found") + } + + var outputBuffer bytes.Buffer + ctx := printer.WithContext(t.Context(), printer.New(&outputBuffer, &outputBuffer)) + + mockFileSystem := filesys.MakeFsInMemory() + assert.NoError(t, mockFileSystem.Mkdir(rootString)) + assert.NoError(t, mockFileSystem.WriteFile(filepath.Join(rootString, "Kptfile"), []byte(`apiVersion: kpt.dev/v1 +kind: Kptfile +metadata: + name: root-package +pipeline: + mutators: + - exec: sleep 30 + timeout: 10s +`))) + assert.NoError(t, mockFileSystem.WriteFile(filepath.Join(rootString, "resources.yaml"), []byte(`apiVersion: v1 +kind: ConfigMap +metadata: + name: root-cm +`))) + + opts := runneroptions.RunnerOptions{} + opts.InitDefaults(runneroptions.GHCRImagePrefix) + opts.AllowExec = true + opts.Timeout = 100 * time.Millisecond + renderer := &Renderer{ + PkgPath: rootString, + FileSystem: mockFileSystem, + RunnerOptions: opts, + Output: &bytes.Buffer{}, + } + + start := time.Now() + _, err := renderer.Execute(ctx) + assert.Error(t, err) + assert.Contains(t, outputBuffer.String(), "Timed out: pipeline execution timeout of 100ms exceeded") + assert.Less(t, time.Since(start), 10*time.Second) +} + +func TestRenderer_Execute_StepTimeout_Precedence_StepFirst(t *testing.T) { + if _, err := exec.LookPath("sleep"); err != nil { + t.Skip("skipping render timeout test: no sleep binary found") + } + + var outputBuffer bytes.Buffer + ctx := printer.WithContext(t.Context(), printer.New(&outputBuffer, &outputBuffer)) + + mockFileSystem := filesys.MakeFsInMemory() + assert.NoError(t, mockFileSystem.Mkdir(rootString)) + assert.NoError(t, mockFileSystem.WriteFile(filepath.Join(rootString, "Kptfile"), []byte(`apiVersion: kpt.dev/v1 +kind: Kptfile +metadata: + name: root-package +pipeline: + mutators: + - exec: sleep 30 + timeout: 100ms +`))) + assert.NoError(t, mockFileSystem.WriteFile(filepath.Join(rootString, "resources.yaml"), []byte(`apiVersion: v1 +kind: ConfigMap +metadata: + name: root-cm +`))) + + opts := runneroptions.RunnerOptions{} + opts.InitDefaults(runneroptions.GHCRImagePrefix) + opts.AllowExec = true + opts.Timeout = 10 * time.Second + renderer := &Renderer{ + PkgPath: rootString, + FileSystem: mockFileSystem, + RunnerOptions: opts, + Output: &bytes.Buffer{}, + } + + start := time.Now() + _, err := renderer.Execute(ctx) + assert.Error(t, err) + assert.Contains(t, outputBuffer.String(), "Timed out: function execution timeout of 100ms exceeded") + assert.Less(t, time.Since(start), 10*time.Second) +} + func TestRunPipeline_SubpkgNameDisplay(t *testing.T) { tests := []struct { name string