enable kafka auth credentials

docker-init

@@ -3,4 +3,6 @@ echo "Start consumer with kafka.brokers $KAFKA_BROKERS, kafka.topic $KAFKA_TOPIC
 /consumer --kafka.brokers "$KAFKA_BROKERS" \
   --kafka.topic "$KAFKA_TOPIC" \
   --kafka.consumer_group "$KAFKA_CONSUMER_GROUP" \
-  --customerid "$CONSUMER_FILTER_CUSTOMERID"
+  --customerid "$CONSUMER_FILTER_CUSTOMERID" \
+  --kafka.user "$KAFKA_USER" \
+  --kafka.pass "$KAFKA_PASS"

dockerfile

@@ -7,4 +7,6 @@ ENV KAFKA_BROKERS="127.0.0.1:9092,[::1]:9092"
 ENV KAFKA_TOPIC="flow-messages-enriched"
 ENV KAFKA_CONSUMER_GROUP="dashboard"
 ENV CONSUMER_FILTER_CUSTOMERID="10109"
+ENV KAFKA_USER=""
+ENV KAFKA_PASS=""
 ENTRYPOINT [ "/docker-init" ]

main.go

@@ -44,6 +44,13 @@ func main() {
 	var exportInterval = 15 * time.Second
 	tophostExporter.Initialize(promExporter, maxHosts, exportInterval)
 
+	// Set kafka auth
+	if *kafkaUser != "" {
+		kafkaConn.SetAuth(*kafkaUser, *kafkaPass)
+	} else {
+		kafkaConn.SetAuthAnon()
+	}
+
 	// Establish Kafka Connection
 	kafkaConn.StartConsumer(*kafkaBroker, []string{*kafkaInTopic}, *kafkaConsumerGroup, sarama.OffsetNewest)
 	defer kafkaConn.Close()

options.go

@@ -11,6 +11,8 @@ var (
 	kafkaConsumerGroup = flag.String("kafka.consumer_group", "dashboard", "Kafka Consumer Group")
 	kafkaInTopic       = flag.String("kafka.topic", "flow-messages-enriched", "Kafka topic to consume from")
 	kafkaBroker        = flag.String("kafka.brokers", "127.0.0.1:9092,[::1]:9092", "Kafka brokers separated by commas")
+	kafkaUser          = flag.String("kafka.user", "", "Kafka username to authenticate with")
+	kafkaPass          = flag.String("kafka.pass", "", "Kafka password to authenticate with")
 
 	// prometheus options
 	// TODO listen on addr