Merge pull request #10125 from chlins/fix/hook-exec-timeout-and-leak

Cancel hook exec stream on timeout and bound hook timeouts

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,26 @@ 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 canceled, 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
+		streamErr := executor.StreamWithContext(ctx, streamOptions)
+		// Inspect the local context as soon as the stream returns. Otherwise a stream error
+		// completed before the deadline could be misclassified if this goroutine sends its
+		// result before the caller is scheduled to receive it.
+		errCh <- normalizeExecHookError(streamErr, ctx.Err(), localHook.Timeout.Duration)
 	}()
 
-	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:
+	case <-ctx.Done():
 		return errors.Errorf("timed out after %v", localHook.Timeout.Duration)
 	}
 
@@ -184,6 +197,14 @@ func (e *defaultPodCommandExecutor) ExecutePodCommand(log logrus.FieldLogger, it
 	return err
 }
 
+func normalizeExecHookError(streamErr, contextErr error, timeout time.Duration) error {
+	if errors.Is(contextErr, context.DeadlineExceeded) {
+		return errors.Errorf("timed out after %v", timeout)
+	}
+
+	return streamErr
+}
+
 func ensureContainerExists(pod *corev1api.Pod, container string) error {
 	existsAsMainContainer := slices.ContainsFunc(pod.Spec.Containers, func(c corev1api.Container) bool {
 		return c.Name == container

pkg/podexec/pod_command_executor_test.go

@@ -177,6 +177,33 @@ func TestExecutePodCommand(t *testing.T) {
 			hookError:             errors.New("hook error"),
 			expectedError:         "hook error",
 		},
+		{
+			name:                  "stream deadline exceeded before local timeout",
+			command:               []string{"some", "command"},
+			expectedContainerName: "foo",
+			expectedErrorMode:     v1.HookErrorModeFail,
+			expectedTimeout:       defaultTimeout,
+			hookError:             context.DeadlineExceeded,
+			expectedError:         context.DeadlineExceeded.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 {
@@ -246,6 +273,54 @@ func TestExecutePodCommand(t *testing.T) {
 	}
 }
 
+func TestNormalizeExecHookError(t *testing.T) {
+	hookErr := errors.New("hook error")
+	tests := []struct {
+		name              string
+		streamErr         error
+		contextErr        error
+		expectedError     string
+		preserveStreamErr bool
+	}{
+		{
+			name:          "local context deadline exceeded",
+			streamErr:     context.DeadlineExceeded,
+			contextErr:    context.DeadlineExceeded,
+			expectedError: "timed out after 30s",
+		},
+		{
+			name:              "stream deadline exceeded before local timeout",
+			streamErr:         context.DeadlineExceeded,
+			expectedError:     context.DeadlineExceeded.Error(),
+			preserveStreamErr: true,
+		},
+		{
+			name:              "ordinary hook error",
+			streamErr:         hookErr,
+			expectedError:     hookErr.Error(),
+			preserveStreamErr: true,
+		},
+		{
+			name: "no errors",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			err := normalizeExecHookError(test.streamErr, test.contextErr, defaultTimeout)
+			if test.expectedError == "" {
+				require.NoError(t, err)
+				return
+			}
+
+			require.EqualError(t, err, test.expectedError)
+			if test.preserveStreamErr && err != test.streamErr {
+				t.Fatalf("expected stream error to be returned unchanged")
+			}
+		})
+	}
+}
+
 func TestEnsureContainerExists(t *testing.T) {
 	pod := &corev1api.Pod{
 		Spec: corev1api.PodSpec{

pkg/podexec/pod_command_executor_timeout_test.go

@@ -0,0 +1,181 @@
+/*
+Copyright 2026 the Velero contributors.
+
+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 podexec
+
+import (
+	"context"
+	"net/url"
+	"runtime"
+	"sync"
+	"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 canceled, like the SPDY executor does.
+type contextAwareExecutor struct {
+	canceled     chan struct{}
+	canceledOnce 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.canceledOnce {
+		e.canceledOnce = true
+		close(e.canceled)
+	}
+	return ctx.Err()
+}
+
+// contextIgnoringExecutor lets the outer timeout path return before the stream does.
+// Once released, the stream goroutine can only exit if its result channel is buffered.
+type contextIgnoringExecutor struct {
+	release  <-chan struct{}
+	returned *sync.WaitGroup
+}
+
+func (e *contextIgnoringExecutor) Stream(options remotecommand.StreamOptions) error { return nil }
+
+func (e *contextIgnoringExecutor) StreamWithContext(ctx context.Context, options remotecommand.StreamOptions) error {
+	defer e.returned.Done()
+	<-e.release
+	return nil
+}
+
+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 canceled, otherwise the command keeps
+// running on the API server after ExecutePodCommand has returned.
+func TestExecutePodCommandCancelsStreamOnTimeout(t *testing.T) {
+	exec := &contextAwareExecutor{canceled: 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.canceled:
+	case <-time.After(2 * time.Second):
+		t.Fatal("stream was not canceled 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
+		expectedError = "timed out after 1ms"
+	)
+
+	messages := map[string]int{}
+	for range rounds {
+		exec := &contextAwareExecutor{canceled: 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")
+		}
+		if err.Error() != expectedError {
+			t.Fatalf("expected %q, got %q", expectedError, err)
+		}
+		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()
+
+	release := make(chan struct{})
+	returned := &sync.WaitGroup{}
+	for range rounds {
+		returned.Add(1)
+		exec := &contextIgnoringExecutor{release: release, returned: returned}
+		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")
+		}
+	}
+
+	// Every ExecutePodCommand call has already taken the timeout path. Releasing the
+	// streams now forces their goroutines to send into an errCh with no receiver.
+	close(release)
+	returned.Wait()
+	time.Sleep(200 * time.Millisecond)
+	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)
+	}
+}