Request and handle server-side printing when watching with kubectl

pkg/kubectl/cmd/get/BUILD

@@ -41,6 +41,7 @@ go_library(
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/meta:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/internalversion:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1/unstructured:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1beta1:go_default_library",

pkg/kubectl/cmd/get/get.go

@@ -29,6 +29,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	kapierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
+	metainternal "k8s.io/apimachinery/pkg/apis/meta/internalversion"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	metav1beta1 "k8s.io/apimachinery/pkg/apis/meta/v1beta1"
@@ -622,6 +623,7 @@ func (o *GetOptions) watch(f cmdutil.Factory, cmd *cobra.Command, args []string)
 		ResourceTypeOrNameArgs(true, args...).
 		SingleResourceType().
 		Latest().
+		TransformRequests(o.transformRequests).
 		Do()
 	if err := r.Err(); err != nil {
 		return err
@@ -662,6 +664,8 @@ func (o *GetOptions) watch(f cmdutil.Factory, cmd *cobra.Command, args []string)
 
 	writer := utilprinters.GetNewTabWriter(o.Out)
 
+	tableGK := metainternal.SchemeGroupVersion.WithKind("Table").GroupKind()
+
 	// print the current object
 	if !o.WatchOnly {
 		var objsToPrint []runtime.Object
@@ -672,8 +676,8 @@ func (o *GetOptions) watch(f cmdutil.Factory, cmd *cobra.Command, args []string)
 			objsToPrint = append(objsToPrint, obj)
 		}
 		for _, objToPrint := range objsToPrint {
-			if o.IsHumanReadablePrinter {
-				// printing always takes the internal version, but the watch event uses externals
+			if o.IsHumanReadablePrinter && objToPrint.GetObjectKind().GroupVersionKind().GroupKind() != tableGK {
+				// printing anything other than tables always takes the internal version, but the watch event uses externals
 				internalGV := mapping.GroupVersionKind.GroupKind().WithVersion(runtime.APIVersionInternal).GroupVersion()
 				objToPrint = attemptToConvertToInternal(objToPrint, legacyscheme.Scheme, internalGV)
 			}
@@ -705,7 +709,7 @@ func (o *GetOptions) watch(f cmdutil.Factory, cmd *cobra.Command, args []string)
 			// printing always takes the internal version, but the watch event uses externals
 			// TODO fix printing to use server-side or be version agnostic
 			objToPrint := e.Object
-			if o.IsHumanReadablePrinter {
+			if o.IsHumanReadablePrinter && objToPrint.GetObjectKind().GroupVersionKind().GroupKind() != tableGK {
 				internalGV := mapping.GroupVersionKind.GroupKind().WithVersion(runtime.APIVersionInternal).GroupVersion()
 				objToPrint = attemptToConvertToInternal(e.Object, legacyscheme.Scheme, internalGV)
 			}

pkg/kubectl/cmd/get/get_test.go

@@ -1403,6 +1403,113 @@ foo    0/0              0          <unknown>
 	}
 }
 
+func TestWatchResourceTable(t *testing.T) {
+	columns := []metav1beta1.TableColumnDefinition{
+		{Name: "Name", Type: "string", Format: "name", Description: "the name", Priority: 0},
+		{Name: "Active", Type: "boolean", Description: "active", Priority: 0},
+	}
+
+	listTable := &metav1beta1.Table{
+		TypeMeta:          metav1.TypeMeta{APIVersion: "meta.k8s.io/v1beta1", Kind: "Table"},
+		ColumnDefinitions: columns,
+		Rows: []metav1beta1.TableRow{
+			{
+				Cells: []interface{}{"a", true},
+				Object: runtime.RawExtension{
+					Object: &corev1.Pod{
+						TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
+						ObjectMeta: metav1.ObjectMeta{Name: "a", Namespace: "test", ResourceVersion: "10"},
+					},
+				},
+			},
+			{
+				Cells: []interface{}{"b", true},
+				Object: runtime.RawExtension{
+					Object: &corev1.Pod{
+						TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
+						ObjectMeta: metav1.ObjectMeta{Name: "b", Namespace: "test", ResourceVersion: "20"},
+					},
+				},
+			},
+		},
+	}
+
+	events := []watch.Event{
+		{
+			Type: watch.Added,
+			Object: &metav1beta1.Table{
+				TypeMeta:          metav1.TypeMeta{APIVersion: "meta.k8s.io/v1beta1", Kind: "Table"},
+				ColumnDefinitions: columns, // first event includes the columns
+				Rows: []metav1beta1.TableRow{{
+					Cells: []interface{}{"a", false},
+					Object: runtime.RawExtension{
+						Object: &corev1.Pod{
+							TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
+							ObjectMeta: metav1.ObjectMeta{Name: "a", Namespace: "test", ResourceVersion: "30"},
+						},
+					},
+				}},
+			},
+		},
+		{
+			Type: watch.Deleted,
+			Object: &metav1beta1.Table{
+				ColumnDefinitions: []metav1beta1.TableColumnDefinition{},
+				Rows: []metav1beta1.TableRow{{
+					Cells: []interface{}{"b", false},
+					Object: runtime.RawExtension{
+						Object: &corev1.Pod{
+							TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
+							ObjectMeta: metav1.ObjectMeta{Name: "b", Namespace: "test", ResourceVersion: "40"},
+						},
+					},
+				}},
+			},
+		},
+	}
+
+	tf := cmdtesting.NewTestFactory().WithNamespace("test")
+	defer tf.Cleanup()
+	codec := scheme.Codecs.LegacyCodec(scheme.Scheme.PrioritizedVersionsAllGroups()...)
+
+	tf.UnstructuredClient = &fake.RESTClient{
+		NegotiatedSerializer: resource.UnstructuredPlusDefaultContentConfig().NegotiatedSerializer,
+		Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
+			switch req.URL.Path {
+			case "/namespaces/test/pods":
+				if req.URL.Query().Get("watch") != "true" && req.URL.Query().Get("fieldSelector") == "" {
+					return &http.Response{StatusCode: 200, Header: cmdtesting.DefaultHeader(), Body: cmdtesting.ObjBody(codec, listTable)}, nil
+				}
+				if req.URL.Query().Get("watch") == "true" && req.URL.Query().Get("fieldSelector") == "" {
+					return &http.Response{StatusCode: 200, Header: cmdtesting.DefaultHeader(), Body: watchBody(codec, events)}, nil
+				}
+				t.Fatalf("request url: %#v,and request: %#v", req.URL, req)
+				return nil, nil
+			default:
+				t.Fatalf("request url: %#v,and request: %#v", req.URL, req)
+				return nil, nil
+			}
+		}),
+	}
+
+	streams, _, buf, _ := genericclioptions.NewTestIOStreams()
+	cmd := NewCmdGet("kubectl", tf, streams)
+	cmd.SetOutput(buf)
+
+	cmd.Flags().Set("watch", "true")
+	cmd.Run(cmd, []string{"pods"})
+
+	expected := `NAME   ACTIVE
+a      true
+b      true
+a      false
+b      false
+`
+	if e, a := expected, buf.String(); e != a {
+		t.Errorf("expected\n%v\ngot\n%v", e, a)
+	}
+}
+
 func TestWatchResourceIdentifiedByFile(t *testing.T) {
 	pods, events := watchTestData()
 
@@ -1538,7 +1645,9 @@ func watchBody(codec runtime.Codec, events []watch.Event) io.ReadCloser {
 	buf := bytes.NewBuffer([]byte{})
 	enc := restclientwatch.NewEncoder(streaming.NewEncoder(buf, codec), codec)
 	for i := range events {
-		enc.Encode(&events[i])
+		if err := enc.Encode(&events[i]); err != nil {
+			panic(err)
+		}
 	}
 	return json.Framer.NewFrameReader(ioutil.NopCloser(buf))
 }

pkg/printers/humanreadable.go

@@ -63,6 +63,7 @@ type HumanReadablePrinter struct {
 	defaultHandler *handlerEntry
 	options        PrintOptions
 	lastType       interface{}
+	lastColumns    []metav1beta1.TableColumnDefinition
 	skipTabWriter  bool
 	encoder        runtime.Encoder
 	decoder        runtime.Decoder
@@ -289,10 +290,26 @@ func (h *HumanReadablePrinter) PrintObj(obj runtime.Object, output io.Writer) er
 
 	// display tables following the rules of options
 	if table, ok := obj.(*metav1beta1.Table); ok {
-		if err := DecorateTable(table, h.options); err != nil {
+		// Do not print headers if this table has no column definitions, or they are the same as the last ones we printed
+		localOptions := h.options
+		if len(table.ColumnDefinitions) == 0 || reflect.DeepEqual(table.ColumnDefinitions, h.lastColumns) {
+			localOptions.NoHeaders = true
+		}
+
+		if len(table.ColumnDefinitions) == 0 {
+			// If this table has no column definitions, use the columns from the last table we printed for decoration and layout.
+			// This is done when receiving tables in watch events to save bandwidth.
+			localOptions.NoHeaders = true
+			table.ColumnDefinitions = h.lastColumns
+		} else {
+			// If this table has column definitions, remember them for future use.
+			h.lastColumns = table.ColumnDefinitions
+		}
+
+		if err := DecorateTable(table, localOptions); err != nil {
 			return err
 		}
-		return PrintTable(table, output, h.options)
+		return PrintTable(table, output, localOptions)
 	}
 
 	// check if the object is unstructured. If so, let's attempt to convert it to a type we can understand before