diff --git a/pkg/cioutil/container_io.go b/pkg/cioutil/container_io.go index 22dd6b4a0a5..041b5374f8e 100644 --- a/pkg/cioutil/container_io.go +++ b/pkg/cioutil/container_io.go @@ -32,6 +32,7 @@ import ( "github.com/containerd/containerd/v2/cmd/containerd-shim-runc-v2/process" "github.com/containerd/containerd/v2/defaults" "github.com/containerd/containerd/v2/pkg/cio" + "github.com/containerd/log" ) const binaryIOProcTermTimeout = 12 * time.Second // Give logger process 10 seconds for cleanup @@ -52,6 +53,44 @@ var bufPool = sync.Pool{ }, } +// closeOnce wraps f's Close so that extra calls return the first result +// instead of "file already closed". The logger pipe ends below are closed +// individually on the success path, but also sit in the error-path closers +// list of NewContainerIO. +func closeOnce(f *os.File) func() error { + var once sync.Once + var err error + return func() error { + once.Do(func() { err = f.Close() }) + return err + } +} + +// bestEffortWriter forwards writes to w until one fails, then silently +// discards all further writes. The pipe feeding the logging binary is wrapped +// in this before it joins the stdio tee of a foreground container: logging is +// best-effort there, and a dead logging binary (EPIPE once our read ends are +// closed, see NewContainerIO) must not error the whole tee — that would stop +// the copier that drains the container's stdio FIFO and deadlock the +// container. The attach keeps streaming; the log is what goes incomplete. +// https://github.com/containerd/nerdctl/issues/5137 +type bestEffortWriter struct { + w io.Writer + dead bool +} + +func (b *bestEffortWriter) Write(p []byte) (int, error) { + // Only ever called from the single stdio copy goroutine of its stream, so + // no locking is needed. + if !b.dead { + if _, err := b.w.Write(p); err != nil { + b.dead = true + log.L.WithError(err).Warn("writing container output to the logging binary failed; further output will not be logged") + } + } + return len(p), nil +} + func (c *ncio) Config() cio.Config { return c.config } @@ -156,19 +195,23 @@ func NewContainerIO(namespace string, logURI string, tty bool, stdin io.Reader, if err != nil { return nil, err } - closers = append(closers, stdoutr.Close, stdoutw.Close) + closeStdoutR := closeOnce(stdoutr) + closers = append(closers, closeStdoutR, stdoutw.Close) stderrr, stderrw, err := os.Pipe() if err != nil { return nil, err } - closers = append(closers, stderrr.Close, stderrw.Close) + closeStderrR := closeOnce(stderrr) + closers = append(closers, closeStderrR, stderrw.Close) r, w, err := os.Pipe() if err != nil { return nil, err } - closers = append(closers, r.Close, w.Close) + closeR := closeOnce(r) + closeW := closeOnce(w) + closers = append(closers, closeR, closeW) u, err := url.Parse(logURI) if err != nil { @@ -184,18 +227,44 @@ func NewContainerIO(namespace string, logURI string, tty bool, stdin io.Reader, closers = append(closers, func() error { return cmd.Process.Kill() }) // close our side of the pipe after start - if err := w.Close(); err != nil { + if err := closeW(); err != nil { return nil, fmt.Errorf("failed to close write pipe after start: %w", err) } + // Close our copies of the stdio read ends that were handed to the + // logging binary; the child holds its own duplicates via ExtraFiles. + // This is the equivalent of containerd's binaryIO.CloseAfterStart. + // If this process kept the read ends open, a logging binary that + // stops reading (killed, crashed, ...) would never surface as EPIPE + // on the tee writes below: the stdio copy goroutine would block + // forever on the full pipe, stop draining the container's stdout + // FIFO, and deadlock both the container and `nerdctl run` itself + // (including `nerdctl rm -f` of the wedged container). + // https://github.com/containerd/nerdctl/issues/5137 + if err := closeStdoutR(); err != nil { + return nil, fmt.Errorf("failed to close stdout pipe read end after start: %w", err) + } + if err := closeStderrR(); err != nil { + return nil, fmt.Errorf("failed to close stderr pipe read end after start: %w", err) + } + // wait for the logging binary to be ready + // For binary-v2, readiness requires a byte to be written before close. + // For binary, EOF is treated as ready for backward compatibility. b := make([]byte, 1) - if _, err := r.Read(b); err != nil && err != io.EOF { + n, err := r.Read(b) + if err != nil && err != io.EOF { return nil, fmt.Errorf("failed to read from logging binary: %w", err) } + if u.Scheme == "binary-v2" && n == 0 { + return nil, errors.New("logging binary did not call ready (it may have crashed or exited prematurely)") + } + if err := closeR(); err != nil { + return nil, fmt.Errorf("failed to close ready pipe read end: %w", err) + } - stdoutWriters = append(stdoutWriters, stdoutw) - stderrWriters = append(stderrWriters, stderrw) + stdoutWriters = append(stdoutWriters, &bestEffortWriter{w: stdoutw}) + stderrWriters = append(stderrWriters, &bestEffortWriter{w: stderrw}) } streams.Stdout = io.MultiWriter(stdoutWriters...) diff --git a/pkg/cioutil/container_io_test.go b/pkg/cioutil/container_io_test.go new file mode 100644 index 00000000000..8ba699f7c8b --- /dev/null +++ b/pkg/cioutil/container_io_test.go @@ -0,0 +1,77 @@ +/* + Copyright The containerd Authors. + + 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 cioutil + +import ( + "errors" + "os" + "testing" +) + +type failingWriter struct { + calls int +} + +func (f *failingWriter) Write(p []byte) (int, error) { + f.calls++ + return 0, errors.New("broken pipe") +} + +// TestBestEffortWriter verifies that a failing logger pipe never errors the +// stdio tee: the first failed write disables the writer and every write still +// reports full success, so the copier draining the container's stdio keeps +// running. Regression test for +// https://github.com/containerd/nerdctl/issues/5137 +func TestBestEffortWriter(t *testing.T) { + fw := &failingWriter{} + b := &bestEffortWriter{w: fw} + + for i := 0; i < 3; i++ { + n, err := b.Write([]byte("data")) + if err != nil { + t.Fatalf("write %d: best-effort writer must not return an error, got %v", i, err) + } + if n != 4 { + t.Fatalf("write %d: expected n=4, got %d", i, n) + } + } + if fw.calls != 1 { + t.Fatalf("expected the underlying writer to be abandoned after the first failure, got %d calls", fw.calls) + } +} + +// TestBestEffortWriterClosedPipe exercises the real failure mode: writing to +// an os.Pipe whose read end is closed (EPIPE), as happens when the logging +// binary dies after our copies of its read ends were closed. +func TestBestEffortWriterClosedPipe(t *testing.T) { + r, w, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + r.Close() + defer w.Close() + + b := &bestEffortWriter{w: w} + for i := 0; i < 2; i++ { + if n, err := b.Write([]byte("data")); err != nil || n != 4 { + t.Fatalf("write %d: expected (4, nil), got (%d, %v)", i, n, err) + } + } + if !b.dead { + t.Fatal("expected the writer to be marked dead after EPIPE") + } +} diff --git a/pkg/logging/logging.go b/pkg/logging/logging.go index a93594bd8c0..59bb9cc0db8 100644 --- a/pkg/logging/logging.go +++ b/pkg/logging/logging.go @@ -200,16 +200,37 @@ func getContainerWait(ctx context.Context, address string, config *logging.Confi if err != nil { return nil, err } + // closeAfterDelivery forwards the first delivery from ch and then closes + // the client, so that callers which re-arm the wait (see the wait loop in + // loggingProcessAdapter) do not accumulate open clients. + closeAfterDelivery := func(ch <-chan containerd.ExitStatus) <-chan containerd.ExitStatus { + out := make(chan containerd.ExitStatus, 1) + go func() { + defer close(out) + defer client.Close() + if status, ok := <-ch; ok { + out <- status + } + }() + return out + } con, err := client.LoadContainer(ctx, config.ID) if err != nil { + client.Close() return nil, err } task, err := con.Task(ctx, nil) if err == nil { - return task.Wait(ctx) + exitCh, err := task.Wait(ctx) + if err != nil { + client.Close() + return nil, err + } + return closeAfterDelivery(exitCh), nil } if !errdefs.IsNotFound(err) { + client.Close() return nil, err } @@ -232,16 +253,24 @@ func getContainerWait(ctx context.Context, address string, config *logging.Confi for { select { case <-ctx.Done(): + client.Close() return nil, errors.New("timed out waiting for container task to start") case <-ticker.C: task, err = con.Task(ctx, nil) if err == nil { - return task.Wait(ctx) + exitCh, err := task.Wait(ctx) + if err != nil { + client.Close() + return nil, err + } + return closeAfterDelivery(exitCh), nil } if !errdefs.IsNotFound(err) { + client.Close() return nil, err } if outputSeen() { + client.Close() return alreadyExited(), nil } } @@ -250,6 +279,10 @@ func getContainerWait(ctx context.Context, address string, config *logging.Confi type ContainerWaitFunc func(ctx context.Context, address string, config *logging.Config, outputSeen func() bool) (<-chan containerd.ExitStatus, error) +// containerWaitRetryDelay is how long the logger waits before re-arming the +// container wait after the wait channel delivered an error instead of an exit. +const containerWaitRetryDelay = 1 * time.Second + func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, address string, getContainerWait ContainerWaitFunc, config *logging.Config) error { if err := driver.PreProcess(ctx, dataStore, config); err != nil { return err @@ -374,15 +407,34 @@ func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, addres // keeps the stdio FIFO write ends open (so the container can be // restarted), so the FIFOs may not reach EOF on exit; without this the // read goroutines, and therefore the logger, could block forever. - exitCh, err := getContainerWait(ctx, address, config, outputSeen) - if err != nil { - // We could not determine when the container exits. Do not cancel the - // readers: they will finish on their own when the FIFO reaches EOF. - // Cancelling here could truncate a still-running container. - log.G(ctx).Errorf("failed to get container task wait channel: %v", err) - return + for { + exitCh, err := getContainerWait(ctx, address, config, outputSeen) + if err != nil { + // We could not determine when the container exits. Do not cancel the + // readers: they will finish on their own when the FIFO reaches EOF. + // Cancelling here could truncate a still-running container. + log.G(ctx).Errorf("failed to get container task wait channel: %v", err) + return + } + status := <-exitCh + if status.Error() == nil { + // The container has exited. + break + } + // The channel delivered a Wait RPC error, not a container exit: + // containerd's client sends Wait failures through the same channel + // as a synthetic ExitStatus (client/task.go). Treating that as an + // exit would cancel the readers, and with them all logging, while + // the container is still running. Re-arm the wait instead. + // https://github.com/containerd/nerdctl/issues/5137 + log.G(ctx).WithError(status.Error()).Warn("error while waiting for container exit; retrying") + select { + case <-ctx.Done(): + // SIGTERM: the goroutine above already cancels the readers. + return + case <-time.After(containerWaitRetryDelay): + } } - <-exitCh stdoutR.Cancel() stderrR.Cancel() }() diff --git a/pkg/logging/logging_test.go b/pkg/logging/logging_test.go index ecab183bebc..f5411f7953a 100644 --- a/pkg/logging/logging_test.go +++ b/pkg/logging/logging_test.go @@ -20,6 +20,7 @@ import ( "bufio" "bytes" "context" + "errors" "math/rand" "os" "strings" @@ -148,6 +149,120 @@ func TestLoggingProcessAdapter(t *testing.T) { // stream. The container's stdio FIFOs are modelled with os.Pipe; closing the // write end models the container exiting and containerd closing the FIFO. // Regression test for https://github.com/containerd/nerdctl/issues/5006 + +// TestLoggingProcessAdapterTrailingChunk verifies that the logger forwards all +// of the container's output, including a final chunk that has no trailing +// newline, rather than holding that chunk back until something closes the +// stream. The container's stdio FIFOs are modelled with os.Pipe; closing the +// write end models the container exiting and containerd closing the FIFO. +// Regression test for https://github.com/containerd/nerdctl/issues/5006 + +// TestLoggingProcessAdapterWaitError verifies that the logger does not treat a +// Wait failure as a container exit. containerd's client delivers Wait RPC +// errors through the exit channel as a synthetic ExitStatus carrying an error; +// if the logger cancelled its readers on such a delivery, all logging would +// silently stop while the container keeps running — and, in the foreground +// attach path, wedge `nerdctl run` behind the no-longer-drained logger pipes. +// The logger must instead re-arm the wait and keep reading until a real exit +// arrives. Regression test for +// https://github.com/containerd/nerdctl/issues/5137 +func TestLoggingProcessAdapterWaitError(t *testing.T) { + stdoutR, stdoutW, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + stderrR, stderrW, err := os.Pipe() + if err != nil { + t.Fatal(err) + } + defer stdoutR.Close() + defer stderrR.Close() + defer stderrW.Close() + + driver := &SyncMockDriver{} + config := &logging.Config{ + Stdout: stdoutR, + Stderr: stderrR, + } + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + // The first wait channel delivers a Wait RPC error (a synthetic exit); the + // second delivers a real exit once the test has verified that logging + // survived the first delivery. + rearmed := make(chan struct{}) + realExitCh := make(chan containerd.ExitStatus, 1) + var waitCalls int + var getContainerWaitMock ContainerWaitFunc = func(ctx context.Context, address string, config *logging.Config, outputSeen func() bool) (<-chan containerd.ExitStatus, error) { + waitCalls++ + if waitCalls == 1 { + errChan := make(chan containerd.ExitStatus, 1) + errChan <- *containerd.NewExitStatus(255, time.Time{}, errors.New("transient wait RPC failure")) + return errChan, nil + } + close(rearmed) + return realExitCh, nil + } + + done := make(chan error, 1) + go func() { + done <- loggingProcessAdapter(ctx, driver, "testDataStore", "", getContainerWaitMock, config) + }() + + if _, err := stdoutW.Write([]byte("before wait error\n")); err != nil { + t.Fatal(err) + } + + // The logger must re-arm the wait rather than cancel its readers. + select { + case <-rearmed: + case <-time.After(30 * time.Second): + t.Fatal("logger did not re-arm the container wait after the wait channel delivered an error") + } + + // Output produced after the errored delivery must still be logged. + if _, err := stdoutW.Write([]byte("after wait error\n")); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(30 * time.Second) + for { + driver.mu.Lock() + got := strings.Join(driver.receivedStdout, "") + driver.mu.Unlock() + if strings.Contains(got, "after wait error") { + break + } + if time.Now().After(deadline) { + t.Fatalf("output written after the errored wait delivery was never logged; got stdout: %q", got) + } + time.Sleep(10 * time.Millisecond) + } + + // A real exit must still terminate the logger. Close both write ends so + // the stream readers finish via EOF: on Windows, cancelreader cannot + // cancel a blocked pipe read, so the readers must not be left waiting on + // an open pipe when the exit is delivered. + stdoutW.Close() + stderrW.Close() + realExitCh <- containerd.ExitStatus{} + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(30 * time.Second): + t.Fatal("logger did not terminate on the real container exit") + } + + driver.mu.Lock() + defer driver.mu.Unlock() + stdout := strings.Join(driver.receivedStdout, "") + if !strings.Contains(stdout, "before wait error") || !strings.Contains(stdout, "after wait error") { + t.Fatalf("expected stdout to contain output from before and after the errored wait delivery, got: %q", stdout) + } +} + func TestLoggingProcessAdapterTrailingChunk(t *testing.T) { const expected = "'Hello World!\nThere is no newline'"