Add SASL-PLAINTEXT authentication method to Kafka using sarama package

Author: Haidlir Naqvi

docs/config.md

@@ -103,12 +103,14 @@ The Kafka configuration contains the following key
 |compression          | none        | VFLOW_KAFKA_COMPRESSION      | compression codecs: gzip, snappy, lz4                                              |
 |retry-max            | 2           | VFLOW_KAFKA_RETRY_MAX        | the total number of times to retry                                                 |
 |request-size-max     | 104857600   | VFLOW_KAFKA_REQUEST_SIZE_MAX | the maximum size (in bytes) of any request that will be attempted to send to Kafka |
-|retry-backoff        | 10          | VFLOW_KAFKA_RETRY_BACKOFF    | wait for leader election to occur before retrying in milliseconds              |
-|tls-enabled          | false       | VFLOW_KAFKA_TLS_ENABLED      | connect using TLS                                                                   |
+|retry-backoff        | 10          | VFLOW_KAFKA_RETRY_BACKOFF    | wait for leader election to occur before retrying in milliseconds                  |
+|tls-enabled          | false       | VFLOW_KAFKA_TLS_ENABLED      | connect using TLS                                                                  |
 |tls-cert             | none        | VFLOW_KAFKA_TLS_CERT         | certificate file for client authentication                                         |
 |tls-key              | none        | VFLOW_KAFKA_TLS_KEY          | key file for client authentication                                                 |
 |ca-file              | none        | VFLOW_KAFKA_CA_FILE          | certificate authority file for TLS client authentication                           |
-|tls-skip-verify      | true        | VFLOW_KAFKA_TLS_SKIP_VERIFY  | if true, the server's certificate will not validate                                                      |
+|tls-skip-verify      | true        | VFLOW_KAFKA_TLS_SKIP_VERIFY  | if true, the server's certificate will not validate                                |
+|sasl-username        | none        | VFLOW_KAFKA_SASL_USERNAME    | username for SASL authentication                                                   |
+|sasl-username        | none        | VFLOW_KAFKA_SASL_PASSWORD    | password for SASL authentication                                                   |
 
 ## Example
 ```

producer/sarama.go

@@ -34,7 +34,7 @@ import (
 	"time"
 
 	"github.com/Shopify/sarama"
-	"gopkg.in/yaml.v2"
+	yaml "gopkg.in/yaml.v2"
 )
 
 // KafkaSarama represents kafka producer
@@ -55,7 +55,9 @@ type KafkaSaramaConfig struct {
 	TLSCertFile    string   `yaml:"tls-cert" env:"TLS_CERT"`
 	TLSKeyFile     string   `yaml:"tls-key" env:"TLS_KEY"`
 	CAFile         string   `yaml:"ca-file" env:"CA_FILE"`
-	TLSSkipVerify  bool     `yaml:"tls-skip-verify" env:"TLS-SKIP-VERIFY"`
+	TLSSkipVerify  bool     `yaml:"tls-skip-verify" env:"TLS_SKIP_VERIFY"`
+	SASLUsername   string   `yaml:"sasl-username" env:"SASL_USERNAME"`
+	SASLPassword   string   `yaml:"sasl-password" env:"SASL_PASSWORD"`
 }
 
 func (k *KafkaSarama) setup(configFile string, logger *log.Logger) error {
@@ -109,6 +111,13 @@ func (k *KafkaSarama) setup(configFile string, logger *log.Logger) error {
 		}
 	}
 
+	// Enable SASL Auth Config if username is filled
+	if k.config.SASLUsername != "" {
+		config.Net.SASL.Enable = true
+		config.Net.SASL.User = k.config.SASLUsername
+		config.Net.SASL.Password = k.config.SASLPassword
+	}
+
 	// get env config
 	k.loadEnv("VFLOW_KAFKA")