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 <an.lu.91@googlemail.com>

Author: gsaddict

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...)

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

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