Add filesource provider

changelog/fragments/1737666699-Add-filesource-provider.yaml

@@ -0,0 +1,34 @@
+# 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: feature
+
+# Change summary; a 80ish characters long description of the change.
+summary: Add filesource provider
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# NOTE: This field will be rendered only for breaking-change and known-issue kinds at the moment.
+description: |
+  filesource provider watches for changes of the files and updates the values of the variables
+  when the content of the file changes.
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component:
+
+# PR URL; optional; the PR number 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/elastic/elastic-agent/pull/6587
+
+# 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/elastic/elastic-agent/issues/6362

internal/pkg/agent/cmd/include.go

@@ -9,6 +9,7 @@ import (
 	_ "github.com/elastic/elastic-agent/internal/pkg/composable/providers/agent"
 	_ "github.com/elastic/elastic-agent/internal/pkg/composable/providers/docker"
 	_ "github.com/elastic/elastic-agent/internal/pkg/composable/providers/env"
+	_ "github.com/elastic/elastic-agent/internal/pkg/composable/providers/filesource"
 	_ "github.com/elastic/elastic-agent/internal/pkg/composable/providers/host"
 	_ "github.com/elastic/elastic-agent/internal/pkg/composable/providers/kubernetes"
 	_ "github.com/elastic/elastic-agent/internal/pkg/composable/providers/kubernetesleaderelection"

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

@@ -0,0 +1,177 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package filesource
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"slices"
+
+	"github.com/fsnotify/fsnotify"
+
+	"github.com/elastic/elastic-agent/internal/pkg/composable"
+	"github.com/elastic/elastic-agent/internal/pkg/config"
+	corecomp "github.com/elastic/elastic-agent/internal/pkg/core/composable"
+	"github.com/elastic/elastic-agent/pkg/core/logger"
+)
+
+func init() {
+	// filesource provider reads and watches for changes on files that are defined in the provider configuration.
+	//
+	// To be notified when a file is change the provider will watch the parent directory of the file so if the file
+	// is replaced that it will read the new contents. If a file doesn't exist or the provider is unable to read
+	// the file then it will report the value as an empty string.
+	//
+	// If the provided path happens to be a directory then it just report the value as an empty string.
+	composable.Providers.MustAddContextProvider("filesource", ContextProviderBuilder)
+}
+
+type fileSourceConfig struct {
+	Type string `config:"type"`
+	Path string `config:"path"`
+}
+
+type contextProvider struct {
+	logger *logger.Logger
+
+	sources map[string]fileSourceConfig
+}
+
+// Run runs the filesource context provider.
+func (c *contextProvider) Run(ctx context.Context, comm corecomp.ContextProviderComm) error {
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return fmt.Errorf("failed to create watcher: %w", err)
+	}
+	defer watcher.Close()
+
+	// invert the mapping to map paths to source names
+	inverted := make(map[string][]string, len(c.sources))
+	for sourceName, sourceCfg := range c.sources {
+		sources, ok := inverted[sourceCfg.Path]
+		if !ok {
+			sources = []string{sourceName}
+		} else {
+			sources = append(sources, sourceName)
+		}
+		inverted[sourceCfg.Path] = sources
+	}
+
+	// determine the paths to watch (watch is performed on the directories that contain the file)
+	//
+	// you cannot register the same directory multiple times so this ensures its only registered once
+	paths := make([]string, 0, len(c.sources))
+	for _, cfg := range c.sources {
+		parent := filepath.Dir(cfg.Path)
+		if !slices.Contains(paths, parent) {
+			paths = append(paths, parent)
+		}
+	}
+	for _, path := range paths {
+		err = watcher.Add(path)
+		if err != nil {
+			return fmt.Errorf("failed to watch path %q: %w", path, err)
+		}
+	}
+
+	// read the initial values after the watch has started
+	// this ensures that if the value changed between this code and the loop below
+	// the updated file changes will not be missed
+	current := make(map[string]interface{}, len(c.sources))
+	for path, sources := range inverted {
+		value := c.readContents(path)
+		for _, source := range sources {
+			current[source] = value
+		}
+	}
+	err = comm.Set(current)
+	if err != nil {
+		return fmt.Errorf("failed to set current context: %w", err)
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case err, ok := <-watcher.Errors:
+			if !ok {
+				// watcher was closed
+				return nil
+			}
+			c.logger.Errorf("file watcher errored: %s", err)
+		case e, ok := <-watcher.Events:
+			if !ok { // Channel was closed (i.e. Watcher.Close() was called).
+				// watcher was closed
+				return nil
+			}
+
+			path := filepath.Clean(e.Name)
+			sources, ok := inverted[path]
+			if !ok {
+				// watching the directory, it can contain files that we are not watching
+				// ignore these events unless we are actively watching this file
+				continue
+			}
+
+			switch {
+			case e.Op&(fsnotify.Create|fsnotify.Write|fsnotify.Remove) != 0:
+				// file was created, updated, or deleted (update the value)
+				value := c.readContents(path)
+				for _, source := range sources {
+					current[source] = value
+				}
+				err = comm.Set(current)
+				if err != nil {
+					return fmt.Errorf("failed to set current context from notify event: %w", err)
+				}
+			}
+		}
+	}
+}
+
+func (c *contextProvider) readContents(path string) string {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		c.logger.Errorf("failed to read file %q: %s", path, err)
+		return ""
+	}
+	return string(data)
+}
+
+// ContextProviderBuilder builds the context provider.
+func ContextProviderBuilder(log *logger.Logger, c *config.Config, _ bool) (corecomp.ContextProvider, error) {
+	p := &contextProvider{
+		logger: log,
+	}
+	if c != nil {
+		err := c.UnpackTo(&p.sources)
+		if err != nil {
+			return nil, fmt.Errorf("failed to unpack config: %w", err)
+		}
+	}
+	for sourceName, sourceCfg := range p.sources {
+		if sourceCfg.Type != "" && sourceCfg.Type != "raw" {
+			return nil, fmt.Errorf("%q defined an unsupported type %q", sourceName, sourceCfg.Type)
+		}
+		if sourceCfg.Path == "" {
+			return nil, fmt.Errorf("%q is missing a defined path", sourceName)
+		}
+		// only use an absolute path (convert from relative)
+		if !filepath.IsAbs(sourceCfg.Path) {
+			path, err := filepath.Abs(sourceCfg.Path)
+			if err != nil {
+				return nil, fmt.Errorf("%q failed to determine absolute path for %q: %w", sourceName, sourceCfg.Path, err)
+			}
+			sourceCfg.Path = path
+		}
+		path := filepath.Dir(sourceCfg.Path)
+		if path == "" || path == "." {
+			return nil, fmt.Errorf("%q has a path %q that is invalid", sourceName, sourceCfg.Path)
+		}
+	}
+	return p, nil
+}

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

@@ -0,0 +1,163 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package filesource
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	ctesting "github.com/elastic/elastic-agent/internal/pkg/composable/testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/elastic-agent/internal/pkg/composable"
+	"github.com/elastic/elastic-agent/internal/pkg/config"
+	"github.com/elastic/elastic-agent/pkg/core/logger"
+)
+
+func TestContextProvider_Config(t *testing.T) {
+	scenarios := []struct {
+		Name   string
+		Config *config.Config
+		Err    error
+	}{
+		{
+			Name: "no path",
+			Config: config.MustNewConfigFrom(map[string]interface{}{
+				"one": map[string]interface{}{},
+			}),
+			Err: errors.New(`"one" is missing a defined path`),
+		},
+		{
+			Name: "invalid type",
+			Config: config.MustNewConfigFrom(map[string]interface{}{
+				"one": map[string]interface{}{
+					"type": "json",
+					"path": "/etc/agent/content",
+				},
+			}),
+			Err: errors.New(`"one" defined an unsupported type "json"`),
+		},
+		// other errors in the config validation are hard to validate in a test
+		// they are just very defensive
+		{
+			Name: "valid path",
+			Config: config.MustNewConfigFrom(map[string]interface{}{
+				"one": map[string]interface{}{
+					"path": "/etc/agent/content1",
+				},
+				"two": map[string]interface{}{
+					"path": "/etc/agent/content2",
+				},
+			}),
+		},
+	}
+	for _, s := range scenarios {
+		t.Run(s.Name, func(t *testing.T) {
+			log, err := logger.New("filesource_test", false)
+			require.NoError(t, err)
+
+			builder, _ := composable.Providers.GetContextProvider("filesource")
+			_, err = builder(log, s.Config, true)
+			if s.Err != nil {
+				require.Equal(t, s.Err, err)
+			} else {
+				require.NoError(t, err)
+			}
+		})
+	}
+}
+
+func TestContextProvider(t *testing.T) {
+	const testTimeout = 3 * time.Second
+
+	tmpDir := t.TempDir()
+	value1 := "value1"
+	file1 := filepath.Join(tmpDir, "value1_path")
+	require.NoError(t, os.WriteFile(file1, []byte(value1), 0o644))
+	value2 := "value2"
+	file2 := filepath.Join(tmpDir, "value2_path")
+	require.NoError(t, os.WriteFile(file2, []byte(value2), 0o644))
+
+	log, err := logger.New("filesource_test", false)
+	require.NoError(t, err)
+
+	c, err := config.NewConfigFrom(map[string]interface{}{
+		"one": map[string]interface{}{
+			"path": file1,
+		},
+		"two": map[string]interface{}{
+			"path": file2,
+		},
+	})
+	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{}) {
+		// Forward Set's input to the test channel
+		setChan <- value
+	})
+
+	go func() {
+		_ = provider.Run(ctx, comm)
+	}()
+
+	// wait for it to be called once
+	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["one"])
+	require.Equal(t, value2, current["two"])
+
+	// update the value in one
+	value1 = "update1"
+	require.NoError(t, os.WriteFile(file1, []byte(value1), 0o644))
+
+	// wait for file1 to be updated
+	for {
+		var oneUpdated map[string]interface{}
+		select {
+		case oneUpdated = <-setChan:
+		case <-time.After(testTimeout):
+			require.FailNow(t, "timeout waiting for provider to call Set")
+		}
+
+		if value1 == oneUpdated["one"] && value2 == oneUpdated["two"] {
+			break
+		}
+	}
+
+	// update the value in two
+	value2 = "update2"
+	require.NoError(t, os.WriteFile(file2, []byte(value2), 0o644))
+
+	for {
+		// wait for file2 to be updated
+		var twoUpdated map[string]interface{}
+		select {
+		case twoUpdated = <-setChan:
+		case <-time.After(testTimeout):
+			require.FailNow(t, "timeout waiting for provider to call Set")
+		}
+
+		if value1 == twoUpdated["one"] && value2 == twoUpdated["two"] {
+			break
+		}
+	}
+}