Cancel hook exec stream on timeout and bound hook timeouts

Hook timeouts come from pod annotations via time.ParseDuration, which accepts
negative and arbitrarily large values, and the exec stream was never cancelled.

Signed-off-by: chlins <chlins.zhang@gmail.com>

changelogs/unreleased/10125-chlins

@@ -0,0 +1 @@
+Cancel hook exec stream on timeout and bound hook timeouts

pkg/podexec/pod_command_executor.go

@@ -36,6 +36,10 @@ import (
 
 const defaultTimeout = 30 * time.Second
 
+// maxHookTimeout bounds a user-supplied hook timeout, which can come from a pod
+// annotation, so a single hook cannot hold up a backup for an unbounded time.
+const maxHookTimeout = 4 * time.Hour
+
 // PodCommandExecutor is capable of executing a command in a container in a pod.
 type PodCommandExecutor interface {
 	// ExecutePodCommand executes a command in a container in a pod. If the command takes longer than
@@ -112,9 +116,15 @@ func (e *defaultPodCommandExecutor) ExecutePodCommand(log logrus.FieldLogger, it
 		localHook.OnError = api.HookErrorModeFail
 	}
 
-	if localHook.Timeout.Duration == 0 {
+	// A non-positive timeout is not a valid bound. Timeouts sourced from pod annotations are
+	// parsed with time.ParseDuration, which accepts negative values, and a negative duration
+	// would otherwise leave the hook without any timeout at all.
+	if localHook.Timeout.Duration <= 0 {
 		localHook.Timeout.Duration = defaultTimeout
 	}
+	if localHook.Timeout.Duration > maxHookTimeout {
+		localHook.Timeout.Duration = maxHookTimeout
+	}
 
 	hookLog := log.WithFields(
 		logrus.Fields{
@@ -158,23 +168,28 @@ func (e *defaultPodCommandExecutor) ExecutePodCommand(log logrus.FieldLogger, it
 		Stderr: &stderr,
 	}
 
-	errCh := make(chan error)
+	// The timeout drives the context so the exec stream is actually cancelled, rather than
+	// being left running on the API server after this function has returned.
+	ctx, cancel := context.WithTimeout(context.Background(), localHook.Timeout.Duration)
+	defer cancel()
+
+	// Buffered so the goroutine below can always send its result and exit, even when this
+	// function has already returned on the timeout path.
+	errCh := make(chan error, 1)
 
 	go func() {
-		err = executor.StreamWithContext(context.Background(), streamOptions)
-		errCh <- err
+		errCh <- executor.StreamWithContext(ctx, streamOptions)
 	}()
 
-	var timeoutCh <-chan time.Time
-	if localHook.Timeout.Duration > 0 {
-		timer := time.NewTimer(localHook.Timeout.Duration)
-		defer timer.Stop()
-		timeoutCh = timer.C
-	}
-
 	select {
 	case err = <-errCh:
-	case <-timeoutCh:
+		// On a timeout the stream returns because the context expired, so both this case
+		// and ctx.Done() are ready and the select picks one at random. Report the timeout
+		// either way instead of surfacing the context error only some of the time.
+		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			return errors.Errorf("timed out after %v", localHook.Timeout.Duration)
+		}
+	case <-ctx.Done():
 		return errors.Errorf("timed out after %v", localHook.Timeout.Duration)
 	}
 

pkg/podexec/pod_command_executor_test.go

@@ -177,6 +177,24 @@ func TestExecutePodCommand(t *testing.T) {
 			hookError:             errors.New("hook error"),
 			expectedError:         "hook error",
 		},
+		{
+			// Timeouts from pod annotations go through time.ParseDuration, which accepts
+			// negative values. Without clamping, the hook would run with no timeout at all.
+			name:                  "negative timeout falls back to the default",
+			command:               []string{"some", "command"},
+			expectedContainerName: "foo",
+			expectedErrorMode:     v1.HookErrorModeFail,
+			timeout:               -1 * time.Second,
+			expectedTimeout:       30 * time.Second,
+		},
+		{
+			name:                  "timeout above the maximum is capped",
+			command:               []string{"some", "command"},
+			expectedContainerName: "foo",
+			expectedErrorMode:     v1.HookErrorModeFail,
+			timeout:               100000 * time.Hour,
+			expectedTimeout:       maxHookTimeout,
+		},
 	}
 
 	for _, test := range tests {

pkg/podexec/pod_command_executor_timeout_test.go

@@ -0,0 +1,136 @@
+package podexec
+
+import (
+	"context"
+	"net/url"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/remotecommand"
+
+	v1 "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	velerotest "github.com/vmware-tanzu/velero/pkg/test"
+)
+
+const timeoutTestPodJSON = `{
+	"metadata": {"namespace": "ns", "name": "pod-1"},
+	"spec": {"containers": [{"name": "container-1"}]}
+}`
+
+// contextAwareExecutor returns once its context is cancelled, like the SPDY executor does.
+type contextAwareExecutor struct {
+	cancelled     chan struct{}
+	cancelledOnce bool
+}
+
+func (e *contextAwareExecutor) Stream(options remotecommand.StreamOptions) error { return nil }
+
+func (e *contextAwareExecutor) StreamWithContext(ctx context.Context, options remotecommand.StreamOptions) error {
+	<-ctx.Done()
+	if !e.cancelledOnce {
+		e.cancelledOnce = true
+		close(e.cancelled)
+	}
+	return ctx.Err()
+}
+
+func newTimeoutTestExecutor(t *testing.T, exec remotecommand.Executor) (*defaultPodCommandExecutor, map[string]any) {
+	t.Helper()
+
+	clientConfig := &rest.Config{}
+	poster := &mockPoster{}
+	podCommandExecutor := NewPodCommandExecutor(clientConfig, poster).(*defaultPodCommandExecutor)
+
+	factory := &mockStreamExecutorFactory{}
+	podCommandExecutor.streamExecutorFactory = factory
+
+	baseURL, _ := url.Parse("https://some.server")
+	contentConfig := rest.ClientContentConfig{GroupVersion: schema.GroupVersion{Group: "", Version: "v1"}}
+	poster.On("Post").Return(rest.NewRequestWithClient(baseURL, "/api/v1", contentConfig, nil))
+	factory.On("NewSPDYExecutor", clientConfig, "POST", mock.Anything).Return(exec, nil)
+
+	pod, err := velerotest.GetAsMap(timeoutTestPodJSON)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	return podCommandExecutor, pod
+}
+
+func timeoutTestHook(timeout time.Duration) *v1.ExecHook {
+	return &v1.ExecHook{
+		Container: "container-1",
+		Command:   []string{"sh", "-c", "sleep 60"},
+		Timeout:   metav1.Duration{Duration: timeout},
+	}
+}
+
+// A hook that times out must have its exec stream cancelled, otherwise the command keeps
+// running on the API server after ExecutePodCommand has returned.
+func TestExecutePodCommandCancelsStreamOnTimeout(t *testing.T) {
+	exec := &contextAwareExecutor{cancelled: make(chan struct{})}
+	podCommandExecutor, pod := newTimeoutTestExecutor(t, exec)
+
+	err := podCommandExecutor.ExecutePodCommand(velerotest.NewLogger(), pod, "ns", "pod-1", "hookName", timeoutTestHook(100*time.Millisecond))
+	if err == nil {
+		t.Fatal("expected a timeout error")
+	}
+
+	select {
+	case <-exec.cancelled:
+	case <-time.After(2 * time.Second):
+		t.Fatal("stream was not cancelled after the hook timed out")
+	}
+}
+
+// When the stream returns because the context expired, both select cases are ready and one
+// is picked at random, so the reported error must not depend on which one wins.
+func TestExecutePodCommandTimeoutErrorIsDeterministic(t *testing.T) {
+	const rounds = 50
+
+	messages := map[string]int{}
+	for range rounds {
+		exec := &contextAwareExecutor{cancelled: make(chan struct{})}
+		podCommandExecutor, pod := newTimeoutTestExecutor(t, exec)
+
+		err := podCommandExecutor.ExecutePodCommand(velerotest.NewLogger(), pod, "ns", "pod-1", "hookName", timeoutTestHook(time.Millisecond))
+		if err == nil {
+			t.Fatal("expected a timeout error")
+		}
+		messages[err.Error()]++
+	}
+
+	if len(messages) != 1 {
+		t.Fatalf("expected one error message, got %d: %v", len(messages), messages)
+	}
+}
+
+func TestExecutePodCommandDoesNotLeakOnTimeout(t *testing.T) {
+	const rounds = 10
+
+	runtime.GC()
+	time.Sleep(200 * time.Millisecond)
+	before := runtime.NumGoroutine()
+
+	for range rounds {
+		exec := &contextAwareExecutor{cancelled: make(chan struct{})}
+		podCommandExecutor, pod := newTimeoutTestExecutor(t, exec)
+
+		if err := podCommandExecutor.ExecutePodCommand(velerotest.NewLogger(), pod, "ns", "pod-1", "hookName", timeoutTestHook(50*time.Millisecond)); err == nil {
+			t.Fatal("expected a timeout error")
+		}
+	}
+
+	time.Sleep(time.Second)
+	runtime.GC()
+	time.Sleep(200 * time.Millisecond)
+
+	if leaked := runtime.NumGoroutine() - before; leaked >= rounds {
+		t.Fatalf("%d goroutines leaked over %d timed out hooks", leaked, rounds)
+	}
+}