Ensure unique ID for dynamic variable replacement. (#1866)

* Ensure unique ID for dynamic variable replacement.

* Add changelog.

changelog/fragments/1669928791-Ensure-unique-input-ID-when-variable-substition-occurs-from-a-dynamic-provider.yaml

@@ -0,0 +1,31 @@
+# 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: Ensure unique input ID when variable substition occurs from a dynamic provider
+
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+#description:
+
+# Affected component; a word indicating the component this changeset affects.
+component:
+
+# PR number; 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: 1866
+
+# Issue number; 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: 1751

internal/pkg/agent/transpiler/ast_test.go

@@ -1840,7 +1840,7 @@ func TestLookupString(t *testing.T) {
 }
 
 func mustMakeVars(mapping map[string]interface{}) *Vars {
-	v, err := NewVars(mapping, nil)
+	v, err := NewVars("", mapping, nil)
 	if err != nil {
 		panic(err)
 	}

internal/pkg/agent/transpiler/utils.go

@@ -15,7 +15,7 @@ func RenderInputs(inputs Node, varsArray []*Vars) (Node, error) {
 	if !ok {
 		return nil, fmt.Errorf("inputs must be an array")
 	}
-	nodes := []*Dict{}
+	var nodes []varIDMap
 	nodesMap := map[string]*Dict{}
 	for _, vars := range varsArray {
 		for _, node := range l.Value().([]Node) {
@@ -41,17 +41,50 @@ func RenderInputs(inputs Node, varsArray []*Vars) (Node, error) {
 			_, exists := nodesMap[hash]
 			if !exists {
 				nodesMap[hash] = dict
-				nodes = append(nodes, dict)
+				nodes = append(nodes, varIDMap{vars.ID(), dict})
 			}
 		}
 	}
-	nInputs := []Node{}
+	var nInputs []Node
 	for _, node := range nodes {
-		nInputs = append(nInputs, promoteProcessors(node))
+		if node.id != "" {
+			// vars has unique ID, concat ID onto existing ID
+			idNode, ok := node.d.Find("id")
+			if ok {
+				idKey, _ := idNode.(*Key) // always a Key
+
+				// clone original and update its key to 'original_id'
+				origKey, _ := idKey.Clone().(*Key) // always a Key
+				origKey.name = "original_id"
+				node.d.Insert(origKey)
+
+				// update id field to concat the id of the variable context set
+				switch idVal := idKey.value.(type) {
+				case *StrVal:
+					idVal.value = fmt.Sprintf("%s-%s", idVal.value, node.id)
+				case *IntVal:
+					idKey.value = NewStrVal(fmt.Sprintf("%d-%s", idVal.value, node.id))
+				case *UIntVal:
+					idKey.value = NewStrVal(fmt.Sprintf("%d-%s", idVal.value, node.id))
+				case *FloatVal:
+					idKey.value = NewStrVal(fmt.Sprintf("%f-%s", idVal.value, node.id))
+				default:
+					return nil, fmt.Errorf("id field type invalid, expected string, int, uint, or float got: %T", idKey.value)
+				}
+			} else {
+				node.d.Insert(NewKey("id", NewStrVal(node.id)))
+			}
+		}
+		nInputs = append(nInputs, promoteProcessors(node.d))
 	}
 	return NewList(nInputs), nil
 }
 
+type varIDMap struct {
+	id string
+	d  *Dict
+}
+
 func promoteProcessors(dict *Dict) *Dict {
 	p := dict.Processors()
 	if p == nil {

internal/pkg/agent/transpiler/utils_test.go

@@ -363,6 +363,7 @@ func TestRenderInputs(t *testing.T) {
 		"vars with processors": {
 			input: NewKey("inputs", NewList([]Node{
 				NewDict([]Node{
+					NewKey("id", NewStrVal("initial")),
 					NewKey("type", NewStrVal("logfile")),
 					NewKey("streams", NewList([]Node{
 						NewDict([]Node{
@@ -385,6 +386,7 @@ func TestRenderInputs(t *testing.T) {
 			})),
 			expected: NewList([]Node{
 				NewDict([]Node{
+					NewKey("id", NewStrVal("initial-value1")),
 					NewKey("type", NewStrVal("logfile")),
 					NewKey("streams", NewList([]Node{
 						NewDict([]Node{
@@ -411,8 +413,10 @@ func TestRenderInputs(t *testing.T) {
 							})),
 						}),
 					})),
+					NewKey("original_id", NewStrVal("initial")),
 				}),
 				NewDict([]Node{
+					NewKey("id", NewStrVal("initial-value2")),
 					NewKey("type", NewStrVal("logfile")),
 					NewKey("streams", NewList([]Node{
 						NewDict([]Node{
@@ -439,10 +443,11 @@ func TestRenderInputs(t *testing.T) {
 							})),
 						}),
 					})),
+					NewKey("original_id", NewStrVal("initial")),
 				}),
 			}),
 			varsArray: []*Vars{
-				mustMakeVarsP(map[string]interface{}{
+				mustMakeVarsP("value1", map[string]interface{}{
 					"var1": map[string]interface{}{
 						"name": "value1",
 					},
@@ -458,7 +463,7 @@ func TestRenderInputs(t *testing.T) {
 							},
 						},
 					}),
-				mustMakeVarsP(map[string]interface{}{
+				mustMakeVarsP("value2", map[string]interface{}{
 					"var1": map[string]interface{}{
 						"name": "value2",
 					},
@@ -499,6 +504,7 @@ func TestRenderInputs(t *testing.T) {
 							})),
 						}),
 					})),
+					NewKey("id", NewStrVal("value1")),
 					NewKey("processors", NewList([]Node{
 						NewDict([]Node{
 							NewKey("add_fields", NewDict([]Node{
@@ -519,6 +525,7 @@ func TestRenderInputs(t *testing.T) {
 							})),
 						}),
 					})),
+					NewKey("id", NewStrVal("value2")),
 					NewKey("processors", NewList([]Node{
 						NewDict([]Node{
 							NewKey("add_fields", NewDict([]Node{
@@ -532,7 +539,7 @@ func TestRenderInputs(t *testing.T) {
 				}),
 			}),
 			varsArray: []*Vars{
-				mustMakeVarsP(map[string]interface{}{
+				mustMakeVarsP("value1", map[string]interface{}{
 					"var1": map[string]interface{}{
 						"name": "value1",
 					},
@@ -548,7 +555,7 @@ func TestRenderInputs(t *testing.T) {
 							},
 						},
 					}),
-				mustMakeVarsP(map[string]interface{}{
+				mustMakeVarsP("value2", map[string]interface{}{
 					"var1": map[string]interface{}{
 						"name": "value2",
 					},
@@ -599,6 +606,7 @@ func TestRenderInputs(t *testing.T) {
 							NewKey("invalid", NewStrVal("value")),
 						})),
 					})),
+					NewKey("id", NewStrVal("value1")),
 				}),
 				NewDict([]Node{
 					NewKey("type", NewStrVal("logfile")),
@@ -614,10 +622,11 @@ func TestRenderInputs(t *testing.T) {
 							NewKey("invalid", NewStrVal("value")),
 						})),
 					})),
+					NewKey("id", NewStrVal("value2")),
 				}),
 			}),
 			varsArray: []*Vars{
-				mustMakeVarsP(map[string]interface{}{
+				mustMakeVarsP("value1", map[string]interface{}{
 					"var1": map[string]interface{}{
 						"name": "value1",
 					},
@@ -633,7 +642,7 @@ func TestRenderInputs(t *testing.T) {
 							},
 						},
 					}),
-				mustMakeVarsP(map[string]interface{}{
+				mustMakeVarsP("value2", map[string]interface{}{
 					"var1": map[string]interface{}{
 						"name": "value2",
 					},
@@ -674,6 +683,7 @@ func TestRenderInputs(t *testing.T) {
 							})),
 						}),
 					})),
+					NewKey("id", NewStrVal("value1")),
 					NewKey("processors", NewList([]Node{
 						NewDict([]Node{
 							NewKey("add_fields", NewDict([]Node{
@@ -687,7 +697,7 @@ func TestRenderInputs(t *testing.T) {
 				}),
 			}),
 			varsArray: []*Vars{
-				mustMakeVarsP(map[string]interface{}{
+				mustMakeVarsP("value1", map[string]interface{}{
 					"var1": map[string]interface{}{
 						"name": "value1",
 					},
@@ -703,7 +713,7 @@ func TestRenderInputs(t *testing.T) {
 							},
 						},
 					}),
-				mustMakeVarsP(map[string]interface{}{
+				mustMakeVarsP("value2", map[string]interface{}{
 					"var1": map[string]interface{}{
 						"name": "value1",
 					},
@@ -736,8 +746,8 @@ func TestRenderInputs(t *testing.T) {
 	}
 }
 
-func mustMakeVarsP(mapping map[string]interface{}, processorKey string, processors Processors) *Vars {
-	v, err := NewVarsWithProcessors(mapping, processorKey, processors, nil)
+func mustMakeVarsP(id string, mapping map[string]interface{}, processorKey string, processors Processors) *Vars {
+	v, err := NewVarsWithProcessors(id, mapping, processorKey, processors, nil)
 	if err != nil {
 		panic(err)
 	}

internal/pkg/agent/transpiler/vars.go

@@ -21,24 +21,25 @@ var ErrNoMatch = fmt.Errorf("no matching vars")
 
 // Vars is a context of variables that also contain a list of processors that go with the mapping.
 type Vars struct {
+	id                    string
 	tree                  *AST
 	processorsKey         string
 	processors            Processors
 	fetchContextProviders mapstr.M
 }
 
 // NewVars returns a new instance of vars.
-func NewVars(mapping map[string]interface{}, fetchContextProviders mapstr.M) (*Vars, error) {
-	return NewVarsWithProcessors(mapping, "", nil, fetchContextProviders)
+func NewVars(id string, mapping map[string]interface{}, fetchContextProviders mapstr.M) (*Vars, error) {
+	return NewVarsWithProcessors(id, mapping, "", nil, fetchContextProviders)
 }
 
 // NewVarsWithProcessors returns a new instance of vars with attachment of processors.
-func NewVarsWithProcessors(mapping map[string]interface{}, processorKey string, processors Processors, fetchContextProviders mapstr.M) (*Vars, error) {
+func NewVarsWithProcessors(id string, mapping map[string]interface{}, processorKey string, processors Processors, fetchContextProviders mapstr.M) (*Vars, error) {
 	tree, err := NewAST(mapping)
 	if err != nil {
 		return nil, err
 	}
-	return &Vars{tree, processorKey, processors, fetchContextProviders}, nil
+	return &Vars{id, tree, processorKey, processors, fetchContextProviders}, nil
 }
 
 // Replace returns a new value based on variable replacement.
@@ -91,6 +92,11 @@ func (v *Vars) Replace(value string) (Node, error) {
 	return NewStrValWithProcessors(result+value[lastIndex:], processors), nil
 }
 
+// ID returns the unique ID for the vars.
+func (v *Vars) ID() string {
+	return v.id
+}
+
 // Lookup returns the value from the vars.
 func (v *Vars) Lookup(name string) (interface{}, bool) {
 	// lookup in the AST tree

internal/pkg/agent/transpiler/vars_test.go

@@ -227,6 +227,7 @@ func TestVars_ReplaceWithProcessors(t *testing.T) {
 		},
 	}
 	vars, err := NewVarsWithProcessors(
+		"",
 		map[string]interface{}{
 			"testing": map[string]interface{}{
 				"key1": "data1",
@@ -293,6 +294,7 @@ func TestVars_ReplaceWithFetchContextProvider(t *testing.T) {
 		"kubernetes_secrets": mockFetchProvider,
 	}
 	vars, err := NewVarsWithProcessors(
+		"id",
 		map[string]interface{}{
 			"testing": map[string]interface{}{
 				"key1": "data1",

internal/pkg/composable/controller.go

@@ -193,15 +193,16 @@ func (c *controller) Run(ctx context.Context) error {
 			mapping[name] = state.Current()
 		}
 		// this is ensured not to error, by how the mappings states are verified
-		vars[0], _ = transpiler.NewVars(mapping, fetchContextProviders)
+		vars[0], _ = transpiler.NewVars("", mapping, fetchContextProviders)
 
 		// add to the vars list for each dynamic providers mappings
 		for name, state := range c.dynamicProviders {
 			for _, mappings := range state.Mappings() {
 				local, _ := cloneMap(mapping) // will not fail; already been successfully cloned once
 				local[name] = mappings.mapping
+				id := fmt.Sprintf("%s-%s", name, mappings.id)
 				// this is ensured not to error, by how the mappings states are verified
-				v, _ := transpiler.NewVarsWithProcessors(local, name, mappings.processors, fetchContextProviders)
+				v, _ := transpiler.NewVarsWithProcessors(id, local, name, mappings.processors, fetchContextProviders)
 				vars = append(vars, v)
 			}
 		}
@@ -237,7 +238,7 @@ func (c *contextProviderState) Set(mapping map[string]interface{}) error {
 		return err
 	}
 	// ensure creating vars will not error
-	_, err = transpiler.NewVars(mapping, nil)
+	_, err = transpiler.NewVars("", mapping, nil)
 	if err != nil {
 		return err
 	}
@@ -262,6 +263,7 @@ func (c *contextProviderState) Current() map[string]interface{} {
 }
 
 type dynamicProviderMapping struct {
+	id         string
 	priority   int
 	mapping    map[string]interface{}
 	processors transpiler.Processors
@@ -292,7 +294,7 @@ func (c *dynamicProviderState) AddOrUpdate(id string, priority int, mapping map[
 		return err
 	}
 	// ensure creating vars will not error
-	_, err = transpiler.NewVars(mapping, nil)
+	_, err = transpiler.NewVars("", mapping, nil)
 	if err != nil {
 		return err
 	}
@@ -305,6 +307,7 @@ func (c *dynamicProviderState) AddOrUpdate(id string, priority int, mapping map[
 		return nil
 	}
 	c.mappings[id] = dynamicProviderMapping{
+		id:         id,
 		priority:   priority,
 		mapping:    mapping,
 		processors: processors,

pkg/component/component.go

@@ -145,7 +145,7 @@ func (r *RuntimeSpecs) PolicyToComponents(policy map[string]interface{}) ([]Comp
 	if err != nil {
 		return nil, nil, err
 	}
-	vars, err := transpiler.NewVars(map[string]interface{}{
+	vars, err := transpiler.NewVars("", map[string]interface{}{
 		"runtime": map[string]interface{}{
 			"platform": r.platform.String(),
 			"os":       r.platform.OS,