RecognizingDecoder didn't handle ambiguous input

YAML is not guaranteed to be recognizable, so we need to bump JSON and
protobuf above it in the decoding order. Add a unit test.

pkg/api/serialization_proto_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package api_test
 
 import (
+	"bytes"
 	"encoding/hex"
 	"math/rand"
 	"testing"
@@ -41,6 +42,28 @@ func init() {
 	})
 }
 
+func TestUniversalDeserializer(t *testing.T) {
+	expected := &v1.Pod{ObjectMeta: v1.ObjectMeta{Name: "test"}}
+	d := api.Codecs.UniversalDeserializer()
+	for _, mediaType := range []string{"application/json", "application/yaml", "application/vnd.kubernetes.protobuf"} {
+		e, ok := api.Codecs.SerializerForMediaType(mediaType, nil)
+		if !ok {
+			t.Fatal(mediaType)
+		}
+		buf := &bytes.Buffer{}
+		if err := e.EncodeToStream(expected, buf); err != nil {
+			t.Fatalf("%s: %v", mediaType, err)
+		}
+		obj, _, err := d.Decode(buf.Bytes(), &unversioned.GroupVersionKind{Kind: "Pod", Version: "v1"}, nil)
+		if err != nil {
+			t.Fatalf("%s: %v", mediaType, err)
+		}
+		if !api.Semantic.DeepEqual(expected, obj) {
+			t.Fatalf("%s: %#v", mediaType, obj)
+		}
+	}
+}
+
 func TestProtobufRoundTrip(t *testing.T) {
 	obj := &v1.Pod{}
 	apitesting.FuzzerFor(t, v1.SchemeGroupVersion, rand.NewSource(benchmarkSeed)).Fuzz(obj)

pkg/runtime/serializer/json/json.go

@@ -186,12 +186,13 @@ func (s *Serializer) EncodeToStream(obj runtime.Object, w io.Writer, overrides .
 }
 
 // RecognizesData implements the RecognizingDecoder interface.
-func (s *Serializer) RecognizesData(peek io.Reader) (bool, error) {
+func (s *Serializer) RecognizesData(peek io.Reader) (ok, unknown bool, err error) {
-	_, ok := utilyaml.GuessJSONStream(peek, 2048)
 	if s.yaml {
-		return !ok, nil
+		// we could potentially look for '---'
+		return false, true, nil
 	}
-	return ok, nil
+	_, ok = utilyaml.GuessJSONStream(peek, 2048)
+	return ok, false, nil
 }
 
 // EncodesAsText returns true because both JSON and YAML are considered textual representations

pkg/runtime/serializer/protobuf/protobuf.go

@@ -419,12 +419,6 @@ func (s *RawSerializer) EncodeToStream(obj runtime.Object, w io.Writer, override
 	}
 }
 
-// RecognizesData implements the RecognizingDecoder interface - objects encoded with this serializer
-// have no innate identifying information and so cannot be recognized.
-func (s *RawSerializer) RecognizesData(peek io.Reader) (bool, error) {
-	return false, nil
-}
-
 var LengthDelimitedFramer = lengthDelimitedFramer{}
 
 type lengthDelimitedFramer struct{}

pkg/runtime/serializer/recognizer/recognizer.go

@@ -27,51 +27,98 @@ import (
 
 type RecognizingDecoder interface {
 	runtime.Decoder
-	RecognizesData(peek io.Reader) (bool, error)
+	// RecognizesData should return true if the input provided in the provided reader
+	// belongs to this decoder, or an error if the data could not be read or is ambiguous.
+	// Unknown is true if the data could not be determined to match the decoder type.
+	// Decoders should assume that they can read as much of peek as they need (as the caller
+	// provides) and may return unknown if the data provided is not sufficient to make a
+	// a determination. When peek returns EOF that may mean the end of the input or the
+	// end of buffered input - recognizers should return the best guess at that time.
+	RecognizesData(peek io.Reader) (ok, unknown bool, err error)
 }
 
+// NewDecoder creates a decoder that will attempt multiple decoders in an order defined
+// by:
+//
+// 1. The decoder implements RecognizingDecoder and identifies the data
+// 2. All other decoders, and any decoder that returned true for unknown.
+//
+// The order passed to the constructor is preserved within those priorities.
 func NewDecoder(decoders ...runtime.Decoder) runtime.Decoder {
-	recognizing, blind := []RecognizingDecoder{}, []runtime.Decoder{}
-	for _, d := range decoders {
-		if r, ok := d.(RecognizingDecoder); ok {
-			recognizing = append(recognizing, r)
-		} else {
-			blind = append(blind, d)
-		}
-	}
 	return &decoder{
-		recognizing: recognizing,
+		decoders: decoders,
-		blind:       blind,
 	}
 }
 
 type decoder struct {
-	recognizing []RecognizingDecoder
+	decoders []runtime.Decoder
-	blind       []runtime.Decoder
+}
+
+var _ RecognizingDecoder = &decoder{}
+
+func (d *decoder) RecognizesData(peek io.Reader) (bool, bool, error) {
+	var (
+		lastErr    error
+		anyUnknown bool
+	)
+	data, _ := bufio.NewReaderSize(peek, 1024).Peek(1024)
+	for _, r := range d.decoders {
+		switch t := r.(type) {
+		case RecognizingDecoder:
+			ok, unknown, err := t.RecognizesData(bytes.NewBuffer(data))
+			if err != nil {
+				lastErr = err
+				continue
+			}
+			anyUnknown = anyUnknown || unknown
+			if !ok {
+				continue
+			}
+			return true, false, nil
+		}
+	}
+	return false, anyUnknown, lastErr
 }
 
 func (d *decoder) Decode(data []byte, gvk *unversioned.GroupVersionKind, into runtime.Object) (runtime.Object, *unversioned.GroupVersionKind, error) {
-	var lastErr error
+	var (
-	for _, r := range d.recognizing {
+		lastErr error
+		skipped []runtime.Decoder
+	)
+
+	// try recognizers, record any decoders we need to give a chance later
+	for _, r := range d.decoders {
+		switch t := r.(type) {
+		case RecognizingDecoder:
 			buf := bytes.NewBuffer(data)
-		ok, err := r.RecognizesData(buf)
+			ok, unknown, err := t.RecognizesData(buf)
 			if err != nil {
 				lastErr = err
 				continue
 			}
+			if unknown {
+				skipped = append(skipped, t)
+				continue
+			}
 			if !ok {
 				continue
 			}
 			return r.Decode(data, gvk, into)
+		default:
+			skipped = append(skipped, t)
 		}
-	for _, d := range d.blind {
+	}
-		out, actual, err := d.Decode(data, gvk, into)
+
+	// try recognizers that returned unknown or didn't recognize their data
+	for _, r := range skipped {
+		out, actual, err := r.Decode(data, gvk, into)
 		if err != nil {
 			lastErr = err
 			continue
 		}
 		return out, actual, nil
 	}
+
 	if lastErr == nil {
 		lastErr = fmt.Errorf("no serialization format matched the provided data")
 	}