resolve links on start file watcher (#733)

* resolve links on start file watcher

* fix resolve links in watcher

Author: DmitryRomanov

plugin/input/file/watcher.go

@@ -44,10 +44,25 @@ func NewWatcher(
 }
 
 func (w *watcher) start() {
-	for _, pattern := range w.paths.Include {
+	for i, pattern := range w.paths.Include {
 		// /var/lib/docker/containers/**/*-json.log -> /var/lib/docker/containers
 		basePattern, _ := doublestar.SplitPattern(pattern)
-		w.basePaths = append(w.basePaths, basePattern)
+		allLinksResolvedPath, err := resolvePathLinks(basePattern)
+		if err != nil {
+			panic(err)
+		}
+		w.paths.Include[i] = strings.Replace(w.paths.Include[i], basePattern, allLinksResolvedPath, 1)
+		w.basePaths = append(w.basePaths, allLinksResolvedPath)
+	}
+
+	for i, pattern := range w.paths.Exclude {
+		// /var/lib/docker/containers/**/*-json.log -> /var/lib/docker/containers
+		basePattern, _ := doublestar.SplitPattern(pattern)
+		allLinksResolvedPath, err := resolvePathLinks(basePattern)
+		if err != nil {
+			panic(err)
+		}
+		w.paths.Exclude[i] = strings.Replace(w.paths.Exclude[i], basePattern, allLinksResolvedPath, 1)
 	}
 	w.commonPath = commonPathPrefix(w.basePaths)
 
@@ -215,3 +230,60 @@ func (w *watcher) watch() {
 		w.notify(event.Event(), event.Path())
 	}
 }
+
+func resolvePathLinks(basePath string) (string, error) {
+	resolvedPath := basePath
+	components := filepath.SplitList(resolvedPath)
+
+	var finalPath string
+	for _, component := range components {
+		if component == "" {
+			continue
+		}
+
+		finalPath = filepath.Join(finalPath, component)
+
+		info, err := os.Lstat(finalPath)
+		if err != nil {
+			upDir := filepath.Dir(basePath)
+			resolvedPath, err := resolvePathLinks(upDir)
+			return filepath.Join(
+				resolvedPath,
+				filepath.Base(basePath),
+			), err
+		}
+
+		if info.Mode()&os.ModeSymlink != 0 {
+			target, err := os.Readlink(finalPath)
+			if err != nil {
+				return "", err
+			}
+
+			if !filepath.IsAbs(target) {
+				finalPath = filepath.Join(filepath.Dir(finalPath), target)
+			} else {
+				finalPath = target
+			}
+		}
+	}
+
+	getParentDir := func(path string) string {
+		normalizedPath := strings.TrimSuffix(path, string(os.PathSeparator))
+		parentDir := filepath.Dir(normalizedPath)
+		if parentDir == "" || parentDir == string(os.PathSeparator) {
+			return string(os.PathSeparator)
+		}
+		return parentDir
+	}
+
+	upDir := getParentDir(finalPath)
+	if upDir == string(os.PathSeparator) || upDir == filepath.VolumeName(finalPath)+string(os.PathSeparator) {
+		return finalPath, nil
+	} else {
+		resolvedPath, err := resolvePathLinks(upDir)
+		return filepath.Join(
+			resolvedPath,
+			filepath.Base(finalPath),
+		), err
+	}
+}

plugin/input/file/watcher_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -90,7 +91,15 @@ func TestWatcher(t *testing.T) {
 
 // nolint:gocritic
 func TestWatcherPaths(t *testing.T) {
-	dir := t.TempDir()
+	originalDir := t.TempDir()
+
+	dirWithLink := t.TempDir()
+	dir := filepath.Join(dirWithLink, "symlink")
+	err := os.Symlink(originalDir, dir)
+	if err != nil {
+		t.Fatalf("Failed to create symlink: %v", err)
+	}
+
 	shouldCreate := atomic.Int64{}
 	notifyFn := func(_ notify.Event, _ string, _ os.FileInfo) {
 		shouldCreate.Inc()
@@ -124,47 +133,47 @@ func TestWatcherPaths(t *testing.T) {
 		shouldNotify bool
 	}{
 		{
-			filename:     filepath.Join(dir, "nginx-ingress-0/error.log"),
+			filename:     filepath.Join(originalDir, "nginx-ingress-0/error.log"),
 			shouldNotify: true,
 		},
 		{
-			filename:     filepath.Join(dir, "log/errors.log"),
+			filename:     filepath.Join(originalDir, "log/errors.log"),
 			shouldNotify: true,
 		},
 		{
-			filename:     filepath.Join(dir, "log/0/errors.log"),
+			filename:     filepath.Join(originalDir, "log/0/errors.log"),
 			shouldNotify: true,
 		},
 		{
-			filename:     filepath.Join(dir, "log/0/0/errors.log"),
+			filename:     filepath.Join(originalDir, "log/0/0/errors.log"),
 			shouldNotify: true,
 		},
 		{
-			filename:     filepath.Join(dir, "access.log"),
+			filename:     filepath.Join(originalDir, "access.log"),
 			shouldNotify: true,
 		},
 		{
-			filename:     filepath.Join(dir, "sub_access.log"),
+			filename:     filepath.Join(originalDir, "sub_access.log"),
 			shouldNotify: true,
 		},
 		{
-			filename:     filepath.Join(dir, "access1.log"),
+			filename:     filepath.Join(originalDir, "access1.log"),
 			shouldNotify: false,
 		},
 		{
-			filename:     filepath.Join(dir, "nginx/errors.log"),
+			filename:     filepath.Join(originalDir, "nginx/errors.log"),
 			shouldNotify: false,
 		},
 		{
-			filename:     filepath.Join(dir, "log/payments/errors.log"),
+			filename:     filepath.Join(originalDir, "log/payments/errors.log"),
 			shouldNotify: false,
 		},
 		{
-			filename:     filepath.Join(dir, "log/payments/nginx-ingress-0/errors.log"),
+			filename:     filepath.Join(originalDir, "log/payments/nginx-ingress-0/errors.log"),
 			shouldNotify: false,
 		},
 		{
-			filename:     filepath.Join(dir, "nginx-ingress-payments/error.log"),
+			filename:     filepath.Join(originalDir, "nginx-ingress-payments/error.log"),
 			shouldNotify: false,
 		},
 	}
@@ -204,3 +213,25 @@ func TestCommonPathPrefix(t *testing.T) {
 	result := commonPathPrefix(paths)
 	a.Equal("/var", result)
 }
+
+func TestResolvePathLinks(t *testing.T) {
+	a := assert.New(t)
+	originalDir := t.TempDir()
+
+	dirWithLink := t.TempDir()
+	dirLink := filepath.Join(dirWithLink, "symlink")
+	err := os.Symlink(originalDir, dirLink)
+	if err != nil {
+		t.Fatalf("Failed to create symlink: %v", err)
+	}
+	linkSubDir := filepath.Join(dirLink, "dir")
+	os.Mkdir(linkSubDir, os.FileMode(0o755))
+
+	resultDir := filepath.Join(linkSubDir, "dir")
+	result, err := resolvePathLinks(resultDir)
+	a.Equal(
+		strings.Replace(resultDir, dirLink, originalDir, 1),
+		result,
+	)
+	a.Equal(nil, err)
+}