Update all kyaml filter uses to process results especially

kyaml/filtersutil/filtersutil.go

@@ -5,8 +5,10 @@ package filtersutil
 
 import (
 	"encoding/json"
+	goerrors "errors"
 
 	"sigs.k8s.io/kustomize/kyaml/errors"
+	"sigs.k8s.io/kustomize/kyaml/fn/framework"
 	"sigs.k8s.io/kustomize/kyaml/kio"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
 )
@@ -30,9 +32,9 @@ func ApplyToJSON(filter kio.Filter, objs ...marshalerUnmarshaler) error {
 		nodes = append(nodes, node)
 	}
 
-	// apply the filter
+	// apply the filter, skip error type "Results"
 	nodes, err := filter.Filter(nodes)
-	if err != nil {
+	if err != nil && !goerrors.Is(err, &framework.Results{}) {
 		return err
 	}
 	if len(nodes) != len(objs) {

kyaml/fn/framework/patch.go

@@ -5,6 +5,7 @@ package framework
 
 import (
 	"bytes"
+	goerrors "errors"
 	"fmt"
 	"strings"
 	"text/template"
@@ -50,7 +51,7 @@ func (t ResourcePatchTemplate) Filter(items []*yaml.RNode) ([]*yaml.RNode, error
 	target := items
 	if t.Selector != nil {
 		target, err = t.Selector.Filter(items)
-		if err != nil {
+		if err != nil && !goerrors.Is(err, &Results{}) {
 			return nil, err
 		}
 	}
@@ -136,7 +137,7 @@ func (cpt ContainerPatchTemplate) Filter(items []*yaml.RNode) ([]*yaml.RNode, er
 	target := items
 	if cpt.Selector != nil {
 		target, err = cpt.Selector.Filter(items)
-		if err != nil {
+		if err != nil && !goerrors.Is(err, &Results{}) {
 			return nil, err
 		}
 	}

kyaml/fn/framework/processors.go

@@ -4,6 +4,7 @@
 package framework
 
 import (
+	goerrors "errors"
 	"strings"
 
 	"k8s.io/kube-openapi/pkg/validation/spec"
@@ -278,19 +279,18 @@ func (tp *TemplateProcessor) doPreProcess(items []*yaml.RNode) ([]*yaml.RNode, e
 		filter := tp.PreProcessFilters[i]
 		var err error
 		items, err = filter.Filter(items)
-		if err != nil {
+		if err != nil && !goerrors.Is(err, &Results{}) {
 			return nil, err
 		}
 	}
 	return items, nil
 }
 
 func (tp *TemplateProcessor) doMerge(items []*yaml.RNode) ([]*yaml.RNode, error) {
-	var err error
 	if tp.MergeResources {
-		items, err = filters.MergeFilter{}.Filter(items)
+		return filters.MergeFilter{}.Filter(items)
 	}
-	return items, err
+	return items, nil
 }
 
 func (tp *TemplateProcessor) doPostProcess(items []*yaml.RNode) ([]*yaml.RNode, error) {
@@ -301,7 +301,7 @@ func (tp *TemplateProcessor) doPostProcess(items []*yaml.RNode) ([]*yaml.RNode,
 		filter := tp.PostProcessFilters[i]
 		var err error
 		items, err = filter.Filter(items)
-		if err != nil {
+		if err != nil && !goerrors.Is(err, &Results{}) {
 			return nil, err
 		}
 	}

kyaml/fn/runtime/starlark/starlark.go

@@ -5,13 +5,16 @@ package starlark
 
 import (
 	"bytes"
+	"encoding/json"
+	goerrors "errors"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
 
 	"go.starlark.net/starlark"
 	"sigs.k8s.io/kustomize/kyaml/errors"
+	"sigs.k8s.io/kustomize/kyaml/fn/framework"
 	"sigs.k8s.io/kustomize/kyaml/fn/runtime/runtimeutil"
 	"sigs.k8s.io/kustomize/kyaml/internal/forked/github.com/qri-io/starlib/util"
 	"sigs.k8s.io/kustomize/kyaml/kio/filters"
@@ -161,16 +164,35 @@ func (sf *Filter) writeResourceList(value starlark.Value, writer io.Writer) erro
 	if err != nil {
 		return errors.Wrap(err)
 	}
+
+	var allResults framework.Results
 	err = items.VisitElements(func(node *yaml.RNode) error {
 		// starlark will serialize the resources sorting the fields alphabetically,
 		// format them to have a better ordering
 		_, err := filters.FormatFilter{}.Filter([]*yaml.RNode{node})
-		return err
+		if err != nil {
+			var results framework.Results
+			if !goerrors.As(err, &results) {
+				return err
+			}
+			allResults = append(allResults, results...)
+		}
+		return nil
 	})
 	if err != nil {
 		return errors.Wrap(err)
 	}
 
+	bResults, err := json.Marshal(allResults)
+	if err != nil {
+		return errors.Wrap(err)
+	}
+
+	rl, err = yaml.SetField("results", yaml.MustParse(string(bResults))).Filter(rl)
+	if err != nil {
+		return errors.Wrap(err)
+	}
+
 	s, err := rl.String()
 	if err != nil {
 		return errors.Wrap(err)

kyaml/kio/filters/filters.go

@@ -4,10 +4,12 @@
 package filters
 
 import (
+	goerrors "errors"
 	"fmt"
 	"sort"
 	"strings"
 
+	"sigs.k8s.io/kustomize/kyaml/fn/framework"
 	"sigs.k8s.io/kustomize/kyaml/kio"
 	"sigs.k8s.io/kustomize/kyaml/kio/kioutil"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
@@ -86,11 +88,11 @@ type MatchModifyFilter struct {
 var _ kio.Filter = &MatchModifyFilter{}
 
 func (f MatchModifyFilter) Filter(input []*yaml.RNode) ([]*yaml.RNode, error) {
-	var matches = input
+	matches := input
 	var err error
 	for _, filter := range f.MatchFilters {
 		matches, err = MatchFilter{Filters: filter}.Filter(matches)
-		if err != nil {
+		if err != nil && !goerrors.Is(err, &framework.Results{}) {
 			return nil, err
 		}
 	}

kyaml/kio/kio.go

@@ -6,9 +6,12 @@
 package kio
 
 import (
+	"encoding/json"
+	goerrors "errors"
 	"fmt"
 
 	"sigs.k8s.io/kustomize/kyaml/errors"
+	"sigs.k8s.io/kustomize/kyaml/fn/framework"
 	"sigs.k8s.io/kustomize/kyaml/kio/kioutil"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
 )
@@ -51,6 +54,9 @@ type ReaderWriter interface {
 // When possible, Filters should be serializable to yaml so that they can be described
 // as either data or code.
 //
+// A Filter can return an error type "Result" which must be handled as a non-terminating
+// spacial case, i.e. a benign error value similar to EOF.
+//
 // Analogous to http://www.linfo.org/filters.html
 type Filter interface {
 	Filter([]*yaml.RNode) ([]*yaml.RNode, error)
@@ -102,24 +108,26 @@ type PipelineExecuteCallbackFunc = func(op Filter)
 // ExecuteWithCallback executes each step in the sequence, returning immediately after encountering
 // any error as part of the Pipeline. The callback will be called each time a step succeeds.
 func (p Pipeline) ExecuteWithCallback(callback PipelineExecuteCallbackFunc) error {
-	var result []*yaml.RNode
+	var rnodes []*yaml.RNode
 
 	// read from the inputs
 	for _, i := range p.Inputs {
 		nodes, err := i.Read()
 		if err != nil {
 			return errors.Wrap(err)
 		}
-		result = append(result, nodes...)
+		rnodes = append(rnodes, nodes...)
 	}
 
+	var allResults framework.Results
+
 	// apply operations
 	var err error
 	for i := range p.Filters {
 		// Not all RNodes passed through kio.Pipeline have metadata nor should
 		// they all be required to.
 		var nodeAnnos map[string]map[string]string
-		nodeAnnos, err = storeInternalAnnotations(result)
+		nodeAnnos, err = storeInternalAnnotations(rnodes)
 		if err != nil && err != yaml.ErrMissingMetadata {
 			return err
 		}
@@ -128,25 +136,50 @@ func (p Pipeline) ExecuteWithCallback(callback PipelineExecuteCallbackFunc) erro
 		if callback != nil {
 			callback(op)
 		}
-		result, err = op.Filter(result)
+		rnodes, err = op.Filter(rnodes)
+
+		if err != nil {
+			var results framework.Results
+			if !goerrors.As(err, &results) {
+				return errors.Wrap(err)
+			}
+			allResults = append(allResults, results...)
+		}
+
 		// TODO (issue 2872): This len(result) == 0 should be removed and empty result list should be
 		// handled by outputs. However currently some writer like LocalPackageReadWriter
 		// will clear the output directory and which will cause unpredictable results
-		if len(result) == 0 && !p.ContinueOnEmptyResult || err != nil {
-			return errors.Wrap(err)
+		if len(rnodes) == 0 && !p.ContinueOnEmptyResult {
+			return errors.Wrap(fmt.Errorf("empty results on filter index %d", i))
 		}
 
 		// If either the internal annotations for path, index, and id OR the legacy
 		// annotations for path, index, and id are changed, we have to update the other.
-		err = reconcileInternalAnnotations(result, nodeAnnos)
+		err = reconcileInternalAnnotations(rnodes, nodeAnnos)
 		if err != nil && err != yaml.ErrMissingMetadata {
 			return err
 		}
 	}
 
 	// write to the outputs
 	for _, o := range p.Outputs {
-		if err := o.Write(result); err != nil {
+		switch o.(type) {
+		case *ByteReadWriter, *ByteWriter:
+			bResults, err := json.Marshal(allResults)
+			if err != nil {
+				return errors.Wrap(err)
+			}
+			rNodeResults := yaml.MustParse(string(bResults))
+
+			switch w := o.(type) {
+			case *ByteReadWriter:
+				w.Results = rNodeResults
+			case *ByteWriter:
+				w.Results = rNodeResults
+			}
+		}
+
+		if err := o.Write(rnodes); err != nil {
 			return errors.Wrap(err)
 		}
 	}
@@ -156,13 +189,18 @@ func (p Pipeline) ExecuteWithCallback(callback PipelineExecuteCallbackFunc) erro
 // FilterAll runs the yaml.Filter against all inputs
 func FilterAll(filter yaml.Filter) Filter {
 	return FilterFunc(func(nodes []*yaml.RNode) ([]*yaml.RNode, error) {
+		var results framework.Results
 		for i := range nodes {
 			_, err := filter.Filter(nodes[i])
 			if err != nil {
-				return nil, errors.Wrap(err)
+				var filterResults framework.Results
+				if !goerrors.As(err, &filterResults) {
+					return nil, errors.Wrap(err)
+				}
+				results = append(results, filterResults...)
 			}
 		}
-		return nodes, nil
+		return nodes, results
 	})
 }
 

kyaml/kio/kio_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"sigs.k8s.io/kustomize/kyaml/fn/framework"
 	"sigs.k8s.io/kustomize/kyaml/kio/kioutil"
 	"sigs.k8s.io/kustomize/kyaml/yaml"
 
@@ -574,7 +575,77 @@ data:
 				assert.Error(t, err)
 				assert.Equal(t, tc.expectedErr, err.Error())
 			}
-
 		})
 	}
 }
+
+func TestPipelineWithResults(t *testing.T) {
+	var out bytes.Buffer
+	rw := ByteReadWriter{
+		Reader: strings.NewReader(`
+apiVersion: config.kubernetes.io/v1
+kind: ResourceList
+items:
+- kind: Deployment
+  spec:
+    replicas: 1
+- kind: Service
+  spec:
+    selectors:
+      foo: bar
+`),
+		Writer: &out,
+	}
+	p := Pipeline{
+		Inputs: []Reader{&rw},
+		Filters: []Filter{
+			FilterFunc(func(in []*yaml.RNode) ([]*yaml.RNode, error) {
+				return in, framework.Results{{
+					Message:  "result 1",
+					Severity: framework.Warning,
+				}}
+			}),
+			FilterFunc(func(in []*yaml.RNode) ([]*yaml.RNode, error) {
+				return in, framework.Results{
+					{
+						Message:  "result 2",
+						Severity: framework.Error,
+					},
+					{Message: "result 3"},
+				}
+			}),
+			FilterFunc(func(in []*yaml.RNode) ([]*yaml.RNode, error) {
+				return in, framework.Results{{
+					Message:  "result 4",
+					Severity: framework.Info,
+				}}
+			}),
+		},
+		Outputs: []Writer{&rw},
+	}
+
+	err := p.Execute()
+	assert.NoError(t, err)
+
+	want := `apiVersion: config.kubernetes.io/v1
+kind: ResourceList
+items:
+- kind: Deployment
+  spec:
+    replicas: 1
+- kind: Service
+  spec:
+    selectors:
+      foo: bar
+results:
+- message: result 1
+  severity: warning
+- message: result 2
+  severity: error
+- message: result 3
+- message: result 4
+  severity: info
+`
+
+	assert.Equal(t, want, out.String())
+}