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) + } +}