Address review feedback

Signed-off-by: Vlad Volodkin <vlaad@amazon.com>

Author: Vlad Volodkin

cmd/aws-s3-csi-daemonset-mounter/main.go

@@ -27,6 +27,7 @@
 package main
 
 import (
+	"errors"
 	"flag"
 	"net"
 	"os"
@@ -42,6 +43,7 @@ var (
 	commDir          = flag.String("comm-dir", "/comm", "Directory for communication socket and error files")
 	mountpointBinDir = flag.String("mountpoint-bin-dir", os.Getenv("MOUNTPOINT_BIN_DIR"), "Directory of mount-s3 binary")
 	recvTimeout      = flag.Duration("recv-timeout", 30*time.Second, "Timeout for receiving mount options from a connection")
+	stderrCapacity   = flag.Uint("stderr-capacity", 1024*1024, "Maximum bytes of stderr to retain per Mountpoint process (tail)")
 )
 
 const (
@@ -67,7 +69,7 @@ func main() {
 
 	klog.Infof("Listening on %s, mountpoint binary: %s", sockPath, mountpointPath)
 
-	pm := NewProcessManager(*commDir, &defaultProcessRunner{})
+	pm := NewProcessManager(*commDir, &defaultProcessRunner{stderrCapacity: *stderrCapacity})
 
 	// Handle shutdown signals: terminate all MP processes gracefully
 	sigCh := make(chan os.Signal, 1)
@@ -86,7 +88,7 @@ func main() {
 		conn, err := listener.Accept()
 		if err != nil {
 			// Check if listener was closed (shutdown)
-			if opErr, ok := err.(*net.OpError); ok && opErr.Err.Error() == "use of closed network connection" {
+			if errors.Is(err, net.ErrClosed) {
 				klog.Info("Listener closed, exiting accept loop")
 				break
 			}

cmd/aws-s3-csi-daemonset-mounter/process_runner.go

@@ -5,10 +5,12 @@
 package main
 
 import (
-	"bytes"
 	"io"
 	"os"
 	"os/exec"
+	"sync"
+
+	"k8s.io/klog/v2"
 )
 
 // ProcessHandle represents a started process that can be waited on.
@@ -24,41 +26,96 @@ type ProcessRunner interface {
 }
 
 // defaultProcessRunner is the real implementation that starts OS processes.
-type defaultProcessRunner struct{}
+type defaultProcessRunner struct {
+	stderrCapacity uint
+}
 
 func (r *defaultProcessRunner) Start(cmd *exec.Cmd) (ProcessHandle, error) {
-	var stderrBuf bytes.Buffer
+	stderrBuf := newTailBuf(r.stderrCapacity)
 	if cmd.Stderr != nil {
-		cmd.Stderr = io.MultiWriter(cmd.Stderr, &stderrBuf)
+		cmd.Stderr = io.MultiWriter(cmd.Stderr, stderrBuf)
 	} else {
-		cmd.Stderr = &stderrBuf
+		cmd.Stderr = stderrBuf
 	}
 	if err := cmd.Start(); err != nil {
 		return nil, err
 	}
-	return &realProcessHandle{cmd: cmd, stderrBuf: &stderrBuf}, nil
+	return &defaultProcessHandle{cmd: cmd, stderrBuf: stderrBuf}, nil
 }
 
-type realProcessHandle struct {
+type defaultProcessHandle struct {
 	cmd       *exec.Cmd
-	stderrBuf *bytes.Buffer
+	stderrBuf *tailBuf
 }
 
-func (h *realProcessHandle) Pid() int { return h.cmd.Process.Pid }
+func (h *defaultProcessHandle) Pid() int { return h.cmd.Process.Pid }
 
-func (h *realProcessHandle) Wait() (int, []byte) {
+func (h *defaultProcessHandle) Wait() (int, []byte) {
 	err := h.cmd.Wait()
 	exitCode := 0
 	if err != nil {
 		if exitErr, ok := err.(*exec.ExitError); ok {
 			exitCode = exitErr.ExitCode()
+		} else {
+			klog.Errorf("Unexpected error waiting for process %d: %v", h.cmd.Process.Pid, err)
+			exitCode = 1
 		}
 	} else {
 		exitCode = h.cmd.ProcessState.ExitCode()
 	}
 	return exitCode, h.stderrBuf.Bytes()
 }
 
-func (h *realProcessHandle) Signal(sig os.Signal) error {
+func (h *defaultProcessHandle) Signal(sig os.Signal) error {
 	return h.cmd.Process.Signal(sig)
 }
+
+// tailBuf is a fixed-capacity ring buffer that retains the last N bytes written.
+type tailBuf struct {
+	mu   sync.Mutex
+	buf  []byte
+	pos  int
+	full bool
+}
+
+func newTailBuf(capacity uint) *tailBuf {
+	return &tailBuf{buf: make([]byte, capacity)}
+}
+
+func (tb *tailBuf) Write(p []byte) (int, error) {
+	tb.mu.Lock()
+	defer tb.mu.Unlock()
+	n := len(p)
+	cap := len(tb.buf)
+	if cap == 0 {
+		return n, nil
+	}
+	if len(p) >= cap {
+		copy(tb.buf, p[len(p)-cap:])
+		tb.pos = 0
+		tb.full = true
+		return n, nil
+	}
+	for len(p) > 0 {
+		space := cap - tb.pos
+		copied := copy(tb.buf[tb.pos:], p[:min(len(p), space)])
+		tb.pos = (tb.pos + copied) % cap
+		if tb.pos == 0 || copied == space {
+			tb.full = true
+		}
+		p = p[copied:]
+	}
+	return n, nil
+}
+
+func (tb *tailBuf) Bytes() []byte {
+	tb.mu.Lock()
+	defer tb.mu.Unlock()
+	if !tb.full {
+		return append([]byte(nil), tb.buf[:tb.pos]...)
+	}
+	out := make([]byte, len(tb.buf))
+	n := copy(out, tb.buf[tb.pos:])
+	copy(out[n:], tb.buf[:tb.pos])
+	return out
+}

cmd/aws-s3-csi-daemonset-mounter/process_runner_test.go

@@ -0,0 +1,75 @@
+package main
+
+import (
+	"testing"
+)
+
+func TestTailBuf_UnderCapacity(t *testing.T) {
+	tb := newTailBuf(10)
+	tb.Write([]byte("hello"))
+	got := string(tb.Bytes())
+	if got != "hello" {
+		t.Fatalf("expected %q, got %q", "hello", got)
+	}
+}
+
+func TestTailBuf_ExactCapacity(t *testing.T) {
+	tb := newTailBuf(5)
+	tb.Write([]byte("abcde"))
+	got := string(tb.Bytes())
+	if got != "abcde" {
+		t.Fatalf("expected %q, got %q", "abcde", got)
+	}
+}
+
+func TestTailBuf_OverflowSingleWrite(t *testing.T) {
+	tb := newTailBuf(5)
+	tb.Write([]byte("abcdefgh"))
+	got := string(tb.Bytes())
+	if got != "defgh" {
+		t.Fatalf("expected %q, got %q", "defgh", got)
+	}
+}
+
+func TestTailBuf_OverflowMultipleWrites(t *testing.T) {
+	tb := newTailBuf(5)
+	tb.Write([]byte("abc"))
+	tb.Write([]byte("defgh"))
+	got := string(tb.Bytes())
+	if got != "defgh" {
+		t.Fatalf("expected %q, got %q", "defgh", got)
+	}
+}
+
+func TestTailBuf_WrapAround(t *testing.T) {
+	tb := newTailBuf(5)
+	tb.Write([]byte("abcd"))
+	tb.Write([]byte("ef"))
+	got := string(tb.Bytes())
+	if got != "bcdef" {
+		t.Fatalf("expected %q, got %q", "bcdef", got)
+	}
+}
+
+func TestTailBuf_ManySmallWrites(t *testing.T) {
+	tb := newTailBuf(4)
+	for _, b := range []byte("abcdefghij") {
+		tb.Write([]byte{b})
+	}
+	got := string(tb.Bytes())
+	if got != "ghij" {
+		t.Fatalf("expected %q, got %q", "ghij", got)
+	}
+}
+
+func TestTailBuf_ZeroCapacity(t *testing.T) {
+	tb := newTailBuf(0)
+	n, err := tb.Write([]byte("data"))
+	if err != nil || n != 4 {
+		t.Fatalf("expected n=4 err=nil, got n=%d err=%v", n, err)
+	}
+	got := string(tb.Bytes())
+	if got != "" {
+		t.Fatalf("expected empty, got %q", got)
+	}
+}