Add sasl plain auth kafka plugin

pkg/kafka/auth/config.go

@@ -23,22 +23,25 @@ import (
 )
 
 const (
-	none     = "none"
-	kerberos = "kerberos"
-	tls      = "tls"
+	none      = "none"
+	kerberos  = "kerberos"
+	tls       = "tls"
+	saslPlain = "plain"
 )
 
 var authTypes = []string{
 	none,
 	kerberos,
 	tls,
+	saslPlain,
 }
 
 // AuthenticationConfig describes the configuration properties needed authenticate with kafka cluster
 type AuthenticationConfig struct {
 	Authentication string
 	Kerberos       KerberosConfig
 	TLS            TLSConfig
+	SASLPlain      SASLPlainConfig
 }
 
 //SetConfiguration set configure authentication into sarama config structure
@@ -55,6 +58,8 @@ func (config *AuthenticationConfig) SetConfiguration(saramaConfig *sarama.Config
 		return nil
 	case tls:
 		return setTLSConfiguration(&config.TLS, saramaConfig)
+	case saslPlain:
+		return setSASLPlainConfiguration(&config.SASLPlain, saramaConfig)
 	default:
 		return errors.Errorf("Unknown/Unsupported authentication method %s to kafka cluster.", config.Authentication)
 	}
@@ -74,4 +79,7 @@ func (config *AuthenticationConfig) InitFromViper(configPrefix string, v *viper.
 	config.TLS.CaPath = v.GetString(configPrefix + tlsPrefix + suffixTLSCA)
 	config.TLS.CertPath = v.GetString(configPrefix + tlsPrefix + suffixTLSCert)
 	config.TLS.KeyPath = v.GetString(configPrefix + tlsPrefix + suffixTLSKey)
+
+	config.SASLPlain.UserName = v.GetString(configPrefix + saslPlainPrefix + suffixSASLUsername)
+	config.SASLPlain.Password = v.GetString(configPrefix + saslPlainPrefix + suffixSASLPassword)
 }

pkg/kafka/auth/options.go

@@ -50,6 +50,13 @@ const (
 	defaultCAPath   = ""
 	defaultCertPath = ""
 	defaultKeyPath  = ""
+
+	// SASL Plain configuration options
+	saslPlainPrefix     = ".sasl.plain"
+	suffixSASLUsername  = ".username"
+	suffixSASLPassword  = ".password"
+	defaultSASLUsername = ""
+	defaultSASLPassword = ""
 )
 
 func addKerberosFlags(configPrefix string, flagSet *flag.FlagSet) {
@@ -99,6 +106,18 @@ func addTLSFlags(configPrefix string, flagSet *flag.FlagSet) {
 		"Path to the TLS Key for the Kafka connection")
 }
 
+// Flags for SASL Plain authentication options
+func addSASLPlainFlags(configPrefix string, flagSet *flag.FlagSet) {
+	flagSet.String(
+		configPrefix+saslPlainPrefix+suffixSASLUsername,
+		defaultSASLUsername,
+		"Username for SASL Plain authentication")
+	flagSet.String(
+		configPrefix+saslPlainPrefix+suffixSASLPassword,
+		defaultSASLPassword,
+		"Password for SASL Plain authentication")
+}
+
 // AddFlags add configuration flags to a flagSet.
 func AddFlags(configPrefix string, flagSet *flag.FlagSet) {
 	flagSet.String(
@@ -108,4 +127,5 @@ func AddFlags(configPrefix string, flagSet *flag.FlagSet) {
 	)
 	addKerberosFlags(configPrefix, flagSet)
 	addTLSFlags(configPrefix, flagSet)
+	addSASLPlainFlags(configPrefix, flagSet)
 }

pkg/kafka/auth/sasl_plain.go

@@ -0,0 +1,37 @@
+// Copyright (c) 2019 The Jaeger Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package auth
+
+import (
+	"errors"
+	"github.com/Shopify/sarama"
+)
+
+// SASLPlainConfig defines configurations required for SASL Plain authentication (Refer: https://kafka.apache.org/documentation/#security_sasl_plain)
+type SASLPlainConfig struct {
+	UserName string
+	Password string
+}
+
+func setSASLPlainConfiguration(config *SASLPlainConfig, saramaConfig *sarama.Config) error {
+	if len(config.UserName) == 0 || len(config.Password) == 0 {
+		return errors.New("invalid username/password supplied for SASL Plain authentication. username/password cannot be empty")
+	}
+	saramaConfig.Net.SASL.Enable = true
+	saramaConfig.Net.SASL.Mechanism = sarama.SASLTypePlaintext
+	saramaConfig.Net.SASL.User = config.UserName
+	saramaConfig.Net.SASL.Password = config.Password
+	return nil
+}