diff --git a/pkg/compose/hook.go b/pkg/compose/hook.go index ce2ae9568..0fe587cc2 100644 --- a/pkg/compose/hook.go +++ b/pkg/compose/hook.go @@ -17,6 +17,7 @@ package compose import ( + "bytes" "context" "fmt" "io" @@ -39,27 +40,33 @@ const ( ) func (s *composeService) runHook(ctx context.Context, ctr container.Summary, service types.ServiceConfig, hook types.ServiceHook, listener api.ContainerEventListener) error { - // The output of a hook is the only explanation of why it failed, so keep the - // tail of it even when no listener is attached: the exec used to run - // unattached in that case, the daemon discarded both streams, and the caller - // was left with a bare exit code. Only surfaced on failure, and bounded. - tail := newOutputTail(hookOutputTailLines, hookOutputTailBytes) + // Keep stdout and stderr in separate tails so the error message can prefer + // stderr (which usually holds the real error) over stdout progress noise. + tailOut := newOutputTail(hookOutputTailLines, hookOutputTailBytes) + tailErr := newOutputTail(hookOutputTailLines, hookOutputTailBytes) - var out io.Writer = tail + // wOut/wErr receive the demultiplexed stdout/stderr from StdCopy. When a + // listener is attached they also forward each line to it. + var wOut, wErr io.Writer + wOut = tailOut + wErr = tailErr if listener != nil { - wOut := utils.GetWriter(func(line string) { + source := getContainerNameWithoutProject(ctr) + " ->" + // stdout and stderr share one listener writer: ContainerEvent has no stream + // field, so both appear identically in the live display. + lw := utils.GetWriter(func(line string) { listener(api.ContainerEvent{ Type: api.HookEventLog, - Source: getContainerNameWithoutProject(ctr) + " ->", + Source: source, ID: ctr.ID, Service: service.Name, Line: line, }) }) - defer wOut.Close() //nolint:errcheck - - out = io.MultiWriter(wOut, tail) + defer lw.Close() //nolint:errcheck + wOut = io.MultiWriter(lw, tailOut) + wErr = io.MultiWriter(lw, tailErr) } exec, err := s.apiClient().ExecCreate(ctx, ctr.ID, client.ExecCreateOptions{ @@ -92,13 +99,31 @@ func (s *composeService) runHook(ctx context.Context, ctr container.Summary, ser } defer attach.Close() - if service.Tty { - _, err = io.Copy(out, attach.Reader) - } else { - _, err = stdcopy.StdCopy(out, out, attach.Reader) - } - if err != nil { - return err + // Run the copy in a goroutine so that a context cancellation (Ctrl+C) can + // interrupt it. Without this the blocked Read inside StdCopy/io.Copy would + // prevent the first Ctrl+C from taking effect. + copyDone := make(chan error, 1) + go func() { + var cErr error + if service.Tty { + _, cErr = io.Copy(wOut, attach.Reader) + } else { + _, cErr = stdcopy.StdCopy(wOut, wErr, attach.Reader) + } + copyDone <- cErr + }() + + select { + case err = <-copyDone: + if err != nil { + return err + } + case <-ctx.Done(): + // Close the connection to unblock the blocked Read in the copy goroutine, + // then wait for it to exit before returning so there is no goroutine leak. + attach.Close() + <-copyDone + return ctx.Err() } inspected, err := s.apiClient().ExecInspect(ctx, exec.ID, client.ExecInspectOptions{}) @@ -106,15 +131,25 @@ func (s *composeService) runHook(ctx context.Context, ctr container.Summary, ser return err } if inspected.ExitCode != 0 { - if output := tail.String(); output != "" { - return fmt.Errorf("%s hook exited with status %d: %s", service.Name, inspected.ExitCode, output) - } - - return fmt.Errorf("%s hook exited with status %d", service.Name, inspected.ExitCode) + return hookExitError(service.Name, inspected.ExitCode, tailOut, tailErr) } return nil } +// hookExitError builds a hook failure error. Stderr content is preferred over +// stdout because it usually holds the real failure reason rather than progress +// noise. Falls back to stdout when stderr is empty. +func hookExitError(serviceName string, code int, tailOut, tailErr *outputTail) error { + output := tailErr.String() + if output == "" { + output = tailOut.String() + } + if output == "" { + return fmt.Errorf("%s hook exited with status %d", serviceName, code) + } + return fmt.Errorf("%s hook exited with status %d: %s", serviceName, code, output) +} + // outputTail keeps the last lines written to it, bounded by a line and a byte // count. Writes past the limit are accepted and the oldest content is dropped, // so a chatty hook is never blocked by a full buffer. It is written from a @@ -131,27 +166,44 @@ func newOutputTail(maxLines, maxBytes int) *outputTail { } func (t *outputTail) Write(p []byte) (int, error) { - for _, b := range p { - if b != '\n' { - // Cap the line being assembled, so a hook printing megabytes without - // a newline cannot grow this buffer without bound. - if t.partial.Len() < t.maxBytes { - t.partial.WriteByte(b) - } - - continue + n := len(p) + for len(p) > 0 { + i := bytes.IndexByte(p, '\n') + if i < 0 { + t.appendToPartial(p) + break } - - t.pushLine(t.partial.String()) + t.appendToPartial(p[:i]) + // ToValidUTF8 ensures a byte-truncated multi-byte rune does not corrupt + // the error message. + t.pushLine(strings.ToValidUTF8(t.partial.String(), "")) t.partial.Reset() + p = p[i+1:] } + return n, nil +} - return len(p), nil +// appendToPartial appends b to the in-progress partial line, capped at maxBytes +// so a hook printing megabytes without a newline cannot grow the buffer unbounded. +func (t *outputTail) appendToPartial(b []byte) { + if remaining := t.maxBytes - t.partial.Len(); remaining > 0 { + if len(b) > remaining { + b = b[:remaining] + } + _, _ = t.partial.Write(b) + } } // pushLine appends a line and drops the oldest one when the limit is reached. func (t *outputTail) pushLine(line string) { + // Strip trailing \r (from \r\n sequences). For progress-bar output that uses + // bare \r to rewind the cursor (curl, apt, pip), keep only the last segment + // after the final \r so the visible content is preserved rather than the + // intermediate states. line = strings.TrimRight(line, "\r") + if idx := strings.LastIndex(line, "\r"); idx >= 0 { + line = line[idx+1:] + } if strings.TrimSpace(line) == "" { return } @@ -165,13 +217,15 @@ func (t *outputTail) pushLine(line string) { // String returns the kept output, newest content last, trimmed to maxBytes. func (t *outputTail) String() string { lines := t.lines - if partial := strings.TrimSpace(t.partial.String()); partial != "" { - lines = append(append([]string{}, lines...), partial) + if partial := strings.TrimSpace(strings.ToValidUTF8(t.partial.String(), "")); partial != "" { + lines = append(lines, partial) } output := strings.TrimSpace(strings.Join(lines, "\n")) if len(output) > t.maxBytes { - output = output[len(output)-t.maxBytes:] + // Byte-slicing may split a multi-byte rune; ToValidUTF8 drops the broken + // leading sequence rather than emitting a replacement character. + output = strings.ToValidUTF8(output[len(output)-t.maxBytes:], "") } return output diff --git a/pkg/compose/hook_test.go b/pkg/compose/hook_test.go index d6ff79e08..5072ead03 100644 --- a/pkg/compose/hook_test.go +++ b/pkg/compose/hook_test.go @@ -24,8 +24,10 @@ import ( "io" "net" "os" + "slices" "strings" "testing" + "time" "github.com/compose-spec/compose-go/v2/types" "github.com/creack/pty" @@ -141,11 +143,11 @@ func writeStdcopyFrame(w io.Writer, stream byte, payload string) error { // daemon, leaving no way to find out at all. func TestRunHook_FailureIncludesOutput(t *testing.T) { tests := []struct { - name string - listener api.ContainerEventListener + name string + withListener bool }{ - {name: "with listener", listener: func(api.ContainerEvent) {}}, - {name: "without listener", listener: nil}, + {name: "with listener", withListener: true}, + {name: "without listener", withListener: false}, } for _, tc := range tests { @@ -194,9 +196,23 @@ func TestRunHook_FailureIncludesOutput(t *testing.T) { s, err := NewComposeService(mockCli) assert.NilError(t, err) - err = s.(*composeService).runHook(t.Context(), ctr, service, hook, tc.listener) + var got []string + var listener api.ContainerEventListener + if tc.withListener { + listener = func(event api.ContainerEvent) { + got = append(got, event.Line) + } + } + + err = s.(*composeService).runHook(t.Context(), ctr, service, hook, listener) assert.ErrorContains(t, err, "db hook exited with status 1") assert.ErrorContains(t, err, "SQLSTATE[42S02]: table not found") + if tc.withListener { + assert.Check(t, slices.Contains(got, "migrating"), + "listener must receive stdout lines on failure; got: %v", got) + assert.Check(t, slices.Contains(got, "SQLSTATE[42S02]: table not found"), + "listener must receive stderr lines on failure; got: %v", got) + } }) } } @@ -284,4 +300,137 @@ func TestOutputTail(t *testing.T) { assert.Check(t, len(tail.String()) <= 16, "output must stay within the byte limit") assert.Check(t, strings.HasSuffix(tail.String(), "last line"), "the newest output must survive") }) + + t.Run("strips trailing CR (CRLF lines)", func(t *testing.T) { + tail := newOutputTail(5, 1024) + _, err := tail.Write([]byte("line one\r\nline two\r\n")) + assert.NilError(t, err) + assert.Equal(t, tail.String(), "line one\nline two") + }) + + t.Run("progress bar: keeps last segment after mid-line CR", func(t *testing.T) { + tail := newOutputTail(5, 1024) + _, err := tail.Write([]byte("10%\r20%\r30%\n")) + assert.NilError(t, err) + assert.Equal(t, tail.String(), "30%") + }) + + t.Run("byte-truncation in String is UTF-8 safe", func(t *testing.T) { + // maxBytes=5: joining two € lines gives "€\n€" (7 bytes); the last-5-byte + // slice cuts inside the first €, leaving a broken leading byte. ToValidUTF8 + // must strip it so the result is valid UTF-8. + tail := newOutputTail(10, 5) + _, err := tail.Write([]byte("€\n€\n")) + assert.NilError(t, err) + out := tail.String() + assert.Check(t, out != "", "truncated string must not be empty") + assert.Check(t, strings.ToValidUTF8(out, "\uFFFD") == out, + "String() must return valid UTF-8 after byte truncation; got %q", out) + }) +} + +// TestRunHook_StderrBias verifies that when a failing hook produces output on +// both stdout and stderr, the error message preferentially shows stderr content. +// Stderr is where hooks write their actual error; stdout is progress noise. +func TestRunHook_StderrBias(t *testing.T) { + mockCtrl := gomock.NewController(t) + defer mockCtrl.Finish() + + mockAPI := mocks.NewMockAPIClient(mockCtrl) + mockCli := mocks.NewMockCli(mockCtrl) + mockCli.EXPECT().Client().Return(mockAPI).AnyTimes() + mockCli.EXPECT().Err().Return(streams.NewOut(os.Stderr)).AnyTimes() + mockCli.EXPECT().Out().Return(streams.NewOut(os.Stdout)).AnyTimes() + + serverConn, clientConn := net.Pipe() + + go func() { + // stdout: progress noise that should NOT dominate the error message + assert.NilError(t, writeStdcopyFrame(serverConn, 1, "running migration…\n")) + // stderr: the actual error that SHOULD appear in the error message + assert.NilError(t, writeStdcopyFrame(serverConn, 2, "Table 'service.sites' doesn't exist\n")) + serverConn.Close() //nolint:errcheck + }() + + mockAPI.EXPECT().ExecCreate(gomock.Any(), "ctr-a1b2c3d4e5f6", gomock.Any()). + Return(client.ExecCreateResult{ID: "exec-x"}, nil) + mockAPI.EXPECT().ExecAttach(gomock.Any(), "exec-x", gomock.Any()). + Return(client.ExecAttachResult{ + HijackedResponse: client.NewHijackedResponse(clientConn, ""), + }, nil) + mockAPI.EXPECT().ExecInspect(gomock.Any(), "exec-x", gomock.Any()). + Return(client.ExecInspectResult{ExitCode: 1}, nil) + + s, err := NewComposeService(mockCli) + assert.NilError(t, err) + + var got []string + listener := func(event api.ContainerEvent) { + got = append(got, event.Line) + } + + runErr := s.(*composeService).runHook(t.Context(), + container.Summary{ID: "ctr-a1b2c3d4e5f6"}, + types.ServiceConfig{Name: "db"}, + types.ServiceHook{Command: []string{"migrate"}}, + listener) + + assert.ErrorContains(t, runErr, "doesn't exist", "stderr content must appear in the error") + assert.Assert(t, !strings.Contains(runErr.Error(), "running migration"), + "stdout progress noise must not dominate when stderr has content; got: %s", runErr.Error()) + // The listener receives all lines from both streams regardless of stderr bias. + assert.Check(t, slices.Contains(got, "running migration\u2026"), + "listener must receive stdout lines; got: %v", got) + assert.Check(t, slices.Contains(got, "Table 'service.sites' doesn't exist"), + "listener must receive stderr lines; got: %v", got) +} + +// TestRunHook_ContextCancellation verifies that cancelling the context unblocks +// a hook that is waiting for output (e.g. a hung or slow container) and causes +// runHook to return promptly with context.Canceled. Without the goroutine/select +// fix the first Ctrl+C had no effect because StdCopy blocked on the attach reader. +func TestRunHook_ContextCancellation(t *testing.T) { + mockCtrl := gomock.NewController(t) + defer mockCtrl.Finish() + + mockAPI := mocks.NewMockAPIClient(mockCtrl) + mockCli := mocks.NewMockCli(mockCtrl) + mockCli.EXPECT().Client().Return(mockAPI).AnyTimes() + mockCli.EXPECT().Err().Return(streams.NewOut(os.Stderr)).AnyTimes() + mockCli.EXPECT().Out().Return(streams.NewOut(os.Stdout)).AnyTimes() + + // serverConn is held open and never writes data: StdCopy will block on Read. + serverConn, clientConn := net.Pipe() + defer serverConn.Close() //nolint:errcheck + + mockAPI.EXPECT().ExecCreate(gomock.Any(), "ctr-hang", gomock.Any()). + Return(client.ExecCreateResult{ID: "exec-hang"}, nil) + mockAPI.EXPECT().ExecAttach(gomock.Any(), "exec-hang", gomock.Any()). + Return(client.ExecAttachResult{ + HijackedResponse: client.NewHijackedResponse(clientConn, ""), + }, nil) + // ExecInspect must NOT be called: runHook must return before reaching it. + + ctx, cancel := context.WithCancel(t.Context()) + s, err := NewComposeService(mockCli) + assert.NilError(t, err) + + errCh := make(chan error, 1) + go func() { + errCh <- s.(*composeService).runHook(ctx, + container.Summary{ID: "ctr-hang"}, + types.ServiceConfig{Name: "svc"}, + types.ServiceHook{Command: []string{"sleep", "infinity"}}, + nil) + }() + + cancel() + + select { + case runErr := <-errCh: + assert.ErrorIs(t, runErr, context.Canceled, + "runHook must return context.Canceled after ctx cancellation") + case <-time.After(5 * time.Second): + t.Fatal("runHook did not return within 5 s after ctx cancellation") + } } diff --git a/pkg/compose/pre_start.go b/pkg/compose/pre_start.go index 439969e16..4d9c56830 100644 --- a/pkg/compose/pre_start.go +++ b/pkg/compose/pre_start.go @@ -19,6 +19,7 @@ package compose import ( "context" "fmt" + "io" "strconv" "github.com/compose-spec/compose-go/v2/types" @@ -94,7 +95,7 @@ func (s *composeService) runPreStartHook( // the connection open cannot deadlock `<-logsDone`. logCtx, cancelLogs := context.WithCancel(ctx) defer cancelLogs() - logsDone := s.streamPreStartLogs(logCtx, created.ID, service, index, listener) + logsDone, getTail := s.streamPreStartLogs(logCtx, created.ID, service, index, listener) if _, err := s.apiClient().ContainerStart(ctx, created.ID, client.ContainerStartOptions{}); err != nil { // AutoRemove only fires after a successful start, so the never-started @@ -119,6 +120,11 @@ func (s *composeService) runPreStartHook( waitErr := waitPreStart(ctx, service.Name, index, waitRes) cancelLogs() <-logsDone + if waitErr != nil { + if tail := getTail(); tail != "" { + return fmt.Errorf("%w: %s", waitErr, tail) + } + } return waitErr } @@ -259,12 +265,32 @@ func preStartResultErr(serviceName string, index int, res container.WaitResponse // streamPreStartLogs returns a channel that is closed once the hook log stream // has been fully drained (or never opened). Callers must wait on it before // returning so the goroutine cannot outlive the hook. -func (s *composeService) streamPreStartLogs(ctx context.Context, containerID string, service types.ServiceConfig, index int, listener api.ContainerEventListener) <-chan struct{} { +// streamPreStartLogs opens the hook container's log stream in a background +// goroutine, tees output into the listener (when non-nil) and into per-stream +// tail buffers. It returns: +// - done: closed when the goroutine exits; callers must wait on it +// - getTail: returns the stderr-biased tail (stderr preferred, stdout fallback); +// safe to call only after done is closed +// +// The log stream is always opened even when listener is nil, so the tail is +// populated in detached mode and can appear in error messages. +func (s *composeService) streamPreStartLogs( + ctx context.Context, + containerID string, + service types.ServiceConfig, + index int, + listener api.ContainerEventListener, +) (<-chan struct{}, func() string) { done := make(chan struct{}) - if listener == nil { - close(done) - return done + tailOut := newOutputTail(hookOutputTailLines, hookOutputTailBytes) + tailErr := newOutputTail(hookOutputTailLines, hookOutputTailBytes) + getTail := func() string { + if s := tailErr.String(); s != "" { + return s + } + return tailOut.String() } + source := fmt.Sprintf("%s pre_start[%d] ->", service.Name, index) logs, err := s.apiClient().ContainerLogs(ctx, containerID, client.ContainerLogsOptions{ ShowStdout: true, @@ -272,30 +298,44 @@ func (s *composeService) streamPreStartLogs(ctx context.Context, containerID str Follow: true, }) if err != nil { - listener(api.ContainerEvent{ - Type: api.HookEventLog, - Source: source, - ID: containerID, - Service: service.Name, - Line: fmt.Sprintf("warning: could not attach pre_start log stream: %s", err), - }) - close(done) - return done - } - go func() { - defer close(done) - defer logs.Close() //nolint:errcheck - w := utils.GetWriter(func(line string) { + if listener != nil { listener(api.ContainerEvent{ Type: api.HookEventLog, Source: source, ID: containerID, Service: service.Name, - Line: line, + Line: fmt.Sprintf("warning: could not attach pre_start log stream: %s", err), }) - }) - defer w.Close() //nolint:errcheck - _, _ = stdcopy.StdCopy(w, w, logs) + } + close(done) + return done, getTail + } + go func() { + defer close(done) + defer logs.Close() //nolint:errcheck + + var wOut, wErr io.Writer + wOut = tailOut + wErr = tailErr + + if listener != nil { + // stdout and stderr share one listener writer: ContainerEvent has no stream + // field, so both appear identically in the live display. + lw := utils.GetWriter(func(line string) { + listener(api.ContainerEvent{ + Type: api.HookEventLog, + Source: source, + ID: containerID, + Service: service.Name, + Line: line, + }) + }) + defer lw.Close() //nolint:errcheck + wOut = io.MultiWriter(lw, tailOut) + wErr = io.MultiWriter(lw, tailErr) + } + + _, _ = stdcopy.StdCopy(wOut, wErr, logs) }() - return done + return done, getTail } diff --git a/pkg/compose/pre_start_test.go b/pkg/compose/pre_start_test.go index 005bc5a01..c4277bc61 100644 --- a/pkg/compose/pre_start_test.go +++ b/pkg/compose/pre_start_test.go @@ -367,3 +367,86 @@ func TestWaitPreStart_RaceRealErrorAndResult(t *testing.T) { assert.ErrorContains(t, err, "connection lost") } } + +// stdcopyFrame encodes a single stdcopy-multiplexed frame. +// stream: 1=stdout, 2=stderr. Returns the encoded bytes. +// This helper is kept local (not using writeStdcopyFrame from hook_test.go) +// because hook_test.go carries a //go:build !windows constraint. +func stdcopyFrame(stream byte, data string) []byte { + payload := []byte(data) + header := [8]byte{stream, 0, 0, 0} + header[4] = byte(len(payload) >> 24) + header[5] = byte(len(payload) >> 16) + header[6] = byte(len(payload) >> 8) + header[7] = byte(len(payload)) + return append(header[:], payload...) +} + +// TestPreStart_DetachedModeAttachesLogs verifies that ContainerLogs is opened +// even when listener is nil (detached / compose up -d). Previously +// streamPreStartLogs short-circuited and returned immediately, leaving the tail +// buffer empty and any failure error without output context. +func TestPreStart_DetachedModeAttachesLogs(t *testing.T) { + tested, apiClient := newPreStartTestService(t) + + project := &types.Project{Name: "demo"} + service := types.ServiceConfig{ + Name: "web", + Image: "alpine", + PreStart: []types.ServiceHook{ + {Image: "alpine", Command: types.ShellCommand{"true"}}, + }, + } + ctr := container.Summary{ID: "service-ctr-id"} + + create1 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()). + Return(client.ContainerCreateResult{ID: "hook-1"}, nil) + wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). + Return(waitResultExit(0)).After(create1) + // 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) + + err := tested.runPreStart(t.Context(), project, service, ctr, nil) + assert.NilError(t, err) +} + +// TestPreStart_FailureIncludesTail verifies that when a pre_start hook exits +// with a non-zero code the error message includes the captured stderr output, +// giving operators the context they need to diagnose the failure. +func TestPreStart_FailureIncludesTail(t *testing.T) { + tested, apiClient := newPreStartTestService(t) + + project := &types.Project{Name: "demo"} + service := types.ServiceConfig{ + Name: "db", + Image: "postgres", + PreStart: []types.ServiceHook{ + {Image: "postgres", Command: types.ShellCommand{"migrate"}}, + }, + } + ctr := container.Summary{ID: "service-ctr-id"} + + // Build a stdcopy-multiplexed log stream with a stderr error line. + logContent := append( + stdcopyFrame(1, "starting migration\n"), // stdout noise + stdcopyFrame(2, "table 'sites' doesn't exist\n")..., // stderr error + ) + + create1 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()). + Return(client.ContainerCreateResult{ID: "hook-1"}, nil) + wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()). + Return(waitResultExit(1)).After(create1) + 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) + + err := tested.runPreStart(t.Context(), project, service, ctr, nil) + assert.Assert(t, err != nil) + assert.ErrorContains(t, err, "pre_start[0]") + // Stderr content must be in the error (stderr bias). + assert.ErrorContains(t, err, "doesn't exist") +}