Implement parsing of YAML streams

This adds the ability to parse YAML streams (as produced by helm
template/helmfile template). As those tools might spill out empty
manifests and kubeyaml fails on those, I've addd an option to ignore
them.

Fixes chuckha#7

backend/cmd/kubeyaml/kubeyaml.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"bufio"
 	"bytes"
 	"errors"
 	"flag"
@@ -12,6 +11,7 @@ import (
 	"strings"
 
 	"github.com/chuckha/kubeyaml.com/backend/internal/kubernetes"
+	yaml "gopkg.in/yaml.v2"
 )
 
 /*
@@ -27,9 +27,10 @@ contents
 */
 
 type options struct {
-	Versions []string
-	versions *string
-	silent   *bool
+	Versions    []string
+	versions    *string
+	silent      *bool
+	ignoreEmpty *bool
 }
 
 func (o *options) Validate() error {
@@ -52,6 +53,7 @@ func run(in io.Reader, out io.Writer, args ...string) error {
 	validate := flag.NewFlagSet("validate", flag.ExitOnError)
 	opts.versions = validate.String("versions", "1.19,1.18,1.17,1.16,1.15", "comma separated list of kubernetes versions to validate against")
 	opts.silent = validate.Bool("silent", false, "if true, kubeyaml will not print any output")
+	opts.ignoreEmpty = validate.Bool("ignore-empty", false, "if true, kubeyaml will not treat empty YAML documents as error")
 	validate.Parse(args)
 	err := opts.Validate()
 	if err != nil {
@@ -62,40 +64,57 @@ func run(in io.Reader, out io.Writer, args ...string) error {
 	gf := kubernetes.NewAPIKeyer("io.k8s.api", ".k8s.io")
 
 	// Read the input
-	reader := bufio.NewReader(in)
 	var input bytes.Buffer
-	readerCopy := io.TeeReader(reader, &input)
-	i, err := loader.Load(readerCopy)
-	if err != nil {
-		return &mainError{input.String(), err}
-	}
-
-	aggregatedErrors := &aggErr{}
-	for _, version := range opts.Versions {
-		reslover, err := kubernetes.NewResolver(version)
+	readerCopy := io.TeeReader(in, &input)
+
+	// Split input YAML into separate documents
+	d := yaml.NewDecoder(readerCopy)
+	for {
+		var obj map[interface{}]interface{}
+		err := d.Decode(&obj)
+		if err == io.EOF {
+			break
+		}
 		if err != nil {
-			aggregatedErrors.Add(fmt.Errorf("%s: %v", version, err))
-			continue
+			panic(err)
 		}
-		validator := kubernetes.NewValidator(reslover)
-
-		schema, err := reslover.Resolve(gf.APIKey(i.APIVersion, i.Kind))
+		i, err := loader.LoadManifest(obj)
 		if err != nil {
-			aggregatedErrors.Add(fmt.Errorf("%s: %v", version, err))
-			continue
+			_, ok := err.(*kubernetes.EmptyDocument)
+			if ok && *opts.ignoreEmpty {
+				continue
+			} else {
+				return &mainError{input.String(), err}
+			}
 		}
 
-		if len(aggregatedErrors.errors) > 0 {
-			return aggregatedErrors
-		}
+		aggregatedErrors := &aggErr{}
+		for _, version := range opts.Versions {
+			reslover, err := kubernetes.NewResolver(version)
+			if err != nil {
+				aggregatedErrors.Add(fmt.Errorf("%s: %v", version, err))
+				continue
+			}
+			validator := kubernetes.NewValidator(reslover)
+
+			schema, err := reslover.Resolve(gf.APIKey(i.APIVersion, i.Kind))
+			if err != nil {
+				aggregatedErrors.Add(fmt.Errorf("%s: %v", version, err))
+				continue
+			}
+
+			if len(aggregatedErrors.errors) > 0 {
+				return aggregatedErrors
+			}
 
-		errors := validator.Validate(i.Data, schema)
-		if len(errors) > 0 {
-			if !*opts.silent {
-				fmt.Fprintln(out, string(redbg(errors[0].Error())))
-				fmt.Fprintln(out, colorize(errors[0], input.Bytes()))
+			errors := validator.Validate(i.Data, schema)
+			if len(errors) > 0 {
+				if !*opts.silent {
+					fmt.Fprintln(out, string(redbg(errors[0].Error())))
+					fmt.Fprintln(out, colorize(errors[0], input.Bytes()))
+				}
+				return &aggErr{errors}
 			}
-			return &aggErr{errors}
 		}
 	}
 	return nil

backend/cmd/kubeyaml/kubeyaml_test.go

@@ -10,6 +10,7 @@ import (
 func TestIntegrations(t *testing.T) {
 	testcases := []struct {
 		filename       string
+		extraArgs      string
 		shouldValidate bool
 	}{
 		// missing a selector.
@@ -18,6 +19,13 @@ func TestIntegrations(t *testing.T) {
 		{filename: "issue-8.yaml", shouldValidate: false},
 		// type Airflow is invalid. But we don't validate data
 		{filename: "issue-9.yaml", shouldValidate: true},
+		// first document is valid, second is not
+		{filename: "issue-7.yaml", shouldValidate: false},
+		// first two documents are valid, but there is an empty one (which fails)
+		{filename: "issue-7_2.yaml", shouldValidate: false},
+		// unless -ignore-empty is set
+		{filename: "issue-7_2.yaml", extraArgs: "-ignore-empty", shouldValidate: true},
+		{filename: "kfserving.yaml", extraArgs: "-ignore-empty", shouldValidate: false},
 	}
 
 	for _, tc := range testcases {
@@ -27,7 +35,7 @@ func TestIntegrations(t *testing.T) {
 				t.Fatal(err)
 			}
 			var b bytes.Buffer
-			err = run(f, &b, "-silent")
+			err = run(f, &b, "-silent", tc.extraArgs)
 			if tc.shouldValidate && err != nil {
 				t.Fatal(err)
 			}

backend/cmd/kubeyaml/testdata/issue-7.yaml

@@ -0,0 +1,36 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  annotations:
+    service.beta.kubernetes.io/azure-load-balancer-internal: "true"
+  name: test
+  namespace: tester
+spec:
+  ports:
+    - name: http
+      port: 80
+      protocol: TCP
+      targetPort: 5000
+  selector:
+    app: tes
+  sessionAffinity: None
+  type: Airflow
+---
+apiVersion: v1
+kind: Service
+metadata:
+  annotations:
+    service.beta.kubernetes.io/azure-load-balancer-internal: "true"
+  name: test
+  namespace: tester
+INVALIDspec:
+  ports:
+    - name: http
+      port: 80
+      protocol: TCP
+      targetPort: 5000
+  selector:
+    app: tes
+  sessionAffinity: None
+  type: Airflow

backend/cmd/kubeyaml/testdata/issue-7_2.yaml

@@ -0,0 +1,38 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  annotations:
+    service.beta.kubernetes.io/azure-load-balancer-internal: "true"
+  name: test
+  namespace: tester
+spec:
+  ports:
+    - name: http
+      port: 80
+      protocol: TCP
+      targetPort: 5000
+  selector:
+    app: tes
+  sessionAffinity: None
+  type: Airflow
+---
+apiVersion: v1
+kind: Service
+metadata:
+  annotations:
+    service.beta.kubernetes.io/azure-load-balancer-internal: "true"
+  name: test
+  namespace: tester
+spec:
+  ports:
+    - name: http
+      port: 80
+      protocol: TCP
+      targetPort: 5000
+  selector:
+    app: tes
+  sessionAffinity: None
+  type: Airflow
+---
+

backend/internal/kubernetes/errors.go

@@ -156,3 +156,12 @@ func NewUnknownFormatError(format string) error {
 		Format: format,
 	}
 }
+
+type EmptyDocument struct{}
+
+func (e *EmptyDocument) Error() string {
+	return fmt.Sprintf("empty document")
+}
+func NewEmptyDocument() error {
+	return &EmptyDocument{}
+}

backend/internal/kubernetes/loader.go

@@ -23,17 +23,11 @@ func NewLoader() *Loader {
 	return &Loader{}
 }
 
-// Load reads the input and returns the internal type representing the top level document
+// LoadManifest returns the internal type representing the top level document
 // that is properly cleaned.
-func (l *Loader) Load(reader io.Reader) (*Input, error) {
-	b, err := ioutil.ReadAll(reader)
-	if err != nil {
-		return nil, fmt.Errorf("failed to read incoming reader: %v", err)
-	}
-
-	incoming := map[interface{}]interface{}{}
-	if err := yaml.Unmarshal(b, incoming); err != nil {
-		return nil, fmt.Errorf("failed to unmarshal yaml with error %v", err)
+func (l *Loader) LoadManifest(incoming map[interface{}]interface{}) (*Input, error) {
+	if incoming == nil {
+		return nil, NewEmptyDocument()
 	}
 
 	val, ok := incoming["apiVersion"]
@@ -64,3 +58,18 @@ func (l *Loader) Load(reader io.Reader) (*Input, error) {
 		Data:       incoming,
 	}, nil
 }
+
+// Load reads the input and returns the internal type representing the top level document
+// that is properly cleaned (via LoadManifest)
+func (l *Loader) Load(reader io.Reader) (*Input, error) {
+	b, err := ioutil.ReadAll(reader)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read incoming reader: %v", err)
+	}
+
+	incoming := map[interface{}]interface{}{}
+	if err := yaml.Unmarshal(b, incoming); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal yaml with error %v", err)
+	}
+	return l.LoadManifest(incoming)
+}