summaryrefslogtreecommitdiff
path: root/internal/io/journal
diff options
context:
space:
mode:
Diffstat (limited to 'internal/io/journal')
-rw-r--r--internal/io/journal/filter.go253
-rw-r--r--internal/io/journal/reader.go303
-rw-r--r--internal/io/journal/reader_test.go754
-rw-r--r--internal/io/journal/reader_unsupported.go45
-rw-r--r--internal/io/journal/testhelper/mock.go397
-rw-r--r--internal/io/journal/testhelper/mock_test.go247
6 files changed, 1999 insertions, 0 deletions
diff --git a/internal/io/journal/filter.go b/internal/io/journal/filter.go
new file mode 100644
index 0000000..680c183
--- /dev/null
+++ b/internal/io/journal/filter.go
@@ -0,0 +1,253 @@
+//go:build linux
+
+package journal
+
+import (
+ "bytes"
+ "context"
+
+ "github.com/mimecast/dtail/internal/io/line"
+ "github.com/mimecast/dtail/internal/io/pool"
+ "github.com/mimecast/dtail/internal/lcontext"
+ "github.com/mimecast/dtail/internal/regex"
+)
+
+type journalSink interface {
+ Emit(context.Context, *bytes.Buffer, uint64, int, string) error
+ Full() bool
+}
+
+type processorSink struct {
+ processor line.Processor
+}
+
+func (s processorSink) Emit(_ context.Context, rawLine *bytes.Buffer, count uint64,
+ _ int, sourceID string) error {
+
+ // Per the line.Processor contract, ownership of rawLine transfers to the
+ // processor: it is responsible for recycling the buffer on every return path,
+ // success or error (the journal-path processors installed via makeProcessor
+ // — DirectLineProcessor and AggregateProcessor — recycle unconditionally
+ // before returning a write error, for example). Recycling here
+ // on error would return the same buffer to the shared pool a second time; the
+ // pool would then hand one object to two Get callers and their concurrent
+ // writes would corrupt data and race. So do not recycle rawLine here.
+ return s.processor.ProcessLine(rawLine, count, sourceID)
+}
+
+func (s processorSink) Full() bool {
+ return false
+}
+
+type journalFilter struct {
+ ltx lcontext.LContext
+ sink journalSink
+ re regex.Regex
+ sourceID string
+ stats journalStats
+
+ before []bufferedLine
+ after int
+ maxCount int
+ maxHit int
+ maxClosed bool
+}
+
+type bufferedLine struct {
+ content *bytes.Buffer
+ count uint64
+}
+
+func newJournalFilter(ltx lcontext.LContext, sink journalSink, re regex.Regex,
+ sourceID string) *journalFilter {
+
+ return &journalFilter{
+ ltx: ltx,
+ sink: sink,
+ re: re,
+ sourceID: sourceID,
+ maxCount: ltx.MaxCount,
+ }
+}
+
+func (f *journalFilter) Process(ctx context.Context, rawLine *bytes.Buffer) error {
+ f.stats.updatePosition()
+ if !f.ltx.Has() {
+ return f.processWithoutContext(ctx, rawLine)
+ }
+ return f.processWithContext(ctx, rawLine)
+}
+
+func (f *journalFilter) Close() {
+ for _, line := range f.before {
+ pool.RecycleBytesBuffer(line.content)
+ }
+ f.before = nil
+}
+
+func (f *journalFilter) processWithoutContext(ctx context.Context, rawLine *bytes.Buffer) error {
+ if !f.re.Match(rawLine.Bytes()) {
+ f.stats.updateLineNotMatched()
+ f.stats.updateLineNotTransmitted()
+ pool.RecycleBytesBuffer(rawLine)
+ return nil
+ }
+
+ f.stats.updateLineMatched()
+ if f.sink.Full() {
+ f.stats.updateLineNotTransmitted()
+ pool.RecycleBytesBuffer(rawLine)
+ return nil
+ }
+ f.stats.updateLineTransmitted()
+ return f.sink.Emit(ctx, rawLine, f.stats.totalLineCount(), f.stats.transmittedPerc(), f.sourceID)
+}
+
+func (f *journalFilter) processWithContext(ctx context.Context, rawLine *bytes.Buffer) error {
+ if !f.re.Match(rawLine.Bytes()) {
+ return f.processContextMiss(ctx, rawLine)
+ }
+
+ f.stats.updateLineMatched()
+ if f.maxClosed {
+ pool.RecycleBytesBuffer(rawLine)
+ return errStopReading
+ }
+
+ if err := f.emitBefore(ctx); err != nil {
+ pool.RecycleBytesBuffer(rawLine)
+ return err
+ }
+ f.stats.updateLineTransmitted()
+ if err := f.sink.Emit(ctx, rawLine, f.stats.totalLineCount(), 100, f.sourceID); err != nil {
+ return err
+ }
+
+ if f.maxCount > 0 {
+ f.maxHit++
+ if f.maxHit >= f.maxCount {
+ if f.ltx.AfterContext == 0 {
+ return errStopReading
+ }
+ f.maxClosed = true
+ }
+ }
+ if f.ltx.AfterContext > 0 {
+ f.after = f.ltx.AfterContext
+ }
+ return nil
+}
+
+func (f *journalFilter) processContextMiss(ctx context.Context, rawLine *bytes.Buffer) error {
+ f.stats.updateLineNotMatched()
+ if f.maxClosed && f.after == 0 {
+ pool.RecycleBytesBuffer(rawLine)
+ return errStopReading
+ }
+ if f.after > 0 {
+ f.after--
+ f.stats.updateLineTransmitted()
+ err := f.sink.Emit(ctx, rawLine, f.stats.totalLineCount(), 100, f.sourceID)
+ if err == nil && f.maxClosed && f.after == 0 {
+ return errStopReading
+ }
+ return err
+ }
+ if f.ltx.BeforeContext > 0 {
+ f.rememberBefore(rawLine)
+ f.stats.updateLineNotTransmitted()
+ return nil
+ }
+
+ f.stats.updateLineNotTransmitted()
+ pool.RecycleBytesBuffer(rawLine)
+ return nil
+}
+
+func (f *journalFilter) rememberBefore(rawLine *bytes.Buffer) {
+ if len(f.before) >= f.ltx.BeforeContext {
+ pool.RecycleBytesBuffer(f.before[0].content)
+ copy(f.before, f.before[1:])
+ f.before = f.before[:len(f.before)-1]
+ }
+ f.before = append(f.before, bufferedLine{
+ content: rawLine,
+ count: f.stats.totalLineCount(),
+ })
+}
+
+func (f *journalFilter) emitBefore(ctx context.Context) error {
+ for i, line := range f.before {
+ f.stats.updateLineTransmitted()
+ if err := f.sink.Emit(ctx, line.content, line.count, 100, f.sourceID); err != nil {
+ f.discardBeforeFrom(i + 1)
+ return err
+ }
+ }
+ f.before = f.before[:0]
+ return nil
+}
+
+func (f *journalFilter) discardBeforeFrom(index int) {
+ for _, line := range f.before[index:] {
+ pool.RecycleBytesBuffer(line.content)
+ }
+ f.before = f.before[:0]
+}
+
+type journalStats struct {
+ pos int
+ lineCount uint64
+ matched [100]bool
+ matchCount uint64
+ transmitted [100]bool
+ transmitCount int
+}
+
+func (s *journalStats) totalLineCount() uint64 {
+ return s.lineCount
+}
+
+func (s *journalStats) transmittedPerc() int {
+ return int(percentOf(float64(s.matchCount), float64(s.transmitCount)))
+}
+
+func (s *journalStats) updatePosition() {
+ s.pos = (s.pos + 1) % 100
+ s.lineCount++
+}
+
+func (s *journalStats) updateLineMatched() {
+ if !s.matched[s.pos] {
+ s.matchCount++
+ s.matched[s.pos] = true
+ }
+}
+
+func (s *journalStats) updateLineTransmitted() {
+ if !s.transmitted[s.pos] {
+ s.transmitCount++
+ s.transmitted[s.pos] = true
+ }
+}
+
+func (s *journalStats) updateLineNotMatched() {
+ if s.matched[s.pos] {
+ s.matchCount--
+ s.matched[s.pos] = false
+ }
+}
+
+func (s *journalStats) updateLineNotTransmitted() {
+ if s.transmitted[s.pos] {
+ s.transmitCount--
+ s.transmitted[s.pos] = false
+ }
+}
+
+func percentOf(total float64, value float64) float64 {
+ if total == 0 || total == value {
+ return 100
+ }
+ return value / (total / 100.0)
+}
diff --git a/internal/io/journal/reader.go b/internal/io/journal/reader.go
new file mode 100644
index 0000000..bc27dbf
--- /dev/null
+++ b/internal/io/journal/reader.go
@@ -0,0 +1,303 @@
+//go:build linux
+
+// Package journal provides a journalctl-backed file reader.
+package journal
+
+import (
+ "bufio"
+ "bytes"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "os/exec"
+ "strings"
+ "syscall"
+ "time"
+
+ "github.com/mimecast/dtail/internal/io/fs"
+ "github.com/mimecast/dtail/internal/io/line"
+ "github.com/mimecast/dtail/internal/io/pool"
+ "github.com/mimecast/dtail/internal/lcontext"
+ "github.com/mimecast/dtail/internal/regex"
+)
+
+const (
+ defaultSourceID = "journal"
+ journalctlCommand = "journalctl"
+ maxScannerTokenSize = 1024 * 1024
+ processTerminateGrace = 200 * time.Millisecond
+)
+
+var errStopReading = errors.New("stop journal reading")
+
+// ErrJournalctlNotFound reports that journalctl could not be found on PATH.
+var ErrJournalctlNotFound = errors.New("journalctl not found")
+
+// Reader reads journal entries by executing journalctl.
+type Reader struct {
+ journalctlPath string
+ args []string
+ sourceID string
+ serverMessages chan<- string
+ follow bool
+}
+
+var _ fs.FileReader = (*Reader)(nil)
+
+// NewReader returns a journalctl-backed file reader.
+func NewReader(args []string, sourceID string, follow bool, serverMessages chan<- string) (*Reader, error) {
+ journalctlPath, err := exec.LookPath(journalctlCommand)
+ if err != nil {
+ return nil, fmt.Errorf("%w: %w", ErrJournalctlNotFound, err)
+ }
+ if sourceID == "" {
+ sourceID = defaultSourceID
+ }
+
+ copiedArgs := append([]string(nil), args...)
+ return &Reader{
+ journalctlPath: journalctlPath,
+ args: copiedArgs,
+ sourceID: sourceID,
+ serverMessages: serverMessages,
+ follow: follow,
+ }, nil
+}
+
+// StartWithProcessor reads journalctl stdout and sends matching lines to processor.
+func (r *Reader) StartWithProcessor(ctx context.Context, ltx lcontext.LContext,
+ processor line.Processor, re regex.Regex) error {
+
+ return r.runWithProcessor(ctx, ltx, processor, re)
+}
+
+// StartWithProcessorOptimized reads journalctl stdout and sends matching lines to processor.
+func (r *Reader) StartWithProcessorOptimized(ctx context.Context, ltx lcontext.LContext,
+ processor line.Processor, re regex.Regex) error {
+
+ return r.runWithProcessor(ctx, ltx, processor, re)
+}
+
+// FilePath returns a stable journalctl command description.
+func (r *Reader) FilePath() string {
+ if len(r.args) == 0 {
+ return journalctlCommand
+ }
+ return journalctlCommand + " " + strings.Join(r.args, " ")
+}
+
+// Retry reports whether journalctl should be restarted after it exits.
+func (r *Reader) Retry() bool {
+ return r.follow
+}
+
+func (r *Reader) runWithProcessor(ctx context.Context, ltx lcontext.LContext,
+ processor line.Processor, re regex.Regex) error {
+
+ sink := processorSink{processor: processor}
+
+ // In follow mode r.run blocks until journalctl is stopped, so a batching
+ // processor (the NetworkWriter, which buffers up to 64KB before
+ // sending) would hold live lines in its buffer and the client would never
+ // see interactive output. Flush after every scanned line while following so
+ // journal follow output reaches the client promptly — the same latency
+ // guarantee the file follow path gets from its per-read-chunk flush. A
+ // non-follow snapshot read keeps the batching benefit and flushes once at
+ // the end below.
+ var flushLine func() error
+ if r.follow {
+ flushLine = processor.Flush
+ }
+
+ err := r.run(ctx, ltx, sink, re, flushLine)
+ if flushErr := processor.Flush(); flushErr != nil && err == nil {
+ err = flushErr
+ }
+ return err
+}
+
+func (r *Reader) run(ctx context.Context, ltx lcontext.LContext, sink journalSink,
+ re regex.Regex, flushLine func() error) error {
+
+ cmd := r.command(ctx)
+
+ stdout, err := cmd.StdoutPipe()
+ if err != nil {
+ return fmt.Errorf("open journalctl stdout: %w", err)
+ }
+ stderr, err := cmd.StderrPipe()
+ if err != nil {
+ return fmt.Errorf("open journalctl stderr: %w", err)
+ }
+ if err := cmd.Start(); err != nil {
+ return fmt.Errorf("start journalctl: %w", err)
+ }
+
+ stderrDone := make(chan error, 1)
+ go func() {
+ stderrDone <- r.forwardStderr(ctx, stderr)
+ }()
+
+ filter := newJournalFilter(ltx, sink, re, r.sourceID)
+ scanErr := r.scanStdout(ctx, stdout, filter, flushLine)
+ waitErr := waitForJournalctl(cmd, scanErr != nil)
+ stderrErr := <-stderrDone
+ filter.Close()
+
+ if ctx.Err() != nil || errors.Is(scanErr, errStopReading) {
+ return nil
+ }
+ if scanErr != nil {
+ return scanErr
+ }
+ if stderrErr != nil {
+ return stderrErr
+ }
+ if waitErr != nil {
+ return fmt.Errorf("journalctl failed: %w", waitErr)
+ }
+ return nil
+}
+
+func (r *Reader) command(ctx context.Context) *exec.Cmd {
+ args := r.commandArgs()
+ cmd := exec.CommandContext(ctx, r.journalctlPath, args...)
+ cmd.Cancel = func() error {
+ terminateProcess(cmd.Process)
+ return nil
+ }
+ cmd.WaitDelay = processTerminateGrace
+ return cmd
+}
+
+func (r *Reader) commandArgs() []string {
+ args := append([]string(nil), r.args...)
+ if r.follow {
+ args = append(args, "-f", "-n", "0")
+ }
+ return args
+}
+
+func terminateProcess(process *os.Process) {
+ if process == nil {
+ return
+ }
+ _ = process.Signal(syscall.SIGTERM)
+}
+
+func killProcess(process *os.Process) {
+ if process == nil {
+ return
+ }
+ _ = process.Kill()
+}
+
+func waitForJournalctl(cmd *exec.Cmd, earlyStop bool) error {
+ if !earlyStop {
+ return cmd.Wait()
+ }
+
+ terminateProcess(cmd.Process)
+
+ waitDone := make(chan error, 1)
+ go func() {
+ waitDone <- cmd.Wait()
+ }()
+
+ timer := time.NewTimer(processTerminateGrace)
+ defer timer.Stop()
+
+ select {
+ case err := <-waitDone:
+ return err
+ case <-timer.C:
+ killProcess(cmd.Process)
+ return <-waitDone
+ }
+}
+
+func (r *Reader) scanStdout(ctx context.Context, stdout io.Reader, filter *journalFilter,
+ flushLine func() error) error {
+
+ scanner := bufio.NewScanner(stdout)
+ bufPtr := pool.GetScannerBuffer()
+ defer pool.PutScannerBuffer(bufPtr)
+
+ scanner.Buffer(*bufPtr, maxScannerTokenSize)
+ scanner.Split(scanLinesPreserveEndings)
+
+ for scanner.Scan() {
+ select {
+ case <-ctx.Done():
+ return nil
+ default:
+ }
+
+ lineBuf := pool.BytesBuffer.Get().(*bytes.Buffer)
+ lineBuf.Write(scanner.Bytes())
+ if err := filter.Process(ctx, lineBuf); err != nil {
+ return err
+ }
+
+ // In follow mode, push any buffered line straight to the client so
+ // batching never delays live output. flushLine is nil for non-follow
+ // reads (they batch and flush once at the end) and for the immediate
+ // channel sink.
+ if flushLine != nil {
+ if err := flushLine(); err != nil {
+ return err
+ }
+ }
+ }
+ if err := scanner.Err(); err != nil {
+ return fmt.Errorf("scan journalctl stdout: %w", err)
+ }
+ return nil
+}
+
+func (r *Reader) forwardStderr(ctx context.Context, stderr io.Reader) error {
+ scanner := bufio.NewScanner(stderr)
+ bufPtr := pool.GetScannerBuffer()
+ defer pool.PutScannerBuffer(bufPtr)
+
+ scanner.Buffer(*bufPtr, maxScannerTokenSize)
+ for scanner.Scan() {
+ if !r.sendServerMessage(ctx, fmt.Sprintf("journalctl stderr: %s\n", scanner.Text())) {
+ return nil
+ }
+ }
+ if err := scanner.Err(); err != nil {
+ if errors.Is(err, os.ErrClosed) {
+ return nil
+ }
+ return fmt.Errorf("scan journalctl stderr: %w", err)
+ }
+ return nil
+}
+
+func (r *Reader) sendServerMessage(ctx context.Context, message string) bool {
+ if r.serverMessages == nil {
+ return true
+ }
+ select {
+ case r.serverMessages <- message:
+ return true
+ case <-ctx.Done():
+ return false
+ }
+}
+
+func scanLinesPreserveEndings(data []byte, atEOF bool) (int, []byte, error) {
+ if atEOF && len(data) == 0 {
+ return 0, nil, nil
+ }
+ if i := bytes.IndexByte(data, '\n'); i >= 0 {
+ return i + 1, data[:i+1], nil
+ }
+ if atEOF {
+ return len(data), nil, nil
+ }
+ return 0, nil, nil
+}
diff --git a/internal/io/journal/reader_test.go b/internal/io/journal/reader_test.go
new file mode 100644
index 0000000..a2d084f
--- /dev/null
+++ b/internal/io/journal/reader_test.go
@@ -0,0 +1,754 @@
+//go:build linux
+
+package journal
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "os"
+ "os/exec"
+ "reflect"
+ "strconv"
+ "strings"
+ "sync"
+ "syscall"
+ "testing"
+ "time"
+
+ journaltest "github.com/mimecast/dtail/internal/io/journal/testhelper"
+ "github.com/mimecast/dtail/internal/io/pool"
+ "github.com/mimecast/dtail/internal/lcontext"
+ "github.com/mimecast/dtail/internal/regex"
+)
+
+type captureProcessor struct {
+ lines []string
+}
+
+func (p *captureProcessor) ProcessLine(lineContent *bytes.Buffer, _ uint64, _ string) error {
+ p.lines = append(p.lines, lineContent.String())
+ pool.RecycleBytesBuffer(lineContent)
+ return nil
+}
+
+func (p *captureProcessor) Flush() error {
+ return nil
+}
+
+func (p *captureProcessor) Close() error {
+ return nil
+}
+
+type errorProcessor struct {
+ err error
+}
+
+func (p errorProcessor) ProcessLine(lineContent *bytes.Buffer, _ uint64, _ string) error {
+ pool.RecycleBytesBuffer(lineContent)
+ return p.err
+}
+
+func (p errorProcessor) Flush() error {
+ return nil
+}
+
+func (p errorProcessor) Close() error {
+ return nil
+}
+
+// nonRecyclingErrorProcessor returns an error without recycling the buffer, so a
+// test can observe whether processorSink.Emit wrongly recycles a buffer it does
+// not own.
+type nonRecyclingErrorProcessor struct {
+ err error
+}
+
+func (p nonRecyclingErrorProcessor) ProcessLine(_ *bytes.Buffer, _ uint64, _ string) error {
+ return p.err
+}
+
+func (p nonRecyclingErrorProcessor) Flush() error { return nil }
+
+func (p nonRecyclingErrorProcessor) Close() error { return nil }
+
+// TestProcessorSinkEmitDoesNotRecycleOnError is the regression guard for the
+// bt0 data race. The line.Processor contract transfers buffer ownership to the
+// processor, which recycles it on every return path (the journal-path processors
+// DirectLineProcessor and AggregateProcessor recycle unconditionally before
+// returning a write error). If
+// processorSink.Emit also recycled the buffer on error, the same buffer would be
+// returned to the shared pool twice; the pool would then hand one object to two
+// Get callers whose concurrent writes race and corrupt data. Emit must therefore
+// leave the buffer untouched on error. RecycleBytesBuffer calls buf.Reset(), so
+// a stray recycle would clear the payload — assert it survives.
+func TestProcessorSinkEmitDoesNotRecycleOnError(t *testing.T) {
+ buf := pool.BytesBuffer.Get().(*bytes.Buffer)
+ buf.Reset()
+ buf.WriteString("payload")
+
+ sinkErr := errors.New("processor stopped")
+ sink := processorSink{processor: nonRecyclingErrorProcessor{err: sinkErr}}
+
+ err := sink.Emit(context.Background(), buf, 1, 100, "journal-id")
+ if !errors.Is(err, sinkErr) {
+ t.Fatalf("Emit error = %v, want %v", err, sinkErr)
+ }
+ if got := buf.String(); got != "payload" {
+ t.Fatalf("processorSink.Emit recycled a buffer it does not own (double-recycle regression): buf=%q", got)
+ }
+
+ pool.RecycleBytesBuffer(buf)
+}
+
+func TestNewReaderFailsWhenJournalctlIsMissing(t *testing.T) {
+ t.Setenv("PATH", t.TempDir())
+
+ reader, err := NewReader(nil, "journal", false, nil)
+ if err == nil {
+ t.Fatal("expected missing journalctl error")
+ }
+ if reader != nil {
+ t.Fatalf("expected nil reader, got %#v", reader)
+ }
+ if !errors.Is(err, ErrJournalctlNotFound) {
+ t.Fatalf("missing journalctl error = %v, want ErrJournalctlNotFound", err)
+ }
+ if !errors.Is(err, exec.ErrNotFound) {
+ t.Fatalf("missing journalctl error = %v, want exec.ErrNotFound", err)
+ }
+}
+
+func TestStartReadsJournalctlOutputWithoutFollowFlags(t *testing.T) {
+ mock := journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ Lines: []string{"alpha", "beta"},
+ },
+ })
+
+ reader, err := NewReader([]string{"-u", "ssh.service"}, "journal-id", false, make(chan string, 1))
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ processor := &captureProcessor{}
+ if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+ processor, regex.NewNoop()); err != nil {
+ t.Fatalf("start reader: %v", err)
+ }
+
+ want := []string{"alpha\n", "beta\n"}
+ if !reflect.DeepEqual(processor.lines, want) {
+ t.Fatalf("unexpected lines: got=%v want=%v", processor.lines, want)
+ }
+
+ args := mock.Args(t)
+ if strings.Contains(args, "-f") || strings.Contains(args, "-n 0") {
+ t.Fatalf("non-follow reader passed follow flags: %q", args)
+ }
+ if strings.TrimSpace(args) != "-u ssh.service" {
+ t.Fatalf("unexpected journalctl args: %q", args)
+ }
+ if reader.Retry() {
+ t.Fatal("non-follow reader should not retry")
+ }
+}
+
+func TestStartFollowReadsLinesInOrder(t *testing.T) {
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ FollowLines: []string{"alpha", "beta", "gamma"},
+ InterLineDelay: 5 * time.Millisecond,
+ },
+ })
+
+ reader, err := NewReader([]string{"-u", "ssh.service"}, "journal-id", true, make(chan string, 8))
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ processor := &flushCountingProcessor{}
+ done := make(chan error, 1)
+ go func() {
+ done <- reader.StartWithProcessorOptimized(ctx, lcontext.LContext{}, processor, regex.NewNoop())
+ }()
+
+ // Poll until all three follow lines have arrived (the reader appends from its
+ // goroutine while following).
+ deadline := time.After(2 * time.Second)
+ var got []string
+ for {
+ got = processor.snapshot()
+ if len(got) >= 3 {
+ break
+ }
+ select {
+ case <-deadline:
+ t.Fatalf("follow reader did not deliver 3 lines: got=%v", got)
+ case <-time.After(2 * time.Millisecond):
+ }
+ }
+
+ cancel()
+ select {
+ case err := <-done:
+ if err != nil && !errors.Is(err, context.Canceled) {
+ t.Fatalf("follow reader returned error after cancel: %v", err)
+ }
+ case <-time.After(500 * time.Millisecond):
+ t.Fatal("follow reader did not stop promptly after cancellation")
+ }
+
+ want := []string{"alpha\n", "beta\n", "gamma\n"}
+ if !reflect.DeepEqual(got[:3], want) {
+ t.Fatalf("unexpected follow lines: got=%v want=%v", got, want)
+ }
+}
+
+func TestStartFollowPassesFollowFlagsAndTerminatesOnCancel(t *testing.T) {
+ mock := journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ Lines: []string{"ready"},
+ },
+ })
+
+ reader, err := NewReader([]string{"-u", "ssh.service"}, "journal-id", true, make(chan string, 1))
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ processor := &flushCountingProcessor{}
+ done := make(chan error, 1)
+ go func() {
+ done <- reader.StartWithProcessorOptimized(ctx, lcontext.LContext{}, processor, regex.NewNoop())
+ }()
+
+ // Wait for the first follow line to arrive before probing the process.
+ deadline := time.After(2 * time.Second)
+ for {
+ got := processor.snapshot()
+ if len(got) >= 1 {
+ if got[0] != "ready\n" {
+ t.Fatalf("unexpected first line: %q", got[0])
+ }
+ break
+ }
+ select {
+ case <-deadline:
+ t.Fatal("follow reader did not deliver the first line")
+ case <-time.After(2 * time.Millisecond):
+ }
+ }
+
+ pid := mockPID(t, mock)
+ if !processExists(pid) {
+ t.Fatalf("fake journalctl process %d does not exist before cancel", pid)
+ }
+
+ started := time.Now()
+ cancel()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatalf("follow reader returned error after cancel: %v", err)
+ }
+ case <-time.After(500 * time.Millisecond):
+ t.Fatal("follow reader did not stop promptly after cancellation")
+ }
+ if elapsed := time.Since(started); elapsed > 500*time.Millisecond {
+ t.Fatalf("follow reader returned too slowly after cancel: %s", elapsed)
+ }
+
+ args := mock.Args(t)
+ if !strings.Contains(args, "-f -n 0") {
+ t.Fatalf("follow reader did not pass follow flags: %q", args)
+ }
+ if !mock.Terminated(t) {
+ t.Fatal("journalctl did not observe SIGTERM")
+ }
+ if processExists(pid) {
+ t.Fatalf("fake journalctl process %d still exists after reader returned", pid)
+ }
+ if !reader.Retry() {
+ t.Fatal("follow reader should retry")
+ }
+}
+
+func TestStartSurfacesStderrAsServerMessages(t *testing.T) {
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ Lines: []string{"alpha"},
+ Stderr: []string{"journal warning"},
+ },
+ })
+
+ serverMessages := make(chan string, 1)
+ reader, err := NewReader(nil, "journal-id", false, serverMessages)
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+ &captureProcessor{}, regex.NewNoop()); err != nil {
+ t.Fatalf("start reader: %v", err)
+ }
+
+ select {
+ case message := <-serverMessages:
+ if message != "journalctl stderr: journal warning\n" {
+ t.Fatalf("unexpected server message: %q", message)
+ }
+ default:
+ t.Fatal("expected stderr server message")
+ }
+}
+
+func TestStartReturnsExitErrorAndForwardsStderrOnNonZeroExit(t *testing.T) {
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ Lines: []string{"before failure"},
+ Stderr: []string{"boom"},
+ ExitCode: 17,
+ },
+ })
+
+ serverMessages := make(chan string, 2)
+ reader, err := NewReader(nil, "journal-id", false, serverMessages)
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ processor := &captureProcessor{}
+ err = reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+ processor, regex.NewNoop())
+ if err == nil {
+ t.Fatal("expected non-zero journalctl exit error")
+ }
+ if !strings.Contains(err.Error(), "journalctl failed") {
+ t.Fatalf("unexpected non-zero exit error: %v", err)
+ }
+ var exitErr *exec.ExitError
+ if !errors.As(err, &exitErr) {
+ t.Fatalf("error = %v, want exec.ExitError", err)
+ }
+ if got := exitErr.ExitCode(); got != 17 {
+ t.Fatalf("exit code = %d, want 17", got)
+ }
+ if reader.Retry() {
+ t.Fatal("non-follow reader should not retry after non-zero exit")
+ }
+
+ if want := []string{"before failure\n"}; !reflect.DeepEqual(processor.lines, want) {
+ t.Fatalf("unexpected lines before failure: got=%v want=%v", processor.lines, want)
+ }
+
+ select {
+ case message := <-serverMessages:
+ if message != "journalctl stderr: boom\n" {
+ t.Fatalf("unexpected server message: %q", message)
+ }
+ default:
+ t.Fatal("expected stderr server message")
+ }
+}
+
+func TestStartReadsLongJournalLine(t *testing.T) {
+ const longLineLength = 70 * 1024
+
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ LongLineLength: longLineLength,
+ },
+ })
+
+ reader, err := NewReader(nil, "journal-id", false, nil)
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ processor := &captureProcessor{}
+ if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+ processor, regex.NewNoop()); err != nil {
+ t.Fatalf("start reader: %v", err)
+ }
+
+ want := []string{strings.Repeat("x", longLineLength) + "\n"}
+ if !reflect.DeepEqual(processor.lines, want) {
+ gotLen := 0
+ if len(processor.lines) > 0 {
+ gotLen = len(processor.lines[0])
+ }
+ t.Fatalf("unexpected long line: got len=%d want len=%d", gotLen, len(want[0]))
+ }
+}
+
+func TestStartDropsPartialLineAtShutdown(t *testing.T) {
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ PartialLine: "unterminated",
+ },
+ })
+
+ reader, err := NewReader(nil, "journal-id", false, nil)
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ processor := &captureProcessor{}
+ if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+ processor, regex.NewNoop()); err != nil {
+ t.Fatalf("start reader: %v", err)
+ }
+ if len(processor.lines) != 0 {
+ t.Fatalf("partial line without trailing newline was emitted: %v", processor.lines)
+ }
+}
+
+func TestStartPreservesUTF8Lines(t *testing.T) {
+ want := []string{"żółć 🚀\n", "東京 café\n"}
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ Lines: []string{strings.TrimSuffix(want[0], "\n"), strings.TrimSuffix(want[1], "\n")},
+ },
+ })
+
+ reader, err := NewReader(nil, "journal-id", false, nil)
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ processor := &captureProcessor{}
+ if err := reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+ processor, regex.NewNoop()); err != nil {
+ t.Fatalf("start reader: %v", err)
+ }
+
+ if !reflect.DeepEqual(processor.lines, want) {
+ t.Fatalf("unexpected UTF-8 lines: got=%v want=%v", processor.lines, want)
+ }
+}
+
+func TestConcurrentReadersDifferentUnitsDoNotInterfere(t *testing.T) {
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Units: map[string]journaltest.Invocation{
+ "alpha.service": {
+ Lines: []string{"alpha-1", "alpha-2"},
+ },
+ "beta.service": {
+ Lines: []string{"beta-1", "beta-2"},
+ },
+ },
+ })
+
+ type result struct {
+ lines []string
+ err error
+ }
+
+ runReader := func(unit string) <-chan result {
+ resultCh := make(chan result, 1)
+ go func() {
+ reader, err := NewReader([]string{"-u", unit}, unit, false, nil)
+ if err != nil {
+ resultCh <- result{err: err}
+ return
+ }
+ processor := &captureProcessor{}
+ err = reader.StartWithProcessorOptimized(context.Background(), lcontext.LContext{},
+ processor, regex.NewNoop())
+ resultCh <- result{lines: processor.lines, err: err}
+ }()
+ return resultCh
+ }
+
+ alphaCh := runReader("alpha.service")
+ betaCh := runReader("beta.service")
+
+ alpha := <-alphaCh
+ beta := <-betaCh
+
+ if alpha.err != nil {
+ t.Fatalf("alpha reader: %v", alpha.err)
+ }
+ if beta.err != nil {
+ t.Fatalf("beta reader: %v", beta.err)
+ }
+ if want := []string{"alpha-1\n", "alpha-2\n"}; !reflect.DeepEqual(alpha.lines, want) {
+ t.Fatalf("alpha lines: got=%v want=%v", alpha.lines, want)
+ }
+ if want := []string{"beta-1\n", "beta-2\n"}; !reflect.DeepEqual(beta.lines, want) {
+ t.Fatalf("beta lines: got=%v want=%v", beta.lines, want)
+ }
+}
+
+func TestStartWithProcessorOptimizedAppliesRegexAndLocalContext(t *testing.T) {
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ Lines: []string{"before", "match", "context", "skip"},
+ },
+ })
+
+ reader, err := NewReader(nil, "journal-id", false, nil)
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+ re, err := regex.New("match", regex.Default)
+ if err != nil {
+ t.Fatalf("new regex: %v", err)
+ }
+ processor := &captureProcessor{}
+
+ err = reader.StartWithProcessorOptimized(
+ context.Background(),
+ lcontext.LContext{BeforeContext: 1, AfterContext: 1, MaxCount: 1},
+ processor,
+ re,
+ )
+ if err != nil && !errors.Is(err, context.Canceled) {
+ t.Fatalf("start optimized reader: %v", err)
+ }
+
+ want := []string{"before\n", "match\n", "context\n"}
+ if !reflect.DeepEqual(processor.lines, want) {
+ t.Fatalf("unexpected processed lines: got=%v want=%v", processor.lines, want)
+ }
+}
+
+// flushCountingProcessor records processed lines and how many times Flush was
+// invoked, guarded by a mutex because the reader drives it from a goroutine.
+type flushCountingProcessor struct {
+ mu sync.Mutex
+ lines []string
+ flushCount int
+}
+
+func (p *flushCountingProcessor) ProcessLine(lineContent *bytes.Buffer, _ uint64, _ string) error {
+ p.mu.Lock()
+ p.lines = append(p.lines, lineContent.String())
+ p.mu.Unlock()
+ pool.RecycleBytesBuffer(lineContent)
+ return nil
+}
+
+func (p *flushCountingProcessor) Flush() error {
+ p.mu.Lock()
+ p.flushCount++
+ p.mu.Unlock()
+ return nil
+}
+
+func (p *flushCountingProcessor) Close() error { return nil }
+
+func (p *flushCountingProcessor) counts() (lines, flushes int) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return len(p.lines), p.flushCount
+}
+
+// snapshot returns a copy of the processed lines so a test can inspect ordered
+// content while the follow reader keeps appending from its goroutine.
+func (p *flushCountingProcessor) snapshot() []string {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return append([]string(nil), p.lines...)
+}
+
+// TestStartWithProcessorFollowFlushesEachLine is the regression guard for the
+// output batching fix (task 0t0). A follow read blocks in r.run until journalctl
+// is stopped, so a batching output writer would hold live lines in its 64KB
+// buffer and the client would see nothing until the stream ends. The reader
+// must therefore flush the processor after every line while following. If the
+// per-line flush is dropped, flushCount stays at zero during the follow and
+// this test fails.
+func TestStartWithProcessorFollowFlushesEachLine(t *testing.T) {
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ FollowLines: []string{"one", "two", "three"},
+ InterLineDelay: 5 * time.Millisecond,
+ },
+ })
+
+ reader, err := NewReader([]string{"-u", "ssh.service"}, "journal-id", true, make(chan string, 8))
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+
+ processor := &flushCountingProcessor{}
+ done := make(chan error, 1)
+ go func() {
+ done <- reader.StartWithProcessorOptimized(ctx, lcontext.LContext{}, processor, regex.NewNoop())
+ }()
+
+ // While the follow is still live, each of the three lines must trigger a
+ // flush. Poll until at least three flushes are observed.
+ deadline := time.After(2 * time.Second)
+ for {
+ lines, flushes := processor.counts()