Add ability to output watch events from kubectl get

pkg/kubectl/cmd/get/customcolumn.go

@@ -28,6 +28,8 @@ import (
 	"github.com/liggitt/tabwriter"
 
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/cli-runtime/pkg/printers"
 	"k8s.io/client-go/util/jsonpath"
@@ -205,6 +207,21 @@ func (s *CustomColumnsPrinter) PrintObj(obj runtime.Object, out io.Writer) error
 func (s *CustomColumnsPrinter) printOneObject(obj runtime.Object, parsers []*jsonpath.JSONPath, out io.Writer) error {
 	columns := make([]string, len(parsers))
 	switch u := obj.(type) {
+	case *metav1.WatchEvent:
+		if printers.InternalObjectPreventer.IsForbidden(reflect.Indirect(reflect.ValueOf(u.Object.Object)).Type().PkgPath()) {
+			return fmt.Errorf(printers.InternalObjectPrinterErr)
+		}
+		unstructuredObject, err := runtime.DefaultUnstructuredConverter.ToUnstructured(u.Object.Object)
+		if err != nil {
+			return err
+		}
+		obj = &unstructured.Unstructured{
+			Object: map[string]interface{}{
+				"type":   u.Type,
+				"object": unstructuredObject,
+			},
+		}
+
 	case *runtime.Unknown:
 		if len(u.Raw) > 0 {
 			var err error

pkg/kubectl/cmd/get/get.go

@@ -65,6 +65,8 @@ type GetOptions struct {
 	WatchOnly bool
 	ChunkSize int64
 
+	OutputWatchEvents bool
+
 	LabelSelector     string
 	FieldSelector     string
 	AllNamespaces     bool
@@ -171,6 +173,7 @@ func NewCmdGet(parent string, f cmdutil.Factory, streams genericclioptions.IOStr
 	cmd.Flags().StringVar(&o.Raw, "raw", o.Raw, "Raw URI to request from the server.  Uses the transport specified by the kubeconfig file.")
 	cmd.Flags().BoolVarP(&o.Watch, "watch", "w", o.Watch, "After listing/getting the requested object, watch for changes. Uninitialized objects are excluded if no object name is provided.")
 	cmd.Flags().BoolVar(&o.WatchOnly, "watch-only", o.WatchOnly, "Watch for changes to the requested object(s), without listing/getting first.")
+	cmd.Flags().BoolVar(&o.OutputWatchEvents, "output-watch-events", o.OutputWatchEvents, "Output watch event objects when --watch or --watch-only is used. Existing objects are output as initial ADDED events.")
 	cmd.Flags().Int64Var(&o.ChunkSize, "chunk-size", o.ChunkSize, "Return large lists in chunks rather than all at once. Pass 0 to disable. This flag is beta and may change in the future.")
 	cmd.Flags().BoolVar(&o.IgnoreNotFound, "ignore-not-found", o.IgnoreNotFound, "If the requested object does not exist the command will return exit code 0.")
 	cmd.Flags().StringVarP(&o.LabelSelector, "selector", "l", o.LabelSelector, "Selector (label query) to filter on, supports '=', '==', and '!='.(e.g. -l key1=value1,key2=value2)")
@@ -309,6 +312,9 @@ func (o *GetOptions) Validate(cmd *cobra.Command) error {
 			return fmt.Errorf("--show-labels option cannot be used with %s printer", outputOption)
 		}
 	}
+	if o.OutputWatchEvents && !(o.Watch || o.WatchOnly) {
+		return cmdutil.UsageErrorf(cmd, "--output-watch-events option can only be used with --watch or --watch-only")
+	}
 	return nil
 }
 
@@ -664,6 +670,9 @@ func (o *GetOptions) watch(f cmdutil.Factory, cmd *cobra.Command, args []string)
 		objsToPrint = append(objsToPrint, obj)
 	}
 	for _, objToPrint := range objsToPrint {
+		if o.OutputWatchEvents {
+			objToPrint = &metav1.WatchEvent{Type: string(watch.Added), Object: runtime.RawExtension{Object: objToPrint}}
+		}
 		if err := printer.PrintObj(objToPrint, writer); err != nil {
 			return fmt.Errorf("unable to output the provided object: %v", err)
 		}
@@ -688,7 +697,11 @@ func (o *GetOptions) watch(f cmdutil.Factory, cmd *cobra.Command, args []string)
 	intr := interrupt.New(nil, cancel)
 	intr.Run(func() error {
 		_, err := watchtools.UntilWithoutRetry(ctx, w, func(e watch.Event) (bool, error) {
-			if err := printer.PrintObj(e.Object, writer); err != nil {
+			objToPrint := e.Object
+			if o.OutputWatchEvents {
+				objToPrint = &metav1.WatchEvent{Type: string(e.Type), Object: runtime.RawExtension{Object: objToPrint}}
+			}
+			if err := printer.PrintObj(objToPrint, writer); err != nil {
 				return false, err
 			}
 			writer.Flush()

pkg/kubectl/cmd/get/get_test.go

@@ -2052,6 +2052,220 @@ b      false
 	}
 }
 
+func TestWatchResourceWatchEvents(t *testing.T) {
+
+	testcases := []struct {
+		format   string
+		table    bool
+		expected string
+	}{
+		{
+			format: "",
+			expected: `EVENT      NAMESPACE   NAME      AGE
+ADDED      test        pod/bar   <unknown>
+ADDED      test        pod/foo   <unknown>
+MODIFIED   test        pod/foo   <unknown>
+DELETED    test        pod/foo   <unknown>
+`,
+		},
+		{
+			format: "",
+			table:  true,
+			expected: `EVENT      NAMESPACE   NAME      READY   STATUS   RESTARTS   AGE
+ADDED      test        pod/bar   0/0              0          <unknown>
+ADDED      test        pod/foo   0/0              0          <unknown>
+MODIFIED   test        pod/foo   0/0              0          <unknown>
+DELETED    test        pod/foo   0/0              0          <unknown>
+`,
+		},
+		{
+			format: "wide",
+			table:  true,
+			expected: `EVENT      NAMESPACE   NAME      READY   STATUS   RESTARTS   AGE         IP       NODE     NOMINATED NODE   READINESS GATES
+ADDED      test        pod/bar   0/0              0          <unknown>   <none>   <none>   <none>           <none>
+ADDED      test        pod/foo   0/0              0          <unknown>   <none>   <none>   <none>           <none>
+MODIFIED   test        pod/foo   0/0              0          <unknown>   <none>   <none>   <none>           <none>
+DELETED    test        pod/foo   0/0              0          <unknown>   <none>   <none>   <none>           <none>
+`,
+		},
+		{
+			format: "json",
+			expected: `{"type":"ADDED","object":{"apiVersion":"v1","kind":"Pod","metadata":{"creationTimestamp":null,"name":"bar","namespace":"test","resourceVersion":"9"},"spec":{"containers":null,"dnsPolicy":"ClusterFirst","enableServiceLinks":true,"restartPolicy":"Always","securityContext":{},"terminationGracePeriodSeconds":30},"status":{}}}
+{"type":"ADDED","object":{"apiVersion":"v1","kind":"Pod","metadata":{"creationTimestamp":null,"name":"foo","namespace":"test","resourceVersion":"10"},"spec":{"containers":null,"dnsPolicy":"ClusterFirst","enableServiceLinks":true,"restartPolicy":"Always","securityContext":{},"terminationGracePeriodSeconds":30},"status":{}}}
+{"type":"MODIFIED","object":{"apiVersion":"v1","kind":"Pod","metadata":{"creationTimestamp":null,"name":"foo","namespace":"test","resourceVersion":"11"},"spec":{"containers":null,"dnsPolicy":"ClusterFirst","enableServiceLinks":true,"restartPolicy":"Always","securityContext":{},"terminationGracePeriodSeconds":30},"status":{}}}
+{"type":"DELETED","object":{"apiVersion":"v1","kind":"Pod","metadata":{"creationTimestamp":null,"name":"foo","namespace":"test","resourceVersion":"12"},"spec":{"containers":null,"dnsPolicy":"ClusterFirst","enableServiceLinks":true,"restartPolicy":"Always","securityContext":{},"terminationGracePeriodSeconds":30},"status":{}}}
+`,
+		},
+		{
+			format: "yaml",
+			expected: `object:
+  apiVersion: v1
+  kind: Pod
+  metadata:
+    creationTimestamp: null
+    name: bar
+    namespace: test
+    resourceVersion: "9"
+  spec:
+    containers: null
+    dnsPolicy: ClusterFirst
+    enableServiceLinks: true
+    restartPolicy: Always
+    securityContext: {}
+    terminationGracePeriodSeconds: 30
+  status: {}
+type: ADDED
+---
+object:
+  apiVersion: v1
+  kind: Pod
+  metadata:
+    creationTimestamp: null
+    name: foo
+    namespace: test
+    resourceVersion: "10"
+  spec:
+    containers: null
+    dnsPolicy: ClusterFirst
+    enableServiceLinks: true
+    restartPolicy: Always
+    securityContext: {}
+    terminationGracePeriodSeconds: 30
+  status: {}
+type: ADDED
+---
+object:
+  apiVersion: v1
+  kind: Pod
+  metadata:
+    creationTimestamp: null
+    name: foo
+    namespace: test
+    resourceVersion: "11"
+  spec:
+    containers: null
+    dnsPolicy: ClusterFirst
+    enableServiceLinks: true
+    restartPolicy: Always
+    securityContext: {}
+    terminationGracePeriodSeconds: 30
+  status: {}
+type: MODIFIED
+---
+object:
+  apiVersion: v1
+  kind: Pod
+  metadata:
+    creationTimestamp: null
+    name: foo
+    namespace: test
+    resourceVersion: "12"
+  spec:
+    containers: null
+    dnsPolicy: ClusterFirst
+    enableServiceLinks: true
+    restartPolicy: Always
+    securityContext: {}
+    terminationGracePeriodSeconds: 30
+  status: {}
+type: DELETED
+`,
+		},
+		{
+			format: `jsonpath={.type},{.object.metadata.name},{.object.metadata.resourceVersion}{"\n"}`,
+			expected: `ADDED,bar,9
+ADDED,foo,10
+MODIFIED,foo,11
+DELETED,foo,12
+`,
+		},
+		{
+			format: `go-template={{.type}},{{.object.metadata.name}},{{.object.metadata.resourceVersion}}{{"\n"}}`,
+			expected: `ADDED,bar,9
+ADDED,foo,10
+MODIFIED,foo,11
+DELETED,foo,12
+`,
+		},
+		{
+			format: `custom-columns=TYPE:.type,NAME:.object.metadata.name,RSRC:.object.metadata.resourceVersion`,
+			expected: `TYPE    NAME   RSRC
+ADDED   bar    9
+ADDED   foo    10
+MODIFIED   foo    11
+DELETED    foo    12
+`,
+		},
+		{
+			format: `name`,
+			expected: `pod/bar
+pod/foo
+pod/foo
+pod/foo
+`,
+		},
+	}
+
+	for _, tc := range testcases {
+		t.Run(fmt.Sprintf("%s, table=%v", tc.format, tc.table), func(t *testing.T) {
+			pods, events := watchTestData()
+
+			tf := cmdtesting.NewTestFactory().WithNamespace("test")
+			defer tf.Cleanup()
+			codec := scheme.Codecs.LegacyCodec(scheme.Scheme.PrioritizedVersionsAllGroups()...)
+
+			podList := &corev1.PodList{
+				Items: pods,
+				ListMeta: metav1.ListMeta{
+					ResourceVersion: "10",
+				},
+			}
+
+			tf.UnstructuredClient = &fake.RESTClient{
+				NegotiatedSerializer: resource.UnstructuredPlusDefaultContentConfig().NegotiatedSerializer,
+				Client: fake.CreateHTTPClient(func(req *http.Request) (*http.Response, error) {
+					switch req.URL.Path {
+					case "/pods":
+						if req.URL.Query().Get("watch") == "true" {
+							if tc.table {
+								return &http.Response{StatusCode: 200, Header: cmdtesting.DefaultHeader(), Body: podTableWatchBody(codec, events[2:])}, nil
+							} else {
+								return &http.Response{StatusCode: 200, Header: cmdtesting.DefaultHeader(), Body: watchBody(codec, events[2:])}, nil
+							}
+						}
+
+						if tc.table {
+							return &http.Response{StatusCode: 200, Header: cmdtesting.DefaultHeader(), Body: podTableObjBody(codec, podList.Items...)}, nil
+						} else {
+							return &http.Response{StatusCode: 200, Header: cmdtesting.DefaultHeader(), Body: cmdtesting.ObjBody(codec, podList)}, 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.Flags().Set("all-namespaces", "true")
+			cmd.Flags().Set("show-kind", "true")
+			cmd.Flags().Set("output-watch-events", "true")
+			if len(tc.format) > 0 {
+				cmd.Flags().Set("output", tc.format)
+			}
+
+			cmd.Run(cmd, []string{"pods"})
+			if e, a := tc.expected, buf.String(); e != a {
+				t.Errorf("expected\n%v\ngot\n%v", e, a)
+			}
+		})
+	}
+}
+
 func TestWatchResourceIdentifiedByFile(t *testing.T) {
 	pods, events := watchTestData()
 

pkg/kubectl/cmd/get/table_printer.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"io"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	metav1beta1 "k8s.io/apimachinery/pkg/apis/meta/v1beta1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -45,6 +46,11 @@ func (t *TablePrinter) PrintObj(obj runtime.Object, writer io.Writer) error {
 }
 
 func decodeIntoTable(obj runtime.Object) (runtime.Object, error) {
+	event, isEvent := obj.(*metav1.WatchEvent)
+	if isEvent {
+		obj = event.Object.Object
+	}
+
 	if obj.GetObjectKind().GroupVersionKind().Group != metav1beta1.GroupName {
 		return nil, fmt.Errorf("attempt to decode non-Table object into a v1beta1.Table")
 	}
@@ -73,5 +79,9 @@ func decodeIntoTable(obj runtime.Object) (runtime.Object, error) {
 		row.Object.Object = converted
 	}
 
+	if isEvent {
+		event.Object.Object = table
+		return event, nil
+	}
 	return table, nil
 }

pkg/printers/BUILD

@@ -24,6 +24,7 @@ go_library(
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/duration:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/runtime:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/watch:go_default_library",
         "//vendor/github.com/liggitt/tabwriter:go_default_library",
     ],
 )

pkg/printers/tableprinter.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/duration"
+	"k8s.io/apimachinery/pkg/watch"
 )
 
 var _ ResourcePrinter = &HumanReadablePrinter{}
@@ -56,6 +57,7 @@ var (
 		{Name: "Age", Type: "string", Description: metav1.ObjectMeta{}.SwaggerDoc()["creationTimestamp"]},
 	}
 
+	withEventTypePrefixColumns = []string{"EVENT"}
 	withNamespacePrefixColumns = []string{"NAMESPACE"} // TODO(erictune): print cluster name too.
 )
 
@@ -86,6 +88,12 @@ func (h *HumanReadablePrinter) PrintObj(obj runtime.Object, output io.Writer) er
 		defer w.Flush()
 	}
 
+	var eventType string
+	if event, isEvent := obj.(*metav1.WatchEvent); isEvent {
+		eventType = event.Type
+		obj = event.Object.Object
+	}
+
 	// Case 1: Parameter "obj" is a table from server; print it.
 	// display tables following the rules of options
 	if table, ok := obj.(*metav1beta1.Table); ok {
@@ -112,6 +120,14 @@ func (h *HumanReadablePrinter) PrintObj(obj runtime.Object, output io.Writer) er
 		if err := decorateTable(table, localOptions); err != nil {
 			return err
 		}
+		if len(eventType) > 0 {
+			if err := addColumns(beginning, table,
+				[]metav1beta1.TableColumnDefinition{{Name: "Event", Type: "string"}},
+				[]cellValueFunc{func(metav1beta1.TableRow) (interface{}, error) { return formatEventType(eventType), nil }},
+			); err != nil {
+				return err
+			}
+		}
 		return printTable(table, output, localOptions)
 	}
 
@@ -126,7 +142,7 @@ func (h *HumanReadablePrinter) PrintObj(obj runtime.Object, output io.Writer) er
 			fmt.Fprintln(output)
 		}
 
-		if err := printRowsForHandlerEntry(output, handler, obj, h.options, includeHeaders); err != nil {
+		if err := printRowsForHandlerEntry(output, handler, eventType, obj, h.options, includeHeaders); err != nil {
 			return err
 		}
 		h.lastType = t
@@ -148,7 +164,7 @@ func (h *HumanReadablePrinter) PrintObj(obj runtime.Object, output io.Writer) er
 		fmt.Fprintln(output)
 	}
 
-	if err := printRowsForHandlerEntry(output, handler, obj, h.options, includeHeaders); err != nil {
+	if err := printRowsForHandlerEntry(output, handler, eventType, obj, h.options, includeHeaders); err != nil {
 		return err
 	}
 	h.lastType = handler
@@ -206,6 +222,75 @@ func printTable(table *metav1beta1.Table, output io.Writer, options PrintOptions
 	return nil
 }
 
+type cellValueFunc func(metav1beta1.TableRow) (interface{}, error)
+
+type columnAddPosition int
+
+const (
+	beginning columnAddPosition = 1
+	end       columnAddPosition = 2
+)
+
+func addColumns(pos columnAddPosition, table *metav1beta1.Table, columns []metav1beta1.TableColumnDefinition, valueFuncs []cellValueFunc) error {
+	if len(columns) != len(valueFuncs) {
+		return fmt.Errorf("cannot prepend columns, unmatched value functions")
+	}
+	if len(columns) == 0 {
+		return nil
+	}
+
+	// Compute the new rows
+	newRows := make([][]interface{}, len(table.Rows))
+	for i := range table.Rows {
+		newCells := make([]interface{}, 0, len(columns)+len(table.Rows[i].Cells))
+
+		if pos == end {
+			// If we're appending, start with the existing cells,
+			// then add nil cells to match the number of columns
+			newCells = append(newCells, table.Rows[i].Cells...)
+			for len(newCells) < len(table.ColumnDefinitions) {
+				newCells = append(newCells, nil)
+			}
+		}
+
+		// Compute cells for new columns
+		for _, f := range valueFuncs {
+			newCell, err := f(table.Rows[i])
+			if err != nil {
+				return err
+			}
+			newCells = append(newCells, newCell)
+		}
+
+		if pos == beginning {
+			// If we're prepending, add existing cells
+			newCells = append(newCells, table.Rows[i].Cells...)
+		}
+
+		// Remember the new cells for this row
+		newRows[i] = newCells
+	}
+
+	// All cells successfully computed, now replace columns and rows
+	newColumns := make([]metav1beta1.TableColumnDefinition, 0, len(columns)+len(table.ColumnDefinitions))
+	switch pos {
+	case beginning:
+		newColumns = append(newColumns, columns...)
+		newColumns = append(newColumns, table.ColumnDefinitions...)
+	case end:
+		newColumns = append(newColumns, table.ColumnDefinitions...)
+		newColumns = append(newColumns, columns...)
+	default:
+		return fmt.Errorf("invalid column add position: %v", pos)
+	}
+	table.ColumnDefinitions = newColumns
+	for i := range table.Rows {
+		table.Rows[i].Cells = newRows[i]
+	}
+
+	return nil
+}
+
 // decorateTable takes a table and attempts to add label columns and the
 // namespace column. It will fill empty columns with nil (if the object
 // does not expose metadata). It returns an error if the table cannot
@@ -304,7 +389,7 @@ func decorateTable(table *metav1beta1.Table, options PrintOptions) error {
 // printRowsForHandlerEntry prints the incremental table output (headers if the current type is
 // different from lastType) including all the rows in the object. It returns the current type
 // or an error, if any.
-func printRowsForHandlerEntry(output io.Writer, handler *handlerEntry, obj runtime.Object, options PrintOptions, includeHeaders bool) error {
+func printRowsForHandlerEntry(output io.Writer, handler *handlerEntry, eventType string, obj runtime.Object, options PrintOptions, includeHeaders bool) error {
 	var results []reflect.Value
 
 	args := []reflect.Value{reflect.ValueOf(obj), reflect.ValueOf(options)}
@@ -324,23 +409,46 @@ func printRowsForHandlerEntry(output io.Writer, handler *handlerEntry, obj runti
 		headers = append(headers, formatLabelHeaders(options.ColumnLabels)...)
 		// LABELS is always the last column.
 		headers = append(headers, formatShowLabelsHeader(options.ShowLabels)...)
+		// prepend namespace header
 		if options.WithNamespace {
 			headers = append(withNamespacePrefixColumns, headers...)
 		}
+		// prepend event type header
+		if len(eventType) > 0 {
+			headers = append(withEventTypePrefixColumns, headers...)
+		}
 		printHeader(headers, output)
 	}
 
 	if results[1].IsNil() {
 		rows := results[0].Interface().([]metav1beta1.TableRow)
-		printRows(output, rows, options)
+		printRows(output, eventType, rows, options)
 		return nil
 	}
 	return results[1].Interface().(error)
 }
 
+var formattedEventType = map[string]string{
+	string(watch.Added):    "ADDED   ",
+	string(watch.Modified): "MODIFIED",
+	string(watch.Deleted):  "DELETED ",
+	string(watch.Error):    "ERROR   ",
+}
+
+func formatEventType(eventType string) string {
+	if formatted, ok := formattedEventType[eventType]; ok {
+		return formatted
+	}
+	return string(eventType)
+}
+
 // printRows writes the provided rows to output.
-func printRows(output io.Writer, rows []metav1beta1.TableRow, options PrintOptions) {
+func printRows(output io.Writer, eventType string, rows []metav1beta1.TableRow, options PrintOptions) {
 	for _, row := range rows {
+		if len(eventType) > 0 {
+			fmt.Fprint(output, formatEventType(eventType))
+			fmt.Fprint(output, "\t")
+		}
 		if options.WithNamespace {
 			if obj := row.Object.Object; obj != nil {
 				if m, err := meta.Accessor(obj); err == nil {

pkg/printers/tableprinter_test.go

@@ -52,6 +52,7 @@ func TestPrintRowsForHandlerEntry(t *testing.T) {
 		name          string
 		h             *handlerEntry
 		opt           PrintOptions
+		eventType     string
 		obj           runtime.Object
 		includeHeader bool
 		expectOut     string
@@ -96,6 +97,20 @@ func TestPrintRowsForHandlerEntry(t *testing.T) {
 			includeHeader: true,
 			expectOut:     "NAME\tSTATUS\tAGE\ntest\t\t<unknow>\n",
 		},
+		{
+			name: "with event type",
+			h: &handlerEntry{
+				columnDefinitions: testNamespaceColumnDefinitions,
+				printFunc:         printFunc,
+			},
+			opt:       PrintOptions{},
+			eventType: "ADDED",
+			obj: &corev1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{Name: "test"},
+			},
+			includeHeader: true,
+			expectOut:     "EVENT\tNAME\tSTATUS\tAGE\nADDED   \ttest\t\t<unknow>\n",
+		},
 		{
 			name: "print namespace and withnamespace true, should not print header",
 			h: &handlerEntry{
@@ -116,7 +131,7 @@ func TestPrintRowsForHandlerEntry(t *testing.T) {
 	for _, test := range testCase {
 		t.Run(test.name, func(t *testing.T) {
 			buffer := &bytes.Buffer{}
-			err := printRowsForHandlerEntry(buffer, test.h, test.obj, test.opt, test.includeHeader)
+			err := printRowsForHandlerEntry(buffer, test.h, test.eventType, test.obj, test.opt, test.includeHeader)
 			if err != nil {
 				if err.Error() != test.expectErr {
 					t.Errorf("[%s]expect:\n %v\n but got:\n %v\n", test.name, test.expectErr, err)

staging/src/k8s.io/cli-runtime/pkg/printers/BUILD

@@ -18,6 +18,7 @@ go_library(
     visibility = ["//visibility:public"],
     deps = [
         "//staging/src/k8s.io/apimachinery/pkg/api/meta: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/runtime:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",

staging/src/k8s.io/cli-runtime/pkg/printers/json.go

@@ -22,7 +22,9 @@ import (
 	"fmt"
 	"io"
 	"reflect"
+	"sync/atomic"
 
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 
 	"sigs.k8s.io/yaml"
@@ -41,6 +43,20 @@ func (p *JSONPrinter) PrintObj(obj runtime.Object, w io.Writer) error {
 	}
 
 	switch obj := obj.(type) {
+	case *metav1.WatchEvent:
+		if InternalObjectPreventer.IsForbidden(reflect.Indirect(reflect.ValueOf(obj.Object.Object)).Type().PkgPath()) {
+			return fmt.Errorf(InternalObjectPrinterErr)
+		}
+		data, err := json.Marshal(obj)
+		if err != nil {
+			return err
+		}
+		_, err = w.Write(data)
+		if err != nil {
+			return err
+		}
+		_, err = w.Write([]byte{'\n'})
+		return err
 	case *runtime.Unknown:
 		var buf bytes.Buffer
 		err := json.Indent(&buf, obj.Raw, "", "    ")
@@ -90,6 +106,20 @@ func (p *YAMLPrinter) PrintObj(obj runtime.Object, w io.Writer) error {
 	}
 
 	switch obj := obj.(type) {
+	case *metav1.WatchEvent:
+		if InternalObjectPreventer.IsForbidden(reflect.Indirect(reflect.ValueOf(obj.Object.Object)).Type().PkgPath()) {
+			return fmt.Errorf(InternalObjectPrinterErr)
+		}
+		data, err := json.Marshal(obj)
+		if err != nil {
+			return err
+		}
+		data, err = yaml.JSONToYAML(data)
+		if err != nil {
+			return err
+		}
+		_, err = w.Write(data)
+		return err
 	case *runtime.Unknown:
 		data, err := yaml.JSONToYAML(obj.Raw)
 		if err != nil {

staging/src/k8s.io/cli-runtime/pkg/printers/name.go

@@ -23,6 +23,7 @@ import (
 	"strings"
 
 	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -42,6 +43,11 @@ type NamePrinter struct {
 // PrintObj is an implementation of ResourcePrinter.PrintObj which decodes the object
 // and print "resource/name" pair. If the object is a List, print all items in it.
 func (p *NamePrinter) PrintObj(obj runtime.Object, w io.Writer) error {
+	switch castObj := obj.(type) {
+	case *metav1.WatchEvent:
+		obj = castObj.Object.Object
+	}
+
 	// we use reflect.Indirect here in order to obtain the actual value from a pointer.
 	// using reflect.Indirect indiscriminately is valid here, as all runtime.Objects are supposed to be pointers.
 	// we need an actual value in order to retrieve the package path for an object.