test(tail): Add watch-method behavior matrix for the tailer engine

Add engine-level tests that exercise follow, rename/recreate and
copytruncate rotation, delete-and-recreate, partial lines, symlinks and
offset resume. Each scenario runs under both the native watcher and
polling so coverage is independent of the watch implementation.

Two combinations the current library does not satisfy are recorded as
documented skips: same-length copytruncate under polling, and writes to
a symlink target via the native watcher on macOS.

Author: skartikey

plugins/inputs/tail/tail_behavior_test.go

@@ -0,0 +1,251 @@
+package tail
+
+import (
+	"os"
+	"path/filepath"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/influxdata/telegraf/testutil"
+)
+
+// These tests exercise the tailer engine itself (follow, rotation, truncation,
+// deletion, partial lines, symlinks, offset resume) rather than the parsing
+// glue. Each scenario runs under both the native watcher and polling so the
+// matrix is independent of the underlying watch implementation. They are meant
+// to stay green against the current library and act as an acceptance gate for
+// any replacement of it.
+
+func TestTailFollowAppend(t *testing.T) {
+	runWatchMethods(t, func(t *testing.T, watchMethod string) {
+		f := filepath.Join(t.TempDir(), "test.log")
+		require.NoError(t, os.WriteFile(f, []byte("cpu value=1\n"), 0600))
+
+		plugin := newBehaviorTail(t, watchMethod, "beginning", f)
+		var acc testutil.Accumulator
+		require.NoError(t, plugin.Start(&acc))
+		defer plugin.Stop()
+
+		waitForValues(t, &acc, 1)
+
+		appendLine(t, f, "cpu value=2\n")
+		waitForValues(t, &acc, 1, 2)
+	})
+}
+
+func TestTailRotateRenameRecreate(t *testing.T) {
+	runWatchMethods(t, func(t *testing.T, watchMethod string) {
+		f := filepath.Join(t.TempDir(), "test.log")
+		require.NoError(t, os.WriteFile(f, []byte("cpu value=1\n"), 0600))
+
+		plugin := newBehaviorTail(t, watchMethod, "beginning", f)
+		var acc testutil.Accumulator
+		require.NoError(t, plugin.Start(&acc))
+		defer plugin.Stop()
+
+		waitForValues(t, &acc, 1)
+
+		// Logrotate "create" style: move the active file aside and recreate it.
+		require.NoError(t, os.Rename(f, f+".1"))
+		require.NoError(t, os.WriteFile(f, []byte("cpu value=2\n"), 0600))
+
+		waitForValues(t, &acc, 1, 2)
+	})
+}
+
+func TestTailRotateCopyTruncate(t *testing.T) {
+	runWatchMethods(t, func(t *testing.T, watchMethod string) {
+		if watchMethod == "poll" {
+			// Known gap in the current library: the polling watcher samples the
+			// file size every POLL_DURATION (250ms) and only treats a shrink as
+			// a truncation, so a truncate-and-rewrite back to the same length
+			// within one interval is never detected and the new content is lost.
+			// A replacement should detect truncation reliably under polling.
+			t.Skip("current library misses same-length copytruncate under polling")
+		}
+
+		f := filepath.Join(t.TempDir(), "test.log")
+		require.NoError(t, os.WriteFile(f, []byte("cpu value=1\n"), 0600))
+
+		plugin := newBehaviorTail(t, watchMethod, "beginning", f)
+		var acc testutil.Accumulator
+		require.NoError(t, plugin.Start(&acc))
+		defer plugin.Stop()
+
+		waitForValues(t, &acc, 1)
+
+		// Logrotate "copytruncate" style: truncate the file in place, then write
+		// fresh content from offset zero.
+		require.NoError(t, os.Truncate(f, 0))
+		appendLine(t, f, "cpu value=2\n")
+
+		waitForValues(t, &acc, 1, 2)
+	})
+}
+
+func TestTailDeleteRecreate(t *testing.T) {
+	runWatchMethods(t, func(t *testing.T, watchMethod string) {
+		f := filepath.Join(t.TempDir(), "test.log")
+		require.NoError(t, os.WriteFile(f, []byte("cpu value=1\n"), 0600))
+
+		plugin := newBehaviorTail(t, watchMethod, "beginning", f)
+		var acc testutil.Accumulator
+		require.NoError(t, plugin.Start(&acc))
+		defer plugin.Stop()
+
+		waitForValues(t, &acc, 1)
+
+		// Remove the followed file entirely, then recreate it later.
+		require.NoError(t, os.Remove(f))
+		require.NoError(t, os.WriteFile(f, []byte("cpu value=2\n"), 0600))
+
+		waitForValues(t, &acc, 1, 2)
+	})
+}
+
+func TestTailPartialLine(t *testing.T) {
+	runWatchMethods(t, func(t *testing.T, watchMethod string) {
+		f := filepath.Join(t.TempDir(), "test.log")
+		// Start with a line that has no trailing newline yet.
+		require.NoError(t, os.WriteFile(f, []byte("cpu value="), 0600))
+
+		plugin := newBehaviorTail(t, watchMethod, "beginning", f)
+		var acc testutil.Accumulator
+		require.NoError(t, plugin.Start(&acc))
+		defer plugin.Stop()
+
+		// The line must not be emitted until it is terminated.
+		appendLine(t, f, "1\n")
+		waitForValues(t, &acc, 1)
+	})
+}
+
+func TestTailSymlink(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("symlink creation requires elevated privileges on Windows")
+	}
+
+	runWatchMethods(t, func(t *testing.T, watchMethod string) {
+		if watchMethod == "inotify" && runtime.GOOS == "darwin" {
+			// Known gap in the current library on macOS: the native
+			// (kqueue/FSEvents) watcher registered on the symlink path does not
+			// receive events for appends to the symlink target. A replacement
+			// should follow the resolved target on all platforms.
+			t.Skip("current library misses target writes via the native watcher on macOS")
+		}
+
+		dir := t.TempDir()
+		target := filepath.Join(dir, "real.log")
+		link := filepath.Join(dir, "link.log")
+		require.NoError(t, os.WriteFile(target, []byte("cpu value=1\n"), 0600))
+		require.NoError(t, os.Symlink(target, link))
+
+		plugin := newBehaviorTail(t, watchMethod, "beginning", link)
+		var acc testutil.Accumulator
+		require.NoError(t, plugin.Start(&acc))
+		defer plugin.Stop()
+
+		waitForValues(t, &acc, 1)
+
+		appendLine(t, target, "cpu value=2\n")
+		waitForValues(t, &acc, 1, 2)
+	})
+}
+
+func TestTailResumeOffset(t *testing.T) {
+	runWatchMethods(t, func(t *testing.T, watchMethod string) {
+		f := filepath.Join(t.TempDir(), "test.log")
+		require.NoError(t, os.WriteFile(f, []byte("cpu value=1\ncpu value=2\n"), 0600))
+
+		// First run reads the file and persists its offset on stop.
+		first := newBehaviorTail(t, watchMethod, "saved-or-beginning", f)
+		var firstAcc testutil.Accumulator
+		require.NoError(t, first.Start(&firstAcc))
+		waitForValues(t, &firstAcc, 1, 2)
+		state := first.GetState()
+		first.Stop()
+
+		// New content arrives while no tailer is running.
+		appendLine(t, f, "cpu value=3\n")
+
+		// Second run resumes from the persisted offset and must only see the
+		// new line, never re-reading the already-delivered ones.
+		second := newBehaviorTail(t, watchMethod, "saved-or-beginning", f)
+		require.NoError(t, second.SetState(state))
+		var secondAcc testutil.Accumulator
+		require.NoError(t, second.Start(&secondAcc))
+		defer second.Stop()
+
+		waitForValues(t, &secondAcc, 3)
+		// Reading is sequential, so if 1 or 2 were going to reappear they would
+		// arrive before 3; asserting their absence here needs no extra wait.
+		require.ElementsMatch(t, []float64{3}, tailFieldValues(&secondAcc))
+	})
+}
+
+// runWatchMethods runs fn under each supported watch method so every behavior
+// is verified for both the native watcher and polling.
+func runWatchMethods(t *testing.T, fn func(t *testing.T, watchMethod string)) {
+	t.Helper()
+	methods := []string{"inotify", "poll"}
+	if runtime.GOOS == "windows" {
+		methods = []string{"poll"}
+	}
+	for _, method := range methods {
+		t.Run(method, func(t *testing.T) {
+			fn(t, method)
+		})
+	}
+}
+
+func newBehaviorTail(t *testing.T, watchMethod, initialReadOffset string, files ...string) *Tail {
+	t.Helper()
+	plugin := newTestTail()
+	plugin.Log = testutil.Logger{}
+	plugin.WatchMethod = watchMethod
+	plugin.InitialReadOffset = initialReadOffset
+	plugin.Files = files
+	plugin.SetParserFunc(newInfluxParser)
+	require.NoError(t, plugin.Init())
+	return plugin
+}
+
+func appendLine(t *testing.T, path, line string) {
+	t.Helper()
+	f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0600)
+	require.NoError(t, err)
+	_, err = f.WriteString(line)
+	require.NoError(t, err)
+	require.NoError(t, f.Close())
+}
+
+func tailFieldValues(acc *testutil.Accumulator) []float64 {
+	var out []float64
+	for _, m := range acc.GetTelegrafMetrics() {
+		if v, ok := m.Fields()["value"]; ok {
+			if f, ok := v.(float64); ok {
+				out = append(out, f)
+			}
+		}
+	}
+	return out
+}
+
+func waitForValues(t *testing.T, acc *testutil.Accumulator, want ...float64) {
+	t.Helper()
+	require.Eventuallyf(t, func() bool {
+		got := make(map[float64]bool)
+		for _, v := range tailFieldValues(acc) {
+			got[v] = true
+		}
+		for _, w := range want {
+			if !got[w] {
+				return false
+			}
+		}
+		return true
+	}, 5*time.Second, 50*time.Millisecond, "did not observe values %v, got %v", want, tailFieldValues(acc))
+}