Support per-step timeout field in Kptfile pipeline

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.

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
 }
 

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 {

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()

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 }

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

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

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",

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

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