summaryrefslogtreecommitdiff
path: root/internal/io/journal/reader_test.go
diff options
context:
space:
mode:
Diffstat (limited to 'internal/io/journal/reader_test.go')
-rw-r--r--internal/io/journal/reader_test.go754
1 files changed, 754 insertions, 0 deletions
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()
+ if lines >= 3 && flushes >= 3 {
+ break
+ }
+ select {
+ case <-deadline:
+ t.Fatalf("follow reader did not flush per line: lines=%d flushes=%d", lines, flushes)
+ 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")
+ }
+}
+
+// TestStartWithProcessorNonFollowFlushesOnceAtEnd pins the complementary
+// behavior: a non-follow snapshot read keeps the batching benefit and only
+// flushes once, at the end, rather than per line.
+func TestStartWithProcessorNonFollowFlushesOnceAtEnd(t *testing.T) {
+ journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ Lines: []string{"a", "b", "c", "d"},
+ },
+ })
+
+ reader, err := NewReader(nil, "journal-id", false, nil)
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ processor := &flushCountingProcessor{}
+ if err := reader.StartWithProcessorOptimized(context.Background(),
+ lcontext.LContext{}, processor, regex.NewNoop()); err != nil {
+ t.Fatalf("start optimized reader: %v", err)
+ }
+
+ lines, flushes := processor.counts()
+ if lines != 4 {
+ t.Fatalf("expected 4 processed lines, got %d", lines)
+ }
+ // Only the single terminal flush in runWithProcessor should fire.
+ if flushes != 1 {
+ t.Fatalf("expected exactly 1 flush for a non-follow read, got %d", flushes)
+ }
+}
+
+func TestStartWithProcessorErrorTerminatesJournalctl(t *testing.T) {
+ mock := journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ Lines: []string{"ready"},
+ HoldOpen: true,
+ },
+ })
+
+ reader, err := NewReader(nil, "journal-id", false, nil)
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ processorErr := errors.New("processor stopped")
+ done := make(chan error, 1)
+ go func() {
+ done <- reader.StartWithProcessor(
+ context.Background(),
+ lcontext.LContext{},
+ errorProcessor{err: processorErr},
+ regex.NewNoop(),
+ )
+ }()
+
+ select {
+ case err := <-done:
+ if !errors.Is(err, processorErr) {
+ t.Fatalf("unexpected reader error: %v", err)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("reader hung after processor error")
+ }
+
+ if !mock.Terminated(t) {
+ t.Fatal("journalctl did not observe SIGTERM")
+ }
+}
+
+func TestStartWithProcessorErrorKillsTermIgnoringJournalctl(t *testing.T) {
+ mock := journaltest.InstallMock(t, journaltest.Scenario{
+ Default: journaltest.Invocation{
+ Lines: []string{"ready"},
+ HoldOpen: true,
+ IgnoreSIGTERM: true,
+ },
+ })
+
+ reader, err := NewReader(nil, "journal-id", false, nil)
+ if err != nil {
+ t.Fatalf("new reader: %v", err)
+ }
+
+ processorErr := errors.New("processor stopped")
+ started := time.Now()
+ done := make(chan error, 1)
+ go func() {
+ done <- reader.StartWithProcessor(
+ context.Background(),
+ lcontext.LContext{},
+ errorProcessor{err: processorErr},
+ regex.NewNoop(),
+ )
+ }()
+
+ select {
+ case err := <-done:
+ if !errors.Is(err, processorErr) {
+ t.Fatalf("unexpected reader error: %v", err)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("reader hung after TERM-ignoring journalctl")
+ }
+
+ if elapsed := time.Since(started); elapsed < processTerminateGrace {
+ t.Fatalf("reader returned before kill grace elapsed: %s", elapsed)
+ }
+ if !mock.Terminated(t) {
+ t.Fatal("journalctl did not observe SIGTERM before kill")
+ }
+ pid := mockPID(t, mock)
+ if processExists(pid) {
+ t.Fatalf("fake journalctl process %d still exists after reader returned", pid)
+ }
+}
+
+func readFileString(t *testing.T, path string) string {
+ t.Helper()
+
+ content, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatalf("read %s: %v", path, err)
+ }
+ return string(content)
+}
+
+func mockPID(t *testing.T, mock *journaltest.Mock) int {
+ t.Helper()
+
+ pid, err := strconv.Atoi(strings.TrimSpace(readFileString(t, mock.PIDFile)))
+ if err != nil {
+ t.Fatalf("parse fake journalctl pid: %v", err)
+ }
+ return pid
+}
+
+func processExists(pid int) bool {
+ err := syscall.Kill(pid, 0)
+ return err == nil || errors.Is(err, syscall.EPERM)
+}
generated by cgit v1.2.3 (git 2.25.1) at 2026-08-07 02:44:15 +0000