Remove inputs when all streams are removed (#1869)

* Remove inputs when all streams are removed.

* Add changelog.

* Fix tests.

* Remove processors from test.

Author: blakerouse

changelog/fragments/1669940990-Remove-inputs-when-all-streams-are-removed.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: Remove inputs when all streams are removed
+
+# 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: 1869
+
+# 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: 1868

internal/pkg/agent/transpiler/utils.go

@@ -9,6 +9,13 @@ import (
 	"fmt"
 )
 
+const (
+	// streamsKey is the name of the dictionary key for streams that an input can have. In the case that
+	// an input defines a set of streams and after conditions are applied all the streams are removed then
+	// the entire input is removed.
+	streamsKey = "streams"
+)
+
 // RenderInputs renders dynamic inputs section
 func RenderInputs(inputs Node, varsArray []*Vars) (Node, error) {
 	l, ok := inputs.Value().(*List)
@@ -23,6 +30,10 @@ func RenderInputs(inputs Node, varsArray []*Vars) (Node, error) {
 			if !ok {
 				continue
 			}
+			hadStreams := false
+			if streams := getStreams(dict); streams != nil {
+				hadStreams = true
+			}
 			n, err := dict.Apply(vars)
 			if errors.Is(err, ErrNoMatch) {
 				// has a variable that didn't exist, so we ignore it
@@ -37,6 +48,13 @@ func RenderInputs(inputs Node, varsArray []*Vars) (Node, error) {
 				continue
 			}
 			dict = n.(*Dict)
+			if hadStreams {
+				streams := getStreams(dict)
+				if streams == nil {
+					// conditions removed all streams (input is removed)
+					continue
+				}
+			}
 			hash := string(dict.Hash())
 			_, exists := nodesMap[hash]
 			if !exists {
@@ -85,6 +103,29 @@ type varIDMap struct {
 	d  *Dict
 }
 
+func getStreams(dict *Dict) *List {
+	node, ok := dict.Find(streamsKey)
+	if !ok {
+		return nil
+	}
+	key, ok := node.(*Key)
+	if !ok {
+		return nil
+	}
+	if key.value == nil {
+		return nil
+	}
+	list, ok := key.value.(*List)
+	if !ok {
+		return nil
+	}
+	if len(list.value) == 0 {
+		// didn't have any streams defined in the list (so no removal should be done)
+		return nil
+	}
+	return list
+}
+
 func promoteProcessors(dict *Dict) *Dict {
 	p := dict.Processors()
 	if p == nil {

internal/pkg/agent/transpiler/utils_test.go

@@ -731,6 +731,44 @@ func TestRenderInputs(t *testing.T) {
 					}),
 			},
 		},
+		"input removal with stream conditions": {
+			input: NewKey("inputs", NewList([]Node{
+				NewDict([]Node{
+					NewKey("type", NewStrVal("logfile")),
+					NewKey("streams", NewList([]Node{
+						NewDict([]Node{
+							NewKey("paths", NewList([]Node{
+								NewStrVal("/var/log/${var1.name}.log"),
+							})),
+							NewKey("condition", NewStrVal("${var1.name} != 'value1'")),
+						}),
+						NewDict([]Node{
+							NewKey("paths", NewList([]Node{
+								NewStrVal("/var/log/${var1.name}.log"),
+							})),
+							NewKey("condition", NewStrVal("${var1.name} != 'value1'")),
+						}),
+					})),
+				}),
+			})),
+			expected: NewList([]Node{}),
+			varsArray: []*Vars{
+				mustMakeVarsP("value1", map[string]interface{}{
+					"var1": map[string]interface{}{
+						"name": "value1",
+					},
+				},
+					"var1",
+					nil),
+				mustMakeVarsP("value2", map[string]interface{}{
+					"var1": map[string]interface{}{
+						"name": "value1",
+					},
+				},
+					"var1",
+					nil),
+			},
+		},
 	}
 
 	for name, test := range testcases {