v0.27.2 minor update: long lines, switches, tls

internal/source/file/file.go

@@ -40,6 +40,12 @@ type FileSource struct {
 	// Configuration
 	config *config.FileSourceOptions
 
+	// lw's stdout and stderr when they are files: a console sink writes there,
+	// so following them would feed lw its own output. ownSkipped is per path
+	// already reported, touched only by monitorLoop.
+	own        []os.FileInfo
+	ownSkipped map[string]bool
+
 	// Application
 	subscribers []chan core.LogEntry
 	watchers    map[string]*fileWatcher
@@ -105,8 +111,14 @@ func NewFileSourcePlugin(
 		config:      opts,
 		subscribers: make([]chan core.LogEntry, 0),
 		watchers:    make(map[string]*fileWatcher),
+		ownSkipped:  make(map[string]bool),
 		logger:      logger,
 	}
+	for _, f := range []*os.File{os.Stdout, os.Stderr} {
+		if info, err := f.Stat(); err == nil && info.Mode().IsRegular() {
+			fs.own = append(fs.own, info)
+		}
+	}
 	fs.lastEntryTime.Store(time.Time{})
 
 	fs.session = proxy.CreateSession(
@@ -388,15 +400,39 @@ func (fs *FileSource) scanFile() ([]string, error) {
 			continue
 		}
 
-		name := entry.Name()
-		if re.MatchString(name) {
-			files = append(files, filepath.Join(fs.config.Directory, name))
+		path := filepath.Join(fs.config.Directory, entry.Name())
+		if re.MatchString(entry.Name()) && !fs.isOwn(path) {
+			files = append(files, path)
 		}
 	}
 
 	return files, nil
 }
 
+// isOwn reports whether path is lw's own stdout or stderr, warning once per path
+func (fs *FileSource) isOwn(path string) bool {
+	if len(fs.own) == 0 {
+		return false
+	}
+	info, err := os.Stat(path)
+	if err != nil {
+		return false
+	}
+	for _, own := range fs.own {
+		if os.SameFile(info, own) {
+			if !fs.ownSkipped[path] {
+				fs.ownSkipped[path] = true
+				fs.logger.Warn("msg", "Not following lw's own output",
+					"component", "file_source",
+					"instance_id", fs.id,
+					"path", path)
+			}
+			return true
+		}
+	}
+	return false
+}
+
 // globToRegex converts a simple glob pattern to a regular expression.
 func globToRegex(glob string) string {
 	regex := regexp.QuoteMeta(glob)

internal/source/file/file_test.go

@@ -4,14 +4,41 @@ import (
 	"context"
 	"os"
 	"path/filepath"
+	"slices"
+	"strings"
 	"syscall"
 	"testing"
 	"time"
 
+	"github.com/lixenwraith/logwisp/internal/config"
 	"github.com/lixenwraith/logwisp/internal/core"
 	"github.com/lixenwraith/logwisp/internal/testutil"
+
+	"github.com/lixenwraith/log"
 )
 
+// A file source never follows the file lw's stdout or stderr is: `lw > out.log`
+// in the directory it follows would read its own output back, endlessly
+func TestFileSourceSkipsItsOwnOutput(t *testing.T) {
+	dir := t.TempDir()
+	testutil.WriteFile(t, filepath.Join(dir, "app.log"), "")
+	testutil.WriteFile(t, filepath.Join(dir, "out.log"), "")
+	own, err := os.Stat(filepath.Join(dir, "out.log"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	source := &FileSource{
+		config:     &config.FileSourceOptions{Directory: dir, Pattern: "*.log"},
+		own:        []os.FileInfo{own},
+		ownSkipped: map[string]bool{},
+		logger:     log.NewLogger(),
+	}
+	files, err := source.scanFile()
+	if err != nil || len(files) != 1 || filepath.Base(files[0]) != "app.log" {
+		t.Fatalf("scanFile = %v, %v; want app.log alone", files, err)
+	}
+}
+
 func TestStoppedWatcherReturnsNormally(t *testing.T) {
 	path := filepath.Join(t.TempDir(), "session.jsonl")
 	testutil.WriteFile(t, path, "")
@@ -45,6 +72,25 @@ func TestRemoveWatcherPreservesReplacement(t *testing.T) {
 	}
 }
 
+// A line longer than core.MaxLogEntryBytes continues in the next entry, and the
+// lines around it are read once: the watcher used to stop at it for good
+func TestLongLineContinuesInTheNextEntry(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "app.log")
+	long := strings.Repeat("x", core.MaxLogEntryBytes+10)
+	testutil.WriteFile(t, path, "first\n"+long+"\nafter\n")
+	var got []string
+	w := newFileWatcher(path, true, true, func(e core.LogEntry) { got = append(got, e.Message) }, log.NewLogger())
+	for range 2 {
+		if err := w.checkFile(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	want := []string{"first", long[:core.MaxLogEntryBytes], long[core.MaxLogEntryBytes:], "after"}
+	if !slices.Equal(got, want) {
+		t.Fatalf("read %d entries, want %d: first, the long line in two, after", len(got), len(want))
+	}
+}
+
 // A rotated file reappears under its archive name with the same inode. Its
 // replacement watcher resumes where the original stopped, so a `from = "start"`
 // source does not re-emit every record the file already delivered.

internal/source/file/file_watcher.go

@@ -251,6 +251,7 @@ func (w *fileWatcher) checkFile() error {
 
 		scanner := bufio.NewScanner(file)
 		scanner.Buffer(make([]byte, 0, 64*1024), core.MaxLogEntryBytes)
+		scanner.Split(scanLinesCapped)
 
 		for scanner.Scan() {
 			line := scanner.Text()
@@ -311,6 +312,17 @@ func (w *fileWatcher) checkFile() error {
 
 // initPosition records the file's metadata and, unless the watcher was created
 // to read from the start, sets the initial read position to the end
+// scanLinesCapped is bufio.ScanLines, except that a line longer than
+// core.MaxLogEntryBytes continues in the next one, as on the console source:
+// the scanner would otherwise stop at it and every poll re-read the lines before.
+func scanLinesCapped(data []byte, atEOF bool) (int, []byte, error) {
+	advance, token, err := bufio.ScanLines(data, atEOF)
+	if advance == 0 && token == nil && err == nil && len(data) >= core.MaxLogEntryBytes {
+		return core.MaxLogEntryBytes, data[:core.MaxLogEntryBytes], nil
+	}
+	return advance, token, err
+}
+
 func (w *fileWatcher) initPosition() error {
 	file, err := os.Open(w.directory)
 	if err != nil {