kafka plugins sasl oauth

cfg/config.go

@@ -327,8 +327,8 @@ func Parse(ptr any, values map[string]int) error {
 	return nil
 }
 
-// it isn't just a recursion
-// it also captures values with the same name from parent
+// it isn't just a recursion, it also captures values with the same name from parent
+// if the child value has zero value
 // i.e. take this config:
 //
 //	{
@@ -341,21 +341,24 @@ func Parse(ptr any, values map[string]int) error {
 // this function will set `config.Child.T = config.T`
 // see file.d/cfg/config_test.go:TestHierarchy for an example
 func ParseChild(parent reflect.Value, v reflect.Value, values map[string]int) error {
-	if v.CanAddr() {
-		for i := 0; i < v.NumField(); i++ {
-			name := v.Type().Field(i).Name
-			val := parent.FieldByName(name)
-			if val.CanAddr() {
-				v.Field(i).Set(val)
-			}
+	if !v.CanAddr() {
+		return nil
+	}
+
+	for i := range v.NumField() {
+		// set parent value only if child has zero value
+		if !v.Field(i).IsZero() {
+			continue
 		}
 
-		err := Parse(v.Addr().Interface(), values)
-		if err != nil {
-			return err
+		name := v.Type().Field(i).Name
+		val := parent.FieldByName(name)
+		if val.CanAddr() {
+			v.Field(i).Set(val)
 		}
 	}
-	return nil
+
+	return Parse(v.Addr().Interface(), values)
 }
 
 // ParseSlice recursively parses elements of an slice

cfg/config_test.go

@@ -438,10 +438,18 @@ func TestParseNestedFields(t *testing.T) {
 func TestHierarchy(t *testing.T) {
 	s := &hierarchy{T: "10"}
 	err := Parse(s, map[string]int{})
-
-	assert.Nil(t, err, "shouldn't be an error")
+	assert.NoError(t, err, "shouldn't be an error")
 	assert.Equal(t, "10", s.T, "wrong value")
 	assert.Equal(t, "10", s.Child.T, "wrong value")
+
+	s = &hierarchy{
+		T:     "10",
+		Child: hierarchyChild{T: "20"},
+	}
+	err = Parse(s, map[string]int{})
+	assert.NoError(t, err, "shouldn't be an error")
+	assert.Equal(t, "10", s.T, "wrong value")
+	assert.Equal(t, "20", s.Child.T, "wrong value")
 }
 
 func TestSlice(t *testing.T) {

cfg/kafka_client.go

@@ -1,10 +1,14 @@
 package cfg
 
 import (
+	"context"
+	"errors"
 	"os"
 
+	"github.com/ozontech/file.d/xoauth"
 	"github.com/twmb/franz-go/pkg/kgo"
 	"github.com/twmb/franz-go/pkg/sasl/aws"
+	"github.com/twmb/franz-go/pkg/sasl/oauth"
 	"github.com/twmb/franz-go/pkg/sasl/plain"
 	"github.com/twmb/franz-go/pkg/sasl/scram"
 	"github.com/twmb/franz-go/plugin/kzap"
@@ -27,6 +31,8 @@ type KafkaClientSaslConfig struct {
 	SaslMechanism string
 	SaslUsername  string
 	SaslPassword  string
+
+	SaslOAuth KafkaClientOAuthConfig
 }
 
 type KafkaClientSslConfig struct {
@@ -36,7 +42,63 @@ type KafkaClientSslConfig struct {
 	SslSkipVerify bool
 }
 
-func GetKafkaClientOptions(c KafkaClientConfig, l *zap.Logger) []kgo.Opt {
+type KafkaClientOAuthConfig struct {
+	// static
+	Token string `json:"token"`
+
+	// dynamic
+	ClientID     string   `json:"client_id"`
+	ClientSecret string   `json:"client_secret"`
+	TokenURL     string   `json:"token_url"`
+	Scopes       []string `json:"scopes" slice:"true"`
+	AuthStyle    string   `json:"auth_style" default:"params" options:"params|header"`
+}
+
+func (c *KafkaClientOAuthConfig) isStatic() bool {
+	return c.Token != ""
+}
+
+func (c *KafkaClientOAuthConfig) isDynamic() bool {
+	return c.ClientID != "" && c.ClientSecret != "" && c.TokenURL != ""
+}
+
+func (c *KafkaClientOAuthConfig) isValid() bool {
+	return c.isStatic() || c.isDynamic()
+}
+
+func GetKafkaClientOAuthTokenSource(ctx context.Context, cfg KafkaClientConfig) (xoauth.TokenSource, error) {
+	saslCfg := cfg.GetSaslConfig()
+
+	if !cfg.IsSaslEnabled() || saslCfg.SaslMechanism != "OAUTHBEARER" {
+		return nil, nil
+	}
+
+	saslOAuth := saslCfg.SaslOAuth
+	if !saslOAuth.isValid() {
+		return nil, errors.New("invalid SASL OAUTHBEARER config")
+	}
+
+	if saslOAuth.isStatic() {
+		return xoauth.NewStaticTokenSource(&xoauth.Token{
+			AccessToken: saslOAuth.Token,
+		}), nil
+	}
+
+	authStyle := xoauth.AuthStyleInParams
+	if saslOAuth.AuthStyle == "header" {
+		authStyle = xoauth.AuthStyleInHeader
+	}
+
+	return xoauth.NewReuseTokenSource(ctx, &xoauth.Config{
+		ClientID:     saslOAuth.ClientID,
+		ClientSecret: saslOAuth.ClientSecret,
+		TokenURL:     saslOAuth.TokenURL,
+		Scopes:       saslOAuth.Scopes,
+		AuthStyle:    authStyle,
+	})
+}
+
+func GetKafkaClientOptions(c KafkaClientConfig, l *zap.Logger, tokenSource xoauth.TokenSource) []kgo.Opt {
 	opts := []kgo.Opt{
 		kgo.SeedBrokers(c.GetBrokers()...),
 		kgo.ClientID(c.GetClientID()),
@@ -66,6 +128,18 @@ func GetKafkaClientOptions(c KafkaClientConfig, l *zap.Logger) []kgo.Opt {
 				AccessKey: saslConfig.SaslUsername,
 				SecretKey: saslConfig.SaslPassword,
 			}.AsManagedStreamingIAMMechanism()))
+		case "OAUTHBEARER":
+			authFn := func(ctx context.Context) (oauth.Auth, error) {
+				if tokenSource == nil {
+					return oauth.Auth{}, errors.New("uninitialized token source")
+				}
+				t := tokenSource.Token(ctx)
+				if t == nil {
+					return oauth.Auth{}, errors.New("empty token from token source")
+				}
+				return oauth.Auth{Token: t.AccessToken}, nil
+			}
+			opts = append(opts, kgo.SASL(oauth.Oauth(authFn)))
 		}
 	}
 

e2e/file_elasticsearch/docker-compose.yml

@@ -10,13 +10,14 @@ services:
       - ELASTIC_PASSWORD=elastic
       - xpack.security.enabled=true
       - xpack.security.http.ssl.enabled=false
-    mem_limit: 1073741824
+    mem_limit: 2147483648
     healthcheck:
       test:
         [
           "CMD-SHELL",
-          "curl --output /dev/null --silent --head --fail -u elastic:elastic http://elasticsearch:19200",
+          "curl --output /dev/null --silent --head --fail -u elastic:elastic http://localhost:9200",
         ]
       interval: 10s
       timeout: 10s
       retries: 10
+      start_period: 20s

e2e/file_es_split/docker-compose.yml

@@ -10,13 +10,14 @@ services:
       - http.max_content_length=128b
       - logger.org.elasticsearch.transport=DEBUG
       - logger.org.elasticsearch.http=DEBUG
-    mem_limit: 1073741824
+    mem_limit: 2147483648
     healthcheck:
       test:
         [
           "CMD-SHELL",
-          "curl --output /dev/null --silent --head --fail http://elasticsearch:9200",
+          "curl --output /dev/null --silent --head --fail -u elastic:elastic http://localhost:9200",
         ]
       interval: 10s
       timeout: 10s
       retries: 10
+      start_period: 20s

e2e/file_loki/docker-compose.yml

@@ -1,11 +1,10 @@
 services:
-
-  grafana: 
+  grafana:
     image: grafana/grafana:latest
     container_name: grafana
     ports:
       - "3000:3000"
-    user: '0'
+    user: "0"
     environment:
       - GF_PATHS_PROVISIONING=/etc/grafana/provisioning
       - GF_AUTH_ANONYMOUS_ENABLED=true
@@ -31,9 +30,10 @@ services:
         /run.sh
 
   loki:
-    image: grafana/loki:latest
+    image: grafana/loki:3.5
     container_name: loki
-    command: --config.file=/etc/loki/config.yaml
+    command:
+      - --config.file=/etc/loki/config.yaml
     ports:
       - "3100:3100"
     volumes:
@@ -43,3 +43,4 @@ services:
       interval: 3s
       timeout: 10s
       retries: 10
+      start_period: 15s

e2e/kafka_auth/kafka_auth.go

@@ -125,8 +125,9 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 					config.ClientCert = "./kafka_auth/certs/client_cert.pem"
 				}
 
-				kafka_out.NewClient(config,
+				kafka_out.NewClient(context.Background(), config,
 					zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)),
+					nil,
 				)
 			},
 			func() {
@@ -155,9 +156,9 @@ func (c *Config) Configure(t *testing.T, _ *cfg.Config, _ string) {
 					config.ClientCert = "./kafka_auth/certs/client_cert.pem"
 				}
 
-				kafka_in.NewClient(config,
+				kafka_in.NewClient(context.Background(), config,
 					zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)),
-					Consumer{},
+					Consumer{}, nil,
 				)
 			},
 		}

e2e/kafka_file/kafka_file.go

@@ -53,8 +53,9 @@ func (c *Config) Send(t *testing.T) {
 		Timeout_:         10 * time.Second,
 	}
 
-	client := kafka_out.NewClient(config,
+	client := kafka_out.NewClient(context.Background(), config,
 		zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)),
+		nil,
 	)
 	adminClient := kadm.NewClient(client)
 	_, err := adminClient.CreateTopic(context.TODO(), 1, 1, nil, c.Topics[0])

e2e/split_join/split_join.go

@@ -63,9 +63,9 @@ func (c *Config) Configure(t *testing.T, conf *cfg.Config, pipelineName string)
 		HeartbeatInterval_:   10 * time.Second,
 	}
 
-	c.client = kafka_in.NewClient(config,
+	c.client = kafka_in.NewClient(context.Background(), config,
 		zap.NewNop().WithOptions(zap.WithFatalHook(zapcore.WriteThenPanic)),
-		Consumer{},
+		Consumer{}, nil,
 	)
 
 	adminClient := kadm.NewClient(c.client)

go.mod

@@ -1,6 +1,6 @@
 module github.com/ozontech/file.d
 
-go 1.25
+go 1.25.0
 
 toolchain go1.25.5
 
@@ -47,11 +47,10 @@ require (
 	github.com/twmb/franz-go/plugin/kzap v1.1.2
 	github.com/twmb/tlscfg v1.2.1
 	github.com/valyala/fasthttp v1.48.0
-	github.com/xdg-go/scram v1.1.2
 	go.uber.org/atomic v1.11.0
 	go.uber.org/automaxprocs v1.5.3
 	go.uber.org/zap v1.27.0
-	golang.org/x/net v0.49.0
+	golang.org/x/oauth2 v0.36.0
 	google.golang.org/protobuf v1.36.5
 	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1
@@ -132,8 +131,6 @@ require (
 	github.com/twmb/franz-go/pkg/kmsg v1.12.0 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
-	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
-	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/yuin/gopher-lua v1.1.1 // indirect
 	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
 	go.opentelemetry.io/otel v1.34.0 // indirect
@@ -144,7 +141,7 @@ require (
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/crypto v0.48.0 // indirect
 	golang.org/x/mod v0.32.0 // indirect
-	golang.org/x/oauth2 v0.27.0 // indirect
+	golang.org/x/net v0.49.0 // indirect
 	golang.org/x/sync v0.19.0 // indirect
 	golang.org/x/sys v0.41.0 // indirect
 	golang.org/x/term v0.40.0 // indirect

go.sum

@@ -365,12 +365,6 @@ github.com/valyala/fasthttp v1.48.0 h1:oJWvHb9BIZToTQS3MuQ2R3bJZiNSa2KiNdeI8A+79
 github.com/valyala/fasthttp v1.48.0/go.mod h1:k2zXd82h/7UZc3VOdJ2WaUqt1uZ/XpXAfE9i+HBC3lA=
 github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM=
 github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg=
-github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
-github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
-github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
-github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
-github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
-github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
@@ -453,8 +447,8 @@ golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/net v0.49.0 h1:eeHFmOGUTtaaPSGNmjBKpbng9MulQsJURQUAfUwY++o=
 golang.org/x/net v0.49.0/go.mod h1:/ysNB2EvaqvesRkuLAyjI1ycPZlQHM3q01F02UY/MV8=
-golang.org/x/oauth2 v0.27.0 h1:da9Vo7/tDv5RH/7nZDz1eMGS/q1Vv1N/7FCrBhI9I3M=
-golang.org/x/oauth2 v0.27.0/go.mod h1:onh5ek6nERTohokkhCD/y2cV4Do3fxFHFuAejCkRWT8=
+golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=
+golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q=
 golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -501,7 +495,6 @@ golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.3.8/go.mod h1:E6s5w1FMmriuDzIBO73fBruAKo1PCIq6d2Q6DHfQ8WQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=

plugin/input/kafka/README.md

@@ -140,7 +140,7 @@ If set, the plugin will use SASL authentications mechanism.
 
 <br>
 
-**`sasl_mechanism`** *`string`* *`default=SCRAM-SHA-512`* *`options=PLAIN|SCRAM-SHA-256|SCRAM-SHA-512|AWS_MSK_IAM`* 
+**`sasl_mechanism`** *`string`* *`default=SCRAM-SHA-512`* *`options=PLAIN|SCRAM-SHA-256|SCRAM-SHA-512|AWS_MSK_IAM|OAUTHBEARER`* 
 
 SASL mechanism to use.
 
@@ -158,6 +158,22 @@ SASL password.
 
 <br>
 
+**`sasl_oauth`** *`cfg.KafkaClientOAuthConfig`* 
+
+SASL OAUTHBEARER config. It works only if `sasl_mechanism:"OAUTHBEARER"`.
+> There are 2 options - a static token or a dynamically updated.
+
+`OAuthConfig` params:
+* **`token`** *`string`* - static token
+---
+* **`client_id`** *`string`* - client ID
+* **`client_secret`** *`string`* - client secret
+* **`token_url`** *`string`* - resource server's token endpoint URL
+* **`scopes`** *`[]string`* - optional requested permissions
+* **`auth_style`** *`string`* *`default=params`* *`options=params|header`* - specifies how the endpoint wants the client ID & client secret sent
+
+<br>
+
 **`is_ssl_enabled`** *`bool`* *`default=false`* 
 
 If set, the plugin will use SSL/TLS connections method.