adding kafka sasl support (#865)

Author: i3149

cmd/ktranslate/main.go

@@ -600,6 +600,12 @@ func applyFlags(cfg *ktranslate.Config) error {
 				cfg.KafkaSink.BootstrapServers = val
 			case "kafka.tls.config":
 				cfg.KafkaSink.TlsConfig = val
+			case "kafka.sasl.user":
+				cfg.KafkaSink.SaslUser = val
+			case "kafka.sasl.password":
+				cfg.KafkaSink.SaslPass = val
+			case "kafka.sasl.mechanism":
+				cfg.KafkaSink.SaslMech = val
 			// pkg/sinks/kentik
 			case "kentik_relay_url":
 				cfg.KentikSink.RelayURL = val

config.go

@@ -134,6 +134,9 @@ type KafkaSinkConfig struct {
 	Topic            string
 	BootstrapServers string
 	TlsConfig        string
+	SaslUser         string
+	SaslPass         string
+	SaslMech         string
 }
 
 // KentikSinkConfig is the config for the Kentik sink
@@ -472,6 +475,9 @@ func DefaultConfig() *Config {
 			Topic:            "",
 			BootstrapServers: "",
 			TlsConfig:        "",
+			SaslUser:         "",
+			SaslPass:         "",
+			SaslMech:         "",
 		},
 		KentikSink: &KentikSinkConfig{
 			RelayURL: "",

go.mod

@@ -148,6 +148,8 @@ require (
 	github.com/vmihailenco/msgpack/v5 v5.3.4 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	github.com/xanzy/ssh-agent v0.3.3 // indirect
+	github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c // indirect
+	github.com/xdg/stringprep v1.0.0 // indirect
 	go.mongodb.org/mongo-driver v1.12.1 // indirect
 	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect

pkg/sinks/kafka/kafa.go

@@ -15,18 +15,27 @@ import (
 	"github.com/kentik/ktranslate/pkg/formats"
 	"github.com/kentik/ktranslate/pkg/kt"
 	kafka "github.com/segmentio/kafka-go"
+	"github.com/segmentio/kafka-go/sasl"
+	"github.com/segmentio/kafka-go/sasl/plain"
+	"github.com/segmentio/kafka-go/sasl/scram"
 )
 
 var (
 	topic            string
 	bootstrapServers string
 	tlsConfig        string
+	saslUser         string
+	saslPass         string
+	saslMech         string
 )
 
 func init() {
 	flag.StringVar(&topic, "kafka_topic", "", "kafka topic to produce on")
 	flag.StringVar(&bootstrapServers, "bootstrap.servers", "", "bootstrap.servers")
 	flag.StringVar(&tlsConfig, "kafka.tls.config", "", "tls config to use for kafka. Can be basic|cert.pem,key.pem|cert.pem,key.pem,ca-cert.pem")
+	flag.StringVar(&saslUser, "kafka.sasl.user", "", "kafka user")
+	flag.StringVar(&saslPass, "kafka.sasl.password", "", "kafka password")
+	flag.StringVar(&saslMech, "kafka.sasl.mechanism", "", "plain|scram")
 }
 
 /**
@@ -115,10 +124,36 @@ func (s *KafkaSink) Init(ctx context.Context, format formats.Format, compression
 		Balancer: &kafka.LeastBytes{},
 	}
 
+	var mech sasl.Mechanism
+
+	if s.config.SaslMech != "" {
+		s.Infof("Adding Sasl Support")
+		switch s.config.SaslMech {
+		case "plain":
+			mech = plain.Mechanism{
+				Username: s.config.SaslUser,
+				Password: s.config.SaslPass,
+			}
+		case "scram":
+			mechl, err := scram.Mechanism(scram.SHA512, s.config.SaslUser, s.config.SaslPass)
+			if err != nil {
+				return err
+			}
+			mech = mechl
+		default:
+			return fmt.Errorf("Invalid kafka.sasl.mechanism flag=%s. Valid is plain|scram", s.config.SaslMech)
+		}
+	}
+
 	if s.tlsConf != nil {
 		s.Infof("Adding TLS Support")
 		s.kp.Transport = &kafka.Transport{
-			TLS: s.tlsConf,
+			TLS:  s.tlsConf,
+			SASL: mech,
+		}
+	} else if mech != nil {
+		s.kp.Transport = &kafka.Transport{
+			SASL: mech,
 		}
 	}