Added SASL options for ouput kafka plugin (influxdata#2721)

CHANGELOG.md

@@ -41,6 +41,7 @@ be deprecated eventually.
 
 ### Features
 
+- [#2721](https://github.com/influxdata/telegraf/pull/2721): Added SASL options for kafka output plugin.
 - [#2723](https://github.com/influxdata/telegraf/pull/2723): Added SSL configuration for input haproxy.
 - [#2494](https://github.com/influxdata/telegraf/pull/2494): Add interrupts input plugin.
 - [#2094](https://github.com/influxdata/telegraf/pull/2094): Add generic socket listener & writer.

plugins/outputs/kafka/README.md

@@ -45,6 +45,10 @@ This plugin writes to a [Kafka Broker](http://kafka.apache.org/07/quickstart.htm
   ## Use SSL but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## Optional SASL Config
+  # sasl_username = "kafka"
+  # sasl_password = "secret"
+
   data_format = "influx"
 ```
 

plugins/outputs/kafka/kafka.go

@@ -44,6 +44,11 @@ type Kafka struct {
 	// Skip SSL verification
 	InsecureSkipVerify bool
 
+	// SASL Username
+	SASLUsername string `toml:"sasl_username"`
+	// SASL Password
+	SASLPassword string `toml:"sasl_password"`
+
 	tlsConfig tls.Config
 	producer  sarama.SyncProducer
 
@@ -92,6 +97,10 @@ var sampleConfig = `
   ## Use SSL but skip chain & host verification
   # insecure_skip_verify = false
 
+  ## Optional SASL Config
+  # sasl_username = "kafka"
+  # sasl_password = "secret"
+
   ## Data format to output.
   ## Each data format has its own unique set of configuration options, read
   ## more about them here:
@@ -129,6 +138,12 @@ func (k *Kafka) Connect() error {
 		config.Net.TLS.Enable = true
 	}
 
+	if k.SASLUsername != "" && k.SASLPassword != "" {
+		config.Net.SASL.User = k.SASLUsername
+		config.Net.SASL.Password = k.SASLPassword
+		config.Net.SASL.Enable = true
+	}
+
 	producer, err := sarama.NewSyncProducer(k.Brokers, config)
 	if err != nil {
 		return err