cli for status

cmd/resource/status/cmd/events.go

@@ -0,0 +1,83 @@
+package cmd
+
+import (
+	"context"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+	"sigs.k8s.io/kustomize/kstatus/wait"
+	"sigs.k8s.io/kustomize/kyaml/kio"
+)
+
+func GetEventsRunner() *EventsRunner {
+	r := &EventsRunner{}
+	c := &cobra.Command{
+		Use:   "events",
+		Short: "Events",
+		RunE:  r.runE,
+	}
+	c.Flags().BoolVar(&r.IncludeSubpackages, "include-subpackages", true,
+		"also print resources from subpackages.")
+	c.Flags().DurationVar(&r.Interval, "interval", 2*time.Second,
+		"check every n seconds. Default is every 2 seconds.")
+	c.Flags().DurationVar(&r.Timeout, "timeout", 60*time.Second,
+		"give up after n seconds. Default is 60 seconds.")
+
+	r.Command = c
+	return r
+}
+
+func EventsCommand() *cobra.Command {
+	return GetEventsRunner().Command
+}
+
+type EventsRunner struct {
+	IncludeSubpackages bool
+	Interval           time.Duration
+	Timeout            time.Duration
+	Command            *cobra.Command
+}
+
+func (r *EventsRunner) runE(c *cobra.Command, args []string) error {
+	ctx := context.Background()
+	client, err := getClient()
+	if err != nil {
+		return errors.Wrap(err, "error creating client")
+	}
+
+	resolver := wait.NewResolver(client, r.Interval)
+
+	captureFilter := &CaptureIdentifiersFilter{}
+	filters := []kio.Filter{captureFilter}
+
+	var inputs []kio.Reader
+	for _, a := range args {
+		inputs = append(inputs, kio.LocalPackageReader{
+			PackagePath:        a,
+			IncludeSubpackages: r.IncludeSubpackages,
+		})
+	}
+	if len(inputs) == 0 {
+		inputs = append(inputs, &kio.ByteReader{Reader: c.InOrStdin()})
+	}
+
+	err = kio.Pipeline{
+		Inputs:  inputs,
+		Filters: filters,
+	}.Execute()
+	if err != nil {
+		return errors.Wrap(err, "error reading manifests")
+	}
+
+	printer := newEventPrinter(c.OutOrStdout(), c.OutOrStderr())
+
+	ctx, cancel := context.WithTimeout(ctx, r.Timeout)
+	defer cancel()
+	resChannel := resolver.WaitForStatus(ctx, captureFilter.Identifiers)
+
+	for msg := range resChannel {
+		printer.printEvent(msg)
+	}
+	return nil
+}

cmd/resource/status/cmd/fetch.go

@@ -0,0 +1,99 @@
+package cmd
+
+import (
+	"context"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+	"sigs.k8s.io/kustomize/kstatus/status"
+	"sigs.k8s.io/kustomize/kstatus/wait"
+	"sigs.k8s.io/kustomize/kyaml/kio"
+)
+
+func GetFetchRunner() *FetchRunner {
+	r := &FetchRunner{}
+	c := &cobra.Command{
+		Use:   "fetch",
+		Short: "Fetch",
+		RunE:  r.runE,
+	}
+	c.Flags().BoolVar(&r.IncludeSubpackages, "include-subpackages", true,
+		"also print resources from subpackages.")
+
+	r.Command = c
+	return r
+}
+
+func FetchCommand() *cobra.Command {
+	return GetFetchRunner().Command
+}
+
+type FetchRunner struct {
+	IncludeSubpackages bool
+	Command            *cobra.Command
+}
+
+func (r *FetchRunner) runE(c *cobra.Command, args []string) error {
+	ctx := context.Background()
+	client, err := getClient()
+	if err != nil {
+		return errors.Wrap(err, "error creating client")
+	}
+
+	resolver := wait.NewResolver(client, time.Minute)
+
+	captureFilter := &CaptureIdentifiersFilter{}
+	filters := []kio.Filter{captureFilter}
+
+	var inputs []kio.Reader
+	for _, a := range args {
+		inputs = append(inputs, kio.LocalPackageReader{
+			PackagePath:        a,
+			IncludeSubpackages: r.IncludeSubpackages,
+		})
+	}
+	if len(inputs) == 0 {
+		inputs = append(inputs, &kio.ByteReader{Reader: c.InOrStdin()})
+	}
+
+	err = kio.Pipeline{
+		Inputs:  inputs,
+		Filters: filters,
+	}.Execute()
+	if err != nil {
+		return errors.Wrap(err, "error reading manifests")
+	}
+
+	results := resolver.FetchAndResolve(ctx, captureFilter.Identifiers)
+
+	newTablePrinter(FetchStatusInfo{results}, c.OutOrStdout(), c.OutOrStderr(), false).Print()
+	return nil
+}
+
+type FetchStatusInfo struct {
+	Results []wait.ResourceResult
+}
+
+func (f FetchStatusInfo) CurrentStatus() StatusData {
+	var resourceData []ResourceStatusData
+	for _, res := range f.Results {
+		rsd := ResourceStatusData{
+			Identifier: res.Resource,
+		}
+		if res.Error != nil {
+			rsd.Status = status.UnknownStatus
+			rsd.Message = res.Error.Error()
+		} else {
+			rsd.Status = res.Result.Status
+			rsd.Message = res.Result.Message
+		}
+
+		resourceData = append(resourceData, rsd)
+	}
+
+	return StatusData{
+		AggregateStatus:  status.UnknownStatus,
+		ResourceStatuses: resourceData,
+	}
+}

cmd/resource/status/cmd/print.go

@@ -0,0 +1,339 @@
+package cmd
+
+import (
+	"fmt"
+	"io"
+	"time"
+
+	"github.com/sethgrid/curse"
+
+	"sigs.k8s.io/kustomize/kstatus/status"
+	"sigs.k8s.io/kustomize/kstatus/wait"
+)
+
+const (
+	typeColumn      = "type"
+	namespaceColumn = "namespace"
+	nameColumn      = "name"
+	statusColumn    = "status"
+	messageColumn   = "message"
+)
+
+type colorFunc func(s status.Status) int
+type contentFunc func(resource ResourceStatusData) string
+
+type tableColumnInfo struct {
+	header      string
+	width       int
+	colorFunc   colorFunc
+	contentFunc contentFunc
+}
+
+func defaultColorFunc(_ status.Status) int {
+	return curse.WHITE
+}
+
+var (
+	tableColumns = map[string]tableColumnInfo{
+		typeColumn: {
+			header:    "TYPE",
+			width:     25,
+			colorFunc: defaultColorFunc,
+			contentFunc: func(data ResourceStatusData) string {
+				return fmt.Sprintf("%s/%s", data.Identifier.GetAPIVersion(),
+					data.Identifier.GetKind())
+			},
+		},
+		namespaceColumn: {
+			header:    "NAMESPACE",
+			width:     15,
+			colorFunc: defaultColorFunc,
+			contentFunc: func(data ResourceStatusData) string {
+				return data.Identifier.GetNamespace()
+			},
+		},
+		nameColumn: {
+			header:    "NAME",
+			width:     20,
+			colorFunc: defaultColorFunc,
+			contentFunc: func(data ResourceStatusData) string {
+				return data.Identifier.GetName()
+			},
+		},
+		statusColumn: {
+			header:    "STATUS",
+			width:     10,
+			colorFunc: colorForStatus,
+			contentFunc: func(data ResourceStatusData) string {
+				return data.Status.String()
+			},
+		},
+		messageColumn: {
+			header:    "MESSAGE",
+			width:     40,
+			colorFunc: defaultColorFunc,
+			contentFunc: func(data ResourceStatusData) string {
+				return data.Message
+			},
+		},
+	}
+	tableColumnOrder = []string{typeColumn, namespaceColumn, nameColumn, statusColumn, messageColumn}
+)
+
+type StatusInfo interface {
+	CurrentStatus() StatusData
+}
+
+type StatusData struct {
+	AggregateStatus  status.Status
+	ResourceStatuses []ResourceStatusData
+}
+
+type ResourceStatusData struct {
+	Identifier wait.ResourceIdentifier
+	Status     status.Status
+	Message    string
+}
+
+type TablePrinter struct {
+	statusInfo    StatusInfo
+	out           io.Writer
+	err           io.Writer
+	showAggStatus bool
+}
+
+func newTablePrinter(statusInfo StatusInfo, out io.Writer, err io.Writer, showAggStatus bool) *TablePrinter {
+	return &TablePrinter{
+		statusInfo:    statusInfo,
+		out:           out,
+		err:           err,
+		showAggStatus: showAggStatus,
+	}
+}
+
+func (s *TablePrinter) Print() {
+	c := newCurseOrDie()
+	s.printTable(c, s.statusInfo.CurrentStatus(), false)
+}
+
+func (s *TablePrinter) PrintUntil(stop <-chan struct{}, interval time.Duration) <-chan struct{} {
+	completed := make(chan struct{})
+	go func() {
+		defer close(completed)
+		c := newCurseOrDie()
+		c.SetDefaultStyle()
+		s.printTable(c, s.statusInfo.CurrentStatus(), false)
+		ticker := time.NewTicker(interval)
+		for {
+			select {
+			case <-stop:
+				ticker.Stop()
+				s.printTable(c, s.statusInfo.CurrentStatus(), true)
+				return
+			case <-ticker.C:
+				s.printTable(c, s.statusInfo.CurrentStatus(), true)
+			}
+		}
+	}()
+	return completed
+}
+
+func (s *TablePrinter) printTable(c *curse.Cursor, data StatusData, moveUp bool) {
+	if moveUp {
+		if s.showAggStatus {
+			c.MoveUp(1)
+		}
+		c.MoveUp(1)
+		c.MoveUp(len(data.ResourceStatuses))
+	}
+	c.EraseCurrentLine()
+	if s.showAggStatus {
+		printOrDie(s.out, "AggregateStatus: ")
+		c.SetColor(colorForStatus(data.AggregateStatus))
+		printOrDie(s.out, "%s\n", data.AggregateStatus)
+		c.SetDefaultStyle()
+	}
+	s.printTableRow(c, headers())
+	for _, resource := range data.ResourceStatuses {
+		s.printTableRow(c, row(resource))
+	}
+}
+
+func (s *TablePrinter) printTableRow(c *curse.Cursor, rowData []RowData) {
+	for _, row := range rowData {
+		c.SetColor(row.color)
+		format := fmt.Sprintf("%%-%ds  ", row.width)
+		printOrDie(s.out, format, trimString(row.content, row.width))
+		c.SetDefaultStyle()
+	}
+	printOrDie(s.out, "\n")
+}
+
+type RowData struct {
+	content string
+	color   int
+	width   int
+}
+
+func headers() []RowData {
+	var headers []RowData
+	for _, columnName := range tableColumnOrder {
+		column := tableColumns[columnName]
+		headers = append(headers, RowData{
+			content: column.header,
+			color:   curse.WHITE,
+			width:   column.width,
+		})
+	}
+	return headers
+}
+
+func row(resource ResourceStatusData) []RowData {
+	var row []RowData
+	for _, columnName := range tableColumnOrder {
+		column := tableColumns[columnName]
+		row = append(row, RowData{
+			content: column.contentFunc(resource),
+			color:   column.colorFunc(resource.Status),
+			width:   column.width,
+		})
+	}
+	return row
+}
+
+type eventContentFunc func(wait.Event) string
+
+type eventColumnInfo struct {
+	header                     string
+	width                      int
+	requireResourceUpdateEvent bool
+	contentFunc                eventContentFunc
+}
+
+var (
+	eventColumns = []eventColumnInfo{
+		{
+			header:                     "EVENT TYPE",
+			width:                      15,
+			requireResourceUpdateEvent: false,
+			contentFunc: func(event wait.Event) string {
+				return string(event.Type)
+			},
+		},
+		{
+			header:                     "AGG STATUS",
+			width:                      10,
+			requireResourceUpdateEvent: false,
+			contentFunc: func(event wait.Event) string {
+				return event.AggregateStatus.String()
+			},
+		},
+		{
+			header:                     "TYPE",
+			width:                      20,
+			requireResourceUpdateEvent: true,
+			contentFunc: func(event wait.Event) string {
+				return fmt.Sprintf("%s/%s", event.EventResource.Identifier.GetAPIVersion(),
+					event.EventResource.Identifier.GetKind())
+			},
+		},
+		{
+			header:                     "NAMESPACE",
+			width:                      15,
+			requireResourceUpdateEvent: true,
+			contentFunc: func(event wait.Event) string {
+				return event.EventResource.Identifier.GetNamespace()
+			},
+		},
+		{
+			header:                     "NAME",
+			width:                      20,
+			requireResourceUpdateEvent: true,
+			contentFunc: func(event wait.Event) string {
+				return event.EventResource.Identifier.GetName()
+			},
+		},
+		{
+			header:                     "STATUS",
+			width:                      10,
+			requireResourceUpdateEvent: true,
+			contentFunc: func(event wait.Event) string {
+				return event.EventResource.Status.String()
+			},
+		},
+		{
+			header:                     "MESSAGE",
+			width:                      50,
+			requireResourceUpdateEvent: true,
+			contentFunc: func(event wait.Event) string {
+				if event.EventResource.Error != nil {
+					return event.EventResource.Error.Error()
+				}
+				return event.EventResource.Message
+			},
+		},
+	}
+)
+
+type EventPrinter struct {
+	out io.Writer
+	err io.Writer
+}
+
+func newEventPrinter(out io.Writer, err io.Writer) *EventPrinter {
+	for _, column := range eventColumns {
+		format := fmt.Sprintf("%%-%ds  ", column.width)
+		printOrDie(out, format, column.header)
+	}
+	printOrDie(out, "\n")
+	return &EventPrinter{
+		out: out,
+		err: err,
+	}
+}
+
+func (e *EventPrinter) printEvent(event wait.Event) {
+	for _, column := range eventColumns {
+		if event.Type != wait.ResourceUpdate && column.requireResourceUpdateEvent {
+			continue
+		}
+		format := fmt.Sprintf("%%-%ds  ", column.width)
+		printOrDie(e.out, format, trimString(column.contentFunc(event), column.width))
+	}
+	printOrDie(e.out, "\n")
+}
+
+func newCurseOrDie() *curse.Cursor {
+	// TODO: Handle the issue with creating a new Cursor. For now we
+	// are just ignoring the error (which mostly works).
+	c, _ := curse.New()
+	return c
+}
+
+func printOrDie(w io.Writer, format string, a ...interface{}) {
+	_, err := fmt.Fprintf(w, format, a...)
+	if err != nil {
+		panic(err)
+	}
+}
+
+func colorForStatus(s status.Status) int {
+	switch s {
+	case status.CurrentStatus:
+		return curse.GREEN
+	case status.UnknownStatus:
+		return curse.WHITE
+	case status.InProgressStatus:
+		return curse.YELLOW
+	case status.FailedStatus:
+		return curse.RED
+	}
+	return curse.WHITE
+}
+
+func trimString(str string, maxLength int) string {
+	if len(str) <= maxLength {
+		return str
+	}
+	return str[:maxLength]
+}

cmd/resource/status/cmd/util.go

@@ -0,0 +1,47 @@
+package cmd
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/client/apiutil"
+	"sigs.k8s.io/kustomize/kstatus/wait"
+	"sigs.k8s.io/kustomize/kyaml/kio"
+	"sigs.k8s.io/kustomize/kyaml/yaml"
+)
+
+var (
+	scheme = runtime.NewScheme()
+)
+
+func init() {
+	_ = clientgoscheme.AddToScheme(scheme)
+}
+
+func getClient() (client.Client, error) {
+	config := ctrl.GetConfigOrDie()
+	mapper, err := apiutil.NewDiscoveryRESTMapper(config)
+	if err != nil {
+		return nil, err
+	}
+	return client.New(config, client.Options{Scheme: scheme, Mapper: mapper})
+}
+
+type CaptureIdentifiersFilter struct {
+	Identifiers []wait.ResourceIdentifier
+}
+
+var _ kio.Filter = &CaptureIdentifiersFilter{}
+
+func (f *CaptureIdentifiersFilter) Filter(slice []*yaml.RNode) ([]*yaml.RNode, error) {
+	for i := range slice {
+		meta, err := slice[i].GetMeta()
+		if err != nil {
+			return nil, err
+		}
+		id := meta.GetIdentifier()
+		f.Identifiers = append(f.Identifiers, &id)
+	}
+	return slice, nil
+}

cmd/resource/status/cmd/wait.go

@@ -0,0 +1,174 @@
+package cmd
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/spf13/cobra"
+
+	"sigs.k8s.io/kustomize/kstatus/status"
+	"sigs.k8s.io/kustomize/kstatus/wait"
+	"sigs.k8s.io/kustomize/kyaml/kio"
+)
+
+func GetWaitRunner() *WaitRunner {
+	r := &WaitRunner{}
+	c := &cobra.Command{
+		Use:   "wait",
+		Short: "Wait",
+		RunE:  r.runE,
+	}
+	c.Flags().BoolVar(&r.IncludeSubpackages, "include-subpackages", true,
+		"also print resources from subpackages.")
+	c.Flags().DurationVar(&r.Interval, "interval", 2*time.Second,
+		"check every n seconds. Default is every 2 seconds.")
+	c.Flags().DurationVar(&r.Timeout, "timeout", 60*time.Second,
+		"give up after n seconds. Default is 60 seconds.")
+
+	r.Command = c
+	return r
+}
+
+func WaitCommand() *cobra.Command {
+	return GetWaitRunner().Command
+}
+
+type WaitRunner struct {
+	IncludeSubpackages bool
+	Interval           time.Duration
+	Timeout            time.Duration
+	Command            *cobra.Command
+}
+
+func (r *WaitRunner) runE(c *cobra.Command, args []string) error {
+	ctx := context.Background()
+	client, err := getClient()
+	if err != nil {
+		return errors.Wrap(err, "error creating client")
+	}
+
+	resolver := wait.NewResolver(client, r.Interval)
+
+	captureFilter := &CaptureIdentifiersFilter{}
+	filters := []kio.Filter{captureFilter}
+
+	var inputs []kio.Reader
+	for _, a := range args {
+		inputs = append(inputs, kio.LocalPackageReader{
+			PackagePath:        a,
+			IncludeSubpackages: r.IncludeSubpackages,
+		})
+	}
+	if len(inputs) == 0 {
+		inputs = append(inputs, &kio.ByteReader{Reader: c.InOrStdin()})
+	}
+
+	err = kio.Pipeline{
+		Inputs:  inputs,
+		Filters: filters,
+	}.Execute()
+	if err != nil {
+		return errors.Wrap(err, "error reading manifests")
+	}
+
+	collector := newResourceStatusCollector(captureFilter.Identifiers)
+
+	stop := make(chan struct{})
+	printer := newTablePrinter(CollectorStatusInfo{collector}, c.OutOrStdout(), c.OutOrStderr(), true)
+	printFinished := printer.PrintUntil(stop, 1*time.Second)
+
+	ctx, cancel := context.WithTimeout(ctx, r.Timeout)
+	defer cancel()
+	resChannel := resolver.WaitForStatus(ctx, captureFilter.Identifiers)
+
+	for msg := range resChannel {
+		switch msg.Type {
+		case wait.ResourceUpdate:
+			collector.updateResourceStatus(msg)
+		case wait.Aborted:
+			collector.updateAggregateStatus(msg.AggregateStatus)
+		case wait.Completed:
+			collector.updateAggregateStatus(msg.AggregateStatus)
+		}
+	}
+	close(stop)
+	<-printFinished // Wait for printer to finish work.
+	return nil
+}
+
+type ResourceStatusCollector struct {
+	mux sync.RWMutex
+
+	AggregateStatus  status.Status
+	ResourceStatuses []*ResourceStatus
+}
+
+func (r *ResourceStatusCollector) updateResourceStatus(msg wait.Event) {
+	r.mux.Lock()
+	defer r.mux.Unlock()
+	r.AggregateStatus = msg.AggregateStatus
+	eventResource := msg.EventResource
+	for _, resourceState := range r.ResourceStatuses {
+		if resourceState.Identifier.GetAPIVersion() == eventResource.Identifier.GetAPIVersion() &&
+			resourceState.Identifier.GetKind() == eventResource.Identifier.GetKind() &&
+			resourceState.Identifier.GetNamespace() == eventResource.Identifier.GetNamespace() &&
+			resourceState.Identifier.GetName() == eventResource.Identifier.GetName() {
+			resourceState.Status = eventResource.Status
+			resourceState.Message = eventResource.Message
+		}
+	}
+}
+
+func (r *ResourceStatusCollector) updateAggregateStatus(aggregateStatus status.Status) {
+	r.mux.Lock()
+	defer r.mux.Unlock()
+	r.AggregateStatus = aggregateStatus
+}
+
+type ResourceStatus struct {
+	Identifier wait.ResourceIdentifier
+	Status     status.Status
+	Message    string
+}
+
+func newResourceStatusCollector(identifiers []wait.ResourceIdentifier) *ResourceStatusCollector {
+	var statuses []*ResourceStatus
+
+	for _, id := range identifiers {
+		statuses = append(statuses, &ResourceStatus{
+			Identifier: id,
+			Status:     status.UnknownStatus,
+			Message:    "",
+		})
+	}
+
+	return &ResourceStatusCollector{
+		AggregateStatus:  status.UnknownStatus,
+		ResourceStatuses: statuses,
+	}
+}
+
+type CollectorStatusInfo struct {
+	Collector *ResourceStatusCollector
+}
+
+func (f CollectorStatusInfo) CurrentStatus() StatusData {
+	f.Collector.mux.RLock()
+	defer f.Collector.mux.RUnlock()
+
+	var resourceData []ResourceStatusData
+	for _, res := range f.Collector.ResourceStatuses {
+		resourceData = append(resourceData, ResourceStatusData{
+			Identifier: res.Identifier,
+			Status:     res.Status,
+			Message:    res.Message,
+		})
+	}
+
+	return StatusData{
+		AggregateStatus:  f.Collector.AggregateStatus,
+		ResourceStatuses: resourceData,
+	}
+}

cmd/resource/status/status.go

@@ -0,0 +1,19 @@
+package status
+
+import (
+	"github.com/spf13/cobra"
+	"sigs.k8s.io/kustomize/cmd/resource/status/cmd"
+)
+
+func StatusCommand() *cobra.Command {
+	var status = &cobra.Command{
+		Use:   "status",
+		Short: "status reference command",
+	}
+
+	status.AddCommand(cmd.FetchCommand())
+	status.AddCommand(cmd.WaitCommand())
+	status.AddCommand(cmd.EventsCommand())
+
+	return status
+}