cli_executor: handle pipe read errors

Previously, when reading and logging the stdout/stderr streams, we
ignored potential read errors. After scanner.Scan() returns false,
scanner.Err() may contain any errors encountered while reading.

Handle this error. Stop logging the stdout/stderr (and log a warning
about it) but still consume the rest of the pipe if possible to collect
the entire stdout/stderr.

If direct consumption of the pipe also fails, return the error.

Assisted-by: Claude
Signed-off-by: Adam Cmiel <acmiel@redhat.com>

Author: chmeliik

pkg/cliwrappers/cli_executor.go

@@ -70,34 +70,50 @@ func (e *CliExecutor) Execute(c Cmd) (string, string, int, error) {
 
 	var stdoutBuf, stderrBuf bytes.Buffer
 
-	readStream := func(linePrefix string, r io.Reader, buf *bytes.Buffer) {
-		scanner := bufio.NewScanner(r)
+	readStream := func(linePrefix string, r io.Reader, buf *bytes.Buffer) error {
+		tee := io.TeeReader(r, buf)
+		scanner := bufio.NewScanner(tee)
 		for scanner.Scan() {
-			line := scanner.Text()
-			executorLog.Info(linePrefix + line)
-			buf.WriteString(line + "\n")
+			executorLog.Info(linePrefix + scanner.Text())
 		}
+		if scanner.Err() != nil {
+			executorLog.Warnf("%sstopped logging output: %s", linePrefix, scanner.Err())
+			// Read the rest of the pipe directly into buf.
+			//
+			// At this point, buf contains everything that was read from r via the scanner
+			// (by property of TeeReader - "the write must complete before the read completes").
+			// A second property of TeeReader that could break this:
+			// "Any error encountered while writing is reported as a read error".
+			// Not a problem here - [bytes.Buffer.Write] never errors, only panics.
+			//
+			// Naturally, buf also doesn't contain anything that *wasn't* read via the scanner,
+			// because the tee being read by the scanner is the only thing doing the writing.
+			// So this Copy() picks up exactly where the scanner left off.
+			if _, err := io.Copy(buf, r); err != nil {
+				return fmt.Errorf("failed to read remaining output: %w", err)
+			}
+		}
+		return nil
 	}
 
 	nameInLogs := c.NameInLogs
 	if nameInLogs == "" {
 		nameInLogs = c.Name
 	}
 
-	done := make(chan struct{}, 2)
+	done := make(chan error, 2)
 	go func() {
-		readStream(nameInLogs+" [stdout] ", stdoutPipe, &stdoutBuf)
-		done <- struct{}{}
+		done <- readStream(nameInLogs+" [stdout] ", stdoutPipe, &stdoutBuf)
 	}()
 	go func() {
-		readStream(nameInLogs+" [stderr] ", stderrPipe, &stderrBuf)
-		done <- struct{}{}
+		done <- readStream(nameInLogs+" [stderr] ", stderrPipe, &stderrBuf)
 	}()
 
-	err = cmd.Wait()
-	// Wait for both output streams to finish
-	<-done
-	<-done
+	// Wait for both output streams to finish before calling cmd.Wait().
+	// Per [exec.Cmd.StdoutPipe] docs, Wait closes the pipes, so all reads must complete first.
+	readErr := errors.Join(<-done, <-done)
+	cmdErr := cmd.Wait()
+	err = errors.Join(readErr, cmdErr)
 
 	return stdoutBuf.String(), stderrBuf.String(), getExitCodeFromError(err), err
 }

pkg/cliwrappers/cli_executor_test.go

@@ -2,6 +2,7 @@ package cliwrappers_test
 
 import (
 	"bytes"
+	"fmt"
 	"os"
 	"path/filepath"
 	"runtime"
@@ -329,6 +330,39 @@ func TestCliExecutor_ExecuteWithLogOutput(t *testing.T) {
 		g.Expect(stdout).To(BeEmpty())
 		g.Expect(stderr).To(BeEmpty())
 	})
+
+	t.Run("should capture full output even when a line exceeds the scanner buffer size", func(t *testing.T) {
+		g := NewWithT(t)
+
+		longLineFile := filepath.Join(t.TempDir(), "long_line.txt")
+		// bufio.Scanner's default max token size is 64KB
+		longLine := strings.Repeat("x", 128*1024)
+		fileContent := fmt.Sprintf("before\n%s\nafter\n", longLine)
+
+		err := os.WriteFile(longLineFile, []byte(fileContent), 0644)
+		g.Expect(err).ToNot(HaveOccurred())
+
+		executor := cliwrappers.NewCliExecutor()
+
+		var stdout, stderr string
+		var exitCode int
+		logOutput := captureLogOutput(func() {
+			cmd := cliwrappers.Command("cat", longLineFile)
+			cmd.LogOutput = true
+			stdout, stderr, exitCode, err = executor.Execute(cmd)
+		})
+
+		g.Expect(err).ToNot(HaveOccurred())
+		g.Expect(exitCode).To(Equal(0))
+		g.Expect(stderr).To(BeEmpty())
+		g.Expect(stdout).To(ContainSubstring("before"))
+		g.Expect(stdout).To(ContainSubstring("\n" + longLine + "\n"))
+		g.Expect(stdout).To(ContainSubstring("after"))
+
+		g.Expect(logOutput).To(ContainSubstring("before"))
+		g.Expect(logOutput).To(ContainSubstring("stopped logging output: bufio.Scanner: token too long"))
+		g.Expect(logOutput).ToNot(ContainSubstring(longLine))
+	})
 }
 
 func TestCheckCliToolAvailable(t *testing.T) {