diff --git a/lib/instances/logs.go b/lib/instances/logs.go index c1ee84240..e1907e20a 100644 --- a/lib/instances/logs.go +++ b/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 { diff --git a/lib/instances/logs_long_line_test.go b/lib/instances/logs_long_line_test.go new file mode 100644 index 000000000..4a2d39f40 --- /dev/null +++ b/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)) + } + } +}