From 6bf82524b8e95b729592201f4c6142e95b54c3d6 Mon Sep 17 00:00:00 2001 From: An Lu Date: Thu, 20 Aug 2026 20:26:04 +0800 Subject: [PATCH] Fix foreground stdio deadlock when the internal logging process stops consuming `nerdctl run` in the foreground tees container stdout/stderr through a 64 KiB pipe to a forked logging process (`nerdctl _NERDCTL_INTERNAL_LOGGING`) on the same goroutine that drains the container's stdio FIFOs. If that logging process ever stopped consuming (killed, crashed, OOM), the tee write blocked forever: nerdctl kept its own copies of the logger pipe read ends open, so the kernel never delivered EPIPE. The blocked goroutine stopped draining the stdout FIFO, the container's writer blocked behind the full pipe at exactly pipe-capacity-plus-one-chunk bytes, the container never exited, `nerdctl run` never returned, and `nerdctl rm -f` on the wedged container hung as well. Four changes, from the analysis in #5137: - Close the parent's copies of the logger pipe read ends once the logging process has started (the equivalent of containerd's binaryIO.CloseAfterStart), so that logger death turns into EPIPE on the tee instead of an eternal pipe-buffer block. - Make the logger leg of the stdio tee best-effort: on the first failed write, warn and stop writing to the logger, but keep streaming to the attached stdout/stderr. Closing the read ends alone is not enough: the EPIPE would error the io.MultiWriter, abort the io.CopyBuffer that drains the container's stdio FIFO, and the container would still wedge on the undrained FIFO chain behind it. With both changes the attach and the container survive logger death; the log file is what goes incomplete (with a warning on nerdctl's stderr). - In the logging process, do not treat an errored delivery on the task wait channel as a container exit. containerd's client sends Wait RPC failures through the same channel as a synthetic ExitStatus; cancelling the stdio readers on such a delivery silently stopped all logging while the container was still running - and, before the changes above, wedged the foreground attach permanently. Re-arm the wait instead, and close each containerd client once its wait delivers so re-arming does not accumulate open clients. - Fail IO setup when a binary-v2 logging binary exits before signalling readiness (mirroring containerd's n == 0 check); plain binary:// keeps EOF-as-ready for backward compatibility with third-party logging binaries. Fixes #5137 Signed-off-by: An Lu --- pkg/cioutil/container_io.go | 83 ++++++++++++++++++++-- pkg/cioutil/container_io_test.go | 77 +++++++++++++++++++++ pkg/logging/logging.go | 72 ++++++++++++++++--- pkg/logging/logging_test.go | 115 +++++++++++++++++++++++++++++++ 4 files changed, 330 insertions(+), 17 deletions(-) create mode 100644 pkg/cioutil/container_io_test.go 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'"