logging: fix data race in logging binary that loses container output

The logging binary had a goroutine that closed internal pipe writers
immediately upon container exit (via gRPC notification). This raced
with copyStream, which was still draining data from the external pipe.
When the gRPC notification arrived before copyStream finished, the
pipe writer was closed mid-copy, causing remaining data to be lost.

Fix by making copyStream responsible for closing the pipe writers
after all data has been copied. The container-exit goroutine (and its
getContainerWait infrastructure) is removed — pipe closure now happens
naturally when the parent process exits or the shim closes the pipe.

Fixes #4782

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: AkihiroSuda

pkg/logging/logging.go

@@ -26,14 +26,12 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
-	"strings"
 	"sync"
 	"time"
 
 	"github.com/fsnotify/fsnotify"
 	"github.com/muesli/cancelreader"
 
-	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/core/runtime/v2/logging"
 	"github.com/containerd/errdefs"
 	"github.com/containerd/log"
@@ -165,49 +163,7 @@ func WaitForLogger(dataStore, ns, id string) error {
 	})
 }
 
-func getContainerWait(ctx context.Context, address string, config *logging.Config) (<-chan containerd.ExitStatus, error) {
-	client, err := containerd.New(strings.TrimPrefix(address, "unix://"), containerd.WithDefaultNamespace(config.Namespace))
-	if err != nil {
-		return nil, err
-	}
-	con, err := client.LoadContainer(ctx, config.ID)
-	if err != nil {
-		return nil, err
-	}
-
-	task, err := con.Task(ctx, nil)
-	if err == nil {
-		return task.Wait(ctx)
-	}
-	if !errdefs.IsNotFound(err) {
-		return nil, err
-	}
-
-	// If task was not found, it's possible that the container runtime is still being created.
-	// Retry every 100ms.
-	ticker := time.NewTicker(100 * time.Millisecond)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ctx.Done():
-			return nil, errors.New("timed out waiting for container task to start")
-		case <-ticker.C:
-			task, err = con.Task(ctx, nil)
-			if err != nil {
-				if errdefs.IsNotFound(err) {
-					continue
-				}
-				return nil, err
-			}
-			return task.Wait(ctx)
-		}
-	}
-}
-
-type ContainerWaitFunc func(ctx context.Context, address string, config *logging.Config) (<-chan containerd.ExitStatus, error)
-
-func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, address string, getContainerWait ContainerWaitFunc, config *logging.Config) error {
+func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore string, config *logging.Config) error {
 	if err := driver.PreProcess(ctx, dataStore, config); err != nil {
 		return err
 	}
@@ -236,6 +192,10 @@ func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, addres
 		if err != nil {
 			log.G(ctx).Errorf("failed to copy stream: %s", err)
 		}
+		// Close the pipe writer after all data has been copied.
+		// This signals EOF to the downstream processLogFunc reader,
+		// ensuring all data is drained before the pipe is closed.
+		writer.Close()
 	}
 	go copyStream(stdoutR, pipeStdoutW)
 	go copyStream(stderrR, pipeStderrW)
@@ -269,18 +229,6 @@ func loggingProcessAdapter(ctx context.Context, driver Driver, dataStore, addres
 		defer wg.Done()
 		driver.Process(stdout, stderr)
 	}()
-	go func() {
-		// close pipeStdoutW and pipeStderrW upon container exit
-		defer pipeStdoutW.Close()
-		defer pipeStderrW.Close()
-
-		exitCh, err := getContainerWait(ctx, address, config)
-		if err != nil {
-			log.G(ctx).Errorf("failed to get container task wait channel: %v", err)
-			return
-		}
-		<-exitCh
-	}()
 	wg.Wait()
 	return driver.PostProcess()
 }
@@ -313,8 +261,7 @@ func loggerFunc(dataStore string) (logging.LoggerFunc, error) {
 				if err := ready(); err != nil {
 					return err
 				}
-				// getContainerWait is extracted as parameter to allow mocking in tests.
-				return loggingProcessAdapter(ctx, driver, dataStore, logConfig.Address, getContainerWait, config)
+				return loggingProcessAdapter(ctx, driver, dataStore, config)
 			})
 		} else if !errors.Is(err, os.ErrNotExist) {
 			// the file does not exist if the container was created with nerdctl < 0.20

pkg/logging/logging_test.go

@@ -23,9 +23,7 @@ import (
 	"math/rand"
 	"strings"
 	"testing"
-	"time"
 
-	containerd "github.com/containerd/containerd/v2/client"
 	"github.com/containerd/containerd/v2/core/runtime/v2/logging"
 )
 
@@ -79,21 +77,11 @@ func TestLoggingProcessAdapter(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	var getContainerWaitMock ContainerWaitFunc = func(ctx context.Context, address string, config *logging.Config) (<-chan containerd.ExitStatus, error) {
-		exitChan := make(chan containerd.ExitStatus, 1)
-		time.Sleep(50 * time.Millisecond)
-		exitChan <- containerd.ExitStatus{}
-		return exitChan, nil
-	}
-
-	err := loggingProcessAdapter(ctx, driver, "testDataStore", "", getContainerWaitMock, config)
+	err := loggingProcessAdapter(ctx, driver, "testDataStore", config)
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// let bufio read the buffer
-	time.Sleep(50 * time.Millisecond)
-
 	// Verify that the driver methods were called
 	if !driver.processed {
 		t.Fatal("process should be processed")