diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index b4ca82534..b9baf4b65 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -21,6 +21,15 @@ go get github.com/shirou/gopsutil@latest # updates the version requirement make vendor # updates the vendored code ``` +### Pointing to a fork + +This example points go-tail to our fork on a specific branch: + +```sh +go mod edit -replace github.com/papertrail/go-tail=github.com/pganalyze/go-tail@deadlock-on-rotated-file +make vendor +``` + ### Compiling and running tests To compile the collector and helper binaries: diff --git a/go.mod b/go.mod index 2011e21ce..779fa1a59 100644 --- a/go.mod +++ b/go.mod @@ -115,3 +115,5 @@ require ( ) go 1.26 + +replace github.com/papertrail/go-tail => github.com/pganalyze/go-tail v0.0.0-20260708205228-8fab800841f3 diff --git a/go.sum b/go.sum index 1bccd28a6..e6002ddf7 100644 --- a/go.sum +++ b/go.sum @@ -232,8 +232,8 @@ github.com/microsoft/go-mssqldb v1.8.0 h1:7cyZ/AT7ycDsEoWPIXibd+aVKFtteUNhDGf3ao github.com/microsoft/go-mssqldb v1.8.0/go.mod h1:6znkekS3T2vp0waiMhen4GPU1BiAsrP+iXHcE7a7rFo= github.com/ogier/pflag v0.0.0-20160129220114-45c278ab3607 h1:db+rES1EpSjP45xOU3hgS41oawQiZzqfnl6dUgBdFjY= github.com/ogier/pflag v0.0.0-20160129220114-45c278ab3607/go.mod h1:zkFki7tvTa0tafRvTBIZTvzYyAu6kQhPZFnshFFPE+g= -github.com/papertrail/go-tail v0.0.0-20180509224916-973c153b0431 h1:i1egM7gz4bPxLCIwBJOkpk6TqHpjTnL4dE1xdN/4dcs= -github.com/papertrail/go-tail v0.0.0-20180509224916-973c153b0431/go.mod h1:dMID0RaS2a5rhpOjC4RsAKitU6WGgkFBZnPVffL69b8= +github.com/pganalyze/go-tail v0.0.0-20260708205228-8fab800841f3 h1:B/hNceRye79VJbKTHfolrqWeNNxEGWNs6M3hL16mFFY= +github.com/pganalyze/go-tail v0.0.0-20260708205228-8fab800841f3/go.mod h1:aUf8RteFEultHEEQ5zPUnFIs/hLoI2z8vJHmn0rcuEk= github.com/pganalyze/pg_query_go/v6 v6.2.2 h1:O0L6zMC226R82RF3X5n0Ki6HjytDsoAzuzp4ATVAHNo= github.com/pganalyze/pg_query_go/v6 v6.2.2/go.mod h1:Cn6+j4870kJz3iYNsb0VsNG04vpSWgEvBwc590J4qD0= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ= diff --git a/input/system/selfhosted/logs.go b/input/system/selfhosted/logs.go index c6fa36805..196a42415 100644 --- a/input/system/selfhosted/logs.go +++ b/input/system/selfhosted/logs.go @@ -235,8 +235,16 @@ func tailFile(ctx context.Context, path string, out chan<- SelfHostedLogStreamIt TailLoop: for { select { - case line := <-t.Lines(): - out <- SelfHostedLogStreamItem{Line: line.String()} + case line, ok := <-t.Lines(): + if !ok { + break TailLoop + } + select { + case out <- SelfHostedLogStreamItem{Line: line.String()}: + case <-ctx.Done(): + prefixedLogger.PrintVerbose("Stopping log tail for %s (stop requested)", path) + break TailLoop + } case <-ctx.Done(): prefixedLogger.PrintVerbose("Stopping log tail for %s (stop requested)", path) break TailLoop @@ -329,6 +337,12 @@ func setupLogLocationTail(ctx context.Context, logLocation string, out chan<- Se return fmt.Errorf("fsnotify new: %s", err) } + err = watcher.Add(logLocation) + if err != nil { + watcher.Close() + return fmt.Errorf("fsnotify add \"%s\": %s", logLocation, err) + } + go func() { defer watcher.Close() for { @@ -358,7 +372,7 @@ func setupLogLocationTail(ctx context.Context, logLocation string, out chan<- Se } } } - if event.Op&fsnotify.Remove == fsnotify.Remove || event.Op&fsnotify.Rename == fsnotify.Rename || event.Op&fsnotify.Chmod == fsnotify.Chmod { + if event.Op&fsnotify.Remove == fsnotify.Remove || event.Op&fsnotify.Rename == fsnotify.Rename { tailCancel, ok := openFiles[event.Name] if ok { tailCancel() @@ -371,8 +385,6 @@ func setupLogLocationTail(ctx context.Context, logLocation string, out chan<- Se case <-ctx.Done(): prefixedLogger.PrintVerbose("Log file fsnotify watcher received stop signal") for fileName, tailCancel := range openFiles { - // TODO: This cancel might actually not be necessary since we are - // already canceling the parent context? tailCancel() delete(openFiles, fileName) } @@ -382,11 +394,6 @@ func setupLogLocationTail(ctx context.Context, logLocation string, out chan<- Se } }() - err = watcher.Add(logLocation) - if err != nil { - return fmt.Errorf("fsnotify add \"%s\": %s", logLocation, err) - } - return nil } diff --git a/input/system/selfhosted/logs_test.go b/input/system/selfhosted/logs_test.go new file mode 100644 index 000000000..577b36512 --- /dev/null +++ b/input/system/selfhosted/logs_test.go @@ -0,0 +1,163 @@ +package selfhosted + +import ( + "context" + "io" + "log" + "os" + "path/filepath" + "testing" + "time" + + "github.com/papertrail/go-tail/follower" + "github.com/pganalyze/collector/util" +) + +// Verifies that tailFile reads new lines appended to a file and outputs them. +func TestTailFile_BasicLineReading(t *testing.T) { + dir := t.TempDir() + logPath := filepath.Join(dir, "test.log") + + if err := os.WriteFile(logPath, nil, 0644); err != nil { + t.Fatalf("create file: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + linesCh := make(chan SelfHostedLogStreamItem, 10) + if err := tailFile(ctx, logPath, linesCh, testLogger()); err != nil { + t.Fatalf("tailFile: %v", err) + } + + time.Sleep(100 * time.Millisecond) + if err := os.WriteFile(logPath, []byte("hello tail\n"), 0644); err != nil { + t.Fatalf("append line: %v", err) + } + + lines := drainLines(linesCh, 1*time.Second) + if len(lines) == 0 { + t.Error("expected at least one line, got none") + } +} + +// Verifies that when a log file is renamed (simulating logrotate), the follower detects the +// change and reads from the new file. +func TestTailFile_LogRotation(t *testing.T) { + dir := t.TempDir() + logPath := filepath.Join(dir, "postgresql.log") + + if err := os.WriteFile(logPath, nil, 0644); err != nil { + t.Fatalf("create file: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + linesCh := make(chan SelfHostedLogStreamItem, 100) + if err := tailFile(ctx, logPath, linesCh, testLogger()); err != nil { + t.Fatalf("tailFile: %v", err) + } + time.Sleep(200 * time.Millisecond) + + // Simulate logrotate: rename + create new file + if err := os.Rename(logPath, logPath+".1"); err != nil { + t.Fatalf("rename: %v", err) + } + if err := os.WriteFile(logPath, []byte("after rotation\n"), 0644); err != nil { + t.Fatalf("create new: %v", err) + } + + lines := drainLines(linesCh, 2*time.Second) + found := false + for _, line := range lines { + if line == "after rotation" { + found = true + break + } + } + if !found { + t.Errorf("expected 'after rotation' after log rotation, got: %v", lines) + } +} + +// Verifies that Close() does not block when the follower's reader is stuck waiting for a newline +// (e.g. file has content without a trailing newline and was rotated). +func TestFollower_Close_NoDeadlock_BlockedRead(t *testing.T) { + dir := t.TempDir() + logPath := filepath.Join(dir, "test.log") + + // No trailing newline, so reader blocks on ReadBytes('\n') + if err := os.WriteFile(logPath, []byte("no newline at end"), 0644); err != nil { + t.Fatalf("create file: %v", err) + } + + f, err := follower.New(logPath, follower.Config{Whence: io.SeekEnd, Offset: 0, Reopen: true}) + if err != nil { + t.Fatalf("follower.New: %v", err) + } + + // Rotate the file so the follower holds an fd to a now-deleted file + if err := os.Rename(logPath, logPath+".1"); err != nil { + t.Fatalf("rename: %v", err) + } + + // Close must return promptly; a deadlock here means the fd leak is present + done := make(chan struct{}) + go func() { f.Close(); close(done) }() + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("Close() blocked") + } +} + +// Verifies that Close() returns when the follower's sendLine goroutine is blocked on the unbuffered +// lines channel. Closing the underlying file unblocks the reader, which lets the goroutine exit. +func TestFollower_Close_NoDeadlock_SendLineBlocks(t *testing.T) { + dir := t.TempDir() + logPath := filepath.Join(dir, "test.log") + + // A complete line will cause sendLine to block on the unbuffered channel + if err := os.WriteFile(logPath, []byte("line to trigger sendLine\n"), 0644); err != nil { + t.Fatalf("create file: %v", err) + } + + f, err := follower.New(logPath, follower.Config{Whence: io.SeekStart, Offset: 0, Reopen: true}) + if err != nil { + t.Fatalf("follower.New: %v", err) + } + + // Intentionally don't read from f.Lines() so sendLine blocks + done := make(chan struct{}) + go func() { f.Close(); close(done) }() + + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("Close() blocked - sendLine deadlock not resolved") + } +} + +func testLogger() *util.Logger { + return &util.Logger{ + Verbose: true, + Destination: log.New(os.Stderr, "", 0), + } +} + +func drainLines(ch chan SelfHostedLogStreamItem, timeout time.Duration) []string { + var lines []string + timer := time.NewTimer(timeout) + defer timer.Stop() + for { + select { + case item, ok := <-ch: + if !ok { + return lines + } + lines = append(lines, item.Line) + case <-timer.C: + return lines + } + } +} diff --git a/vendor/github.com/papertrail/go-tail/follower/follower.go b/vendor/github.com/papertrail/go-tail/follower/follower.go index a73297d16..95cdbb365 100644 --- a/vendor/github.com/papertrail/go-tail/follower/follower.go +++ b/vendor/github.com/papertrail/go-tail/follower/follower.go @@ -3,7 +3,6 @@ package follower import ( "bufio" "bytes" - "fmt" "io" "os" "sync" @@ -17,10 +16,6 @@ const ( peekSize = 1024 ) -var ( - _ = fmt.Print -) - type Line struct { bytes []byte discarded int @@ -62,7 +57,7 @@ func New(filename string, config Config) (*Follower, error) { filename: filename, lines: make(chan Line), config: config, - closeCh: make(chan struct{}), + closeCh: make(chan struct{}, 1), } err := t.reopen() @@ -84,6 +79,9 @@ func (t *Follower) Err() error { } func (t *Follower) Close() { + if t.file != nil { + t.file.Close() + } t.closeCh <- struct{}{} } @@ -244,8 +242,6 @@ func (t *Follower) follow() error { continue } } - - return nil } func (t *Follower) rewatch() error { diff --git a/vendor/modules.txt b/vendor/modules.txt index daf282ab2..dabf2380e 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -384,8 +384,8 @@ github.com/mgutz/ansi # github.com/ogier/pflag v0.0.0-20160129220114-45c278ab3607 ## explicit github.com/ogier/pflag -# github.com/papertrail/go-tail v0.0.0-20180509224916-973c153b0431 -## explicit +# github.com/papertrail/go-tail v0.0.0-20180509224916-973c153b0431 => github.com/pganalyze/go-tail v0.0.0-20260708205228-8fab800841f3 +## explicit; go 1.19 github.com/papertrail/go-tail/follower # github.com/pganalyze/pg_query_go/v6 v6.2.2 ## explicit; go 1.20 @@ -820,3 +820,4 @@ gopkg.in/mcuadros/go-syslog.v2/format gopkg.in/mcuadros/go-syslog.v2/internal/syslogparser gopkg.in/mcuadros/go-syslog.v2/internal/syslogparser/rfc3164 gopkg.in/mcuadros/go-syslog.v2/internal/syslogparser/rfc5424 +# github.com/papertrail/go-tail => github.com/pganalyze/go-tail v0.0.0-20260708205228-8fab800841f3