RHINENG-3742: Multiple kafka brokers

base/mqueue/mqueue_impl_gokafka.go

@@ -68,14 +68,13 @@ func (t *kafkaGoWriterImpl) WriteMessages(ctx context.Context, msgs ...KafkaMess
 }
 
 func NewKafkaReaderFromEnv(topic string) Reader {
-	kafkaAddress := utils.FailIfEmpty(utils.Cfg.KafkaAddress, "KAFKA_ADDRESS")
 	kafkaGroup := utils.FailIfEmpty(utils.Cfg.KafkaGroup, "KAFKA_GROUP")
 	minBytes := utils.Cfg.KafkaReaderMinBytes
 	maxBytes := utils.Cfg.KafkaReaderMaxBytes
 	maxAttempts := utils.Cfg.KafkaReaderMaxAttempts
 
 	config := kafka.ReaderConfig{
-		Brokers:     []string{kafkaAddress},
+		Brokers:     utils.Cfg.KafkaServers,
 		Topic:       topic,
 		GroupID:     kafkaGroup,
 		MinBytes:    minBytes,
@@ -90,11 +89,10 @@ func NewKafkaReaderFromEnv(topic string) Reader {
 }
 
 func NewKafkaWriterFromEnv(topic string) Writer {
-	kafkaAddress := utils.FailIfEmpty(utils.Cfg.KafkaAddress, "KAFKA_ADDRESS")
 	maxAttempts := utils.Cfg.KafkaWriterMaxAttempts
 
 	config := kafka.WriterConfig{
-		Brokers: []string{kafkaAddress},
+		Brokers: utils.Cfg.KafkaServers,
 		Topic:   topic,
 		// By default the writer will wait for a second (or until the buffer is filled by different goroutines)
 		// before sending the batch of messages. Disable this, and use it in 'non-batched' mode

base/utils/config.go

@@ -43,6 +43,7 @@ type Config struct {
 	Ratelimit          int
 
 	// kafka
+	KafkaServers           []string
 	KafkaAddress           string
 	KafkaSslEnabled        bool
 	KafkaSslCert           string
@@ -181,6 +182,7 @@ func initAPIromClowder() {
 func initKafkaFromClowder() {
 	if len(clowder.LoadedConfig.Kafka.Brokers) > 0 {
 		Cfg.KafkaSaslType = nil
+		Cfg.KafkaServers = clowder.KafkaServers
 		brokerCfg := clowder.LoadedConfig.Kafka.Brokers[0]
 		kafkaHost := brokerCfg.Hostname
 		kafkaPort := *brokerCfg.Port
@@ -213,6 +215,7 @@ func initKafkaFromClowder() {
 }
 
 type Endpoint clowder.DependencyEndpoint
+type PrivateEndpoint clowder.PrivateDependencyEndpoint
 
 func initServicesFromClowder() {
 	webappName := "webapp-service"
@@ -236,13 +239,13 @@ func initServicesFromClowder() {
 		if e.App == "patchman" {
 			switch e.Name {
 			case "manager":
-				Cfg.ManagerPrivateAddress = (*Endpoint)(&e).buildURL()
+				Cfg.ManagerPrivateAddress = (*PrivateEndpoint)(&e).buildURL()
 			case "listener":
-				Cfg.ListenerPrivateAddress = (*Endpoint)(&e).buildURL()
+				Cfg.ListenerPrivateAddress = (*PrivateEndpoint)(&e).buildURL()
 			case "evaluator-upload":
-				Cfg.EvaluatorUploadPrivateAddress = (*Endpoint)(&e).buildURL()
+				Cfg.EvaluatorUploadPrivateAddress = (*PrivateEndpoint)(&e).buildURL()
 			case "evaluator-recalc":
-				Cfg.EvaluatorRecalcPrivateAddress = (*Endpoint)(&e).buildURL()
+				Cfg.EvaluatorRecalcPrivateAddress = (*PrivateEndpoint)(&e).buildURL()
 			}
 		}
 	}
@@ -260,6 +263,18 @@ func (e *Endpoint) buildURL() string {
 	return fmt.Sprintf("%s://%s:%d", scheme, e.Hostname, port)
 }
 
+func (e *PrivateEndpoint) buildURL() string {
+	port := e.Port
+	scheme := "http"
+	if clowder.LoadedConfig.TlsCAPath != nil {
+		scheme += "s"
+		if e.TlsPort != nil {
+			port = *e.TlsPort
+		}
+	}
+	return fmt.Sprintf("%s://%s:%d", scheme, e.Hostname, port)
+}
+
 func initCloudwatchFromClowder() {
 	cwCfg := clowder.LoadedConfig.Logging.Cloudwatch
 	if cwCfg != nil {

conf/cdappconfig.json

@@ -50,18 +50,21 @@
     },
     "endpoints": [
         {
+            "apiPath": "api/patch/v3",
             "app": "vmaas",
             "hostname": "platform",
             "name": "webapp-service",
             "port": 9001
         },
         {
+            "apiPath": "api/patch/v3",
             "app": "vmaas",
             "hostname": "platform",
             "name": "webapp-go",
             "port": 9001
         },
         {
+            "apiPath": "api/patch/v3",
             "app": "rbac",
             "hostname": "platform",
             "name": "",

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/pkg/errors v0.9.1
 	github.com/pmezard/go-difflib v1.0.0
 	github.com/prometheus/client_golang v1.15.1
-	github.com/redhatinsights/app-common-go v1.6.6
+	github.com/redhatinsights/app-common-go v1.6.7
 	github.com/redhatinsights/identity v0.0.0-20220719174832-36a7b1cbeff1
 	github.com/redhatinsights/platform-go-middlewares v0.20.0
 	github.com/segmentio/kafka-go v0.4.40

go.sum

@@ -217,8 +217,8 @@ github.com/prometheus/common v0.44.0 h1:+5BrQJwiBB9xsMygAB3TNvpQKOwlkc25LbISbrdO
 github.com/prometheus/common v0.44.0/go.mod h1:ofAIvZbQ1e/nugmZGz4/qCb9Ap1VoSTIO7x0VV9VvuY=
 github.com/prometheus/procfs v0.10.1 h1:kYK1Va/YMlutzCGazswoHKo//tZVlFpKYh+PymziUAg=
 github.com/prometheus/procfs v0.10.1/go.mod h1:nwNm2aOCAYw8uTR/9bWRREkZFxAUcWzPHWJq+XBB/FM=
-github.com/redhatinsights/app-common-go v1.6.6 h1:daOwCpGtW6IxGd9iO6TY3yoaV/HaHKjo/j/05dV0hHM=
-github.com/redhatinsights/app-common-go v1.6.6/go.mod h1:6gzRyg8ZyejwMCksukeAhh2ZXOB3uHSmBsbP06fG2PQ=
+github.com/redhatinsights/app-common-go v1.6.7 h1:cXWW0F6ZW53RLRr54gn7Azo9CLTysYOmFDR0D0Qd0Fs=
+github.com/redhatinsights/app-common-go v1.6.7/go.mod h1:6gzRyg8ZyejwMCksukeAhh2ZXOB3uHSmBsbP06fG2PQ=
 github.com/redhatinsights/identity v0.0.0-20220719174832-36a7b1cbeff1 h1:oCJ53EClFw5LdNWLKYmwGeqkUoYLV0Wy6ZaZlwjDYDY=
 github.com/redhatinsights/identity v0.0.0-20220719174832-36a7b1cbeff1/go.mod h1:B0Dwuuaghxyqo8ltmLZyLpKQFKU6r8cE2YRrO0bVdXM=
 github.com/redhatinsights/platform-go-middlewares v0.20.0 h1:qwK9ArGYRlORsZ56PXXLJrGvzTsMe3bk2lR+WN5aIjM=