connectors status command [issue-12]

internal/ctl/connectors/connectors.go

@@ -23,6 +23,7 @@ func Command() *cobra.Command {
 	connectorsCmd.AddCommand(removeConnectorCmd())
 	connectorsCmd.AddCommand(pauseConnectorsCmd())
 	connectorsCmd.AddCommand(resumeConnectorsCmd())
+	connectorsCmd.AddCommand(connectorsStatusCmd())
 
 	return connectorsCmd
 }

internal/ctl/connectors/status.go

@@ -0,0 +1,139 @@
+package connectors
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+
+	"github.com/90poe/connectctl/internal/ctl"
+	"github.com/90poe/connectctl/internal/version"
+	"github.com/90poe/connectctl/pkg/client/connect"
+	"github.com/90poe/connectctl/pkg/manager"
+	"github.com/jedib0t/go-pretty/table"
+	"github.com/pkg/errors"
+
+	"github.com/spf13/cobra"
+)
+
+type connectorsStatusCmdParams struct {
+	ClusterURL string
+	Connectors []string
+	Output     string
+}
+
+func connectorsStatusCmd() *cobra.Command {
+	params := &connectorsStatusCmdParams{}
+
+	statusCmd := &cobra.Command{
+		Use:   "status",
+		Short: "Get status for connectors in a cluster",
+		Long:  "",
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			return doConnectorsStatus(cmd, params)
+		},
+	}
+
+	ctl.AddCommonConnectorsFlags(statusCmd, &params.ClusterURL)
+	ctl.AddConnectorNamesFlags(statusCmd, &params.Connectors)
+	ctl.AddOutputFlags(statusCmd, &params.Output)
+
+	return statusCmd
+}
+
+func doConnectorsStatus(_ *cobra.Command, params *connectorsStatusCmdParams) error {
+	config := &manager.Config{
+		ClusterURL: params.ClusterURL,
+		Version:    version.Version,
+	}
+
+	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
+
+	client, err := connect.NewClient(params.ClusterURL, connect.WithUserAgent(userAgent))
+	if err != nil {
+		return errors.Wrap(err, "error creating connect client")
+	}
+
+	mngr, err := manager.NewConnectorsManager(client, config)
+	if err != nil {
+		return errors.Wrap(err, "error creating connectors manager")
+	}
+
+	statusList, err := mngr.Status(params.Connectors)
+	if err != nil {
+		return errors.Wrap(err, "error getting connectors status")
+	}
+
+	switch params.Output {
+	case "json":
+		if err = printAsJSON(statusList); err != nil {
+			return errors.Wrap(err, "error printing connectors status as JSON")
+		}
+
+	case "table":
+		printAsTable(statusList)
+
+	default:
+		return fmt.Errorf("invalid output format specified: %s", params.Output)
+	}
+
+	return nil
+}
+
+func countFailing(statusList []*connect.ConnectorStatus) (int, int) {
+	connectorCount := 0
+	taskCount := 0
+
+	for _, status := range statusList {
+		if status.Connector.State == "FAILED" {
+			connectorCount++
+		}
+
+		taskCount += countFailingTasks(&status.Tasks)
+	}
+
+	return connectorCount, taskCount
+}
+
+func countFailingTasks(tasks *[]connect.TaskState) int {
+	count := 0
+
+	for _, task := range *tasks {
+		if task.State == "FAILED" {
+			count++
+		}
+	}
+
+	return count
+}
+
+func printAsJSON(statusList []*connect.ConnectorStatus) error {
+	b, err := json.MarshalIndent(statusList, "", "  ")
+	if err != nil {
+		return err
+	}
+
+	os.Stdout.Write(b)
+	return nil
+}
+
+func printAsTable(statusList []*connect.ConnectorStatus) {
+	t := table.NewWriter()
+	t.SetOutputMirror(os.Stdout)
+	t.AppendHeader(table.Row{"Name", "State", "WorkerId", "Tasks"})
+
+	for _, status := range statusList {
+		tasks := ""
+		for _, task := range status.Tasks {
+			tasks += fmt.Sprintf("%d(%s): %s\n", task.ID, task.WorkerID, task.State)
+		}
+
+		t.AppendRow(table.Row{
+			status.Name,
+			status.Connector.State,
+			status.Connector.WorkerID,
+			tasks,
+		})
+	}
+
+	t.Render()
+}

pkg/manager/status.go

@@ -0,0 +1,39 @@
+package manager
+
+import (
+	"github.com/90poe/connectctl/pkg/client/connect"
+	"github.com/pkg/errors"
+)
+
+// Status - gets status of specified (or all) connectors
+func (c *ConnectorManager) Status(connectors []string) ([]*connect.ConnectorStatus, error) {
+	if len(connectors) == 0 {
+		return c.allConnectorsStatus()
+	}
+
+	return c.specifiedConnectorsStatus(connectors)
+}
+
+func (c *ConnectorManager) allConnectorsStatus() ([]*connect.ConnectorStatus, error) {
+	existing, _, err := c.client.ListConnectors()
+	if err != nil {
+		return nil, errors.Wrap(err, "error listing connectors")
+	}
+
+	return c.specifiedConnectorsStatus(existing)
+}
+
+func (c *ConnectorManager) specifiedConnectorsStatus(connectors []string) ([]*connect.ConnectorStatus, error) {
+	statusList := make([]*connect.ConnectorStatus, len(connectors))
+
+	for idx, connectorName := range connectors {
+		status, _, err := c.client.GetConnectorStatus(connectorName)
+		if err != nil {
+			return nil, errors.Wrapf(err, "error getting connector status for %s", connectorName)
+		}
+
+		statusList[idx] = status
+	}
+
+	return statusList, nil
+}