Stream instance log lines longer than 64 KiB

lib/instances/logs.go

@@ -10,6 +10,7 @@ import (
 	"strconv"
 	"strings"
 	"time"
+	"unicode/utf8"
 
 	"github.com/kernel/hypeman/lib/logger"
 )
@@ -118,6 +119,7 @@ func (m *manager) streamInstanceLogs(ctx context.Context, id string, tail int, f
 		defer cmd.Process.Kill()
 
 		scanner := bufio.NewScanner(stdout)
+		scanner.Split(scanLogLines)
 		for scanner.Scan() {
 			line := scanner.Text()
 			if source == LogSourceApp && shouldSkipAppLogLine(line) {
@@ -142,6 +144,31 @@ func (m *manager) streamInstanceLogs(ctx context.Context, id string, tail int, f
 	return out, nil
 }
 
+// maxLogLineBytes is the longest line streamed as a single entry. It matches
+// bufio.Scanner's default token limit.
+const maxLogLineBytes = bufio.MaxScanTokenSize
+
+// scanLogLines is bufio.ScanLines, except that a line longer than
+// maxLogLineBytes is returned in pieces instead of failing the scan with
+// bufio.ErrTooLong, which would end the stream. Pieces do not split a UTF-8
+// sequence.
+func scanLogLines(data []byte, atEOF bool) (advance int, token []byte, err error) {
+	advance, token, err = bufio.ScanLines(data, atEOF)
+	if advance > 0 || token != nil || err != nil || len(data) < maxLogLineBytes {
+		return advance, token, err
+	}
+	n := maxLogLineBytes
+	for i := n - 1; i >= 0 && i > n-utf8.UTFMax; i-- {
+		if utf8.RuneStart(data[i]) {
+			if !utf8.FullRune(data[i:n]) {
+				n = i
+			}
+			break
+		}
+	}
+	return n, data[:n], nil
+}
+
 // rotateLogIfNeeded performs copytruncate rotation if file exceeds maxBytes
 // Keeps up to maxFiles old backups (.1, .2, etc.)
 func rotateLogIfNeeded(path string, maxBytes int64, maxFiles int) error {

lib/instances/logs_long_line_test.go

@@ -0,0 +1,47 @@
+package instances
+
+import (
+	"context"
+	"os"
+	"strings"
+	"testing"
+	"time"
+	"unicode/utf8"
+
+	"github.com/kernel/hypeman/lib/paths"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStreamInstanceLogsLongLine(t *testing.T) {
+	p := paths.New(t.TempDir())
+	m := &manager{paths: p}
+	require.NoError(t, m.ensureDirectories("long-line"))
+	require.NoError(t, m.saveMetadata(&metadata{StoredMetadata: StoredMetadata{Id: "long-line", DataDir: p.InstanceDir("long-line")}}))
+
+	// The odd-length prefix puts piece boundaries inside two-byte runes.
+	longLine := "x" + strings.Repeat("é", 200*1024)
+	content := "before\n" + longLine + "\nafter\n"
+	require.NoError(t, os.WriteFile(p.InstanceAppLog("long-line"), []byte(content), 0644))
+
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	logs, err := m.streamInstanceLogs(ctx, "long-line", 10, false, LogSourceApp)
+	require.NoError(t, err)
+
+	var lines []string
+	for {
+		select {
+		case line, ok := <-logs:
+			if !ok {
+				require.Equal(t, "before", lines[0])
+				require.Equal(t, "after", lines[len(lines)-1])
+				require.Equal(t, longLine, strings.Join(lines[1:len(lines)-1], ""))
+				return
+			}
+			require.True(t, utf8.ValidString(line), "piece %d splits a UTF-8 sequence", len(lines))
+			lines = append(lines, line)
+		case <-ctx.Done():
+			t.Fatalf("log stream did not finish; received %d lines", len(lines))
+		}
+	}
+}