Add kafka describe <name> command

command/kafka/command.go

@@ -83,9 +83,10 @@ func (c *Command) init() error {
 		RunE:  c.list,
 	})
 	c.AddCommand(&cobra.Command{
-		Use:   "describe",
+		Use:   "describe <name>",
 		Short: "Describe a Kafka cluster.",
 		RunE:  c.describe,
+		Args:  cobra.ExactArgs(1),
 	})
 	c.AddCommand(&cobra.Command{
 		Use:   "delete",
@@ -121,7 +122,13 @@ func (c *Command) list(cmd *cobra.Command, args []string) error {
 }
 
 func (c *Command) describe(cmd *cobra.Command, args []string) error {
-	return shared.ErrNotImplemented
+	req := &schedv1.KafkaCluster{AccountId: c.config.Auth.Account.Id, Name: args[0]}
+	cluster, err := c.kafka.Describe(context.Background(), req)
+	if err != nil {
+		return common.HandleError(err)
+	}
+	fmt.Println(cluster)
+	return nil
 }
 
 func (c *Command) update(cmd *cobra.Command, args []string) error {

command/kafka/grpc.go

@@ -21,6 +21,14 @@ func (c *GRPCClient) List(ctx context.Context, cluster *schedv1.KafkaCluster) ([
 	return resp.Clusters, nil
 }
 
+func (c *GRPCClient) Describe(ctx context.Context, cluster *schedv1.KafkaCluster) (*schedv1.KafkaCluster, error) {
+	resp, err := c.client.Describe(ctx, &schedv1.GetKafkaClusterRequest{Cluster: cluster})
+	if err != nil {
+		return nil, shared.ConvertGRPCError(err)
+	}
+	return resp.Cluster, nil
+}
+
 // The gRPC server the GPRClient talks to. Plugin authors implement this if they're using Go.
 type GRPCServer struct {
 	Impl Kafka
@@ -30,3 +38,8 @@ func (s *GRPCServer) List(ctx context.Context, req *schedv1.GetKafkaClustersRequ
 	r, err := s.Impl.List(ctx, req.Cluster)
 	return &schedv1.GetKafkaClustersReply{Clusters: r}, err
 }
+
+func (s *GRPCServer) Describe(ctx context.Context, req *schedv1.GetKafkaClusterRequest) (*schedv1.GetKafkaClusterReply, error) {
+	r, err := s.Impl.Describe(ctx, req.Cluster)
+	return &schedv1.GetKafkaClusterReply{Cluster: r}, err
+}

command/kafka/interface.go

@@ -13,6 +13,7 @@ import (
 
 type Kafka interface {
 	List(ctx context.Context, cluster *schedv1.KafkaCluster) ([]*schedv1.KafkaCluster, error)
+	Describe(ctx context.Context, cluster *schedv1.KafkaCluster) (*schedv1.KafkaCluster, error)
 }
 
 type Plugin struct {

http/kafka.go

@@ -38,3 +38,16 @@ func (s *KafkaService) List(cluster *schedv1.KafkaCluster) ([]*schedv1.KafkaClus
 	}
 	return reply.Clusters, resp, nil
 }
+
+// Describe returns details for a given kafka cluster.
+func (s *KafkaService) Describe(cluster *schedv1.KafkaCluster) (*schedv1.KafkaCluster, *http.Response, error) {
+	reply := new(schedv1.GetKafkaClustersReply)
+	resp, err := s.sling.New().Get("/api/clusters").QueryStruct(cluster).Receive(reply, reply)
+	if err != nil {
+		return nil, resp, errors.Wrap(err, "unable to fetch kafka clusters")
+	}
+	if reply.Error != nil {
+		return nil, resp, errors.Wrap(reply.Error, "error fetching kafka clusters")
+	}
+	return reply.Clusters[0], resp, nil
+}

plugin/confluent-kafka-plugin/main.go

@@ -73,6 +73,12 @@ func (c *Kafka) List(ctx context.Context, cluster *schedv1.KafkaCluster) ([]*sch
 	return ret, shared.ConvertAPIError(err)
 }
 
+func (c *Kafka) Describe(ctx context.Context, cluster *schedv1.KafkaCluster) (*schedv1.KafkaCluster, error) {
+	c.Logger.Log("msg", "kafka.Describe()")
+	ret, _, err := c.Client.Kafka.Describe(cluster)
+	return ret, shared.ConvertAPIError(err)
+}
+
 func check(err error) {
 	if err != nil {
 		golog.Fatal(err)

shared/kafka/kafka.proto

@@ -7,5 +7,6 @@ import "kafka/scheduler/v1/scheduler.proto";
 // TODO: should this definition be in cc-structs?
 service Kafka {
   rpc List(kafka.scheduler.v1.GetKafkaClustersRequest) returns (kafka.scheduler.v1.GetKafkaClustersReply) {}
+  rpc Describe(kafka.scheduler.v1.GetKafkaClusterRequest) returns (kafka.scheduler.v1.GetKafkaClusterReply) {}
   // ...
 }