[cmd/opampsupervisor] Use bufio.Reader instead of bufio.Scanner to avoid buffer limits (#44310)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

When passthrough logging is enabled, the supervisor reads in collector
output and then writes it to its own logger instead of a dedicated log
file. An error can occur when a single collector log is too big (>64kb)
and cause the scanner to fail. Instead we'll use a reader splitting logs
based on newlines to avoid issues with logs being too large.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes #44127

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Hard to trigger with a unit test. Can verify using the hostmetrics
receiver. I was able to trigger the bug when the hostmetrics receiver
tried reading process metrics on my machine. This single log was greater
than 64kb. On the previous version I saw the error `bufio.Scanner: token
too long` but on this branch it prints the log no issue.

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

Author: dpaasman00

.chloggen/fix_supervisor-passthrough-logs-overflow.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: bug_fix
+
+# The name of the component, or a single word describing the area of concern, (e.g. receiver/filelog)
+component: cmd/opampsupervisor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Fix supervisor passthrough logs overflow by using bufio.Reader instead of bufio.Scanner
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [44127]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

cmd/opampsupervisor/e2e_test.go

@@ -1369,6 +1369,40 @@ func createHealthCheckCollectorConf(t *testing.T, nopPipeline bool) (cfg *bytes.
 	return &confmapBuf, h[:], 13133
 }
 
+func createHostMetricsCollectorConf(t *testing.T) (*bytes.Buffer, []byte) {
+	wd, err := os.Getwd()
+	require.NoError(t, err)
+
+	// Create output files
+	// The testing package will automatically clean these up after each test.
+	tempDir := t.TempDir()
+	outputFile, err := os.CreateTemp(tempDir, "output_*.json")
+	require.NoError(t, err)
+	t.Cleanup(func() { outputFile.Close() })
+
+	colCfgTpl, err := os.ReadFile(path.Join(wd, "testdata", "collector", "hostmetrics_pipeline.yaml"))
+	require.NoError(t, err)
+
+	templ, err := template.New("").Parse(string(colCfgTpl))
+	require.NoError(t, err)
+
+	var confmapBuf bytes.Buffer
+	err = templ.Execute(
+		&confmapBuf,
+		map[string]string{
+			"outputLogFile": outputFile.Name(),
+		},
+	)
+	require.NoError(t, err)
+
+	h := sha256.New()
+	if _, err := io.Copy(h, bytes.NewBuffer(confmapBuf.Bytes())); err != nil {
+		log.Fatal(err)
+	}
+
+	return &confmapBuf, h.Sum(nil)
+}
+
 // Wait for the Supervisor to connect to or disconnect from the OpAMP server
 func waitForSupervisorConnection(connection chan bool, connected bool) {
 	select {
@@ -1978,7 +2012,7 @@ func TestSupervisorLogging(t *testing.T) {
 	storageDir := t.TempDir()
 	remoteCfgFilePath := filepath.Join(storageDir, "last_recv_remote_config.dat")
 
-	collectorCfg, hash, _, _ := createSimplePipelineCollectorConf(t)
+	collectorCfg, hash := createHostMetricsCollectorConf(t)
 	remoteCfgProto := &protobufs.AgentRemoteConfig{
 		Config: &protobufs.AgentConfigMap{
 			ConfigMap: map[string]*protobufs.AgentConfigFile{
@@ -1998,15 +2032,16 @@ func TestSupervisorLogging(t *testing.T) {
 		},
 	})
 	defer server.shutdown()
+	server.start()
 
+	// manually create supervisor and logger for this test
 	supervisorLogFilePath := filepath.Join(storageDir, "supervisor_log.log")
 	cfgFile := getSupervisorConfig(t, "logging", map[string]string{
 		"url":         server.addr,
 		"storage_dir": storageDir,
 		"log_level":   "0",
 		"log_file":    supervisorLogFilePath,
 	})
-
 	cfg, err := config.Load(cfgFile.Name())
 	require.NoError(t, err)
 	logger, err := telemetry.NewLogger(cfg.Telemetry.Logs)
@@ -2016,23 +2051,27 @@ func TestSupervisorLogging(t *testing.T) {
 	require.NoError(t, err)
 	require.Nil(t, s.Start(t.Context()))
 
-	// Start the server and wait for the supervisor to connect
-	server.start()
 	waitForSupervisorConnection(server.supervisorConnected, true)
 	require.True(t, connected.Load(), "Supervisor failed to connect")
-
+	// give the collector some time to write to the log file
+	time.Sleep(5 * time.Second)
 	s.Shutdown()
 
 	// Read from log file checking for Info level logs
 	logFile, err := os.Open(supervisorLogFilePath)
 	require.NoError(t, err)
 
-	scanner := bufio.NewScanner(logFile)
+	reader := bufio.NewReader(logFile)
 	seenCollectorLog := false
-	for scanner.Scan() {
-		line := scanner.Bytes()
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			break
+		}
+		line = strings.TrimRight(line, "\r\n")
+
 		var log logEntry
-		err := json.Unmarshal(line, &log)
+		err = json.Unmarshal([]byte(line), &log)
 		require.NoError(t, err)
 
 		level, err := zapcore.ParseLevel(log.Level)

cmd/opampsupervisor/supervisor/commander/commander.go

@@ -8,10 +8,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"slices"
+	"strings"
 	"sync/atomic"
 	"syscall"
 	"time"
@@ -157,24 +159,42 @@ func (c *Commander) startWithPassthroughLogging() error {
 
 	// capture agent output
 	go func() {
-		scanner := bufio.NewScanner(stdoutPipe)
-		for scanner.Scan() {
-			line := scanner.Text()
+		reader := bufio.NewReader(stdoutPipe)
+		for {
+			line, err := reader.ReadString('\n')
+			if err != nil {
+				if err != io.EOF {
+					c.logger.Error("Error reading agent stdout", zap.Error(err))
+				}
+				// Trim and log the last line if it exists
+				if line != "" {
+					line = strings.TrimRight(line, "\r\n")
+					colLogger.Info(line)
+				}
+				break
+			}
+			line = strings.TrimRight(line, "\r\n")
 			colLogger.Info(line)
 		}
-		if err := scanner.Err(); err != nil {
-			c.logger.Error("Error reading agent stdout: %w", zap.Error(err))
-		}
 	}()
 	go func() {
-		scanner := bufio.NewScanner(stderrPipe)
-		for scanner.Scan() {
-			line := scanner.Text()
+		reader := bufio.NewReader(stderrPipe)
+		for {
+			line, err := reader.ReadString('\n')
+			if err != nil {
+				if err != io.EOF {
+					c.logger.Error("Error reading agent stderr", zap.Error(err))
+				}
+				// Trim and log the last line if it exists
+				if line != "" {
+					line = strings.TrimRight(line, "\r\n")
+					colLogger.Error(line)
+				}
+				break
+			}
+			line = strings.TrimRight(line, "\r\n")
 			colLogger.Error(line)
 		}
-		if err := scanner.Err(); err != nil {
-			c.logger.Error("Error reading agent stderr: %w", zap.Error(err))
-		}
 	}()
 
 	c.logger.Debug("Agent process started", zap.Int("pid", c.cmd.Process.Pid))
@@ -226,24 +246,50 @@ func (c *Commander) StartOneShot() ([]byte, []byte, error) {
 	}
 	// capture agent output
 	go func() {
-		scanner := bufio.NewScanner(stdoutPipe)
-		for scanner.Scan() {
-			stdout = append(stdout, scanner.Bytes()...)
+		reader := bufio.NewReader(stdoutPipe)
+		for {
+			line, err := reader.ReadString('\n')
+			if err != nil {
+				if err != io.EOF {
+					c.logger.Error("Error reading agent stdout", zap.Error(err))
+				}
+				// Trim and append the last line if it exists
+				// Normalize line endings to \n
+				if line != "" {
+					line = strings.TrimRight(line, "\r\n")
+					stdout = append(stdout, []byte(line)...)
+					stdout = append(stdout, byte('\n'))
+				}
+				break
+			}
+			// Normalize line endings to \n
+			line = strings.TrimRight(line, "\r\n")
+			stdout = append(stdout, []byte(line)...)
 			stdout = append(stdout, byte('\n'))
 		}
-		if err := scanner.Err(); err != nil {
-			c.logger.Error("Error reading agent stdout: %w", zap.Error(err))
-		}
 	}()
 	go func() {
-		scanner := bufio.NewScanner(stderrPipe)
-		for scanner.Scan() {
-			stderr = append(stderr, scanner.Bytes()...)
+		reader := bufio.NewReader(stderrPipe)
+		for {
+			line, err := reader.ReadString('\n')
+			if err != nil {
+				if err != io.EOF {
+					c.logger.Error("Error reading agent stderr", zap.Error(err))
+				}
+				// Trim and append the last line if it exists
+				// Normalize line endings to \n
+				if line != "" {
+					line = strings.TrimRight(line, "\r\n")
+					stderr = append(stderr, []byte(line)...)
+					stderr = append(stderr, byte('\n'))
+				}
+				break
+			}
+			// Normalize line endings to \n
+			line = strings.TrimRight(line, "\r\n")
+			stderr = append(stderr, []byte(line)...)
 			stderr = append(stderr, byte('\n'))
 		}
-		if err := scanner.Err(); err != nil {
-			c.logger.Error("Error reading agent stderr: %w", zap.Error(err))
-		}
 	}()
 
 	c.logger.Debug("Agent process started", zap.Int("pid", cmd.Process.Pid))

cmd/opampsupervisor/testdata/collector/hostmetrics_pipeline.yaml

@@ -0,0 +1,55 @@
+receivers:
+  hostmetrics:
+    collection_interval: 1s
+    scrapers:
+      process:
+        metrics:
+          process.threads:
+            enabled: true
+          process.signals_pending:
+            enabled: true
+          process.paging.faults:
+            enabled: true
+          process.memory.utilization:
+            enabled: true
+          process.memory.usage:
+            enabled: true
+          process.open_file_descriptors:
+            enabled: true
+          process.memory.virtual:
+            enabled: true
+          process.handles:
+            enabled: true
+          process.disk.operations:
+            enabled: true
+          process.context_switches:
+            enabled: true
+          process.disk.io:
+            enabled: true
+          process.cpu.utilization:
+            enabled: true
+          process.cpu.time:
+            enabled: true
+        mute_process_name_error: true
+        mute_process_exe_error: true
+        mute_process_io_error: true
+        mute_process_user_error: true
+        mute_process_cgroup_error: true  
+
+exporters:
+  file:
+    path: {{.outputLogFile}}
+
+extensions:
+  health_check:
+    endpoint: "localhost:13133"
+
+service:
+  extensions: [health_check]
+  pipelines:
+    logs:
+      receivers: [hostmetrics]
+      exporters: [file]
+  telemetry:
+    logs:
+      level: debug

cmd/opampsupervisor/testdata/supervisor/supervisor_logging.yaml

@@ -21,4 +21,7 @@ agent:
 telemetry:
   logs:
     level: {{.log_level}} # info level logs
-    output_paths: ['{{.log_file}}']
+    # keep stdout so we can see logs in test output
+    output_paths: 
+      - '{{.log_file}}'
+      - 'stdout'