Fix filesource provider with kubernetes symlink (#11050) (#11151)

* 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. (cherry picked from commit 0d50584) Co-authored-by: Blake Rouse <blake.rouse@elastic.co>

Author: mergify[bot]

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")
+}