Add timeout setting to Steps

This feature allows a Task author to specify a Step timeout in a
Taskrun.

An example use case is when a Task author would like to execute a
Step for setting up an execution environment. One may expect this
Step to execute within a few seconds. If the execution time takes
longer than expected one may rather want to fail fast instead of
waiting for the TaskRun timeout to abort the TaskRun.

Closes tektoncd#1690

Author: Peaorl

cmd/entrypoint/main.go

@@ -17,6 +17,7 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"flag"
 	"log"
 	"os"
@@ -40,6 +41,7 @@ var (
 	terminationPath     = flag.String("termination_path", "/tekton/termination", "If specified, file to write upon termination")
 	results             = flag.String("results", "", "If specified, list of file names that might contain task results")
 	waitPollingInterval = time.Second
+	timeout             = flag.String("timeout", "", "If specified, sets timeout for step")
 )
 
 func main() {
@@ -78,7 +80,16 @@ func main() {
 		log.Printf("non-fatal error copying credentials: %q", err)
 	}
 
-	if err := e.Go(); err != nil {
+	// Add timeout to context if a non-zero timeout is specified for a step
+	ctx := context.Background()
+	timeoutContext, _ := time.ParseDuration(*timeout)
+	var cancel context.CancelFunc
+	if timeoutContext != time.Duration(0) {
+		ctx, cancel = context.WithTimeout(ctx, timeoutContext)
+		defer cancel()
+	}
+
+	if err := e.Go(ctx); err != nil {
 		switch t := err.(type) {
 		case skipError:
 			log.Print("Skipping step because a previous step failed")

cmd/entrypoint/runner.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"os"
 	"os/exec"
 	"os/signal"
@@ -19,7 +20,7 @@ type realRunner struct {
 
 var _ entrypoint.Runner = (*realRunner)(nil)
 
-func (rr *realRunner) Run(args ...string) error {
+func (rr *realRunner) Run(ctx context.Context, args ...string) error {
 	if len(args) == 0 {
 		return nil
 	}
@@ -33,7 +34,7 @@ func (rr *realRunner) Run(args ...string) error {
 	signal.Notify(rr.signals)
 	defer signal.Reset()
 
-	cmd := exec.Command(name, args...)
+	cmd := exec.CommandContext(ctx, name, args...)
 	cmd.Stdout = os.Stdout
 	cmd.Stderr = os.Stderr
 	// dedicated PID group used to forward signals to

cmd/entrypoint/runner_test.go

@@ -1,9 +1,11 @@
 package main
 
 import (
+	"context"
 	"os"
 	"syscall"
 	"testing"
+	"time"
 )
 
 // TestRealRunnerSignalForwarding will artificially put an interrupt signal (SIGINT) in the rr.signals chan.
@@ -14,9 +16,26 @@ func TestRealRunnerSignalForwarding(t *testing.T) {
 	rr := realRunner{}
 	rr.signals = make(chan os.Signal, 1)
 	rr.signals <- syscall.SIGINT
-	if err := rr.Run("sleep", "3600"); err.Error() == "signal: interrupt" {
+	if err := rr.Run(context.Background(), "sleep", "3600"); err.Error() == "signal: interrupt" {
 		t.Logf("SIGINT forwarded to Entrypoint")
 	} else {
 		t.Fatalf("Unexpected error received: %v", err)
 	}
 }
+
+// TestRealRunnerTimeout tests whether cmd is killed after a millisecond even though it's supposed to sleep for 10 milliseconds.
+func TestRealRunnerTimeout(t *testing.T) {
+	rr := realRunner{}
+	timeout := time.Millisecond
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	if err := rr.Run(ctx, "sleep", "0.01"); err != nil {
+		if ctx.Err() == context.DeadlineExceeded {
+			t.Logf("killed process since %s deadline was exceeded", timeout.String())
+		} else if ctx.Err() != nil {
+			t.Fatalf("unexpected context error received: %v", ctx.Err())
+		} else {
+			t.Fatalf("unexpected command execution error received: %v", err)
+		}
+	}
+}

docs/tasks.md

@@ -12,6 +12,7 @@ weight: 1
   - [Defining `Steps`](#defining-steps)
     - [Reserved directories](#reserved-directories)
     - [Running scripts within `Steps`](#running-scripts-within-steps)
+    - [Specifying a timeout](#specifying-a-timeout)
   - [Specifying `Parameters`](#specifying-parameters)
   - [Specifying `Resources`](#specifying-resources)
   - [Specifying `Workspaces`](#specifying-workspaces)
@@ -241,7 +242,25 @@ steps:
     #!/usr/bin/env bash
     /bin/my-binary
 ```
+#### Specifying a timeout
 
+A `Step` can specify a `timeout` field. If the `Step` execution time exceeds the specified timeout, this `Step` and any subsequent `Steps` are canceled.
+Specifically, the running process spawned by the `Step` is killed.
+An accompanying log is output under the `status.conditions.message` field of the `TaskRun` YAML.
+
+The format for a timeout is a duration string as specified in the [Go time package](https://golang.org/pkg/time/#ParseDuration) (e.g. 1s or 1ms).
+
+The example `Step` below is supposed to sleep for 60 seconds but will be canceled by the specified 5 second timeout.
+```yaml
+steps:
+  - name: willTimeout
+    image: ubuntu
+    script: | 
+      #!/usr/bin/env bash
+      echo "I am supposed to sleep for 60 seconds!"
+      sleep 60
+    timeout: 5s
+``` 
 ### Specifying `Parameters`
 
 You can specify parameters, such as compilation flags or artifact names, that you want to supply to the `Task` at execution time.

internal/builder/v1beta1/pod.go

@@ -33,6 +33,9 @@ type PodSpecOp func(*corev1.PodSpec)
 // PodStatusOp is an operation which modifies a PodStatus struct.
 type PodStatusOp func(status *corev1.PodStatus)
 
+// PodContainerStatusOp is an operation which modifies a ContainerStatus struct.
+type PodContainerStatusOp func(status *corev1.ContainerStatus)
+
 // Pod creates a Pod with default values.
 // Any number of Pod modifiers can be passed to transform it.
 func Pod(name string, ops ...PodOp) *corev1.Pod {

pkg/apis/pipeline/v1beta1/task_types.go

@@ -125,10 +125,11 @@ type Step struct {
 	//
 	// If Script is not empty, the Step cannot have an Command or Args.
 	Script string `json:"script,omitempty"`
+	// If step times out after Timeout, pod is terminated
+	Timeout string `json:"timeout,omitempty"`
 }
 
-// Sidecar embeds the Container type, which allows it to include fields not
-// provided by Container.
+// Sidecar has nearly the same data structure as Step, consisting of a Container and an optional Script, but does not have the ability to timeout.
 type Sidecar struct {
 	corev1.Container `json:",inline"`
 

pkg/apis/pipeline/v1beta1/task_validation.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"strings"
+	"time"
 
 	"github.com/tektoncd/pipeline/pkg/apis/validate"
 	"github.com/tektoncd/pipeline/pkg/substitution"
@@ -189,6 +190,12 @@ func validateSteps(steps []Step) *apis.FieldError {
 			names.Insert(s.Name)
 		}
 
+		if s.Timeout != "" {
+			if _, err := time.ParseDuration(s.Timeout); err != nil {
+				return apis.ErrInvalidValue(s.Timeout, "timeout")
+			}
+		}
+
 		for _, vm := range s.VolumeMounts {
 			if strings.HasPrefix(vm.MountPath, "/tekton/") &&
 				!strings.HasPrefix(vm.MountPath, "/tekton/home") {

pkg/entrypoint/entrypointer.go

@@ -17,6 +17,7 @@ limitations under the License.
 package entrypoint
 
 import (
+	"context"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -73,7 +74,7 @@ type Waiter interface {
 
 // Runner encapsulates running commands.
 type Runner interface {
-	Run(args ...string) error
+	Run(ctx context.Context, args ...string) error
 }
 
 // PostWriter encapsulates writing a file when complete.
@@ -84,7 +85,7 @@ type PostWriter interface {
 
 // Go optionally waits for a file, runs the command, and writes a
 // post file.
-func (e Entrypointer) Go() error {
+func (e Entrypointer) Go(ctx context.Context) error {
 	prod, _ := zap.NewProduction()
 	logger := prod.Sugar()
 
@@ -120,7 +121,14 @@ func (e Entrypointer) Go() error {
 		ResultType: v1beta1.InternalTektonResultType,
 	})
 
-	err := e.Runner.Run(e.Args...)
+	err := e.Runner.Run(ctx, e.Args...)
+	if ctx.Err() == context.DeadlineExceeded {
+		output = append(output, v1beta1.PipelineResourceResult{
+			Key:        "Reason",
+			Value:      "TimeoutExceeded",
+			ResultType: v1beta1.InternalTektonResultType,
+		})
+	}
 
 	// Write the post file *no matter what*
 	e.WritePostFile(e.PostFile, err)

pkg/entrypoint/entrypointer_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package entrypoint
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"io/ioutil"
@@ -76,7 +77,7 @@ func TestEntrypointerFailures(t *testing.T) {
 				Runner:          fr,
 				PostWriter:      fpw,
 				TerminationPath: "termination",
-			}.Go()
+			}.Go(context.Background())
 			if err == nil {
 				t.Fatalf("Entrypointer didn't fail")
 			}
@@ -139,7 +140,7 @@ func TestEntrypointer(t *testing.T) {
 				Runner:          fr,
 				PostWriter:      fpw,
 				TerminationPath: "termination",
-			}.Go()
+			}.Go(context.Background())
 			if err != nil {
 				t.Fatalf("Entrypointer failed: %v", err)
 			}
@@ -214,7 +215,7 @@ func (f *fakeWaiter) Wait(file string, _ bool) error {
 
 type fakeRunner struct{ args *[]string }
 
-func (f *fakeRunner) Run(args ...string) error {
+func (f *fakeRunner) Run(ctx context.Context, args ...string) error {
 	f.args = &args
 	return nil
 }
@@ -232,7 +233,7 @@ func (f *fakeErrorWaiter) Wait(file string, expectContent bool) error {
 
 type fakeErrorRunner struct{ args *[]string }
 
-func (f *fakeErrorRunner) Run(args ...string) error {
+func (f *fakeErrorRunner) Run(ctx context.Context, args ...string) error {
 	f.args = &args
 	return errors.New("runner failed")
 }

pkg/pod/entrypoint.go

@@ -84,8 +84,6 @@ var (
 // Containers must have Command specified; if the user didn't specify a
 // command, we must have fetched the image's ENTRYPOINT before calling this
 // method, using entrypoint_lookup.go.
-//
-// TODO(#1605): Also use entrypoint injection to order sidecar start/stop.
 func orderContainers(entrypointImage string, extraEntrypointArgs []string, steps []corev1.Container, results []v1beta1.TaskResult) (corev1.Container, []corev1.Container, error) {
 	initContainer := corev1.Container{
 		Name:         "place-tools",
@@ -236,6 +234,6 @@ func isContainerSidecar(name string) bool { return strings.HasPrefix(name, sidec
 // trimStepPrefix returns the container name, stripped of its step prefix.
 func trimStepPrefix(name string) string { return strings.TrimPrefix(name, stepPrefix) }
 
-// trimSidecarPrefix returns the container name, stripped of its sidecar
+// TrimSidecarPrefix returns the container name, stripped of its sidecar
 // prefix.
 func TrimSidecarPrefix(name string) string { return strings.TrimPrefix(name, sidecarPrefix) }

pkg/pod/entrypoint_test.go

@@ -87,7 +87,7 @@ func TestOrderContainers(t *testing.T) {
 		VolumeMounts:           []corev1.VolumeMount{toolsMount},
 		TerminationMessagePath: "/tekton/termination",
 	}}
-	gotInit, got, err := orderContainers(images.EntrypointImage, []string{}, steps, nil)
+	gotInit, got, err := orderContainers(images.EntrypointImage, []string{}, steps, make([]string, len(steps)), nil)
 	if err != nil {
 		t.Fatalf("orderContainers: %v", err)
 	}
@@ -171,7 +171,7 @@ func TestEntryPointResults(t *testing.T) {
 		VolumeMounts:           []corev1.VolumeMount{toolsMount},
 		TerminationMessagePath: "/tekton/termination",
 	}}
-	_, got, err := orderContainers(images.EntrypointImage, []string{}, steps, results)
+	_, got, err := orderContainers(images.EntrypointImage, []string{}, steps, make([]string, len(steps)), results)
 	if err != nil {
 		t.Fatalf("orderContainers: %v", err)
 	}
@@ -209,7 +209,7 @@ func TestEntryPointResultsSingleStep(t *testing.T) {
 		VolumeMounts:           []corev1.VolumeMount{toolsMount, downwardMount},
 		TerminationMessagePath: "/tekton/termination",
 	}}
-	_, got, err := orderContainers(images.EntrypointImage, []string{}, steps, results)
+	_, got, err := orderContainers(images.EntrypointImage, []string{}, steps, make([]string, len(steps)), results)
 	if err != nil {
 		t.Fatalf("orderContainers: %v", err)
 	}
@@ -243,7 +243,7 @@ func TestEntryPointSingleResultsSingleStep(t *testing.T) {
 		VolumeMounts:           []corev1.VolumeMount{toolsMount, downwardMount},
 		TerminationMessagePath: "/tekton/termination",
 	}}
-	_, got, err := orderContainers(images.EntrypointImage, []string{}, steps, results)
+	_, got, err := orderContainers(images.EntrypointImage, []string{}, steps, make([]string, len(steps)), results)
 	if err != nil {
 		t.Fatalf("orderContainers: %v", err)
 	}

pkg/pod/pod.go

@@ -139,6 +139,8 @@ func (b *Builder) Build(ctx context.Context, taskRun *v1beta1.TaskRun, taskSpec
 		return nil, err
 	}
 
+	stepContainers = setStepTimeouts(taskSpec.Steps, stepContainers)
+
 	// Rewrite steps with entrypoint binary. Append the entrypoint init
 	// container to place the entrypoint binary.
 	entrypointInit, stepContainers, err := orderContainers(b.Images.EntrypointImage, credEntrypointArgs, stepContainers, taskSpec.Results)
@@ -398,3 +400,12 @@ func shouldAddReadyAnnotationOnPodCreate(ctx context.Context, sidecars []v1beta1
 	cfg := config.FromContextOrDefaults(ctx)
 	return !cfg.FeatureFlags.RunningInEnvWithInjectedSidecars
 }
+
+func setStepTimeouts(steps []v1beta1.Step, stepContainers []corev1.Container) []corev1.Container {
+	for i := range stepContainers {
+		if steps[i].Timeout != "" {
+			stepContainers[i].Args = append(stepContainers[i].Args, steps[i].Timeout)
+		}
+	}
+	return stepContainers
+}

pkg/pod/pod_test.go

@@ -1085,7 +1085,54 @@ script-heredoc-randomly-generated-78c5n
 				Resources:              corev1.ResourceRequirements{Requests: allZeroQty()},
 				TerminationMessagePath: "/tekton/termination",
 			}},
-		}}} {
+		},
+	}, {
+		desc: "step-with-timeout",
+		ts: v1beta1.TaskSpec{
+			Steps: []v1beta1.Step{{Container: corev1.Container{
+				Name:    "name",
+				Image:   "image",
+				Command: []string{"cmd"}, // avoid entrypoint lookup.
+			},
+				Timeout: "1s",
+			}},
+		},
+		want: &corev1.PodSpec{
+			RestartPolicy:  corev1.RestartPolicyNever,
+			InitContainers: []corev1.Container{placeToolsInit},
+			Containers: []corev1.Container{{
+				Name:    "step-name",
+				Image:   "image",
+				Command: []string{"/tekton/tools/entrypoint"},
+				Args: []string{
+					"-wait_file",
+					"/tekton/downward/ready",
+					"-wait_file_content",
+					"-post_file",
+					"/tekton/tools/0",
+					"-termination_path",
+					"/tekton/termination",
+					"-timeout",
+					"1s",
+					"-entrypoint",
+					"cmd",
+					"--",
+				},
+				Env: implicitEnvVars,
+				VolumeMounts: append([]corev1.VolumeMount{toolsMount, downwardMount, {
+					Name:      "tekton-creds-init-home-9l9zj",
+					MountPath: "/tekton/creds",
+				}}, implicitVolumeMounts...),
+				WorkingDir:             pipeline.WorkspaceDir,
+				Resources:              corev1.ResourceRequirements{Requests: allZeroQty()},
+				TerminationMessagePath: "/tekton/termination",
+			}},
+			Volumes: append(implicitVolumes, toolsVolume, downwardVolume, corev1.Volume{
+				Name:         "tekton-creds-init-home-9l9zj",
+				VolumeSource: corev1.VolumeSource{EmptyDir: &corev1.EmptyDirVolumeSource{Medium: corev1.StorageMediumMemory}},
+			}),
+		},
+	}} {
 		t.Run(c.desc, func(t *testing.T) {
 			names.TestingSeed()
 			store := config.NewStore(logtesting.TestLogger(t))