add kafka scaler sasl

go.mod

@@ -59,6 +59,7 @@ require (
  github.com/spf13/pflag v1.0.5
  github.com/streadway/amqp v0.0.0-20190827072141-edfb9018d271
  github.com/stretchr/testify v1.4.0
+ github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c
  google.golang.org/api v0.10.0
  google.golang.org/genproto v0.0.0-20191002211648-c459b9ce5143
  google.golang.org/grpc v1.24.0

go.sum

@@ -599,7 +599,9 @@ github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8 h1:3SVOIvH7Ae1KRYy
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
 github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs=
 github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
+github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c h1:u40Z8hqBAAQyv+vATcGgV0YCnDjqSL7/q/JyPhhJSPk=
 github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
+github.com/xdg/stringprep v1.0.0 h1:d9X0esnoa3dFsV0FG35rAT0RIhYFlPq7MiP+DW89La0=
 github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
 github.com/xiang90/probing v0.0.0-20160813154853-07dd2e8dfe18/go.mod h1:UETIi67q53MR2AWcXfiuqkDkRtnGDLqkBTpCHuJHxtU=
 github.com/xiang90/probing v0.0.0-20190116061207-43a291ad63a2 h1:eY9dn8+vbi4tKz5Qo6v2eYzo7kUS51QINcR5jNpbZS8=

pkg/handler/scale_handler.go

@@ -273,7 +273,7 @@ func (h *ScaleHandler) getScaler(name, namespace, triggerType string, resolvedEn
  case "aws-cloudwatch":
  return scalers.NewAwsCloudwatchScaler(resolvedEnv, triggerMetadata, authParams)
  case "kafka":
- return scalers.NewKafkaScaler(resolvedEnv, triggerMetadata)
+ return scalers.NewKafkaScaler(resolvedEnv, triggerMetadata, authParams)
  case "rabbitmq":
  return scalers.NewRabbitMQScaler(resolvedEnv, triggerMetadata, authParams)
  case "azure-eventhub":

pkg/scalers/kafka_scaler.go

@@ -2,6 +2,8 @@ package scalers
 
 import (
  "context"
+ "crypto/tls"
+ "crypto/x509"
  "errors"
  "fmt"
  "strconv"
@@ -17,19 +19,38 @@ import (
 )
 
 type kafkaScaler struct {
- resolvedSecrets map[string]string
- metadata kafkaMetadata
- client sarama.Client
- admin sarama.ClusterAdmin
+ metadata kafkaMetadata
+ client sarama.Client
+ admin sarama.ClusterAdmin
 }
 
 type kafkaMetadata struct {
  brokers []string
  group string
  topic string
  lagThreshold int64
+
+ // auth
+ authMode kafkaAuthMode
+ username string
+ password string
+
+ // ssl
+ cert string
+ key string
+ ca string
 }
 
+type kafkaAuthMode string
+
+const (
+ kafkaAuthModeForNone kafkaAuthMode = "none"
+ kafkaAuthModeForSaslPlaintext kafkaAuthMode = "sasl_plaintext"
+ kafkaAuthModeForSaslScramSha256 kafkaAuthMode = "sasl_scram_sha256"
+ kafkaAuthModeForSaslScramSha512 kafkaAuthMode = "sasl_scram_sha512"
+ kafkaAuthModeForSaslSSL kafkaAuthMode = "sasl_ssl"
+)
+
 const (
  lagThresholdMetricName = "lagThreshold"
  kafkaMetricType = "External"
@@ -39,8 +60,8 @@ const (
 var kafkaLog = logf.Log.WithName("kafka_scaler")
 
 // NewKafkaScaler creates a new kafkaScaler
-func NewKafkaScaler(resolvedSecrets, metadata map[string]string) (Scaler, error) {
- kafkaMetadata, err := parseKafkaMetadata(metadata)
+func NewKafkaScaler(resolvedEnv, metadata, authParams map[string]string) (Scaler, error) {
+ kafkaMetadata, err := parseKafkaMetadata(resolvedEnv, metadata, authParams)
  if err != nil {
  return nil, fmt.Errorf("error parsing kafka metadata: %s", err)
  }
@@ -51,14 +72,13 @@ func NewKafkaScaler(resolvedSecrets, metadata map[string]string) (Scaler, error)
  }
 
  return &kafkaScaler{
- client: client,
- admin: admin,
- metadata: kafkaMetadata,
- resolvedSecrets: resolvedSecrets,
+ client: client,
+ admin: admin,
+ metadata: kafkaMetadata,
  }, nil
 }
 
-func parseKafkaMetadata(metadata map[string]string) (kafkaMetadata, error) {
+func parseKafkaMetadata(resolvedEnv, metadata, authParams map[string]string) (kafkaMetadata, error) {
  meta := kafkaMetadata{}
 
  if metadata["brokerList"] == "" {
@@ -86,6 +106,45 @@ func parseKafkaMetadata(metadata map[string]string) (kafkaMetadata, error) {
  meta.lagThreshold = t
  }
 
+ meta.authMode = kafkaAuthModeForNone
+ if val, ok := authParams["authMode"]; ok {
+ mode := kafkaAuthMode(val)
+ if mode != kafkaAuthModeForNone && mode != kafkaAuthModeForSaslPlaintext && mode != kafkaAuthModeForSaslSSL && mode != kafkaAuthModeForSaslScramSha256 && mode != kafkaAuthModeForSaslScramSha512 {
+ return meta, fmt.Errorf("err auth mode %s given", mode)
+ }
+
+ meta.authMode = mode
+ }
+
+ if meta.authMode != kafkaAuthModeForNone {
+ if authParams["username"] == "" {
+ return meta, errors.New("no username given")
+ }
+ meta.username = authParams["username"]
+
+ if authParams["password"] == "" {
+ return meta, errors.New("no password given")
+ }
+ meta.password = authParams["password"]
+ }
+
+ if meta.authMode == kafkaAuthModeForSaslSSL {
+ if authParams["ca"] == "" {
+ return meta, errors.New("no ca given")
+ }
+ meta.ca = authParams["ca"]
+
+ if authParams["cert"] == "" {
+ return meta, errors.New("no cert given")
+ }
+ meta.cert = authParams["cert"]
+
+ if authParams["key"] == "" {
+ return meta, errors.New("no key given")
+ }
+ meta.key = authParams["key"]
+ }
+
  return meta, nil
 }
 
@@ -118,6 +177,40 @@ func getKafkaClients(metadata kafkaMetadata) (sarama.Client, sarama.ClusterAdmin
  config := sarama.NewConfig()
  config.Version = sarama.V1_0_0_0
 
+ if ok := metadata.authMode == kafkaAuthModeForSaslPlaintext || metadata.authMode == kafkaAuthModeForSaslSSL || metadata.authMode == kafkaAuthModeForSaslScramSha256 || metadata.authMode == kafkaAuthModeForSaslScramSha512; ok {
+ config.Net.SASL.Enable = true
+ config.Net.SASL.User = metadata.username
+ config.Net.SASL.Password = metadata.password
+ }
+
+ if metadata.authMode == kafkaAuthModeForSaslSSL {
+ cert, err := tls.X509KeyPair([]byte(metadata.cert), []byte(metadata.key))
+ if err != nil {
+ return nil, nil, fmt.Errorf("error parse X509KeyPair: %s", err)
+ }
+
+ caCertPool := x509.NewCertPool()
+ caCertPool.AppendCertsFromPEM([]byte(metadata.ca))
+
+ tlsConfig := &tls.Config{
+ Certificates: []tls.Certificate{cert},
+ RootCAs: caCertPool,
+ }
+
+ config.Net.TLS.Enable = true
+ config.Net.TLS.Config = tlsConfig
+ }
+
+ if metadata.authMode == kafkaAuthModeForSaslScramSha256 {
+ config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA256} }
+ config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA256)
+ }
+
+ if metadata.authMode == kafkaAuthModeForSaslScramSha512 {
+ config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} }
+ config.Net.SASL.Mechanism = sarama.SASLMechanism(sarama.SASLTypeSCRAMSHA512)
+ }
+
  client, err := sarama.NewClient(metadata.brokers, config)
  if err != nil {
  return nil, nil, fmt.Errorf("error creating kafka client: %s", err)

pkg/scalers/kafka_scaler_test.go

@@ -21,17 +21,23 @@ var validMetadata = map[string]string{
  "topic": "my-topic",
 }
 
+// A complete valid authParams example for sasl, with username and passwd
+var validAuthParams = map[string]string{
+ "authMode": "sasl_plaintext",
+ "username": "admin",
+ "passwd": "admin",
+}
+
 var parseKafkaMetadataTestDataset = []parseKafkaMetadataTestData{
  {map[string]string{}, true, 0, nil, "", ""},
  {map[string]string{"brokerList": "foobar:9092"}, true, 1, []string{"foobar:9092"}, "", ""},
  {map[string]string{"brokerList": "foo:9092,bar:9092"}, true, 2, []string{"foo:9092", "bar:9092"}, "", ""},
  {map[string]string{"brokerList": "a", "consumerGroup": "my-group"}, true, 1, []string{"a"}, "my-group", ""},
- {validMetadata, false, 2, []string{"broker1:9092", "broker2:9092"}, "my-group", "my-topic"},
 }
 
 func TestGetBrokers(t *testing.T) {
  for _, testData := range parseKafkaMetadataTestDataset {
- meta, err := parseKafkaMetadata(testData.metadata)
+ meta, err := parseKafkaMetadata(nil, testData.metadata, validAuthParams)
 
  if err != nil && !testData.isError {
  t.Error("Expected success but got error", err)

pkg/scalers/kafka_scram_client.go

@@ -0,0 +1,36 @@
+package scalers
+
+import (
+ "crypto/sha256"
+ "crypto/sha512"
+ "hash"
+
+ "github.com/xdg/scram"
+)
+
+var SHA256 scram.HashGeneratorFcn = func() hash.Hash { return sha256.New() }
+var SHA512 scram.HashGeneratorFcn = func() hash.Hash { return sha512.New() }
+
+type XDGSCRAMClient struct {
+ *scram.Client
+ *scram.ClientConversation
+ scram.HashGeneratorFcn
+}
+
+func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error) {
+ x.Client, err = x.HashGeneratorFcn.NewClient(userName, password, authzID)
+ if err != nil {
+ return err
+ }
+ x.ClientConversation = x.Client.NewConversation()
+ return nil
+}
+
+func (x *XDGSCRAMClient) Step(challenge string) (response string, err error) {
+ response, err = x.ClientConversation.Step(challenge)
+ return
+}
+
+func (x *XDGSCRAMClient) Done() bool {
+ return x.ClientConversation.Done()
+}