Better handling of YAML that tastes like JSON

For the most part, JSON is a subset of YAML.  This might lead one to
think that we should ALWAYS use YAML processing.  Unfortunately a JSON
"stream" (as defined by Go's encoding/json and many other places, though
not the JSON spec) is a series of JSON objects.  E.g. This:

```
{}{}{}
```

...is a valid JSON stream.

YAML does NOT accept that, insisting on `---` on a new line between YAML
documents.

Before this commit, YAMLOrJSONDecoder tries to detect if the input is
JSON by looking at the first few characters for "{".  Unfortunately,
some perfectly valid YAML also tastes like that.

After this commit, YAMLOrJSONDecoder will detect a failure to parse as
JSON and instead flip to YAML parsing.  This should handle the ambiguous
YAML.

Once we flip to YAML we never flip back, and once we detect a JSON
stream (as defined above) we lose the ability to flip to YAML.  A
multi-document is either all JSON or all YAML, even if we use the JSON
parser to decode the first object (because JSON is YAML for a single
object).

Author: thockin

staging/src/k8s.io/apimachinery/pkg/util/yaml/decoder.go

@@ -20,10 +20,12 @@ import (
 	"bufio"
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"strings"
 	"unicode"
+	"unicode/utf8"
 
 	jsonutil "k8s.io/apimachinery/pkg/util/json"
 
@@ -92,7 +94,7 @@ func UnmarshalStrict(data []byte, v interface{}) error {
 // YAML decoding path is not used (so that error messages are
 // JSON specific).
 func ToJSON(data []byte) ([]byte, error) {
-	if hasJSONPrefix(data) {
+	if IsJSONBuffer(data) {
 		return data, nil
 	}
 	return yaml.YAMLToJSON(data)
@@ -102,7 +104,8 @@ func ToJSON(data []byte) ([]byte, error) {
 // separating individual documents. It first converts the YAML
 // body to JSON, then unmarshals the JSON.
 type YAMLToJSONDecoder struct {
-	reader Reader
+	reader      Reader
+	inputOffset int
 }
 
 // NewYAMLToJSONDecoder decodes YAML documents from the provided
@@ -121,7 +124,7 @@ func NewYAMLToJSONDecoder(r io.Reader) *YAMLToJSONDecoder {
 // yaml.Unmarshal.
 func (d *YAMLToJSONDecoder) Decode(into interface{}) error {
 	bytes, err := d.reader.Read()
-	if err != nil && err != io.EOF {
+	if err != nil && err != io.EOF { //nolint:errorlint
 		return err
 	}
 
@@ -131,9 +134,14 @@ func (d *YAMLToJSONDecoder) Decode(into interface{}) error {
 			return YAMLSyntaxError{err}
 		}
 	}
+	d.inputOffset += len(bytes)
 	return err
 }
 
+func (d *YAMLToJSONDecoder) InputOffset() int {
+	return d.inputOffset
+}
+
 // YAMLDecoder reads chunks of objects and returns ErrShortBuffer if
 // the data is not sufficient.
 type YAMLDecoder struct {
@@ -229,18 +237,20 @@ func splitYAMLDocument(data []byte, atEOF bool) (advance int, token []byte, err
 	return 0, nil, nil
 }
 
-// decoder is a convenience interface for Decode.
-type decoder interface {
-	Decode(into interface{}) error
-}
-
-// YAMLOrJSONDecoder attempts to decode a stream of JSON documents or
-// YAML documents by sniffing for a leading { character.
+// YAMLOrJSONDecoder attempts to decode a stream of JSON or YAML documents.
+// While JSON is YAML, the way Go's JSON decode defines a multi-document stream
+// is a series of JSON objects (e.g. {}{}), but YAML defines a multi-document
+// stream as a series of documents separated by "---".
+//
+// This decoder will attempt to decode the stream as JSON first, and if that
+// fails, it will switch to YAML. Once it determines the stream is JSON (by
+// finding a non-YAML-delimited series of objects), it will not switch to YAML.
+// Once it switches to YAML it will not switch back to JSON.
 type YAMLOrJSONDecoder struct {
-	r          io.Reader
-	bufferSize int
-
-	decoder decoder
+	json   *json.Decoder
+	yaml   *YAMLToJSONDecoder
+	stream *StreamReader
+	count  int // how many objects have been decoded
 }
 
 type JSONSyntaxError struct {
@@ -265,31 +275,108 @@ func (e YAMLSyntaxError) Error() string {
 // how far into the stream the decoder will look to figure out whether this
 // is a JSON stream (has whitespace followed by an open brace).
 func NewYAMLOrJSONDecoder(r io.Reader, bufferSize int) *YAMLOrJSONDecoder {
-	return &YAMLOrJSONDecoder{
-		r:          r,
-		bufferSize: bufferSize,
+	d := &YAMLOrJSONDecoder{}
+
+	reader, _, mightBeJSON := GuessJSONStream(r, bufferSize)
+	d.stream = reader
+	if mightBeJSON {
+		d.json = json.NewDecoder(reader)
+	} else {
+		d.yaml = NewYAMLToJSONDecoder(reader)
 	}
+	return d
 }
 
 // Decode unmarshals the next object from the underlying stream into the
 // provide object, or returns an error.
 func (d *YAMLOrJSONDecoder) Decode(into interface{}) error {
-	if d.decoder == nil {
-		buffer, _, isJSON := GuessJSONStream(d.r, d.bufferSize)
-		if isJSON {
-			d.decoder = json.NewDecoder(buffer)
+	// Because we don't know if this is a JSON or YAML stream, a failure from
+	// both decoders is ambiguous. When in doubt, it will return the error from
+	// the JSON decoder.  Unfortunately, this means that if the first document
+	// is invalid YAML, the error won't be awesome.
+	// TODO: the errors from YAML are not great, we could improve them a lot.
+	var firstErr error
+	if d.json != nil {
+		err := d.json.Decode(into)
+		if err == nil {
+			d.stream.Consume(int(d.json.InputOffset()) - d.stream.Consumed())
+			d.count++
+			return nil
+		}
+		if err == io.EOF { //nolint:errorlint
+			return err
+		}
+		var syntax *json.SyntaxError
+		if ok := errors.As(err, &syntax); ok {
+			firstErr = JSONSyntaxError{
+				Offset: syntax.Offset,
+				Err:    syntax,
+			}
 		} else {
-			d.decoder = NewYAMLToJSONDecoder(buffer)
+			firstErr = err
+		}
+		if d.count > 1 {
+			// If we found 0 or 1 JSON object(s), this stream is still
+			// ambiguous.  But if we found more than 1 JSON object, then this
+			// is an unambiguous JSON stream, and we should not switch to YAML.
+			return err
+		}
+		// If JSON decoding hits the end of one object and then fails on the
+		// next, it leaves any leading whitespace in the buffer, which can
+		// confuse the YAML decoder. We just eat any whitespace we find, up to
+		// and including the first newline.
+		d.stream.Rewind()
+		if err := d.consumeWhitespace(); err == nil {
+			d.yaml = NewYAMLToJSONDecoder(d.stream)
+		}
+		d.json = nil
+	}
+	if d.yaml != nil {
+		err := d.yaml.Decode(into)
+		if err == nil {
+			d.stream.Consume(d.yaml.InputOffset() - d.stream.Consumed())
+			d.count++
+			return nil
+		}
+		if err == io.EOF { //nolint:errorlint
+			return err
+		}
+		if firstErr == nil {
+			firstErr = err
 		}
 	}
-	err := d.decoder.Decode(into)
-	if syntax, ok := err.(*json.SyntaxError); ok {
-		return JSONSyntaxError{
-			Offset: syntax.Offset,
-			Err:    syntax,
+	if firstErr != nil {
+		return firstErr
+	}
+	return fmt.Errorf("decoding failed as both JSON and YAML")
+}
+
+func (d *YAMLOrJSONDecoder) consumeWhitespace() error {
+	consumed := 0
+	for {
+		buf, err := d.stream.ReadN(4)
+		if err != nil && err == io.EOF { //nolint:errorlint
+			return err
+		}
+		r, sz := utf8.DecodeRune(buf)
+		if r == utf8.RuneError || sz == 0 {
+			return fmt.Errorf("invalid utf8 rune")
+		}
+		d.stream.RewindN(len(buf) - sz)
+		if !unicode.IsSpace(r) {
+			d.stream.RewindN(sz)
+			d.stream.Consume(consumed)
+			return nil
+		}
+		if r == '\n' {
+			d.stream.Consume(consumed)
+			return nil
+		}
+		if err == io.EOF { //nolint:errorlint
+			break
 		}
 	}
-	return err
+	return io.EOF
 }
 
 type Reader interface {
@@ -311,7 +398,7 @@ func (r *YAMLReader) Read() ([]byte, error) {
 	var buffer bytes.Buffer
 	for {
 		line, err := r.reader.Read()
-		if err != nil && err != io.EOF {
+		if err != nil && err != io.EOF { //nolint:errorlint
 			return nil, err
 		}
 
@@ -329,11 +416,11 @@ func (r *YAMLReader) Read() ([]byte, error) {
 			if buffer.Len() != 0 {
 				return buffer.Bytes(), nil
 			}
-			if err == io.EOF {
+			if err == io.EOF { //nolint:errorlint
 				return nil, err
 			}
 		}
-		if err == io.EOF {
+		if err == io.EOF { //nolint:errorlint
 			if buffer.Len() != 0 {
 				// If we're at EOF, we have a final, non-terminated line. Return it.
 				return buffer.Bytes(), nil
@@ -369,26 +456,20 @@ func (r *LineReader) Read() ([]byte, error) {
 // GuessJSONStream scans the provided reader up to size, looking
 // for an open brace indicating this is JSON. It will return the
 // bufio.Reader it creates for the consumer.
-func GuessJSONStream(r io.Reader, size int) (io.Reader, []byte, bool) {
-	buffer := bufio.NewReaderSize(r, size)
+func GuessJSONStream(r io.Reader, size int) (*StreamReader, []byte, bool) {
+	buffer := NewStreamReader(r, size)
 	b, _ := buffer.Peek(size)
-	return buffer, b, hasJSONPrefix(b)
+	return buffer, b, IsJSONBuffer(b)
 }
 
 // IsJSONBuffer scans the provided buffer, looking
 // for an open brace indicating this is JSON.
 func IsJSONBuffer(buf []byte) bool {
-	return hasJSONPrefix(buf)
+	return hasPrefix(buf, jsonPrefix)
 }
 
 var jsonPrefix = []byte("{")
 
-// hasJSONPrefix returns true if the provided buffer appears to start with
-// a JSON open brace.
-func hasJSONPrefix(buf []byte) bool {
-	return hasPrefix(buf, jsonPrefix)
-}
-
 // Return true if the first non-whitespace bytes in buf is
 // prefix.
 func hasPrefix(buf []byte, prefix []byte) bool {

staging/src/k8s.io/apimachinery/pkg/util/yaml/decoder_test.go

@@ -19,7 +19,6 @@ package yaml
 import (
 	"bufio"
 	"bytes"
-	"encoding/json"
 	"fmt"
 	"io"
 	"math/rand"
@@ -105,7 +104,7 @@ stuff: 1
 	}
 	b = make([]byte, 15)
 	n, err = r.Read(b)
-	if err != io.EOF || n != 0 {
+	if err != io.EOF || n != 0 { //nolint:errorlint
 		t.Fatalf("expected EOF: %d / %v", n, err)
 	}
 }
@@ -205,7 +204,7 @@ stuff: 1
 		t.Fatalf("unexpected object: %#v", obj)
 	}
 	obj = generic{}
-	if err := s.Decode(&obj); err != io.EOF {
+	if err := s.Decode(&obj); err != io.EOF { //nolint:errorlint
 		t.Fatalf("unexpected error: %v", err)
 	}
 }
@@ -319,6 +318,11 @@ func TestYAMLOrJSONDecoder(t *testing.T) {
 			{"foo": "bar"},
 			{"baz": "biz"},
 		}},
+		// Spaces for indent, tabs are not allowed in YAML.
+		{"foo:\n  field: bar\n---\nbaz:\n  field: biz", 100, false, false, []generic{
+			{"foo": map[string]any{"field": "bar"}},
+			{"baz": map[string]any{"field": "biz"}},
+		}},
 		{"foo: bar\n---\n", 100, false, false, []generic{
 			{"foo": "bar"},
 		}},
@@ -334,6 +338,38 @@ func TestYAMLOrJSONDecoder(t *testing.T) {
 		{"foo: bar\n", 100, false, false, []generic{
 			{"foo": "bar"},
 		}},
+		// First document is JSON, second is YAML
+		{"{\"foo\": \"bar\"}\n---\n{baz: biz}", 100, false, false, []generic{
+			{"foo": "bar"},
+			{"baz": "biz"},
+		}},
+		// First document is JSON, second is YAML, longer than the buffer
+		{"{\"foo\": \"bar\"}\n---\n{baz: biz0123456780123456780123456780123456780123456789}", 20, false, false, []generic{
+			{"foo": "bar"},
+			{"baz": "biz0123456780123456780123456780123456780123456789"},
+		}},
+		// First document is JSON, then whitespace, then YAML
+		{"{\"foo\": \"bar\"}    \n---\n{baz: biz}", 100, false, false, []generic{
+			{"foo": "bar"},
+			{"baz": "biz"},
+		}},
+		// First document is YAML, second is JSON
+		{"{foo: bar}\n---\n{\"baz\": \"biz\"}", 100, false, false, []generic{
+			{"foo": "bar"},
+			{"baz": "biz"},
+		}},
+		// First document is JSON, second is YAML, using spaces
+		{"{\n  \"foo\": \"bar\"\n}\n---\n{\n  baz: biz\n}", 100, false, false, []generic{
+			{"foo": "bar"},
+			{"baz": "biz"},
+		}},
+		// First document is JSON, second is YAML, using tabs
+		{"{\n\t\"foo\": \"bar\"\n}\n---\n{\n\tbaz: biz\n}", 100, false, false, []generic{
+			{"foo": "bar"},
+			{"baz": "biz"},
+		}},
+		// First 2 documents are JSON, third is YAML (stream is JSON)
+		{"{\"foo\": \"bar\"}\n{\"baz\": \"biz\"}\n---\n{qux: zrb}", 100, true, true, nil},
 	}
 	for i, testCase := range testCases {
 		decoder := NewYAMLOrJSONDecoder(bytes.NewReader([]byte(testCase.input)), testCase.buffer)
@@ -348,7 +384,7 @@ func TestYAMLOrJSONDecoder(t *testing.T) {
 			}
 			objs = append(objs, out)
 		}
-		if err != io.EOF {
+		if err != io.EOF { //nolint:errorlint
 			switch {
 			case testCase.err && err == nil:
 				t.Errorf("%d: unexpected non-error", i)
@@ -360,12 +396,12 @@ func TestYAMLOrJSONDecoder(t *testing.T) {
 				continue
 			}
 		}
-		switch decoder.decoder.(type) {
-		case *YAMLToJSONDecoder:
+		switch {
+		case decoder.yaml != nil:
 			if testCase.isJSON {
 				t.Errorf("%d: expected JSON decoder, got YAML", i)
 			}
-		case *json.Decoder:
+		case decoder.json != nil:
 			if !testCase.isJSON {
 				t.Errorf("%d: expected YAML decoder, got JSON", i)
 			}
@@ -419,7 +455,7 @@ func testReadLines(t *testing.T, lineLengths []int) {
 	var readLines [][]byte
 	for range lines {
 		bytes, err := lineReader.Read()
-		if err != nil && err != io.EOF {
+		if err != nil && err != io.EOF { //nolint:errorlint
 			t.Fatalf("failed to read lines: %v", err)
 		}
 		readLines = append(readLines, bytes)