Use armon/circbuf

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

Author: Vlad Volodkin

Dockerfile.local

@@ -76,5 +76,6 @@ COPY --from=builder /go/src/github.com/awslabs/mountpoint-s3-csi-driver/LICENSES
 COPY --from=builder /go/src/github.com/awslabs/mountpoint-s3-csi-driver/bin/aws-s3-csi-driver /bin/aws-s3-csi-driver
 COPY --from=builder /go/src/github.com/awslabs/mountpoint-s3-csi-driver/bin/aws-s3-csi-controller /bin/aws-s3-csi-controller
 COPY --from=builder /go/src/github.com/awslabs/mountpoint-s3-csi-driver/bin/aws-s3-csi-mounter /bin/aws-s3-csi-mounter
+COPY --from=builder /go/src/github.com/awslabs/mountpoint-s3-csi-driver/bin/aws-s3-csi-daemonset-mounter /bin/aws-s3-csi-daemonset-mounter
 
 ENTRYPOINT ["/bin/aws-s3-csi-driver"]

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

@@ -27,7 +27,7 @@ const errorFileExt = ".error"
 // ProcessManager tracks and manages Mountpoint child processes.
 type ProcessManager struct {
 	commDir   string
-	runner    ProcessRunner
+	runner    ProcessRunner // interface for spawning processes; substituted in tests
 	mu        sync.Mutex
 	processes map[string]ProcessHandle // mountId -> process handle
 	wg        sync.WaitGroup           // tracks waiter goroutines
@@ -143,9 +143,15 @@ func (pw *prefixWriter) Write(p []byte) (int, error) {
 		if len(line) == 0 && i == len(lines)-1 {
 			break
 		}
-		pw.w.Write([]byte(pw.prefix))
-		pw.w.Write(line)
-		pw.w.Write([]byte("\n"))
+		if _, err := pw.w.Write([]byte(pw.prefix)); err != nil {
+			return 0, err
+		}
+		if _, err := pw.w.Write(line); err != nil {
+			return 0, err
+		}
+		if _, err := pw.w.Write([]byte("\n")); err != nil {
+			return 0, err
+		}
 	}
 	return len(p), nil
 }

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

@@ -8,8 +8,8 @@ import (
 	"io"
 	"os"
 	"os/exec"
-	"sync"
 
+	"github.com/armon/circbuf"
 	"k8s.io/klog/v2"
 )
 
@@ -31,7 +31,10 @@ type defaultProcessRunner struct {
 }
 
 func (r *defaultProcessRunner) Start(cmd *exec.Cmd) (ProcessHandle, error) {
-	stderrBuf := newTailBuf(r.stderrCapacity)
+	stderrBuf, err := circbuf.NewBuffer(int64(r.stderrCapacity))
+	if err != nil {
+		return nil, err
+	}
 	if cmd.Stderr != nil {
 		cmd.Stderr = io.MultiWriter(cmd.Stderr, stderrBuf)
 	} else {
@@ -45,7 +48,7 @@ func (r *defaultProcessRunner) Start(cmd *exec.Cmd) (ProcessHandle, error) {
 
 type defaultProcessHandle struct {
 	cmd       *exec.Cmd
-	stderrBuf *tailBuf
+	stderrBuf *circbuf.Buffer
 }
 
 func (h *defaultProcessHandle) Pid() int { return h.cmd.Process.Pid }
@@ -69,53 +72,3 @@ func (h *defaultProcessHandle) Wait() (int, []byte) {
 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

@@ -3,6 +3,7 @@ module github.com/awslabs/mountpoint-s3-csi-driver
 go 1.26.2
 
 require (
+	github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2
 	github.com/aws/aws-sdk-go-v2 v1.39.2
 	github.com/aws/aws-sdk-go-v2/config v1.31.12
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.9

go.mod

@@ -1,5 +1,7 @@
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2 h1:7Ip0wMmLHLRJdrloDxZfhMm0xrLXZS8+COSu2bXmEQs=
+github.com/armon/circbuf v0.0.0-20190214190532-5111143e8da2/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
 github.com/aws/aws-sdk-go-v2 v1.39.2 h1:EJLg8IdbzgeD7xgvZ+I8M1e0fL0ptn/M47lianzth0I=
 github.com/aws/aws-sdk-go-v2 v1.39.2/go.mod h1:sDioUELIUO9Znk23YVmIk86/9DOpkbyyVb1i/gUNFXY=
 github.com/aws/aws-sdk-go-v2/config v1.31.12 h1:pYM1Qgy0dKZLHX2cXslNacbcEFMkDMl+Bcj5ROuS6p8=