Fix filesource provider with kubernetes symlink (#11050)

* Add a unit tests to validate kubernetes symlink behavior works with the filesource provider.

* Fix filesource to work on Linux.

* Remove the extra test.

* Skip test on Windows.

* Add changelog entry.

* Improvement from code review.

Author: blakerouse

changelog/fragments/1762438673-fix-filesource-provider-to-work-with-kubernetes-secret-mounts.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: fix filesource provider to work with kubernetes secret mounts
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

internal/pkg/composable/providers/filesource/filesource.go

@@ -63,16 +63,40 @@ func (c *contextProvider) Run(ctx context.Context, comm corecomp.ContextProvider
 	}
 	defer watcher.Close()
 
-	// invert the mapping to map paths to source names
+	// Build mapping from paths (including intermediate symlinks) to source names
+	// and keep track of the original source paths for reading
+	// For Kubernetes secrets: both "token" and "..data/" will map to the same source,
+	// but we always read from the original "token" path
 	inverted := make(map[string][]string, len(c.cfg.Sources))
+	sourcePaths := make(map[string]string, len(c.cfg.Sources)) // sourceName -> original path
+
 	for sourceName, sourceCfg := range c.cfg.Sources {
+		// Store the original path for this source
+		sourcePaths[sourceName] = sourceCfg.Path
+
+		// Add the direct path
 		sources, ok := inverted[sourceCfg.Path]
 		if !ok {
 			sources = []string{sourceName}
 		} else {
 			sources = append(sources, sourceName)
 		}
 		inverted[sourceCfg.Path] = sources
+
+		// Also add any intermediate symlinks in the resolution chain
+		// This handles Kubernetes secrets where token -> ..data/token -> ..2024_01_01/token
+		intermediates := resolveSymlinkChain(sourceCfg.Path)
+		for _, intermediate := range intermediates {
+			sources, ok := inverted[intermediate]
+			if !ok {
+				sources = []string{sourceName}
+			} else {
+				if !slices.Contains(sources, sourceName) {
+					sources = append(sources, sourceName)
+				}
+			}
+			inverted[intermediate] = sources
+		}
 	}
 
 	// determine the paths to watch (watch is performed on the directories that contain the file)
@@ -97,11 +121,9 @@ func (c *contextProvider) Run(ctx context.Context, comm corecomp.ContextProvider
 	// the updated file changes will not be missed
 	current := make(map[string]interface{}, len(c.cfg.Sources))
 	readAll := func() error {
-		for path, sources := range inverted {
+		for sourceName, path := range sourcePaths {
 			value := c.readContents(path)
-			for _, source := range sources {
-				current[source] = value
-			}
+			current[sourceName] = value
 		}
 		err = comm.Set(current)
 		if err != nil {
@@ -153,12 +175,15 @@ func (c *contextProvider) Run(ctx context.Context, comm corecomp.ContextProvider
 				switch {
 				case e.Op&(fsnotify.Create|fsnotify.Write|fsnotify.Remove) != 0:
 					// file was created, updated, or deleted (update the value)
+					// when an intermediate symlink changes (e.g., ..data), we need to
+					// re-read all sources that depend on it using their original paths
 					changed := false
-					value := c.readContents(path)
-					for _, source := range sources {
-						previous := current[source]
+					for _, sourceName := range sources {
+						sourcePath := sourcePaths[sourceName]
+						value := c.readContents(sourcePath)
+						previous := current[sourceName]
 						if previous != value {
-							current[source] = value
+							current[sourceName] = value
 							changed = true
 						}
 					}
@@ -284,3 +309,64 @@ func drainQueue(e <-chan fsnotify.Event) {
 		}
 	}
 }
+
+// resolveSymlinkChain resolves a path and returns all intermediate symlink paths
+// that should be watched to detect changes. This is critical for Kubernetes secrets
+// where the structure is: token -> ..data/token -> ..2024_01_01/token
+// When Kubernetes updates the secret, it replaces ..data, so we need to watch it.
+func resolveSymlinkChain(path string) []string {
+	var intermediates []string
+	dir := filepath.Dir(path)
+
+	// Check if the file itself is a symlink
+	target, err := os.Readlink(path)
+	if err != nil {
+		// Not a symlink or doesn't exist, nothing to track
+		return intermediates
+	}
+
+	// If it's a relative symlink, resolve it relative to the parent directory
+	if !filepath.IsAbs(target) {
+		target = filepath.Join(dir, target)
+	}
+
+	// Now walk through the target path and find intermediate symlinks
+	// For example, if target is "/var/secrets/..data/token", we want to check if "..data" is a symlink
+	targetDir := filepath.Dir(target)
+
+	// Check each component of the target directory path for symlinks
+	components := strings.Split(filepath.Clean(targetDir), string(filepath.Separator))
+	currentPath := ""
+	if filepath.IsAbs(targetDir) {
+		currentPath = string(filepath.Separator)
+	}
+
+	for _, component := range components {
+		if component == "" {
+			continue
+		}
+		currentPath = filepath.Join(currentPath, component)
+
+		// Check if this component is a symlink
+		_, err = os.Readlink(currentPath)
+		if err == nil {
+			// This is a symlink, add it to our watch list
+			cleanPath := filepath.Clean(currentPath)
+			// Windows paths are case-insensitive
+			if runtime.GOOS == "windows" {
+				cleanPath = strings.ToLower(cleanPath)
+			}
+			intermediates = append(intermediates, cleanPath)
+		}
+	}
+
+	// Also check if the target file itself is a symlink (nested symlinks)
+	_, err = os.Readlink(target)
+	if err == nil {
+		// The target itself is also a symlink, recursively resolve it
+		nestedIntermediates := resolveSymlinkChain(target)
+		intermediates = append(intermediates, nestedIntermediates...)
+	}
+
+	return intermediates
+}

internal/pkg/composable/providers/filesource/filesource_test.go

@@ -181,3 +181,134 @@ func TestContextProvider(t *testing.T) {
 		}
 	}
 }
+
+func TestContextProvider_KubernetesSymlinks(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("Skipping Kubernetes symlink test on Windows, because atomic replacing a symlink using os.Rename doesn't work")
+	}
+
+	const testTimeout = 3 * time.Second
+
+	// Create directory structure that mimics Kubernetes secrets
+	tmpDir := t.TempDir()
+
+	// Create initial timestamped directory with secret content
+	dataDir1 := filepath.Join(tmpDir, "..2024_01_01_12_00")
+	require.NoError(t, os.Mkdir(dataDir1, 0o755))
+
+	value1 := "secret-token-v1"
+	tokenFile1 := filepath.Join(dataDir1, "token")
+	require.NoError(t, os.WriteFile(tokenFile1, []byte(value1), 0o644))
+
+	value2 := "secret-cert-v1"
+	certFile1 := filepath.Join(dataDir1, "ca.crt")
+	require.NoError(t, os.WriteFile(certFile1, []byte(value2), 0o644))
+
+	// Create ..data symlink pointing to the timestamped directory
+	dataSymlink := filepath.Join(tmpDir, "..data")
+	require.NoError(t, os.Symlink(dataDir1, dataSymlink))
+
+	// Create top-level symlinks (what the user actually references)
+	tokenSymlink := filepath.Join(tmpDir, "token")
+	require.NoError(t, os.Symlink(filepath.Join("..data", "token"), tokenSymlink))
+
+	certSymlink := filepath.Join(tmpDir, "ca.crt")
+	require.NoError(t, os.Symlink(filepath.Join("..data", "ca.crt"), certSymlink))
+
+	// Setup logger and provider
+	log, err := logger.New("filesource_test", false)
+	require.NoError(t, err)
+
+	osPath := func(path string) string {
+		return path
+	}
+	if runtime.GOOS == "windows" {
+		osPath = func(path string) string {
+			return strings.ToLower(path)
+		}
+	}
+
+	c, err := config.NewConfigFrom(map[string]interface{}{
+		"sources": map[string]interface{}{
+			"token": map[string]interface{}{
+				"path": osPath(tokenSymlink),
+			},
+			"cert": map[string]interface{}{
+				"path": osPath(certSymlink),
+			},
+		},
+	})
+	require.NoError(t, err)
+
+	builder, _ := composable.Providers.GetContextProvider("filesource")
+	provider, err := builder(log, c, true)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	comm := ctesting.NewContextComm(ctx)
+	setChan := make(chan map[string]interface{})
+	comm.CallOnSet(func(value map[string]interface{}) {
+		t.Logf("Set called with: token=%v, cert=%v", value["token"], value["cert"])
+		setChan <- value
+	})
+
+	go func() {
+		_ = provider.Run(ctx, comm)
+	}()
+
+	// Wait for initial values
+	var current map[string]interface{}
+	select {
+	case current = <-setChan:
+	case <-time.After(testTimeout):
+		require.FailNow(t, "timeout waiting for provider to call Set")
+	}
+
+	require.Equal(t, value1, current["token"], "initial token value should match")
+	require.Equal(t, value2, current["cert"], "initial cert value should match")
+
+	// Simulate Kubernetes secret update:
+	// 1. Create new timestamped directory with updated content
+	dataDir2 := filepath.Join(tmpDir, "..2024_01_01_13_00")
+	require.NoError(t, os.Mkdir(dataDir2, 0o755))
+
+	value1Updated := "secret-token-v2"
+	tokenFile2 := filepath.Join(dataDir2, "token")
+	require.NoError(t, os.WriteFile(tokenFile2, []byte(value1Updated), 0o644))
+
+	value2Updated := "secret-cert-v2"
+	certFile2 := filepath.Join(dataDir2, "ca.crt")
+	require.NoError(t, os.WriteFile(certFile2, []byte(value2Updated), 0o644))
+
+	// 2. Atomically replace ..data symlink (this is what Kubernetes does)
+	// Create temporary symlink, then rename it to replace the old one atomically
+	dataTmpSymlink := filepath.Join(tmpDir, "..data_tmp")
+	require.NoError(t, os.Symlink(dataDir2, dataTmpSymlink))
+	require.NoError(t, os.Rename(dataTmpSymlink, dataSymlink))
+
+	// Note: The top-level symlinks (token, ca.crt) are NOT modified
+	// They still point to ..data/token and ..data/ca.crt
+	// Only the ..data symlink target changed
+
+	// Wait for the provider to detect the update
+	// This should happen because fsnotify should see the ..data symlink change
+	updateDetected := false
+	deadline := time.After(testTimeout)
+	for !updateDetected {
+		select {
+		case updated := <-setChan:
+			// Check if we got the updated values
+			if updated["token"] == value1Updated && updated["cert"] == value2Updated {
+				updateDetected = true
+				t.Log("Successfully detected Kubernetes-style symlink update")
+			} else {
+				t.Logf("Got update but values don't match yet: token=%v, cert=%v", updated["token"], updated["cert"])
+			}
+		case <-deadline:
+			require.FailNow(t, "timeout waiting for provider to detect Kubernetes-style symlink update")
+		}
+	}
+
+	require.True(t, updateDetected, "provider should detect Kubernetes-style symlink updates")
+}