Add json output format for kpt live status command (#2362)

Author: mortent

commands/livecmd.go

@@ -57,7 +57,7 @@ func GetLiveCommand(ctx context.Context, _, version string) *cobra.Command {
 	initCmd := cmdliveinit.NewCommand(ctx, f, ioStreams)
 	applyCmd := cmdapply.NewCommand(ctx, rgProvider, ioStreams)
 	destroyCmd := cmddestroy.NewCommand(ctx, rgProvider, ioStreams)
-	statusCmd := status.NewCommand(ctx, rgProvider, ioStreams)
+	statusCmd := status.NewCommand(ctx, rgProvider)
 	installRGCmd := GetInstallRGRunner(f, ioStreams).Command
 	liveCmd.AddCommand(initCmd, applyCmd, destroyCmd, statusCmd, installRGCmd)
 

thirdparty/cli-utils/status/cmdstatus.go

@@ -10,9 +10,11 @@ import (
 	"time"
 
 	"github.com/GoogleContainerTools/kpt/internal/docs/generated/livedocs"
+	"github.com/GoogleContainerTools/kpt/internal/printer"
 	"github.com/GoogleContainerTools/kpt/internal/util/strings"
 	"github.com/GoogleContainerTools/kpt/pkg/live"
-	"github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/status/printers"
+	"github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/printers"
+	statusprinters "github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/status/printers"
 	"github.com/go-errors/errors"
 	"github.com/spf13/cobra"
 	"k8s.io/cli-runtime/pkg/genericclioptions"
@@ -40,13 +42,11 @@ var (
 	PollUntilOptions = []string{Known, Current, Deleted, Forever}
 )
 
-func NewRunner(ctx context.Context, provider provider.Provider,
-	ioStreams genericclioptions.IOStreams) *Runner {
+func NewRunner(ctx context.Context, provider provider.Provider) *Runner {
 	r := &Runner{
 		ctx:               ctx,
 		provider:          provider,
 		pollerFactoryFunc: pollerFactoryFunc,
-		ioStreams:         ioStreams,
 	}
 	c := &cobra.Command{
 		Use:     "status [PKG_PATH | -]",
@@ -67,18 +67,16 @@ func NewRunner(ctx context.Context, provider provider.Provider,
 	return r
 }
 
-func NewCommand(ctx context.Context, provider provider.Provider,
-	ioStreams genericclioptions.IOStreams) *cobra.Command {
-	return NewRunner(ctx, provider, ioStreams).Command
+func NewCommand(ctx context.Context, provider provider.Provider) *cobra.Command {
+	return NewRunner(ctx, provider).Command
 }
 
 // Runner captures the parameters for the command and contains
 // the run function.
 type Runner struct {
-	ctx       context.Context
-	Command   *cobra.Command
-	ioStreams genericclioptions.IOStreams
-	provider  provider.Provider
+	ctx      context.Context
+	Command  *cobra.Command
+	provider provider.Provider
 
 	period    time.Duration
 	pollUntil string
@@ -93,13 +91,18 @@ func (r *Runner) preRunE(*cobra.Command, []string) error {
 		return fmt.Errorf("pollUntil must be one of %s",
 			strings.JoinStringsWithQuotes(PollUntilOptions))
 	}
+
+	if found := printers.ValidatePrinterType(r.output); !found {
+		return fmt.Errorf("unknown output type %q", r.output)
+	}
 	return nil
 }
 
 // runE implements the logic of the command and will delegate to the
 // poller to compute status for each of the resources. One of the printer
 // implementations takes care of printing the output.
 func (r *Runner) runE(c *cobra.Command, args []string) error {
+	pr := printer.FromContextOrDie(r.ctx)
 	if len(args) == 0 {
 		// default to the current working directory
 		cwd, err := os.Getwd()
@@ -138,7 +141,7 @@ func (r *Runner) runE(c *cobra.Command, args []string) error {
 
 	// Exit here if the inventory is empty.
 	if len(identifiers) == 0 {
-		_, _ = fmt.Fprint(r.ioStreams.Out, "no resources found in the inventory\n")
+		pr.Printf("no resources found in the inventory\n")
 		return nil
 	}
 
@@ -149,7 +152,10 @@ func (r *Runner) runE(c *cobra.Command, args []string) error {
 
 	// Fetch a printer implementation based on the desired output format as
 	// specified in the output flag.
-	printer, err := printers.CreatePrinter(r.output, r.ioStreams)
+	printer, err := statusprinters.CreatePrinter(r.output, genericclioptions.IOStreams{
+		Out:    pr.OutStream(),
+		ErrOut: pr.ErrStream(),
+	})
 	if err != nil {
 		return errors.WrapPrefix(err, "error creating printer", 1)
 	}
@@ -186,8 +192,7 @@ func (r *Runner) runE(c *cobra.Command, args []string) error {
 		UseCache:     true,
 	})
 
-	printer.Print(eventChannel, identifiers, cancelFunc)
-	return nil
+	return printer.Print(eventChannel, identifiers, cancelFunc)
 }
 
 // desiredStatusNotifierFunc returns an Observer function for the

thirdparty/cli-utils/status/cmdstatus_test.go

@@ -4,6 +4,7 @@
 package status
 
 import (
+	"bytes"
 	"context"
 	"path/filepath"
 	"strings"
@@ -17,7 +18,6 @@ import (
 	"github.com/GoogleContainerTools/kpt/pkg/live"
 	"github.com/stretchr/testify/assert"
 	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/cli-runtime/pkg/genericclioptions"
 	cmdtesting "k8s.io/kubectl/pkg/cmd/testing"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 	"sigs.k8s.io/cli-utils/pkg/apply/poller"
@@ -239,7 +239,6 @@ deployment.apps/foo is InProgress: inProgress
 		t.Run(tn, func(t *testing.T) {
 			tf := cmdtesting.NewTestFactory().WithNamespace("namespace")
 			defer tf.Cleanup()
-			ioStreams, _, outBuf, _ := genericclioptions.NewTestIOStreams() //nolint:dogsled
 
 			w, clean := testutil.SetupWorkspace(t)
 			defer clean()
@@ -250,8 +249,9 @@ deployment.apps/foo is InProgress: inProgress
 			revert := testutil.Chdir(t, w.WorkspaceDirectory)
 			defer revert()
 
+			var outBuf bytes.Buffer
 			provider := live.NewFakeResourceGroupProvider(tf, tc.inventory)
-			runner := NewRunner(fake.CtxWithDefaultPrinter(), provider, ioStreams)
+			runner := NewRunner(fake.CtxWithPrinter(&outBuf, &outBuf), provider)
 			runner.pollerFactoryFunc = func(c cmdutil.Factory) (poller.Poller, error) {
 				return &fakePoller{tc.events}, nil
 			}

thirdparty/cli-utils/status/printers/event/event_printer.go

@@ -0,0 +1,60 @@
+package list
+
+import (
+	"sigs.k8s.io/cli-utils/pkg/apply/event"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/polling/collector"
+	pollevent "sigs.k8s.io/cli-utils/pkg/kstatus/polling/event"
+	"sigs.k8s.io/cli-utils/pkg/object"
+	"sigs.k8s.io/cli-utils/pkg/print/list"
+)
+
+// BaseListPrinter implements the Printer interface and outputs the resource
+// status information as a list of events as they happen.
+type BaseListPrinter struct {
+	Formatter list.Formatter
+}
+
+// Print takes an event channel and outputs the status events on the channel
+// until the channel is closed. The provided cancelFunc is consulted on
+// every event and is responsible for stopping the poller when appropriate.
+// This function will block.
+func (ep *BaseListPrinter) Print(ch <-chan pollevent.Event, identifiers []object.ObjMetadata,
+	cancelFunc collector.ObserverFunc) error {
+	coll := collector.NewResourceStatusCollector(identifiers)
+	// The actual work is done by the collector, which will invoke the
+	// callback on every event. In the callback we print the status
+	// information and call the cancelFunc which is responsible for
+	// stopping the poller at the correct time.
+	done := coll.ListenWithObserver(ch, collector.ObserverFunc(
+		func(statusCollector *collector.ResourceStatusCollector, e pollevent.Event) {
+			err := ep.printStatusEvent(e)
+			if err != nil {
+				panic(err)
+			}
+			cancelFunc(statusCollector, e)
+		}),
+	)
+	// Block until the done channel is closed.
+	<-done
+	if o := coll.LatestObservation(); o.Error != nil {
+		return o.Error
+	}
+	return nil
+}
+
+func (ep *BaseListPrinter) printStatusEvent(se pollevent.Event) error {
+	switch se.EventType {
+	case pollevent.ResourceUpdateEvent:
+		id := se.Resource.Identifier
+		return ep.Formatter.FormatStatusEvent(event.StatusEvent{
+			Identifier:       id,
+			Resource:         se.Resource.Resource,
+			PollResourceInfo: se.Resource,
+		})
+	case pollevent.ErrorEvent:
+		return ep.Formatter.FormatErrorEvent(event.ErrorEvent{
+			Err: se.Error,
+		})
+	}
+	return nil
+}

thirdparty/cli-utils/status/printers/list/base.go

@@ -21,5 +21,5 @@ type Printer interface {
 	// needs to, and that it has completed shutting down. The latter is important
 	// to make sure the printer has a chance to output all data before the
 	// program terminates.
-	Print(ch <-chan event.Event, identifiers []object.ObjMetadata, cancelFunc collector.ObserverFunc)
+	Print(ch <-chan event.Event, identifiers []object.ObjMetadata, cancelFunc collector.ObserverFunc) error
 }

thirdparty/cli-utils/status/printers/printer/printer.go

@@ -4,20 +4,29 @@
 package printers
 
 import (
-	"github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/status/printers/event"
+	"github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/printers"
+	"github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/printers/events"
+	"github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/printers/json"
+	"github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/status/printers/list"
 	"github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/status/printers/printer"
 	"github.com/GoogleContainerTools/kpt/thirdparty/cli-utils/status/printers/table"
-
 	"k8s.io/cli-runtime/pkg/genericclioptions"
+	"sigs.k8s.io/cli-utils/pkg/common"
 )
 
 // CreatePrinter return an implementation of the Printer interface. The
 // actual implementation is based on the printerType requested.
 func CreatePrinter(printerType string, ioStreams genericclioptions.IOStreams) (printer.Printer, error) {
 	switch printerType {
-	case "table":
+	case printers.TablePrinter:
 		return table.NewTablePrinter(ioStreams), nil
+	case printers.JSONPrinter:
+		return &list.BaseListPrinter{
+			Formatter: json.NewFormatter(ioStreams, common.DryRunNone),
+		}, nil
 	default:
-		return event.NewEventPrinter(ioStreams), nil
+		return &list.BaseListPrinter{
+			Formatter: events.NewFormatter(ioStreams, common.DryRunNone),
+		}, nil
 	}
 }

thirdparty/cli-utils/status/printers/printers.go

@@ -35,7 +35,7 @@ func NewTablePrinter(ioStreams genericclioptions.IOStreams) *tablePrinter {
 // until the channel is closed .
 //nolint:interfacer
 func (t *tablePrinter) Print(ch <-chan event.Event, identifiers []object.ObjMetadata,
-	cancelFunc collector.ObserverFunc) {
+	cancelFunc collector.ObserverFunc) error {
 	coll := collector.NewResourceStatusCollector(identifiers)
 	stop := make(chan struct{})
 
@@ -51,6 +51,10 @@ func (t *tablePrinter) Print(ch <-chan event.Event, identifiers []object.ObjMeta
 	// Block until all the collector has shut down. This means the
 	// eventChannel has been closed and all events have been processed.
 	<-done
+	var err error
+	if o := coll.LatestObservation(); o.Error != nil {
+		err = o.Error
+	}
 
 	// Close the stop channel to notify the print goroutine that it should
 	// shut down.
@@ -60,6 +64,7 @@ func (t *tablePrinter) Print(ch <-chan event.Event, identifiers []object.ObjMeta
 	// the printer has updated the UI with the latest state and
 	// exited from the goroutine.
 	<-printCompleted
+	return err
 }
 
 var columns = []table.ColumnDefinition{