From 4220c7abe8ac9c6e424811ef3b38a7ee14b0986d Mon Sep 17 00:00:00 2001 From: chlins Date: Thu, 30 Jul 2026 16:37:18 +0800 Subject: [PATCH 1/2] 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 --- changelogs/unreleased/10125-chlins | 1 + pkg/podexec/pod_command_executor.go | 39 +++-- pkg/podexec/pod_command_executor_test.go | 18 +++ .../pod_command_executor_timeout_test.go | 136 ++++++++++++++++++ 4 files changed, 182 insertions(+), 12 deletions(-) create mode 100644 changelogs/unreleased/10125-chlins create mode 100644 pkg/podexec/pod_command_executor_timeout_test.go diff --git a/changelogs/unreleased/10125-chlins b/changelogs/unreleased/10125-chlins new file mode 100644 index 000000000..1a1b00371 --- /dev/null +++ b/changelogs/unreleased/10125-chlins @@ -0,0 +1 @@ +Cancel hook exec stream on timeout and bound hook timeouts diff --git a/pkg/podexec/pod_command_executor.go b/pkg/podexec/pod_command_executor.go index 4ba4d4dc9..71894a489 100644 --- a/pkg/podexec/pod_command_executor.go +++ b/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) } diff --git a/pkg/podexec/pod_command_executor_test.go b/pkg/podexec/pod_command_executor_test.go index 13b00877a..de32fc605 100644 --- a/pkg/podexec/pod_command_executor_test.go +++ b/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 { diff --git a/pkg/podexec/pod_command_executor_timeout_test.go b/pkg/podexec/pod_command_executor_timeout_test.go new file mode 100644 index 000000000..88cb3ed92 --- /dev/null +++ b/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) + } +} From 2649b2554c05ba4dc35d8a9facf5943b7d1e45e3 Mon Sep 17 00:00:00 2001 From: chlins Date: Mon, 3 Aug 2026 13:28:31 +0800 Subject: [PATCH 2/2] Fix hook timeout review feedback Signed-off-by: chlins --- pkg/podexec/pod_command_executor.go | 22 ++++-- pkg/podexec/pod_command_executor_test.go | 57 +++++++++++++++ .../pod_command_executor_timeout_test.go | 73 +++++++++++++++---- 3 files changed, 130 insertions(+), 22 deletions(-) diff --git a/pkg/podexec/pod_command_executor.go b/pkg/podexec/pod_command_executor.go index 71894a489..997795c35 100644 --- a/pkg/podexec/pod_command_executor.go +++ b/pkg/podexec/pod_command_executor.go @@ -168,7 +168,7 @@ func (e *defaultPodCommandExecutor) ExecutePodCommand(log logrus.FieldLogger, it Stderr: &stderr, } - // The timeout drives the context so the exec stream is actually cancelled, rather than + // 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() @@ -178,17 +178,15 @@ func (e *defaultPodCommandExecutor) ExecutePodCommand(log logrus.FieldLogger, it errCh := make(chan error, 1) go func() { - errCh <- executor.StreamWithContext(ctx, streamOptions) + 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) }() select { case err = <-errCh: - // 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) } @@ -199,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 diff --git a/pkg/podexec/pod_command_executor_test.go b/pkg/podexec/pod_command_executor_test.go index de32fc605..e30911a20 100644 --- a/pkg/podexec/pod_command_executor_test.go +++ b/pkg/podexec/pod_command_executor_test.go @@ -177,6 +177,15 @@ 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. @@ -264,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{ diff --git a/pkg/podexec/pod_command_executor_timeout_test.go b/pkg/podexec/pod_command_executor_timeout_test.go index 88cb3ed92..a79389481 100644 --- a/pkg/podexec/pod_command_executor_timeout_test.go +++ b/pkg/podexec/pod_command_executor_timeout_test.go @@ -1,9 +1,26 @@ +/* +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" @@ -22,23 +39,38 @@ const timeoutTestPodJSON = `{ "spec": {"containers": [{"name": "container-1"}]} }` -// contextAwareExecutor returns once its context is cancelled, like the SPDY executor does. +// contextAwareExecutor returns once its context is canceled, like the SPDY executor does. type contextAwareExecutor struct { - cancelled chan struct{} - cancelledOnce bool + 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.cancelledOnce { - e.cancelledOnce = true - close(e.cancelled) + 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() @@ -70,10 +102,10 @@ func timeoutTestHook(timeout time.Duration) *v1.ExecHook { } } -// A hook that times out must have its exec stream cancelled, otherwise the command keeps +// 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{cancelled: make(chan struct{})} + 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)) @@ -82,26 +114,32 @@ func TestExecutePodCommandCancelsStreamOnTimeout(t *testing.T) { } select { - case <-exec.cancelled: + case <-exec.canceled: case <-time.After(2 * time.Second): - t.Fatal("stream was not cancelled after the hook timed out") + 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 + const ( + rounds = 50 + expectedError = "timed out after 1ms" + ) messages := map[string]int{} for range rounds { - exec := &contextAwareExecutor{cancelled: make(chan struct{})} + 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()]++ } @@ -117,8 +155,11 @@ func TestExecutePodCommandDoesNotLeakOnTimeout(t *testing.T) { time.Sleep(200 * time.Millisecond) before := runtime.NumGoroutine() + release := make(chan struct{}) + returned := &sync.WaitGroup{} for range rounds { - exec := &contextAwareExecutor{cancelled: make(chan struct{})} + 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 { @@ -126,7 +167,11 @@ func TestExecutePodCommandDoesNotLeakOnTimeout(t *testing.T) { } } - time.Sleep(time.Second) + // 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)