Handle long log lines from scrapers and plugins

pkg/logger/plugin.go

@@ -187,25 +187,57 @@ func (log *PluginLogger) handleStderrLine(line string) {
 	}
 }
 
+// maxLogLineLength is the maximum length of a single log line read from a plugin.
+// Longer lines are truncated to this length.
+const maxLogLineLength = 64 * 1024
+
+const truncatedSuffix = " [truncated]"
+
 // ReadLogMessages reads plugin log messages from src, forwarding them to the PluginLoggers Logger.
 // ProgressLevel messages are parsed as float64 and forwarded to ProgressChan. If ProgressChan is full,
 // then the progress message is not forwarded.
+// Log lines longer than maxLogLineLength are truncated.
 // This method only returns when it reaches the end of src or encounters an error while reading src.
 // This method closes src before returning.
 func (log *PluginLogger) ReadLogMessages(src io.ReadCloser) {
-	// pipe plugin stderr to our logging
-	scanner := bufio.NewScanner(src)
-	for scanner.Scan() {
-		str := scanner.Text()
-		if str != "" {
-			log.handleStderrLine(str)
+	defer src.Close()
+
+	// pipe plugin stderr to our logging.
+	// bufio.Scanner is unsuitable here: it aborts permanently on lines longer than
+	// its buffer, which would stop us reading the remainder of the plugin's output
+	// and break the plugin's stderr pipe.
+	reader := bufio.NewReader(src)
+
+	var line strings.Builder
+	truncated := false
+
+	for {
+		chunk, isPrefix, err := reader.ReadLine()
+
+		if remaining := maxLogLineLength - line.Len(); remaining < len(chunk) {
+			line.Write(chunk[:max(remaining, 0)])
+			truncated = true
+		} else {
+			line.Write(chunk)
 		}
-	}
 
-	str := scanner.Text()
-	if str != "" {
-		log.handleStderrLine(str)
-	}
+		// isPrefix indicates that the line was too long for the read buffer and
+		// will be continued by the next read
+		if !isPrefix {
+			str := line.String()
+			if truncated {
+				str += truncatedSuffix
+			}
+			if str != "" {
+				log.handleStderrLine(str)
+			}
+
+			line.Reset()
+			truncated = false
+		}
 
-	src.Close()
+		if err != nil {
+			return
+		}
+	}
 }

pkg/logger/plugin_test.go

@@ -0,0 +1,66 @@
+package logger
+
+import (
+	"fmt"
+	"io"
+	"strings"
+	"testing"
+)
+
+type captureLogger struct {
+	LoggerImpl
+	messages []string
+}
+
+func (l *captureLogger) Debug(args ...interface{}) {
+	l.messages = append(l.messages, fmt.Sprint(args[len(args)-1]))
+}
+
+func TestReadLogMessages(t *testing.T) {
+	const prefix = "\x01d\x02"
+
+	tests := []struct {
+		name  string
+		input string
+		want  []string
+	}{
+		{
+			name:  "short lines",
+			input: prefix + "one\n" + prefix + "two\n",
+			want:  []string{"one", "two"},
+		},
+		{
+			name:  "no trailing newline",
+			input: prefix + "one\n" + prefix + "two",
+			want:  []string{"one", "two"},
+		},
+		{
+			// a line longer than bufio.MaxScanTokenSize used to abort the read
+			// entirely, discarding every subsequent line
+			name:  "over-long line does not stop subsequent lines",
+			input: prefix + strings.Repeat("x", maxLogLineLength*3) + "\n" + prefix + "after\n",
+			want: []string{
+				strings.Repeat("x", maxLogLineLength-len(prefix)) + truncatedSuffix,
+				"after",
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			l := &captureLogger{}
+			pl := PluginLogger{Logger: l}
+			pl.ReadLogMessages(io.NopCloser(strings.NewReader(tt.input)))
+
+			if len(l.messages) != len(tt.want) {
+				t.Fatalf("got %d messages, want %d", len(l.messages), len(tt.want))
+			}
+			for i, want := range tt.want {
+				if l.messages[i] != want {
+					t.Errorf("message %d: got %.80q (len %d), want %.80q (len %d)",
+						i, l.messages[i], len(l.messages[i]), want, len(want))
+				}
+			}
+		})
+	}
+}