feat: auto-use kafka cluster after creation (#142)

Author: Enda

.vscode/launch.json

@@ -16,7 +16,7 @@
         "--url=localhost:8000"
       ]
     },
-     {
+    {
       "name": "Logout",
       "type": "go",
       "request": "launch",
@@ -27,6 +27,19 @@
         "logout"
       ]
     },
+    {
+      "name": "Create Kafka",
+      "type": "go",
+      "request": "launch",
+      "mode": "auto",
+      "program": "${workspaceFolder}/cmd/rhoas",
+      "env": {},
+      "args": [
+        "kafka",
+        "create",
+        "--name=my-kafka-instance"
+      ]
+    },
     {
       "name": "List Kafkas",
       "type": "go",

CONTRIBUTING.md

@@ -191,9 +191,7 @@ Your config should look as follows:
   ...
   "services": {
     "kafka": {
-      "clusterHost": "localhost:9092",
-      "clusterId": "1iSY6RQ3JKI8Q0OTmjQFd3ocFRg",
-      "clusterName": "serviceapi"
+      "clusterId": "1iSY6RQ3JKI8Q0OTmjQFd3ocFRg"
     }
   }
   ...

pkg/cmd/kafka/connect/connect.go

@@ -127,6 +127,10 @@ func connectToCluster(connection *pkgConnection.Connection) {
 
 	currentNamespace, _, _ := kubeClientconfig.Namespace()
 	clicfg, err := config.Load()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Could not load config: %v\n", err)
+		return
+	}
 
 	if !clicfg.HasKafka() || forceKafkaSelect {
 		clicfg = useKafka(clicfg, connection)
@@ -135,12 +139,22 @@ func connectToCluster(connection *pkgConnection.Connection) {
 		}
 	}
 
+	kafkaCfg := clicfg.Services.Kafka
+
+	managedservices := connection.NewMASClient()
+	kafkaInstance, _, err := managedservices.DefaultApi.GetKafkaById(context.TODO(), kafkaCfg.ClusterID)
+
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Could not get Kafka cluster with ID '%v': %v", kafkaCfg.ClusterID, err)
+		return
+	}
+
 	if err != nil {
 		fmt.Fprint(os.Stderr, "\nInvalid configuration file", err)
 		return
 	}
 
-	fmt.Fprintf(os.Stderr, statusMsg, color.HiGreenString(clicfg.Services.Kafka.ClusterName), color.HiGreenString(currentNamespace), color.HiGreenString(secretName))
+	fmt.Fprintf(os.Stderr, statusMsg, color.HiGreenString(kafkaInstance.Name), color.HiGreenString(currentNamespace), color.HiGreenString(secretName))
 	if shouldContinue := showQuestion("Do you want to continue?"); shouldContinue == false {
 		return
 	}
@@ -150,7 +164,7 @@ func connectToCluster(connection *pkgConnection.Connection) {
 		return
 	}
 	createSecret(credentials, currentNamespace, clientset)
-	createCR(clicfg, clientset, currentNamespace)
+	createCR(clientset, &kafkaInstance, currentNamespace)
 }
 
 func fileExists(path string) bool {
@@ -241,8 +255,8 @@ func createSecret(credentials *managedservices.TokenResponse, currentNamespace s
 	return secret
 }
 
-func createCR(clicfg *config.Config, clientset *kubernetes.Clientset, namespace string) {
-	crName := secretName + "-" + clicfg.Services.Kafka.ClusterName
+func createCR(clientset *kubernetes.Clientset, kafkaInstance *managedservices.KafkaRequest, namespace string) {
+	crName := secretName + "-" + kafkaInstance.Name
 	crInstance := &connection.ManagedKafkaConnection{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      crName,
@@ -254,7 +268,7 @@ func createCR(clicfg *config.Config, clientset *kubernetes.Clientset, namespace
 		},
 		Spec: connection.ManagedKafkaConnectionSpec{
 			BootstrapServer: connection.BootstrapServerSpec{
-				Host: clicfg.Services.Kafka.ClusterHost,
+				Host: kafkaInstance.BootstrapServerHost,
 			},
 			Credentials: connection.CredentialsSpec{
 				Kind:       connection.ClientCredentials,
@@ -313,7 +327,7 @@ func useKafka(cliconfig *config.Config, connection *pkgConnection.Connection) *c
 	index, _, err := prompt.Run()
 	if err == nil {
 		selectedKafka := response.Items[index]
-		var kafkaConfig config.KafkaConfig = config.KafkaConfig{ClusterID: selectedKafka.Id, ClusterName: selectedKafka.Name, ClusterHost: selectedKafka.BootstrapServerHost}
+		var kafkaConfig config.KafkaConfig = config.KafkaConfig{ClusterID: selectedKafka.Id}
 		cliconfig.Services.SetKafka(&kafkaConfig)
 
 		return cliconfig

pkg/cmd/kafka/create/create.go

@@ -47,8 +47,7 @@ func NewCreateCommand() *cobra.Command {
 			}
 			opts.cfg = cfg
 
-			err = kafka.ValidateName(opts.name)
-			if err != nil {
+			if err = kafka.ValidateName(opts.name); err != nil {
 				return err
 			}
 
@@ -59,8 +58,6 @@ func NewCreateCommand() *cobra.Command {
 	cmd.Flags().StringVarP(&opts.name, flags.FlagName, "n", "", "Name of the new Kafka cluster")
 	cmd.Flags().StringVar(&opts.provider, flags.FlagProvider, "aws", "Cloud provider ID")
 	cmd.Flags().StringVar(&opts.region, flags.FlagRegion, "us-east-1", "Cloud Provider Region ID")
-	// Hardcoded as only true is possible
-	// cmd.Flags().BoolVar(&opts.multiAZ, flags.FlagMultiAZ, true, "Determines if cluster should be provisioned across multiple Availability Zones")
 	cmd.Flags().StringVarP(&opts.outputFormat, "output", "o", "json", "Format to display the Kafka cluster. Choose from: \"json\", \"yaml\", \"yml\"")
 
 	_ = cmd.MarkFlagRequired(flags.FlagName)
@@ -69,29 +66,42 @@ func NewCreateCommand() *cobra.Command {
 }
 
 func runCreate(opts *options) error {
-	connection, err := opts.cfg.Connection()
+	cfg := opts.cfg
+
+	connection, err := cfg.Connection()
 	if err != nil {
 		return fmt.Errorf("Can't create connection: %w", err)
 	}
 
 	client := connection.NewMASClient()
 
+	fmt.Fprintln(os.Stderr, "Creating Kafka cluster")
+
 	kafkaRequest := managedservices.KafkaRequestPayload{Name: opts.name, Region: opts.region, CloudProvider: opts.provider, MultiAz: true}
 	response, _, err := client.DefaultApi.CreateKafka(context.Background(), true, kafkaRequest)
 
 	if err != nil {
 		return fmt.Errorf("Error while requesting new Kafka cluster: %w", err)
 	}
 
-	fmt.Fprintf(os.Stderr, "Created new Kafka cluster:\n")
+	fmt.Fprintln(os.Stderr, "Created Kafka cluster:")
 
 	switch opts.outputFormat {
 	case "json":
 		data, _ := json.MarshalIndent(response, "", cmdutil.DefaultJSONIndent)
-		fmt.Print(string(data))
+		fmt.Println(string(data))
 	case "yaml", "yml":
 		data, _ := yaml.Marshal(response)
-		fmt.Print(string(data))
+		fmt.Println(string(data))
+	}
+
+	kafkaCfg := &config.KafkaConfig{
+		ClusterID: response.Id,
+	}
+
+	cfg.Services.SetKafka(kafkaCfg)
+	if err := config.Save(cfg); err != nil {
+		return fmt.Errorf("Unable to automatically use Kafka cluster: %w", err)
 	}
 
 	return nil

pkg/cmd/kafka/use/use.go

@@ -63,9 +63,7 @@ func runUse(opts *options) error {
 
 	// build Kafka config object from the response
 	var kafkaConfig config.KafkaConfig = config.KafkaConfig{
-		ClusterID:   res.Id,
-		ClusterName: res.Name,
-		ClusterHost: res.BootstrapServerHost,
+		ClusterID: res.Id,
 	}
 
 	cfg.Services.SetKafka(&kafkaConfig)

pkg/cmd/serviceaccount/create/create.go

@@ -16,21 +16,18 @@ import (
 // Templates
 var (
 	templateProperties = heredoc.Doc(`
-## Credentials for Kafka cluster: '%v' 
 ## Generated by rhoas cli
 kafka_user=%v
 kafka_password=%v
 `)
 
 	templateEnv = heredoc.Doc(`
-## Credentials for Kafka cluster: '%v' 
 ## Generated by rhoas cli
 KAFKA_USER=%v
 KAFKA_PASSWORD=%v
 `)
 
 	templateKafkaPlain = heredoc.Doc(`
-## Credentials for Kafka cluster: '%v' 
 ## Generated by rhoas cli
 kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="%v" password="%v";
 kafka.sasl.mechanism=PLAIN
@@ -149,7 +146,7 @@ func runCreate(opts *options) error {
 	}
 
 	fmt.Fprintf(os.Stderr, "Writing credentials to %v \n", fileName)
-	fileContent := fmt.Sprintf(fileFormat, cfg.Services.Kafka.ClusterName, response.ClientID, response.ClientSecret)
+	fileContent := fmt.Sprintf(fileFormat, response.ClientID, response.ClientSecret)
 
 	dataToWrite := []byte(fileContent)
 

pkg/config/config.go

@@ -32,9 +32,7 @@ type ServiceConfigMap struct {
 
 // KafkaConfig is the config for the managed Kafka service
 type KafkaConfig struct {
-	ClusterHost string `json:"clusterHost"`
-	ClusterID   string `json:"clusterId"`
-	ClusterName string `json:"clusterName"`
+	ClusterID string `json:"clusterId"`
 }
 
 func (c *Config) SetAccessToken(accessToken string) {
@@ -77,9 +75,7 @@ func (s *ServiceConfigMap) SetKafka(k *KafkaConfig) {
 // Remove the current Kafka cluster from the config
 func (s *ServiceConfigMap) RemoveKafka() {
 	s.Kafka = &KafkaConfig{
-		ClusterID:   "",
-		ClusterHost: "",
-		ClusterName: "",
+		ClusterID: "",
 	}
 }
 

pkg/sdk/kafka/topics/topics.go

@@ -1,6 +1,7 @@
 package topics
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"os"
@@ -31,15 +32,28 @@ func brokerConnect() (broker *kafka.Conn, ctl *kafka.Conn) {
 		fmt.Fprint(os.Stderr, err)
 	}
 
-	if cfg.Services.Kafka.ClusterHost == "" {
+	if cfg.Services.Kafka.ClusterID == "" {
 		fmt.Fprint(os.Stderr, "No Kafka selected. Run rhoas kafka use")
 		panic("Missing config")
 	}
+
+	connection, err := cfg.Connection()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Could not create connection: %v\n", err)
+	}
+
+	managedservices := connection.NewMASClient()
+	kafkaInstance, _, err := managedservices.DefaultApi.GetKafkaById(context.TODO(), cfg.Services.Kafka.ClusterID)
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Could not get Kafka instance: %v\n", err)
+		return
+	}
+
 	var clusterURL string
-	if strings.HasPrefix(cfg.Services.Kafka.ClusterHost, "localhost") {
-		clusterURL = cfg.Services.Kafka.ClusterHost
+	if strings.HasPrefix(kafkaInstance.BootstrapServerHost, "localhost") {
+		clusterURL = kafkaInstance.BootstrapServerHost
 	} else {
-		clusterURL = cfg.Services.Kafka.ClusterHost + ":443"
+		clusterURL = kafkaInstance.BootstrapServerHost + ":443"
 	}
 
 	conn, err := dialer.Dial("tcp", clusterURL)