diff --git a/pkg/compose/executor_start_test.go b/pkg/compose/executor_start_test.go index 392e056fb8a..411aa6bac28 100644 --- a/pkg/compose/executor_start_test.go +++ b/pkg/compose/executor_start_test.go @@ -1047,11 +1047,11 @@ func TestExecutePlanStartPhaseFullChainThroughCreateNode(t *testing.T) { scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0)).After(driftCheck) wait := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). Return(waitResultExit(0)).After(scan) - logs := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). - Return(emptyLogs(), nil).After(wait) hookStart := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). - Return(client.ContainerStartResult{}, nil).After(logs) - expectSuccessRemove(apiClient, "hook-1").After(hookStart) + Return(client.ContainerStartResult{}, nil).After(wait) + logs := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). + Return(emptyLogs(), nil).After(hookStart) + expectSuccessRemove(apiClient, "hook-1").After(logs) apiClient.EXPECT().ContainerStart(gomock.Any(), "new-id", gomock.Any()). Return(client.ContainerStartResult{}, nil) diff --git a/pkg/compose/pre_start.go b/pkg/compose/pre_start.go index d3b4b009824..b3aba4069df 100644 --- a/pkg/compose/pre_start.go +++ b/pkg/compose/pre_start.go @@ -22,6 +22,7 @@ import ( "io" "strconv" "strings" + "time" "github.com/compose-spec/compose-go/v2/types" "github.com/moby/moby/api/pkg/stdcopy" @@ -198,14 +199,6 @@ func (s *composeService) execPreStartHook( Condition: container.WaitConditionNextExit, }) - // Open the log stream before ContainerStart so a fast-exiting hook cannot - // race us to a 404. The dedicated logCtx lets us force the follow stream - // closed once the hook has exited, so a daemon that keeps the connection - // open cannot deadlock `<-logsDone`. - logCtx, cancelLogs := context.WithCancel(ctx) - defer cancelLogs() - logsDone, getTail := s.streamPreStartLogs(logCtx, containerID, service, index, listener) - if _, err := s.apiClient().ContainerStart(ctx, containerID, client.ContainerStartOptions{}); err != nil { // AutoRemove is false, so we must remove the never-started container // explicitly. A failed removal is logged so the orphan is visible. @@ -219,12 +212,30 @@ func (s *composeService) execPreStartHook( case <-waitRes.Result: case <-ctx.Done(): } - cancelLogs() - <-logsDone return err } + // Open the log stream once the hook is started: on a container still in + // "created" state a follow stream ends at once, empty, so a stream opened + // before ContainerStart would never carry a line. A fast-exiting hook can't + // race us to a 404: the runner is retained after it exits (AutoRemove is + // off, runPreStart removes it once we return). The dedicated logCtx lets us + // force the follow stream closed, so a daemon that keeps the connection + // open cannot deadlock `<-logsDone`. + logCtx, cancelLogs := context.WithCancel(ctx) + defer cancelLogs() + logsDone, getTail := s.streamPreStartLogs(logCtx, containerID, service, index, listener) + waitErr := waitPreStart(ctx, service.Name, index, waitRes) + if ctx.Err() == nil { + // The daemon ends the follow stream itself once the hook has exited, + // after sending what it buffered: cancelling right away would cut that + // output off. Bounded, for a daemon that keeps the connection open. + select { + case <-logsDone: + case <-time.After(s.drainTimeout()): + } + } cancelLogs() <-logsDone if waitErr != nil { diff --git a/pkg/compose/pre_start_test.go b/pkg/compose/pre_start_test.go index dc3e62af8d5..7fc70d32a0f 100644 --- a/pkg/compose/pre_start_test.go +++ b/pkg/compose/pre_start_test.go @@ -23,7 +23,9 @@ import ( "io" "strconv" "strings" + "sync" "testing" + "time" "github.com/compose-spec/compose-go/v2/types" "github.com/docker/cli/cli/config/configfile" @@ -122,20 +124,20 @@ func TestPreStart_SuccessTwoHooksInOrder(t *testing.T) { // Hook 1: wait (subscribe) → logs (subscribe) → start → remove. wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). Return(waitResultExit(0)).After(scan) - logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). - Return(emptyLogs(), nil).After(wait1) start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). - Return(client.ContainerStartResult{}, nil).After(logs1) - remove1 := expectSuccessRemove(apiClient, "hook-1").After(start1) + Return(client.ContainerStartResult{}, nil).After(wait1) + logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). + Return(emptyLogs(), nil).After(start1) + remove1 := expectSuccessRemove(apiClient, "hook-1").After(logs1) // Hook 2 only runs after hook 1 has been removed. wait2 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-2", gomock.Any()). Return(waitResultExit(0)).After(remove1) - logs2 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-2", gomock.Any()). - Return(emptyLogs(), nil).After(wait2) start2 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-2", gomock.Any()). - Return(client.ContainerStartResult{}, nil).After(logs2) - expectSuccessRemove(apiClient, "hook-2").After(start2) + Return(client.ContainerStartResult{}, nil).After(wait2) + logs2 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-2", gomock.Any()). + Return(emptyLogs(), nil).After(start2) + expectSuccessRemove(apiClient, "hook-2").After(logs2) err := tested.runPreStart(t.Context(), project, service, func(api.ContainerEvent) {}) assert.NilError(t, err) @@ -157,10 +159,10 @@ func TestPreStart_FirstHookFailsStopsExecution(t *testing.T) { scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0), runnerSummary("hook-2", 1)) wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). Return(waitResultExit(42)).After(scan) - logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). - Return(emptyLogs(), nil).After(wait1) - apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). - Return(client.ContainerStartResult{}, nil).After(logs1) + start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). + Return(client.ContainerStartResult{}, nil).After(wait1) + apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). + Return(emptyLogs(), nil).After(start1) // Hook container is retained on failure — no ContainerRemove expected, and // hook-2's runner is never touched. @@ -284,11 +286,11 @@ func TestPreStart_SecondRunnerMissing(t *testing.T) { scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0)) wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). Return(waitResultExit(0)).After(scan) - logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). - Return(emptyLogs(), nil).After(wait1) start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). - Return(client.ContainerStartResult{}, nil).After(logs1) - expectSuccessRemove(apiClient, "hook-1").After(start1) + Return(client.ContainerStartResult{}, nil).After(wait1) + logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). + Return(emptyLogs(), nil).After(start1) + expectSuccessRemove(apiClient, "hook-1").After(logs1) err := tested.runPreStart(t.Context(), project, service, nil) assert.ErrorContains(t, err, `service "web" pre_start[1]`) @@ -524,10 +526,9 @@ func TestPreStart_ContainerStartFailurePropagates(t *testing.T) { scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0)) wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). Return(waitResultExit(0)).After(scan) - logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). - Return(emptyLogs(), nil).After(wait1) start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). - Return(client.ContainerStartResult{}, errors.New("daemon: container start failed")).After(logs1) + Return(client.ContainerStartResult{}, errors.New("daemon: container start failed")).After(wait1) + // No ContainerLogs: the log stream is only opened once the hook started. // AutoRemove never fires when start fails, so the hook must drop the ghost // container explicitly. This is distinct from the success-path removal // (RemoveVolumes:true) — the never-started container has no logs to preserve. @@ -640,12 +641,12 @@ func TestPreStart_DetachedModeAttachesLogs(t *testing.T) { scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0)) wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). Return(waitResultExit(0)).After(scan) + start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). + Return(client.ContainerStartResult{}, nil).After(wait1) // ContainerLogs MUST be called even with a nil listener. logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). - Return(emptyLogs(), nil).After(wait1) - apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). - Return(client.ContainerStartResult{}, nil).After(logs1) - expectSuccessRemove(apiClient, "hook-1") + Return(emptyLogs(), nil).After(start1) + expectSuccessRemove(apiClient, "hook-1").After(logs1) err := tested.runPreStart(t.Context(), project, service, nil) assert.NilError(t, err) @@ -675,10 +676,10 @@ func TestPreStart_FailureIncludesTail(t *testing.T) { scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0)) wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). Return(waitResultExit(1)).After(scan) - logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). - Return(io.NopCloser(bytes.NewReader(logContent)), nil).After(wait1) - apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). - Return(client.ContainerStartResult{}, nil).After(logs1) + start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). + Return(client.ContainerStartResult{}, nil).After(wait1) + apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). + Return(io.NopCloser(bytes.NewReader(logContent)), nil).After(start1) // Hook container is retained on failure — no ContainerRemove expected. err := tested.runPreStart(t.Context(), project, service, nil) @@ -688,6 +689,167 @@ func TestPreStart_FailureIncludesTail(t *testing.T) { assert.ErrorContains(t, err, "doesn't exist") } +// blockingLogs is a log stream that delivers its content, then stays open until +// its context is canceled, like a daemon that does not end the follow stream +// when the container exits. +type blockingLogs struct { + io.Reader + ctx context.Context +} + +func (b *blockingLogs) Read(p []byte) (int, error) { + n, err := b.Reader.Read(p) + if err == io.EOF { + <-b.ctx.Done() + } + return n, err +} + +func (b *blockingLogs) Close() error { return nil } + +// slowLogs is a log stream that delivers its content only after a delay, +// honoring its context, like a daemon still flushing the output of a hook that +// has already exited. +type slowLogs struct { + io.Reader + ctx context.Context + delay time.Duration + done bool +} + +func (s *slowLogs) Read(p []byte) (int, error) { + if !s.done { + s.done = true + select { + case <-time.After(s.delay): + case <-s.ctx.Done(): + return 0, s.ctx.Err() + } + } + return s.Reader.Read(p) +} + +func (s *slowLogs) Close() error { return nil } + +// TestPreStart_OutputReachesListener pins that a hook's output reaches the +// listener: the log stream is opened once the hook is started (ordering +// enforced by the expectations), since a follow stream opened on a runner +// still in "created" state ends at once, empty. +func TestPreStart_OutputReachesListener(t *testing.T) { + tested, apiClient := newPreStartTestService(t) + + project := &types.Project{Name: "demo"} + service := types.ServiceConfig{ + Name: "web", + ContainerSpec: types.ContainerSpec{Image: "alpine"}, + PreStart: []types.PreStartHook{ + {ContainerSpec: types.ContainerSpec{Image: "alpine", Command: types.ShellCommand{"echo", "hello"}}}, + }, + } + + scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0)) + wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). + Return(waitResultExit(0)).After(scan) + start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). + Return(client.ContainerStartResult{}, nil).After(wait1) + logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). + Return(io.NopCloser(bytes.NewReader(stdcopyFrame(1, "hello\n"))), nil).After(start1) + expectSuccessRemove(apiClient, "hook-1").After(logs1) + + var lines []string + err := tested.runPreStart(t.Context(), project, service, func(e api.ContainerEvent) { + if e.Type == api.HookEventLog { + lines = append(lines, e.Line) + } + }) + assert.NilError(t, err) + assert.DeepEqual(t, lines, []string{"hello"}) +} + +// TestPreStart_FinishedHookOutputIsDrainedBeforeCutting pins the drain: a +// hook that already exited must not have its log stream cut while the daemon +// is still delivering the output it buffered. +func TestPreStart_FinishedHookOutputIsDrainedBeforeCutting(t *testing.T) { + tested, apiClient := newPreStartTestService(t) + tested.logStreamDrainTimeout = 5 * time.Second + + project := &types.Project{Name: "demo"} + service := types.ServiceConfig{ + Name: "web", + ContainerSpec: types.ContainerSpec{Image: "alpine"}, + PreStart: []types.PreStartHook{ + {ContainerSpec: types.ContainerSpec{Image: "alpine", Command: types.ShellCommand{"echo", "hello"}}}, + }, + } + + scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0)) + wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). + Return(waitResultExit(0)).After(scan) + start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). + Return(client.ContainerStartResult{}, nil).After(wait1) + logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). + DoAndReturn(func(ctx context.Context, _ string, _ client.ContainerLogsOptions) (io.ReadCloser, error) { + return &slowLogs{Reader: bytes.NewReader(stdcopyFrame(1, "hello\n")), ctx: ctx, delay: 20 * time.Millisecond}, nil + }).After(start1) + expectSuccessRemove(apiClient, "hook-1").After(logs1) + + var lines []string + var mu sync.Mutex + err := tested.runPreStart(t.Context(), project, service, func(e api.ContainerEvent) { + mu.Lock() + defer mu.Unlock() + if e.Type == api.HookEventLog { + lines = append(lines, e.Line) + } + }) + assert.NilError(t, err) + mu.Lock() + defer mu.Unlock() + assert.DeepEqual(t, lines, []string{"hello"}) +} + +// TestPreStart_FinishedHookDoesNotWaitForAnOpenLogStream pins that a daemon +// keeping the log stream open after the hook exited cannot hang the run: the +// stream is cut once the drain delay elapsed, with what it carried so far kept. +func TestPreStart_FinishedHookDoesNotWaitForAnOpenLogStream(t *testing.T) { + tested, apiClient := newPreStartTestService(t) + tested.logStreamDrainTimeout = 50 * time.Millisecond + + project := &types.Project{Name: "demo"} + service := types.ServiceConfig{ + Name: "web", + ContainerSpec: types.ContainerSpec{Image: "alpine"}, + PreStart: []types.PreStartHook{ + {ContainerSpec: types.ContainerSpec{Image: "alpine", Command: types.ShellCommand{"echo", "hello"}}}, + }, + } + + scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0)) + wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). + Return(waitResultExit(0)).After(scan) + start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). + Return(client.ContainerStartResult{}, nil).After(wait1) + logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). + DoAndReturn(func(ctx context.Context, _ string, _ client.ContainerLogsOptions) (io.ReadCloser, error) { + return &blockingLogs{Reader: bytes.NewReader(stdcopyFrame(1, "hello\n")), ctx: ctx}, nil + }).After(start1) + expectSuccessRemove(apiClient, "hook-1").After(logs1) + + var lines []string + var mu sync.Mutex + err := tested.runPreStart(t.Context(), project, service, func(e api.ContainerEvent) { + mu.Lock() + defer mu.Unlock() + if e.Type == api.HookEventLog { + lines = append(lines, e.Line) + } + }) + assert.NilError(t, err) + mu.Lock() + defer mu.Unlock() + assert.DeepEqual(t, lines, []string{"hello"}) +} + // --------------------------------------------------------------------------- // Feature tests: pre_start container retention on failure // --------------------------------------------------------------------------- @@ -1068,12 +1230,10 @@ func TestPreStart_ContainerStartFailureAndRemoveFails(t *testing.T) { } scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0)) - apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). + wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). Return(waitResultExit(0)).After(scan) - logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()). - Return(emptyLogs(), nil) start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()). - Return(client.ContainerStartResult{}, errors.New("start failed")).After(logs1) + Return(client.ContainerStartResult{}, errors.New("start failed")).After(wait1) // Both the start AND the cleanup removal fail. apiClient.EXPECT().ContainerRemove(gomock.Any(), "hook-1", client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}). Return(client.ContainerRemoveResult{}, errors.New("removal failed")).After(start1) diff --git a/pkg/e2e/hooks_test.go b/pkg/e2e/hooks_test.go index a58d53babdf..107c398eeb1 100644 --- a/pkg/e2e/hooks_test.go +++ b/pkg/e2e/hooks_test.go @@ -106,6 +106,16 @@ func TestPreStartHookInError(t *testing.T) { ServiceState("sample", "created")) } +// The output of a failing hook is what tells why it failed: it must reach the +// error, which is built from the log stream opened on the running hook. +func TestPreStartHookFailureShowsOutput(t *testing.T) { + NewScenario(t, "a failing pre_start hook must report what it printed"). + Step("up fails, and the error carries the hook's output", + ComposeCmd("up", "-d").MayFail(), + ExitCode(1), + OutputContains("hook-diagnostic-marker")) +} + // A retained runner from a failed run must not block the next one: it is a // stale hook container the reconciler has to observe and purge before // creating a fresh one under the same deterministic name, or the second up diff --git a/pkg/e2e/testdata/TestPreStartHookFailureShowsOutput/compose.yaml b/pkg/e2e/testdata/TestPreStartHookFailureShowsOutput/compose.yaml new file mode 100644 index 00000000000..138c5eab139 --- /dev/null +++ b/pkg/e2e/testdata/TestPreStartHookFailureShowsOutput/compose.yaml @@ -0,0 +1,7 @@ +services: + sample: + image: alpine + command: sh -c 'sleep 30' + pre_start: + - image: alpine + command: sh -c 'echo hook-diagnostic-marker >&2; exit 5'