Allow resource.Builder to stream YAML from the CLI

hack/test-cmd.sh

@@ -121,10 +121,10 @@ for version in "${kube_api_versions[@]}"; do
   kubectl get pods "${kube_flags[@]}"
   kubectl get pod redis-master "${kube_flags[@]}"
   [ "$(kubectl get pod redis-master -o template --output-version=v1beta1 -t '{{ .id }}' "${kube_flags[@]}")" == "redis-master" ]
-  output_pod=$(kubectl get pod redis-master -o json --output-version=v1beta1 "${kube_flags[@]}")
+  output_pod=$(kubectl get pod redis-master -o yaml --output-version=v1beta1 "${kube_flags[@]}")
   kubectl delete pod redis-master "${kube_flags[@]}"
   before="$(kubectl get pods -o template -t "{{ len .items }}" "${kube_flags[@]}")"
-  echo $output_pod | kubectl create -f - "${kube_flags[@]}"
+  echo "${output_pod}" | kubectl create -f - "${kube_flags[@]}"
   after="$(kubectl get pods -o template -t "{{ len .items }}" "${kube_flags[@]}")"
   [ "$((${after} - ${before}))" -eq 1 ]
   kubectl get pods -o yaml --output-version=v1beta1 "${kube_flags[@]}" | grep -q "id: redis-master"
@@ -135,6 +135,10 @@ for version in "${kube_api_versions[@]}"; do
   kubectl get services "${kube_flags[@]}"
   kubectl create -f examples/guestbook/frontend-service.json "${kube_flags[@]}"
   kubectl get services "${kube_flags[@]}"
+  output_service=$(kubectl get service frontend -o json --output-version=v1beta3 "${kube_flags[@]}")
+  kubectl delete service frontend "${kube_flags[@]}"
+  echo "${output_service}" | kubectl create -f - "${kube_flags[@]}"
+  kubectl get services "${kube_flags[@]}"
   kubectl delete service frontend "${kube_flags[@]}"
 
   kube::log::status "Testing kubectl(${version}:replicationcontrollers)"

pkg/kubectl/resource/builder_test.go

@@ -25,6 +25,8 @@ import (
 	"reflect"
 	"testing"
 
+	"github.com/ghodss/yaml"
+
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/latest"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
@@ -115,6 +117,26 @@ func streamTestData() (io.Reader, *api.PodList, *api.ServiceList) {
 	return r, pods, svc
 }
 
+func JSONToYAMLOrDie(in []byte) []byte {
+	data, err := yaml.JSONToYAML(in)
+	if err != nil {
+		panic(err)
+	}
+	return data
+}
+
+func streamYAMLTestData() (io.Reader, *api.PodList, *api.ServiceList) {
+	pods, svc := testData()
+	r, w := io.Pipe()
+	go func() {
+		defer w.Close()
+		w.Write(JSONToYAMLOrDie([]byte(runtime.EncodeOrDie(latest.Codec, pods))))
+		w.Write([]byte("\n---\n"))
+		w.Write(JSONToYAMLOrDie([]byte(runtime.EncodeOrDie(latest.Codec, svc))))
+	}()
+	return r, pods, svc
+}
+
 type testVisitor struct {
 	InjectErr error
 	Infos     []*Info
@@ -370,6 +392,23 @@ func TestStream(t *testing.T) {
 	}
 }
 
+func TestYAMLStream(t *testing.T) {
+	r, pods, rc := streamYAMLTestData()
+	b := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).
+		NamespaceParam("test").Stream(r, "STDIN").Flatten()
+
+	test := &testVisitor{}
+	singular := false
+
+	err := b.Do().IntoSingular(&singular).Visit(test.Handle)
+	if err != nil || singular || len(test.Infos) != 3 {
+		t.Fatalf("unexpected response: %v %f %#v", err, singular, test.Infos)
+	}
+	if !reflect.DeepEqual([]runtime.Object{&pods.Items[0], &pods.Items[1], &rc.Items[0]}, test.Objects()) {
+		t.Errorf("unexpected visited objects: %#v", test.Objects())
+	}
+}
+
 func TestMultipleObject(t *testing.T) {
 	r, pods, svc := streamTestData()
 	obj, err := NewBuilder(latest.RESTMapper, api.Scheme, fakeClient()).

pkg/kubectl/resource/visitor.go

@@ -17,7 +17,6 @@ limitations under the License.
 package resource
 
 import (
-	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -31,6 +30,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/meta"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/errors"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util/yaml"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
 )
 
@@ -344,7 +344,7 @@ func NewStreamVisitor(r io.Reader, mapper *Mapper, source string, ignoreErrors b
 
 // Visit implements Visitor over a stream.
 func (v *StreamVisitor) Visit(fn VisitorFunc) error {
-	d := json.NewDecoder(v.Reader)
+	d := yaml.NewYAMLOrJSONDecoder(v.Reader, 4096)
 	for {
 		ext := runtime.RawExtension{}
 		if err := d.Decode(&ext); err != nil {