Added ability to get Kafka CA for first broker

README.md

@@ -24,6 +24,8 @@ The ``clowder`` library also comes with several other helpers
 
 * ``clowder.LoadedConfig.RdsCa()`` - creates a temporary file with the RDSCa and 
   returns the filename.
+* ``clowder.LoadedConfig.KafkaCa(<BrokerConfig>)`` - creates a temporary file with the KafkaCa and 
+  returns the filename, if broker not givne, first is chosen.
 * ``clowder.KafkaTopics`` - returns a map of KafkaTopics using the requestedName
   as the key and the topic object as the value.
 * ``clowder.KafkaServers`` - returns a list of Kafka Broker URLs.

go.mod

@@ -1,3 +1,11 @@
 module github.com/redhatinsights/app-common-go
 
-go 1.13
+go 1.17
+
+require github.com/stretchr/testify v1.8.0
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

go.sum

@@ -0,0 +1,15 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0 h1:pSgiaMZlXftHpm5L7V1+rVB+AZJydKsMxsQBIJw4PKk=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

pkg/api/v1/config.go

@@ -97,8 +97,22 @@ func (a AppConfig) RdsCa() (string, error) {
 
 // KafkaCa writes the Kafka CA from the JSON config to a temporary file and returns
 // the path
-func (a AppConfig) KafkaCa(broker BrokerConfig) (string, error) {
-	return writeContent("kafkaca", "kafka", broker.Cacert)
+func (a AppConfig) KafkaCa(brokers ...BrokerConfig) (string, error) {
+	if len(brokers) == 0 {
+		if len(LoadedConfig.Kafka.Brokers) == 0 {
+			return "", fmt.Errorf("no broker availabl")
+		}
+		brokers = LoadedConfig.Kafka.Brokers
+	}
+	return writeContent("kafkaca", "kafka", brokers[0].Cacert)
+}
+
+func (a AppConfig) KafkaFirstCa() (string, error) {
+	if a.Kafka == nil || len(a.Kafka.Brokers) == 0 || a.Kafka.Brokers[0].Cacert == nil {
+		return "", fmt.Errorf("could not find ca for first broker")
+	}
+	file := a.Kafka.Brokers[0].Cacert
+	return writeContent("kafkaca", "kafka", file)
 }
 
 func writeContent(dir string, file string, contentString *string) (string, error) {
@@ -109,7 +123,7 @@ func writeContent(dir string, file string, contentString *string) (string, error
 	}
 
 	if contentString == nil {
-		return "", fmt.Errorf("No RDS available")
+		return "", fmt.Errorf("no RDS available")
 	}
 
 	content := []byte(*contentString)

pkg/api/v1/config_test.go

@@ -4,72 +4,47 @@ import (
 	"io/ioutil"
 	"log"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestClientLoad(t *testing.T) {
-	if LoadedConfig == nil {
-		log.Fatal("Config didn't load in init()")
-	}
+	assert.NotNil(t, LoadedConfig, "Config didn't load in init()")
+	assert.Len(t, LoadedConfig.Kafka.Brokers, 1, "Kafka brokers not loaded")
+	assert.Equal(t, 27015, *(LoadedConfig.Kafka.Brokers[0].Port), "Kafka port was not loaded")
+	assert.Contains(t, KafkaTopics, "originalName", "Kafka Topic not found")
+	assert.Equal(t, "someTopic", KafkaTopics["originalName"].Name, "Wrong topic name")
+	assert.Contains(t, ObjectBuckets, "reqname", "ObjectBucket not found")
+	assert.Equal(t, "name", ObjectBuckets["reqname"].Name, "Wrong bucket name")
 
-	if len(LoadedConfig.Kafka.Brokers) < 1 {
-		log.Fatal("Kafka brokers not loaded")
-	}
-	if *(LoadedConfig.Kafka.Brokers[0].Port) != 27015 {
-		log.Fatal("Kafka port was not loaded")
-	}
-	val, ok := KafkaTopics["originalName"]
-	if !ok {
-		log.Fatal("Kafka Topic not found")
-	}
-	if val.Name != "someTopic" {
-		log.Fatal("Wrong topic name")
-	}
-	bucket, ok := ObjectBuckets["reqname"]
-	if !ok {
-		log.Fatal("Object bucket not found")
-	}
-	if bucket.Name != "name" {
-		log.Fatal("Wrong bucket name")
-	}
-	if KafkaServers[0] != "broker-host:27015" {
-		log.Fatal("Wrong broker host")
-	}
-	if IsClowderEnabled() == false {
-		log.Fatal("Should be true if env var ACG_CONFIG is present")
-	}
-	if LoadedConfig.FeatureFlags.Hostname != "ff-server.server.example.com" {
-		log.Fatal("Wrong feature flag hostname")
-	}
-	if LoadedConfig.FeatureFlags.Scheme != "http" {
-		log.Fatal("Wrong feature flag scheme")
-	}
-	if DependencyEndpoints["app1"]["endpoint1"].Port != 8000 {
-		log.Fatal("endpoint had wrong port")
-	}
-	if DependencyEndpoints["app2"]["endpoint2"].Name != "endpoint2" {
-		log.Fatal("endpoint had wrong name")
-	}
-	if PrivateDependencyEndpoints["app1"]["endpoint1"].Port != 10000 {
-		log.Fatal("endpoint had wrong port")
-	}
-	if PrivateDependencyEndpoints["app2"]["endpoint2"].Name != "endpoint2" {
-		log.Fatal("endpoint had wrong name")
-	}
+	assert.ElementsMatch(t, []string{"broker-host:27015"}, KafkaServers)
+	assert.True(t, IsClowderEnabled(), "Should be true if env var ACG_CONFIG is present")
 
-	rdsFilename, err := LoadedConfig.RdsCa()
+	assert.Equal(t, "ff-server.server.example.com", LoadedConfig.FeatureFlags.Hostname, "Wrong feature flag hostname")
+	assert.Equal(t, "http", string(LoadedConfig.FeatureFlags.Scheme), "Wrong feature flag scheme")
 
-	if err != nil {
-		log.Fatal("error in creating RDSCa file")
-	}
+	assert.Equal(t, 8000, DependencyEndpoints["app1"]["endpoint1"].Port, "endpoint had wrong port")
+	assert.Equal(t, "endpoint2", DependencyEndpoints["app2"]["endpoint2"].Name, "endpoint had wrong name")
+	assert.Equal(t, 10000, PrivateDependencyEndpoints["app1"]["endpoint1"].Port, "endpoint had wrong port")
+	assert.Equal(t, "endpoint2", PrivateDependencyEndpoints["app2"]["endpoint2"].Name, "endpoint had wrong name")
 
+	rdsFilename, err := LoadedConfig.RdsCa()
+	assert.Nil(t, err, "error in creating RDSCa file")
 	content, err := ioutil.ReadFile(rdsFilename)
-	if err != nil {
-		log.Fatal("error reading ca")
-	}
-
-	if string(content) != *LoadedConfig.Database.RdsCa {
-		log.Fatal("ca didn't match")
-	}
+	assert.Nil(t, err, "error reading ca")
+	assert.Equal(t, *LoadedConfig.Database.RdsCa, string(content), "rds ca didn't match")
+
+	kafkaFilename, err := LoadedConfig.KafkaCa(LoadedConfig.Kafka.Brokers[0])
+	assert.Nil(t, err, "error in creating KafkaCa file")
+	content, err = ioutil.ReadFile(kafkaFilename)
+	assert.Nil(t, err, "error reading ca")
+	assert.Equal(t, *LoadedConfig.Kafka.Brokers[0].Cacert, string(content), "kafka ca didn't match")
+
+	kafkaFilename, err = LoadedConfig.KafkaCa()
+	assert.Nil(t, err, "error in creating KafkaCa file")
+	content, err = ioutil.ReadFile(kafkaFilename)
+	assert.Nil(t, err, "error reading ca")
+	assert.Equal(t, *LoadedConfig.Kafka.Brokers[0].Cacert, string(content), "kafka ca didn't match")
 }
 
 func TestEmptyRDSCa(t *testing.T) {

tests/test.json

@@ -15,7 +15,8 @@
         "brokers": [
             {
                 "hostname": "broker-host",
-                "port": 27015
+                "port": 27015,
+                "cacert": "kafkaca"
             }
         ],
         "topics": [