docker-compose.yml: Add proxy config for kafka-ui

.env.sample

@@ -130,6 +130,7 @@ MESSAGE_BROKER_USERNAME=username
 MESSAGE_BROKER_PASSWORD=password
 MESSAGE_BROKER_ADDRESS=kafka:9094
 KAFKA_UI_PORT=8011
+KAFKA_UI_PRODUCTION_HOST=kafka-ui.test.com
 
 REPORT_GENERATED_BROKER_TOPIC=report_generated
 REPORT_REQUEST_FAILED_BROKER_TOPIC=report_request_failed

docker-compose.dev.yml

@@ -144,7 +144,6 @@ services:
 volumes:
   redisdata:
   redisinsight:
-  kafkaui_data:
 
 networks:
   default:

docker-compose.yml

@@ -296,6 +296,9 @@ services:
       - ${KAFKA_UI_PORT:-8011}:8080
     environment:
       DYNAMIC_CONFIG_ENABLED: "true"
+      VIRTUAL_HOST: ${KAFKA_UI_PRODUCTION_HOST}
+      LETSENCRYPT_HOST: ${KAFKA_UI_PRODUCTION_HOST}
+      VIRTUAL_PORT: 8080
     volumes:
       - kafkaui-data:/etc/kafkaui
 

go/pkg/message-broker/broker.go

@@ -2,5 +2,5 @@ package messagebroker
 
 type MessageBroker[T any] interface {
 	Publish(key string, message T) error
-	Subscribe(ch chan<- T, errCh chan<- error)
+	Subscribe(messages chan<- T, errors chan<- error)
 }

go/pkg/message-broker/json_kafka.go

@@ -29,24 +29,24 @@ func (b *KafkaJsonMessageBroker[T]) Publish(key string, message T) error {
 	return b.broker.Publish(context.Background(), []byte(key), encodedMessage)
 }
 
-func (b *KafkaJsonMessageBroker[T]) Subscribe(ch chan<- T, errChn chan<- error) {
+func (b *KafkaJsonMessageBroker[T]) Subscribe(messages chan<- T, errors chan<- error) {
 
 	msgChannel := make(chan []byte)
 
-	go b.broker.Subscribe(context.Background(), msgChannel, errChn)
+	go b.broker.Subscribe(context.Background(), msgChannel, errors)
 
 	for {
 		message := <-msgChannel
 		var decodedMessage T
 
 		err := json.Unmarshal(message, &decodedMessage)
 		if err != nil {
-			b.logger.Error("Failed to decode a broker json message", zap.Error(err),
+			b.logger.Error("Failed to decode broker json message", zap.Error(err),
 				zap.Any("messsage", message))
-			errChn <- err
+			errors <- err
 		}
 
-		ch <- decodedMessage
+		messages <- decodedMessage
 	}
 
 }

go/pkg/message-broker/kafka.go

@@ -65,16 +65,16 @@ func (b *KafkaMessageBroker) Publish(ctx context.Context, key []byte, value []by
 	return nil
 }
 
-func (b *KafkaMessageBroker) Subscribe(ctx context.Context, channel chan<- []byte, errChannel chan<- error) {
+func (b *KafkaMessageBroker) Subscribe(ctx context.Context, messages chan<- []byte, errors chan<- error) {
 	for {
 		msg, err := b.reader.ReadMessage(ctx)
 
 		if err != nil {
 			b.logger.Error("Failed to read message", zap.Error(err))
-			errChannel <- err
+			errors <- err
 		}
 
-		channel <- msg.Value
+		messages <- msg.Value
 	}
 }
 

go/services/reports/api/openapi.yaml

@@ -317,10 +317,11 @@ components:
           type: string
         customPrompt:
           type: string
-
     reportsPostParams:
       type: object
       properties:
+        correlationId:
+          type: string
         clusterId:
           type: string
         sinceMs:

go/services/reports/internal/brokers/report_request_failed.go

@@ -22,7 +22,7 @@ const (
 
 type ReportRequestFailed struct {
 	CorrelationId string             `json:"correlationId"`
-	ErrorType     ReportRequestError `json:"reportRequest"`
+	ErrorType     ReportRequestError `json:"errorType"`
 	ErrorMessage  string             `json:"errorMessage"`
 	TimestampMs   int64              `json:"timestampMs"`
 }