feat: support sasl/scram authentication (#46)

* Added: support sasl/scram authentication

* chore: remove unused variable kafkaSslValidation

Author: mingqing

README.md

@@ -56,6 +56,13 @@ To connect to Kafka over SSL define the following additonal environment variable
 - `KAFKA_SSL_CLIENT_KEY_PASS`: Kafka SSL client certificate key password (optional), defaults to `""`
 - `KAFKA_SSL_CA_CERT_FILE`: Kafka SSL broker CA certificate file, defaults to `""`
 
+To connect to Kafka over SASL/SCRAM authentication define the following additonal environment variables:
+
+- `KAFKA_SECURITY_PROTOCOL`: Kafka client used protocol to communicate with brokers, must be set if SASL is going to be used, either plain or with SSL
+- `KAFKA_SASL_MECHANISM`: SASL mechanism to use for authentication, defaults to `""`
+- `KAFKA_SASL_USERNAME`: SASL username for use with the PLAIN and SASL-SCRAM-.. mechanisms, defaults to `""`
+- `KAFKA_SASL_PASSWORD`: SASL password for use with the PLAIN and SASL-SCRAM-.. mechanism, defaults to `""`
+
 When deployed in a Kubernetes cluster using Helm and using a Kafka external to the cluster, it might be necessary to define the kafka hostname resolution locally (this fills the /etc/hosts of the container). Use a custom values.yaml file with section `hostAliases` (as mentioned in default values.yaml).
 
 ### prometheus

config.go

@@ -35,7 +35,10 @@ var (
 	kafkaSslClientKeyFile  = ""
 	kafkaSslClientKeyPass  = ""
 	kafkaSslCACertFile     = ""
-	kafkaSslValidation     = false
+	kafkaSecurityProtocol  = ""
+	kafkaSaslMechanism     = ""
+	kafkaSaslUsername      = ""
+	kafkaSaslPassword      = ""
 	serializer             Serializer
 )
 
@@ -88,6 +91,22 @@ func init() {
 		kafkaSslCACertFile = value
 	}
 
+	if value := os.Getenv("KAFKA_SECURITY_PROTOCOL"); value != "" {
+		kafkaSecurityProtocol = strings.ToLower(value)
+	}
+
+	if value := os.Getenv("KAFKA_SASL_MECHANISM"); value != "" {
+		kafkaSaslMechanism = value
+	}
+
+	if value := os.Getenv("KAFKA_SASL_USERNAME"); value != "" {
+		kafkaSaslUsername = value
+	}
+
+	if value := os.Getenv("KAFKA_SASL_PASSWORD"); value != "" {
+		kafkaSaslPassword = value
+	}
+
 	var err error
 	serializer, err = parseSerializationFormat(os.Getenv("SERIALIZATION_FORMAT"))
 	if err != nil {

main.go

@@ -37,14 +37,32 @@ func main() {
 	}
 
 	if kafkaSslClientCertFile != "" && kafkaSslClientKeyFile != "" && kafkaSslCACertFile != "" {
-		kafkaSslValidation = true
-		kafkaConfig["security.protocol"] = "ssl"
+		if kafkaSecurityProtocol == "" {
+			kafkaSecurityProtocol = "ssl"
+		}
+
+		if kafkaSecurityProtocol != "ssl" && kafkaSecurityProtocol != "sasl_ssl" {
+			logrus.Fatal("invalid config: kafka security protocol is not ssl based but ssl config is provided")
+		}
+
+		kafkaConfig["security.protocol"] = kafkaSecurityProtocol
 		kafkaConfig["ssl.ca.location"] = kafkaSslCACertFile              // CA certificate file for verifying the broker's certificate.
 		kafkaConfig["ssl.certificate.location"] = kafkaSslClientCertFile // Client's certificate
 		kafkaConfig["ssl.key.location"] = kafkaSslClientKeyFile          // Client's key
 		kafkaConfig["ssl.key.password"] = kafkaSslClientKeyPass          // Key password, if any.
 	}
 
+	if kafkaSaslMechanism != "" && kafkaSaslUsername != "" && kafkaSaslPassword != "" {
+		if kafkaSecurityProtocol != "sasl_ssl" && kafkaSecurityProtocol != "sasl_plaintext" {
+			logrus.Fatal("invalid config: kafka security protocol is not sasl based but sasl config is provided")
+		}
+
+		kafkaConfig["security.protocol"] = kafkaSecurityProtocol
+		kafkaConfig["sasl.mechanism"] = kafkaSaslMechanism
+		kafkaConfig["sasl.username"] = kafkaSaslUsername
+		kafkaConfig["sasl.password"] = kafkaSaslPassword
+	}
+
 	producer, err := kafka.NewProducer(&kafkaConfig)
 
 	if err != nil {
@@ -66,5 +84,5 @@ func main() {
 		r.POST("/receive", receiveHandler(producer, serializer))
 	}
 
-	r.Run()
+	log.Fatal(r.Run())
 }