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 37ac34b3b4..f7c62680eb 100644 --- a/commands/fn/render/cmdrender.go +++ b/commands/fn/render/cmdrender.go @@ -66,6 +66,9 @@ 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 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, or for its timeout from the Kptfile if it sets one.") cmdutil.FixDocs("kpt", parent, c) r.Command = c return r @@ -97,6 +100,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..c8bf87993d 100644 --- a/documentation/content/en/reference/cli/fn/render/_index.md +++ b/documentation/content/en/reference/cli/fn/render/_index.md @@ -80,6 +80,15 @@ 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 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, 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/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 0b2b24a16e..69d2fddca0 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,15 @@ 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 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, 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/container.go b/pkg/fn/runtime/container.go index 06db5e732c..95cc9f274d 100644 --- a/pkg/fn/runtime/container.go +++ b/pkg/fn/runtime/container.go @@ -82,8 +82,8 @@ type ContainerFn struct { Image string // ImagePullPolicy controls the image pulling behavior. ImagePullPolicy runneroptions.ImagePullPolicy - // Container function will be killed after this timeour. - // The default value is 5 minutes. + // Container function will be killed after this timeout. + // 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 @@ -140,13 +140,18 @@ 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 + ctx, cancel := fnContext(f.Ctx, f.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 newTimeoutExecError(ctx, err, filterCLIOutputFn(&errSink)) + } if exitErr, ok := goerrors.AsType[*exec.ExitError](err); ok { return &ExecError{ OriginalErr: exitErr, @@ -166,7 +171,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 +203,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/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 e5bdf881a3..036ceb2f2f 100644 --- a/pkg/fn/runtime/exec.go +++ b/pkg/fn/runtime/exec.go @@ -27,6 +27,10 @@ import ( "github.com/kptdev/kpt/pkg/printer" ) +// 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 { // Path is the os specific path to the executable // file. It can be relative or absolute. @@ -35,8 +39,9 @@ type ExecFn struct { Args []string Env map[string]string - // Container function will be killed after this timeour. - // The default value is 5 minutes. + // Exec function will be killed after this timeout. + // 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. @@ -46,15 +51,20 @@ 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 - timeout := defaultLongTimeout - if f.Timeout != 0 { - timeout = f.Timeout - } - ctx, cancel := context.WithTimeout(context.Background(), timeout) + ctx, cancel := fnContext(ctx, f.Timeout) 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 +76,9 @@ 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 newTimeoutExecError(ctx, err, errSink.String()) + } 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..9e8412e54b --- /dev/null +++ b/pkg/fn/runtime/exec_test.go @@ -0,0 +1,118 @@ +// 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" + "context" + "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 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, + 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, + 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", + 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) { + ctx, cancel := WithRunTimeout(context.Background(), tc.runTimeout) + defer cancel() + fn := &ExecFn{ + Path: shBin, + Args: []string{"-c", tc.script}, + Timeout: tc.timeout, + FnResult: &fnresultv1.Result{}, + } + + start := time.Now() + err := fn.RunContext(ctx, &bytes.Buffer{}, &bytes.Buffer{}) + if tc.expectedTimeout == 0 { + assert.NoError(t, err) + return + } + + var execErr *ExecError + require.ErrorAs(t, err, &execErr) + 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 + 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..8909611af0 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,14 @@ type ExecError struct { // ExitCode is the exit code returned from function ExitCode int `yaml:"exitCode,omitempty"` + + // 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. @@ -55,9 +64,25 @@ func (fe *ExecError) String() string { } b.WriteString(errLine.String()) fmt.Fprintf(&b, " Exit Code: %d\n", fe.ExitCode) + 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("pipeline execution timeout of %v exceeded", fe.Timeout) + default: + return fmt.Sprintf("function execution timeout of %v exceeded", 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 08dc4cd4ea..ae7035cec4 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,31 @@ 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: function execution timeout of 30s exceeded +`, + }, + { + 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: pipeline execution timeout of 30s exceeded `, }, { diff --git a/pkg/fn/runtime/runner.go b/pkg/fn/runtime/runner.go index 51629dd741..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,9 +189,12 @@ func NewRunner( eFn := &ExecFn{ Path: execPath, Args: execArgs, + Timeout: fnTimeout, 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") @@ -567,6 +580,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 msg := fnErr.timeoutMessage(); msg != "" { + pr.Printf(" Timed out: %s\n", msg) + } } // 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..0a31a53131 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,106 @@ func TestPrintFnResult(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, 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"}, + "", 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(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/fn/runtime/timeout.go b/pkg/fn/runtime/timeout.go new file mode 100644 index 0000000000..9158997c5d --- /dev/null +++ b/pkg/fn/runtime/timeout.go @@ -0,0 +1,87 @@ +// 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() { + // Nothing to cancel: ctx is returned as it is. + } + } + 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..d2846397f5 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,178 @@ 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: 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) +} + +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 diff --git a/pkg/lib/runneroptions/runneroptions.go b/pkg/lib/runneroptions/runneroptions.go index 421250168d..02f08f03d7 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,14 @@ type RunnerOptions struct { CelCostLimit uint64 ExtraTlsCerts [][]byte + + // 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 } type PackageIDType int8 @@ -170,6 +179,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..694f4f3044 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: function execution timeout of 100ms exceeded") + require.Len(t, r.fnResults.Items, 1) + assert.Equal(t, -1, r.fnResults.Items[0].ExitCode) +}