Add support for variables in outputs and default provider

changelog/fragments/1737816507-Add-context-variable-support-to-outputs.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 context variable support to outputs
+
+# 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: |
+  Adds support for using context variable providers in the outputs section of a policy. Includes fallback support
+  to reference env provider when no provider prefix is provided in the variable.
+
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: elastic-agent
+
+# 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/6602
+
+# 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/6376

internal/pkg/agent/application/coordinator/coordinator.go

@@ -172,8 +172,11 @@ type ConfigManager interface {
 type VarsManager interface {
 	Runner
 
+	// DefaultProvider returns the default provider that the variable manager is configured to use.
+	DefaultProvider() string
+
 	// Observe instructs the variables to observe.
-	Observe([]string)
+	Observe(context.Context, []string) ([]*transpiler.Vars, error)
 
 	// Watch returns the chanel to watch for variable changes.
 	Watch() <-chan []*transpiler.Vars
@@ -1244,7 +1247,11 @@ func (c *Coordinator) processConfigAgent(ctx context.Context, cfg *config.Config
 	}
 
 	// pass the observed vars from the AST to the varsMgr
-	c.observeASTVars()
+	err = c.observeASTVars(ctx)
+	if err != nil {
+		// only possible error here is the context being cancelled
+		return err
+	}
 
 	// Disabled for 8.8.0 release in order to limit the surface
 	// https://github.com/elastic/security-team/issues/6501
@@ -1327,25 +1334,32 @@ func (c *Coordinator) generateAST(cfg *config.Config) (err error) {
 // observeASTVars identifies the variables that are referenced in the computed AST and passed to
 // the varsMgr so it knows what providers are being referenced. If a providers is not being
 // referenced then the provider does not need to be running.
-func (c *Coordinator) observeASTVars() {
+func (c *Coordinator) observeASTVars(ctx context.Context) error {
 	if c.varsMgr == nil {
 		// No varsMgr (only happens in testing)
-		return
+		return nil
 	}
-	if c.ast == nil {
-		// No AST; no vars
-		c.varsMgr.Observe(nil)
-		return
+	var vars []string
+	if c.ast != nil {
+		inputs, ok := transpiler.Lookup(c.ast, "inputs")
+		if ok {
+			vars = inputs.Vars(vars, c.varsMgr.DefaultProvider())
+		}
+		outputs, ok := transpiler.Lookup(c.ast, "outputs")
+		if ok {
+			vars = outputs.Vars(vars, c.varsMgr.DefaultProvider())
+		}
 	}
-	inputs, ok := transpiler.Lookup(c.ast, "inputs")
-	if !ok {
-		// No inputs; no vars
-		c.varsMgr.Observe(nil)
-		return
+	updated, err := c.varsMgr.Observe(ctx, vars)
+	if err != nil {
+		// context cancel
+		return err
 	}
-	var vars []string
-	vars = inputs.Vars(vars)
-	c.varsMgr.Observe(vars)
+	if updated != nil {
+		// provided an updated set of vars (observed changed)
+		c.vars = updated
+	}
+	return nil
 }
 
 // processVars updates the transpiler vars in the Coordinator.
@@ -1421,6 +1435,8 @@ func (c *Coordinator) generateComponentModel() (err error) {
 	}()
 
 	ast := c.ast.ShallowClone()
+
+	// perform variable substitution for inputs
 	inputs, ok := transpiler.Lookup(ast, "inputs")
 	if ok {
 		renderedInputs, err := transpiler.RenderInputs(inputs, c.vars)
@@ -1433,6 +1449,20 @@ func (c *Coordinator) generateComponentModel() (err error) {
 		}
 	}
 
+	// perform variable substitution for outputs
+	// outputs only support the context variables (dynamic provides are not provide to the outputs)
+	outputs, ok := transpiler.Lookup(ast, "outputs")
+	if ok {
+		renderedOutputs, err := transpiler.RenderOutputs(outputs, c.vars)
+		if err != nil {
+			return fmt.Errorf("rendering outputs failed: %w", err)
+		}
+		err = transpiler.Insert(ast, renderedOutputs, "outputs")
+		if err != nil {
+			return fmt.Errorf("inserting rendered outputs failed: %w", err)
+		}
+	}
+
 	cfg, err := ast.Map()
 	if err != nil {
 		return fmt.Errorf("failed to convert ast to map[string]interface{}: %w", err)

internal/pkg/agent/application/coordinator/coordinator_test.go

@@ -928,7 +928,7 @@ func BenchmarkCoordinator_generateComponentModel(b *testing.B) {
 	require.NoError(b, err)
 	vars := make([]*transpiler.Vars, len(varsMaps))
 	for i, vm := range varsMaps {
-		vars[i], err = transpiler.NewVars(fmt.Sprintf("%d", i), vm, mapstr.M{})
+		vars[i], err = transpiler.NewVars(fmt.Sprintf("%d", i), vm, mapstr.M{}, "")
 		require.NoError(b, err)
 	}
 
@@ -1188,6 +1188,9 @@ func (l *configChange) Fail(err error) {
 }
 
 type fakeVarsManager struct {
+	varsMx sync.RWMutex
+	vars   []*transpiler.Vars
+
 	varsCh chan []*transpiler.Vars
 	errCh  chan error
 
@@ -1222,19 +1225,29 @@ func (f *fakeVarsManager) Watch() <-chan []*transpiler.Vars {
 	return f.varsCh
 }
 
-func (f *fakeVarsManager) Observe(observed []string) {
+func (f *fakeVarsManager) Observe(ctx context.Context, observed []string) ([]*transpiler.Vars, error) {
 	f.observedMx.Lock()
 	defer f.observedMx.Unlock()
 	f.observed = observed
+	f.varsMx.RLock()
+	defer f.varsMx.RUnlock()
+	return f.vars, nil
 }
 
 func (f *fakeVarsManager) Vars(ctx context.Context, vars []*transpiler.Vars) {
+	f.varsMx.Lock()
+	f.vars = vars
+	f.varsMx.Unlock()
 	select {
 	case <-ctx.Done():
 	case f.varsCh <- vars:
 	}
 }
 
+func (f *fakeVarsManager) DefaultProvider() string {
+	return ""
+}
+
 type fakeOTelManager struct {
 	updateCallback func(*confmap.Conf) error
 	result         error

internal/pkg/agent/application/coordinator/coordinator_unit_test.go

@@ -1104,7 +1104,7 @@ inputs:
 	components = nil
 	vars, err := transpiler.NewVars("", map[string]interface{}{
 		"TEST_VAR": "input-id",
-	}, nil)
+	}, nil, "")
 	require.NoError(t, err, "Vars creation must succeed")
 	varsChan <- []*transpiler.Vars{vars}
 	coord.runLoopIteration(ctx)
@@ -1121,7 +1121,7 @@ inputs:
 	components = nil
 	vars, err = transpiler.NewVars("", map[string]interface{}{
 		"TEST_VAR": "changed-input-id",
-	}, nil)
+	}, nil, "")
 	require.NoError(t, err, "Vars creation must succeed")
 	varsChan <- []*transpiler.Vars{vars}
 	coord.runLoopIteration(ctx)
@@ -1239,7 +1239,7 @@ func TestCoordinatorInitiatesUpgrade(t *testing.T) {
 // (Coordinator will only regenerate its component model when it has non-nil
 // vars).
 func emptyVars(t *testing.T) []*transpiler.Vars {
-	vars, err := transpiler.NewVars("", map[string]interface{}{}, nil)
+	vars, err := transpiler.NewVars("", map[string]interface{}{}, nil, "")
 	require.NoError(t, err, "Vars creation must succeed")
 	return []*transpiler.Vars{vars}
 }

internal/pkg/agent/application/coordinator/diagnostics_test.go

@@ -196,7 +196,7 @@ func TestDiagnosticVariables(t *testing.T) {
 		map[string]interface{}{
 			"testvar": "testvalue",
 		},
-		nil)
+		nil, "")
 	require.NoError(t, err)
 
 	expected := `

internal/pkg/agent/transpiler/ast.go

@@ -64,7 +64,7 @@ type Node interface {
 
 	// Vars adds to the array with the variables identified in the node. Returns the array in-case
 	// the capacity of the array had to be changed.
-	Vars([]string) []string
+	Vars([]string, string) []string
 
 	// Apply apply the current vars, returning the new value for the node. This does not modify the original Node.
 	Apply(*Vars) (Node, error)
@@ -182,10 +182,10 @@ func (d *Dict) Hash64With(h *xxhash.Digest) error {
 }
 
 // Vars returns a list of all variables referenced in the dictionary.
-func (d *Dict) Vars(vars []string) []string {
+func (d *Dict) Vars(vars []string, defaultProvider string) []string {
 	for _, v := range d.value {
 		k := v.(*Key)
-		vars = k.Vars(vars)
+		vars = k.Vars(vars, defaultProvider)
 	}
 	return vars
 }
@@ -318,11 +318,11 @@ func (k *Key) Hash64With(h *xxhash.Digest) error {
 }
 
 // Vars returns a list of all variables referenced in the value.
-func (k *Key) Vars(vars []string) []string {
+func (k *Key) Vars(vars []string, defaultProvider string) []string {
 	if k.value == nil {
 		return vars
 	}
-	return k.value.Vars(vars)
+	return k.value.Vars(vars, defaultProvider)
 }
 
 // Apply applies the vars to the value. This does not modify the original node.
@@ -463,9 +463,9 @@ func (l *List) ShallowClone() Node {
 }
 
 // Vars returns a list of all variables referenced in the list.
-func (l *List) Vars(vars []string) []string {
+func (l *List) Vars(vars []string, defaultProvider string) []string {
 	for _, v := range l.value {
-		vars = v.Vars(vars)
+		vars = v.Vars(vars, defaultProvider)
 	}
 	return vars
 }
@@ -552,12 +552,12 @@ func (s *StrVal) Hash64With(h *xxhash.Digest) error {
 }
 
 // Vars returns a list of all variables referenced in the string.
-func (s *StrVal) Vars(vars []string) []string {
+func (s *StrVal) Vars(vars []string, defaultProvider string) []string {
 	// errors are ignored (if there is an error determine the vars it will also error computing the policy)
 	_, _ = replaceVars(s.value, func(variable string) (Node, Processors, bool) {
 		vars = append(vars, variable)
 		return nil, nil, false
-	}, false)
+	}, false, defaultProvider)
 	return vars
 }
 
@@ -613,7 +613,7 @@ func (s *IntVal) ShallowClone() Node {
 }
 
 // Vars does nothing. Cannot have variable in an IntVal.
-func (s *IntVal) Vars(vars []string) []string {
+func (s *IntVal) Vars(vars []string, defaultProvider string) []string {
 	return vars
 }
 
@@ -691,7 +691,7 @@ func (s *UIntVal) Hash64With(h *xxhash.Digest) error {
 }
 
 // Vars does nothing. Cannot have variable in an UIntVal.
-func (s *UIntVal) Vars(vars []string) []string {
+func (s *UIntVal) Vars(vars []string, defaultProvider string) []string {
 	return vars
 }
 
@@ -764,7 +764,7 @@ func (s *FloatVal) hashString() string {
 }
 
 // Vars does nothing. Cannot have variable in an FloatVal.
-func (s *FloatVal) Vars(vars []string) []string {
+func (s *FloatVal) Vars(vars []string, defaultProvider string) []string {
 	return vars
 }
 
@@ -843,7 +843,7 @@ func (s *BoolVal) Hash64With(h *xxhash.Digest) error {
 }
 
 // Vars does nothing. Cannot have variable in an BoolVal.
-func (s *BoolVal) Vars(vars []string) []string {
+func (s *BoolVal) Vars(vars []string, defaultProvider string) []string {
 	return vars
 }