chore: add scram client for logs sending to kafka

logs/client/kafka/destination.go

@@ -140,6 +140,15 @@ func newDestination(endpoint logsconfig.Endpoint, contentType string, destinatio
 		coreconfig.Config.Logs.Config.Net.SASL.Version = coreconfig.Config.Logs.SaslVersion
 		coreconfig.Config.Logs.Config.Net.SASL.Handshake = coreconfig.Config.Logs.SaslHandshake
 		coreconfig.Config.Logs.Config.Net.SASL.AuthIdentity = coreconfig.Config.Logs.SaslAuthIdentity
+
+		if coreconfig.Config.Logs.Config.Net.SASL.Mechanism == sarama.SASLTypeSCRAMSHA256 {
+			coreconfig.Config.Logs.Config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
+				return &XDGSCRAMClient{HashGeneratorFcn: SHA256}
+			}
+		}
+		if coreconfig.Config.Logs.Config.Net.SASL.Mechanism == sarama.SASLTypeSCRAMSHA512 {
+			coreconfig.Config.Logs.Config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &XDGSCRAMClient{HashGeneratorFcn: SHA512} }
+		}
 	}
 
 	if len(coreconfig.Config.Logs.KafkaVersion) != 0 {

logs/client/kafka/scram_client.go

@@ -0,0 +1,37 @@
+package kafka
+
+import (
+	"crypto/sha256"
+	"crypto/sha512"
+
+	"github.com/xdg-go/scram"
+)
+
+var (
+	SHA256 scram.HashGeneratorFcn = sha256.New
+	SHA512 scram.HashGeneratorFcn = 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()
+}