Merge pull request #76161 from liggitt/kubectl-watch-table

use server-side printing in `kubectl get -w`

pkg/kubectl/cmd/get/BUILD

@@ -23,6 +23,7 @@ go_library(
         "get_flags.go",
         "humanreadable_flags.go",
         "sorter.go",
+        "table_printer.go",
     ],
     importpath = "k8s.io/kubernetes/pkg/kubectl/cmd/get",
     visibility = ["//visibility:public"],
@@ -40,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"
@@ -204,11 +205,11 @@ func (o *GetOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, args []stri
 		o.ExplicitNamespace = false
 	}
 
-	isSorting, err := cmd.Flags().GetString("sort-by")
+	sortBy, err := cmd.Flags().GetString("sort-by")
 	if err != nil {
 		return err
 	}
-	o.Sort = len(isSorting) > 0
+	o.Sort = len(sortBy) > 0
 
 	o.NoHeaders = cmdutil.GetFlagBool(cmd, "no-headers")
 
@@ -253,12 +254,20 @@ func (o *GetOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, args []stri
 			return nil, err
 		}
 
-		printer = maybeWrapSortingPrinter(printer, isSorting)
+		if o.Sort {
+			printer = &SortingPrinter{Delegate: printer, SortField: sortBy}
+		}
+		if o.ServerPrint {
+			printer = &TablePrinter{Delegate: printer}
+		}
 		return printer.PrintObj, nil
 	}
 
 	switch {
 	case o.Watch || o.WatchOnly:
+		if o.Sort {
+			fmt.Fprintf(o.IOStreams.ErrOut, "warning: --watch or --watch-only requested, --sort-by will be ignored\n")
+		}
 	default:
 		if len(args) == 0 && cmdutil.IsFilenameSliceEmpty(o.Filenames, o.Kustomize) {
 			fmt.Fprintf(o.ErrOut, "You must specify the type of resource to get. %s\n\n", cmdutil.SuggestAPIResources(o.CmdParent))
@@ -271,6 +280,12 @@ func (o *GetOptions) Complete(f cmdutil.Factory, cmd *cobra.Command, args []stri
 			return cmdutil.UsageErrorf(cmd, usageString)
 		}
 	}
+
+	// openapi printing is mutually exclusive with server side printing
+	if o.PrintWithOpenAPICols && o.ServerPrint {
+		fmt.Fprintf(o.IOStreams.ErrOut, "warning: --%s requested, --%s will be ignored\n", useOpenAPIPrintColumnFlagLabel, useServerPrintColumns)
+	}
+
 	return nil
 }
 
@@ -398,6 +413,27 @@ func NewRuntimeSorter(objects []runtime.Object, sortBy string) *RuntimeSorter {
 	}
 }
 
+func (o *GetOptions) transformRequests(req *rest.Request) {
+	// We need full objects if printing with openapi columns
+	if o.PrintWithOpenAPICols {
+		return
+	}
+	if !o.ServerPrint || !o.IsHumanReadablePrinter {
+		return
+	}
+
+	group := metav1beta1.GroupName
+	version := metav1beta1.SchemeGroupVersion.Version
+
+	tableParam := fmt.Sprintf("application/json;as=Table;v=%s;g=%s, application/json", version, group)
+	req.SetHeader("Accept", tableParam)
+
+	// if sorting, ensure we receive the full object in order to introspect its fields via jsonpath
+	if o.Sort {
+		req.Param("includeObject", "Object")
+	}
+}
+
 // Run performs the get operation.
 // TODO: remove the need to pass these arguments, like other commands.
 func (o *GetOptions) Run(f cmdutil.Factory, cmd *cobra.Command, args []string) error {
@@ -408,11 +444,6 @@ func (o *GetOptions) Run(f cmdutil.Factory, cmd *cobra.Command, args []string) e
 		return o.watch(f, cmd, args)
 	}
 
-	// openapi printing is mutually exclusive with server side printing
-	if o.PrintWithOpenAPICols && o.ServerPrint {
-		fmt.Fprintf(o.IOStreams.ErrOut, "warning: --%s requested, --%s will be ignored\n", useOpenAPIPrintColumnFlagLabel, useServerPrintColumns)
-	}
-
 	chunkSize := o.ChunkSize
 	if o.Sort {
 		// TODO(juanvallejo): in the future, we could have the client use chunking
@@ -432,26 +463,7 @@ func (o *GetOptions) Run(f cmdutil.Factory, cmd *cobra.Command, args []string) e
 		ContinueOnError().
 		Latest().
 		Flatten().
-		TransformRequests(func(req *rest.Request) {
+		TransformRequests(o.transformRequests).
-			// We need full objects if printing with openapi columns
-			if o.PrintWithOpenAPICols {
-				return
-			}
-			if !o.ServerPrint || !o.IsHumanReadablePrinter {
-				return
-			}
-
-			group := metav1beta1.GroupName
-			version := metav1beta1.SchemeGroupVersion.Version
-
-			tableParam := fmt.Sprintf("application/json;as=Table;v=%s;g=%s, application/json", version, group)
-			req.SetHeader("Accept", tableParam)
-
-			// if sorting, ensure we receive the full object in order to introspect its fields via jsonpath
-			if o.Sort {
-				req.Param("includeObject", "Object")
-			}
-		}).
 		Do()
 
 	if o.IgnoreNotFound {
@@ -475,17 +487,6 @@ func (o *GetOptions) Run(f cmdutil.Factory, cmd *cobra.Command, args []string) e
 
 	objs := make([]runtime.Object, len(infos))
 	for ix := range infos {
-		if o.ServerPrint {
-			table, err := o.decodeIntoTable(infos[ix].Object)
-			if err == nil {
-				infos[ix].Object = table
-			} else {
-				// if we are unable to decode server response into a v1beta1.Table,
-				// fallback to client-side printing with whatever info the server returned.
-				klog.V(2).Infof("Unable to decode server response into a Table. Falling back to hardcoded types: %v", err)
-			}
-		}
-
 		objs[ix] = infos[ix].Object
 	}
 
@@ -505,8 +506,11 @@ func (o *GetOptions) Run(f cmdutil.Factory, cmd *cobra.Command, args []string) e
 
 	var printer printers.ResourcePrinter
 	var lastMapping *meta.RESTMapping
-	nonEmptyObjCount := 0
+
-	w := utilprinters.GetNewTabWriter(o.Out)
+	// track if we write any output
+	trackingWriter := &trackingWriterWrapper{Delegate: o.Out}
+
+	w := utilprinters.GetNewTabWriter(trackingWriter)
 	for ix := range objs {
 		var mapping *meta.RESTMapping
 		var info *resource.Info
@@ -518,16 +522,6 @@ func (o *GetOptions) Run(f cmdutil.Factory, cmd *cobra.Command, args []string) e
 			mapping = info.Mapping
 		}
 
-		// if dealing with a table that has no rows, skip remaining steps
-		// and avoid printing an unnecessary newline
-		if table, isTable := info.Object.(*metav1beta1.Table); isTable {
-			if len(table.Rows) == 0 {
-				continue
-			}
-		}
-
-		nonEmptyObjCount++
-
 		printWithNamespace := o.AllNamespaces
 
 		if mapping != nil && mapping.Scope.Name() == meta.RESTScopeNameRoot {
@@ -574,12 +568,23 @@ func (o *GetOptions) Run(f cmdutil.Factory, cmd *cobra.Command, args []string) e
 		}
 	}
 	w.Flush()
-	if nonEmptyObjCount == 0 && !o.IgnoreNotFound && len(allErrs) == 0 {
+	if trackingWriter.Written == 0 && !o.IgnoreNotFound && len(allErrs) == 0 {
+		// if we wrote no output, and had no errors, and are not ignoring NotFound, be sure we output something
 		fmt.Fprintln(o.ErrOut, "No resources found.")
 	}
 	return utilerrors.NewAggregate(allErrs)
 }
 
+type trackingWriterWrapper struct {
+	Delegate io.Writer
+	Written  int
+}
+
+func (t *trackingWriterWrapper) Write(p []byte) (n int, err error) {
+	t.Written += len(p)
+	return t.Delegate.Write(p)
+}
+
 // raw makes a simple HTTP request to the provided path on the server using the default
 // credentials.
 func (o *GetOptions) raw(f cmdutil.Factory) error {
@@ -615,6 +620,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
@@ -655,6 +661,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
@@ -665,8 +673,8 @@ func (o *GetOptions) watch(f cmdutil.Factory, cmd *cobra.Command, args []string)
 			objsToPrint = append(objsToPrint, obj)
 		}
 		for _, objToPrint := range objsToPrint {
-			if o.IsHumanReadablePrinter {
+			if o.IsHumanReadablePrinter && objToPrint.GetObjectKind().GroupVersionKind().GroupKind() != tableGK {
-				// printing always takes the internal version, but the watch event uses externals
+				// 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)
 			}
@@ -698,7 +706,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)
 			}
@@ -723,35 +731,6 @@ func attemptToConvertToInternal(obj runtime.Object, converter runtime.ObjectConv
 	return internalObject
 }
 
-func (o *GetOptions) decodeIntoTable(obj runtime.Object) (runtime.Object, error) {
-	if obj.GetObjectKind().GroupVersionKind().Kind != "Table" {
-		return nil, fmt.Errorf("attempt to decode non-Table object into a v1beta1.Table")
-	}
-
-	unstr, ok := obj.(*unstructured.Unstructured)
-	if !ok {
-		return nil, fmt.Errorf("attempt to decode non-Unstructured object")
-	}
-	table := &metav1beta1.Table{}
-	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstr.Object, table); err != nil {
-		return nil, err
-	}
-
-	for i := range table.Rows {
-		row := &table.Rows[i]
-		if row.Object.Raw == nil || row.Object.Object != nil {
-			continue
-		}
-		converted, err := runtime.Decode(unstructured.UnstructuredJSONScheme, row.Object.Raw)
-		if err != nil {
-			return nil, err
-		}
-		row.Object.Object = converted
-	}
-
-	return table, nil
-}
-
 func (o *GetOptions) printGeneric(r *resource.Result) error {
 	// we flattened the data from the builder, so we have individual items, but now we'd like to either:
 	// 1. if there is more than one item, combine them all into a single list
@@ -863,16 +842,6 @@ func cmdSpecifiesOutputFmt(cmd *cobra.Command) bool {
 	return cmdutil.GetFlagString(cmd, "output") != ""
 }
 
-func maybeWrapSortingPrinter(printer printers.ResourcePrinter, sortBy string) printers.ResourcePrinter {
-	if len(sortBy) != 0 {
-		return &SortingPrinter{
-			Delegate:  printer,
-			SortField: fmt.Sprintf("%s", sortBy),
-		}
-	}
-	return printer
-}
-
 func multipleGVKsRequested(infos []*resource.Info) bool {
 	if len(infos) < 2 {
 		return false

pkg/kubectl/cmd/get/get_test.go

@@ -351,6 +351,44 @@ foo    0/0              0          <unknown>   <none>
 	}
 }
 
+func TestGetEmptyTable(t *testing.T) {
+	tf := cmdtesting.NewTestFactory().WithNamespace("test")
+	defer tf.Cleanup()
+
+	emptyTable := ioutil.NopCloser(bytes.NewBufferString(`{
+"kind":"Table",
+"apiVersion":"meta.k8s.io/v1beta1",
+"metadata":{
+	"selfLink":"/api/v1/namespaces/default/pods",
+	"resourceVersion":"346"
+},
+"columnDefinitions":[
+	{"name":"Name","type":"string","format":"name","description":"the name","priority":0}
+],
+"rows":[]
+}`))
+
+	tf.UnstructuredClient = &fake.RESTClient{
+		NegotiatedSerializer: resource.UnstructuredPlusDefaultContentConfig().NegotiatedSerializer,
+		Resp:                 &http.Response{StatusCode: 200, Header: cmdtesting.DefaultHeader(), Body: emptyTable},
+	}
+
+	streams, _, buf, errbuf := genericclioptions.NewTestIOStreams()
+	cmd := NewCmdGet("kubectl", tf, streams)
+	cmd.SetOutput(buf)
+	cmd.Run(cmd, []string{"pods"})
+
+	expected := ``
+	if e, a := expected, buf.String(); e != a {
+		t.Errorf("expected\n%v\ngot\n%v", e, a)
+	}
+	expectedErr := `No resources found.
+`
+	if e, a := expectedErr, errbuf.String(); e != a {
+		t.Errorf("expectedErr\n%v\ngot\n%v", e, a)
+	}
+}
+
 func TestGetObjectIgnoreNotFound(t *testing.T) {
 	cmdtesting.InitTestErrorHandler(t)
 
@@ -464,9 +502,49 @@ c      0/0              0          <unknown>
 	}
 }
 
+func TestGetSortedObjectsUnstructuredTable(t *testing.T) {
+	unstructuredMap, err := runtime.DefaultUnstructuredConverter.ToUnstructured(sortTestTableData()[0])
+	if err != nil {
+		t.Fatal(err)
+	}
+	unstructuredBytes, err := encjson.MarshalIndent(unstructuredMap, "", "  ")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// t.Log(string(unstructuredBytes))
+	body := ioutil.NopCloser(bytes.NewReader(unstructuredBytes))
+
+	tf := cmdtesting.NewTestFactory().WithNamespace("test")
+	defer tf.Cleanup()
+
+	tf.UnstructuredClient = &fake.RESTClient{
+		NegotiatedSerializer: resource.UnstructuredPlusDefaultContentConfig().NegotiatedSerializer,
+		Resp:                 &http.Response{StatusCode: 200, Header: cmdtesting.DefaultHeader(), Body: body},
+	}
+	tf.ClientConfigVal = &restclient.Config{ContentConfig: restclient.ContentConfig{GroupVersion: &corev1.SchemeGroupVersion}}
+
+	streams, _, buf, _ := genericclioptions.NewTestIOStreams()
+	cmd := NewCmdGet("kubectl", tf, streams)
+	cmd.SetOutput(buf)
+
+	// sorting with metedata.name
+	cmd.Flags().Set("sort-by", ".metadata.name")
+	cmd.Run(cmd, []string{"pods"})
+
+	expected := `NAME   CUSTOM
+a      custom-a
+b      custom-b
+c      custom-c
+`
+	if e, a := expected, buf.String(); e != a {
+		t.Errorf("expected\n%v\ngot\n%v", e, a)
+	}
+}
+
 func sortTestData() []runtime.Object {
 	return []runtime.Object{
 		&corev1.Pod{
+			TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
 			ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "test", ResourceVersion: "10"},
 			Spec: corev1.PodSpec{
 				RestartPolicy:                 corev1.RestartPolicyAlways,
@@ -477,6 +555,7 @@ func sortTestData() []runtime.Object {
 			},
 		},
 		&corev1.Pod{
+			TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
 			ObjectMeta: metav1.ObjectMeta{Name: "b", Namespace: "test", ResourceVersion: "11"},
 			Spec: corev1.PodSpec{
 				RestartPolicy:                 corev1.RestartPolicyAlways,
@@ -487,6 +566,7 @@ func sortTestData() []runtime.Object {
 			},
 		},
 		&corev1.Pod{
+			TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
 			ObjectMeta: metav1.ObjectMeta{Name: "a", Namespace: "test", ResourceVersion: "9"},
 			Spec: corev1.PodSpec{
 				RestartPolicy:                 corev1.RestartPolicyAlways,
@@ -502,11 +582,17 @@ func sortTestData() []runtime.Object {
 func sortTestTableData() []runtime.Object {
 	return []runtime.Object{
 		&metav1beta1.Table{
-			TypeMeta: metav1.TypeMeta{Kind: "Table"},
+			TypeMeta: metav1.TypeMeta{APIVersion: "meta.k8s.io/v1beta1", Kind: "Table"},
+			ColumnDefinitions: []metav1beta1.TableColumnDefinition{
+				{Name: "NAME", Type: "string", Format: "name"},
+				{Name: "CUSTOM", Type: "string", Format: ""},
+			},
 			Rows: []metav1beta1.TableRow{
 				{
+					Cells: []interface{}{"c", "custom-c"},
 					Object: runtime.RawExtension{
 						Object: &corev1.Pod{
+							TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
 							ObjectMeta: metav1.ObjectMeta{Name: "c", Namespace: "test", ResourceVersion: "10"},
 							Spec: corev1.PodSpec{
 								RestartPolicy:                 corev1.RestartPolicyAlways,
@@ -519,8 +605,10 @@ func sortTestTableData() []runtime.Object {
 					},
 				},
 				{
+					Cells: []interface{}{"b", "custom-b"},
 					Object: runtime.RawExtension{
 						Object: &corev1.Pod{
+							TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
 							ObjectMeta: metav1.ObjectMeta{Name: "b", Namespace: "test", ResourceVersion: "11"},
 							Spec: corev1.PodSpec{
 								RestartPolicy:                 corev1.RestartPolicyAlways,
@@ -533,8 +621,10 @@ func sortTestTableData() []runtime.Object {
 					},
 				},
 				{
+					Cells: []interface{}{"a", "custom-a"},
 					Object: runtime.RawExtension{
 						Object: &corev1.Pod{
+							TypeMeta:   metav1.TypeMeta{APIVersion: "v1", Kind: "Pod"},
 							ObjectMeta: metav1.ObjectMeta{Name: "a", Namespace: "test", ResourceVersion: "9"},
 							Spec: corev1.PodSpec{
 								RestartPolicy:                 corev1.RestartPolicyAlways,
@@ -1313,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()
 
@@ -1448,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/kubectl/cmd/get/sorter.go

@@ -46,13 +46,27 @@ type SortingPrinter struct {
 }
 
 func (s *SortingPrinter) PrintObj(obj runtime.Object, out io.Writer) error {
-	if !meta.IsListType(obj) {
+	if table, isTable := obj.(*metav1beta1.Table); isTable && len(table.Rows) > 1 {
+		parsedField, err := RelaxedJSONPathExpression(s.SortField)
+		if err != nil {
+			parsedField = s.SortField
+		}
+
+		if sorter, err := NewTableSorter(table, parsedField); err != nil {
+			return err
+		} else if err := sorter.Sort(); err != nil {
+			return err
+		}
+		return s.Delegate.PrintObj(table, out)
+	}
+
+	if meta.IsListType(obj) {
+		if err := s.sortObj(obj); err != nil {
+			return err
+		}
 		return s.Delegate.PrintObj(obj, out)
 	}
 
-	if err := s.sortObj(obj); err != nil {
-		return err
-	}
 	return s.Delegate.PrintObj(obj, out)
 }
 

pkg/kubectl/cmd/get/table_printer.go

@@ -0,0 +1,77 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package get
+
+import (
+	"fmt"
+	"io"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	metav1beta1 "k8s.io/apimachinery/pkg/apis/meta/v1beta1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/cli-runtime/pkg/printers"
+	"k8s.io/klog"
+)
+
+// TablePrinter decodes table objects into typed objects before delegating to another printer.
+// Non-table types are simply passed through
+type TablePrinter struct {
+	Delegate printers.ResourcePrinter
+}
+
+func (t *TablePrinter) PrintObj(obj runtime.Object, writer io.Writer) error {
+	table, err := decodeIntoTable(obj)
+	if err == nil {
+		return t.Delegate.PrintObj(table, writer)
+	}
+	// if we are unable to decode server response into a v1beta1.Table,
+	// fallback to client-side printing with whatever info the server returned.
+	klog.V(2).Infof("Unable to decode server response into a Table. Falling back to hardcoded types: %v", err)
+	return t.Delegate.PrintObj(obj, writer)
+}
+
+func decodeIntoTable(obj runtime.Object) (runtime.Object, error) {
+	if obj.GetObjectKind().GroupVersionKind().Group != metav1beta1.GroupName {
+		return nil, fmt.Errorf("attempt to decode non-Table object into a v1beta1.Table")
+	}
+	if obj.GetObjectKind().GroupVersionKind().Kind != "Table" {
+		return nil, fmt.Errorf("attempt to decode non-Table object into a v1beta1.Table")
+	}
+
+	unstr, ok := obj.(*unstructured.Unstructured)
+	if !ok {
+		return nil, fmt.Errorf("attempt to decode non-Unstructured object")
+	}
+	table := &metav1beta1.Table{}
+	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(unstr.Object, table); err != nil {
+		return nil, err
+	}
+
+	for i := range table.Rows {
+		row := &table.Rows[i]
+		if row.Object.Raw == nil || row.Object.Object != nil {
+			continue
+		}
+		converted, err := runtime.Decode(unstructured.UnstructuredJSONScheme, row.Object.Raw)
+		if err != nil {
+			return nil, err
+		}
+		row.Object.Object = converted
+	}
+
+	return table, nil
+}

pkg/kubectl/scheme/BUILD

@@ -38,6 +38,7 @@ go_library(
         "//staging/src/k8s.io/api/storage/v1beta1: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",
         "//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/serializer:go_default_library",

pkg/kubectl/scheme/install.go

@@ -45,6 +45,7 @@ import (
 	storagev1 "k8s.io/api/storage/v1"
 	storagev1beta1 "k8s.io/api/storage/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	metav1beta1 "k8s.io/apimachinery/pkg/apis/meta/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -56,6 +57,7 @@ import (
 func init() {
 	// Register external types for Scheme
 	metav1.AddToGroupVersion(Scheme, schema.GroupVersion{Version: "v1"})
+	utilruntime.Must(metav1beta1.AddToScheme(Scheme))
 	utilruntime.Must(scheme.AddToScheme(Scheme))
 
 	utilruntime.Must(Scheme.SetVersionPriority(corev1.SchemeGroupVersion))

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

staging/src/k8s.io/apimachinery/pkg/apis/meta/v1beta1/register.go

@@ -39,6 +39,12 @@ var scheme = runtime.NewScheme()
 var ParameterCodec = runtime.NewParameterCodec(scheme)
 
 func init() {
+	if err := AddToScheme(scheme); err != nil {
+		panic(err)
+	}
+}
+
+func AddToScheme(scheme *runtime.Scheme) error {
 	scheme.AddKnownTypes(SchemeGroupVersion,
 		&Table{},
 		&TableOptions{},
@@ -46,11 +52,9 @@ func init() {
 		&PartialObjectMetadataList{},
 	)
 
-	if err := scheme.AddConversionFuncs(
+	return scheme.AddConversionFuncs(
 		Convert_Slice_string_To_v1beta1_IncludeObjectPolicy,
-	); err != nil {
+	)
-		panic(err)
-	}
 
 	// register manually. This usually goes through the SchemeBuilder, which we cannot use here.
 	//scheme.AddGeneratedDeepCopyFuncs(GetGeneratedDeepCopyFuncs()...)