From 4220c7abe8ac9c6e424811ef3b38a7ee14b0986d Mon Sep 17 00:00:00 2001 From: chlins Date: Thu, 30 Jul 2026 16:37:18 +0800 Subject: [PATCH] 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) + } +}