get cluster info for version cmd

internal/ctl/flags.go

@@ -1,15 +1,26 @@
 package ctl
 
 import (
+	"fmt"
 	"time"
 
 	"github.com/spf13/cobra"
 	"github.com/spf13/pflag"
 	"github.com/spf13/viper"
 )
 
+func AddClusterFlag(cmd *cobra.Command, required bool, clusterURL *string) {
+	description := "the URL of the connect cluster to manage"
+
+	if required {
+		description = requiredDescription(&description)
+	}
+
+	BindStringVarP(cmd.Flags(), clusterURL, "", "cluster", "c", description)
+}
+
 func AddCommonConnectorsFlags(cmd *cobra.Command, clusterURL *string) {
-	BindStringVarP(cmd.Flags(), clusterURL, "", "cluster", "c", "the URL of the connect cluster to manage (required)")
+	AddClusterFlag(cmd, true, clusterURL)
 }
 
 func AddOutputFlags(cmd *cobra.Command, output *string) {
@@ -67,3 +78,7 @@ func BindIntVar(f *pflag.FlagSet, p *int, value int, long, description string) {
 	_ = viper.BindPFlag(long, f.Lookup(long))
 	viper.SetDefault(long, value)
 }
+
+func requiredDescription(desc *string) string {
+	return fmt.Sprintf("%s (required)", *desc)
+}

internal/ctl/version/version.go

@@ -4,28 +4,86 @@ import (
 	"fmt"
 	"runtime"
 
+	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
 
+	"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"
 )
 
+type versionCmdParams struct {
+	ClusterURL string
+}
+
 // Command creates the the management commands
 func Command() *cobra.Command {
+	params := &versionCmdParams{}
+
 	versionCmd := &cobra.Command{
 		Use:   "version",
 		Short: "Display version information",
 		Long:  "",
-		Run:   doVersion,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			return doVersion(cmd, params)
+		},
 	}
 
+	ctl.AddClusterFlag(versionCmd, false, &params.ClusterURL)
+
 	return versionCmd
 }
 
-func doVersion(cmd *cobra.Command, args []string) {
+func doVersion(cmd *cobra.Command, params *versionCmdParams) error {
+	var (
+		clusterInfo *connect.ClusterInfo
+		err         error
+	)
+
+	if params.ClusterURL != "" {
+		clusterInfo, err = getClusterInfo(params.ClusterURL)
+		if err != nil {
+			return err
+		}
+	}
+
 	fmt.Printf("Version: %s\n", version.Version)
 	fmt.Printf("Commit: %s\n", version.GitHash)
 	fmt.Printf("Build Date: %s\n", version.BuildDate)
 	fmt.Printf("GO Version: %s\n", runtime.Version())
 	fmt.Printf("GOOS: %s\n", runtime.GOOS)
 	fmt.Printf("GOARCH: %s\n", runtime.GOARCH)
+
+	if clusterInfo != nil {
+		fmt.Printf("Connect Worker Version: %s\n", clusterInfo.Version)
+	}
+
+	return nil
+}
+
+func getClusterInfo(clusterURL string) (*connect.ClusterInfo, error) {
+	config := &manager.Config{
+		ClusterURL: clusterURL,
+		Version:    version.Version,
+	}
+
+	userAgent := fmt.Sprintf("90poe.io/connectctl/%s", version.Version)
+
+	client, err := connect.NewClient(config.ClusterURL, connect.WithUserAgent(userAgent))
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating connect client")
+	}
+
+	mngr, err := manager.NewConnectorsManager(client, config)
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating connectors manager")
+	}
+
+	clusterInfo, err := mngr.GetClusterInfo()
+	if err != nil {
+		return nil, errors.Wrap(err, "error getting cluster info")
+	}
+
+	return clusterInfo, nil
 }

pkg/client/connect/cluster.go

@@ -0,0 +1,23 @@
+package connect
+
+import (
+	"net/http"
+)
+
+// This is new and not from the original author
+
+type ClusterInfo struct {
+	Version        string `json:"version"`
+	Commit         string `json:"commit"`
+	KafkaClusterID string `json:"kafka_cluster_id"`
+}
+
+// GetClusterInfo retrieves information about a cluster
+//
+// See: https://docs.confluent.io/current/connect/references/restapi.html#kconnect-cluster
+func (c *Client) GetClusterInfo() (*ClusterInfo, *http.Response, error) {
+	path := ""
+	clusterInfo := new(ClusterInfo)
+	response, err := c.get(path, &clusterInfo)
+	return clusterInfo, response, err
+}

pkg/manager/cluster.go

@@ -0,0 +1,18 @@
+package manager
+
+import (
+	"github.com/pkg/errors"
+
+	"github.com/90poe/connectctl/pkg/client/connect"
+)
+
+// GetClusterInfo returns kafka cluster info
+func (c *ConnectorManager) GetClusterInfo() (*connect.ClusterInfo, error) {
+	clusterInfo, _, err := c.client.GetClusterInfo()
+
+	if err != nil {
+		return nil, errors.Wrap(err, "error getting cluster info")
+	}
+
+	return clusterInfo, nil
+}

pkg/manager/manager.go

@@ -13,6 +13,7 @@ import (
 type ConnectorSource func() ([]connect.Connector, error)
 
 type client interface {
+	GetClusterInfo() (*connect.ClusterInfo, *http.Response, error)
 	CreateConnector(conn connect.Connector) (*http.Response, error)
 	ListConnectors() ([]string, *http.Response, error)
 	GetConnector(name string) (*connect.Connector, *http.Response, error)