fix: only store kafka ID in config

Author: Enda Phelan

CONTRIBUTING.md

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

go.sum

@@ -554,6 +554,8 @@ k8s.io/apimachinery v0.19.4 h1:+ZoddM7nbzrDCp0T3SWnyxqf8cbWPT2fkZImoyvHUG0=
 k8s.io/apimachinery v0.19.4/go.mod h1:DnPGDnARWFvYa3pMHgSxtbZb7gpzzAZ1pTfaUNDVlmA=
 k8s.io/client-go v0.19.4 h1:85D3mDNoLF+xqpyE9Dh/OtrJDyJrSRKkHmDXIbEzer8=
 k8s.io/client-go v0.19.4/go.mod h1:ZrEy7+wj9PjH5VMBCuu/BDlvtUAku0oVFk4MmnW9mWA=
+k8s.io/client-go v1.5.1 h1:XaX/lo2/u3/pmFau8HN+sB5C/b4dc4Dmm2eXjBH4p1E=
+k8s.io/client-go v11.0.0+incompatible h1:LBbX2+lOwY9flffWlJM7f1Ct8V2SRNiMRDFeiwnJo9o=
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.2.0 h1:XRvcwJozkgZ1UQJmfMGpvRthQHOvihEhYtDfAaxMz/A=

pkg/cmd/kafka/connect/connect.go

@@ -135,12 +135,21 @@ 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': %w", kafkaCfg.ClusterID)
+	}
+
 	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 +159,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 +250,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 +263,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 +322,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

@@ -96,8 +96,6 @@ func runCreate(opts *options) error {
 	}
 
 	kafkaCfg := &config.KafkaConfig{
-		ClusterName: response.Name,
-		ClusterHost: response.BootstrapServerHost,
 		ClusterID:   response.Id,
 	}
 

pkg/cmd/kafka/use/use.go

@@ -64,8 +64,6 @@ 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,
 	}
 
 	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)