Add authentication support to kafka

Signed-off-by: Pavol Loffay <ploffay@redhat.com>

exporter/kafkaexporter/README.md

@@ -14,6 +14,23 @@ The following settings can be optionally configured:
   - `otlp_proto`: the payload is serialized to `ExportTraceServiceRequest`.
   - `jaeger_proto`: the payload is serialized to a single Jaeger proto `Span`.
   - `jaeger_json`: the payload is serialized to a single Jaeger JSON Span using `jsonpb`.
+- `authentication`
+  - `type` (default = none): The authentication type. Supported types are `plain_text`, `tls`, `kerberos`.
+  - `plain_text`
+    - `username`: The username to use.
+    - `password`: The password to use
+  - `tls`
+    - `ca_file` path to the CA cert. For a client this verifies the server certificate. Should
+      only be used if `insecure` is set to true.
+    - `cert_file` path to the TLS cert to use for TLS required connections. Should
+      only be used if `insecure` is set to true.
+    - `key_file` path to the TLS key to use for TLS required connections. Should
+      only be used if `insecure` is set to true.
+    - `insecure`  (default = false): Disable verifying the server's certificate chain and host 
+      name (`InsecureSkipVerify` in the tls config)
+    - `server_name_override`: ServerName indicates the name of the server requested by the client
+      in order to support virtual hosting.
+>>>>>>> Add authentication support to kafka
 - `metadata`
   - `full` (default = true): Whether to maintain a full set of metadata. 
                                     When disabled the client does not make the initial request to broker at the startup.

exporter/kafkaexporter/authentication.go

@@ -0,0 +1,74 @@
+// Copyright  The OpenTelemetry 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 kafkaexporter
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/Shopify/sarama"
+
+	"go.opentelemetry.io/collector/config/configtls"
+)
+
+// ConfigureAuthentication configures authentication in sarama.Config.
+func ConfigureAuthentication(config Authentication, saramaConfig *sarama.Config) error {
+	switch AuthType(strings.TrimSpace(string(config.Type))) {
+	case AuthTypeNone, "":
+		return nil
+	case AuthTypeTLS:
+		return configureTLS(config.TLS, saramaConfig)
+	case AuthTypeKerberos:
+		configureKerberos(config.Kerberos, saramaConfig)
+		return nil
+	case AuthTypePlaintext:
+		configurePlaintext(config.PlainText, saramaConfig)
+		return nil
+	default:
+		return fmt.Errorf("unknown/unsupported authentication method %v to kafka cluster", config.Type)
+	}
+}
+
+func configurePlaintext(config PlainTextConfig, saramaConfig *sarama.Config) {
+	saramaConfig.Net.SASL.Enable = true
+	saramaConfig.Net.SASL.User = config.Username
+	saramaConfig.Net.SASL.Password = config.Password
+}
+
+func configureTLS(config configtls.TLSClientSetting, saramaConfig *sarama.Config) error {
+	tlsConfig, err := config.LoadTLSConfig()
+	if err != nil {
+		return fmt.Errorf("error loading tls config: %w", err)
+	}
+	saramaConfig.Net.TLS.Enable = true
+	saramaConfig.Net.TLS.Config = tlsConfig
+	return nil
+}
+
+func configureKerberos(config KerberosConfig, saramaConfig *sarama.Config) {
+	saramaConfig.Net.SASL.Mechanism = sarama.SASLTypeGSSAPI
+	saramaConfig.Net.SASL.Enable = true
+	if config.UseKeyTab {
+		saramaConfig.Net.SASL.GSSAPI.KeyTabPath = config.KeyTabPath
+		saramaConfig.Net.SASL.GSSAPI.AuthType = sarama.KRB5_KEYTAB_AUTH
+	} else {
+		saramaConfig.Net.SASL.GSSAPI.AuthType = sarama.KRB5_USER_AUTH
+		saramaConfig.Net.SASL.GSSAPI.Password = config.Password
+	}
+	saramaConfig.Net.SASL.GSSAPI.KerberosConfigPath = config.ConfigPath
+	saramaConfig.Net.SASL.GSSAPI.Username = config.Username
+	saramaConfig.Net.SASL.GSSAPI.Realm = config.Realm
+	saramaConfig.Net.SASL.GSSAPI.ServiceName = config.ServiceName
+}

exporter/kafkaexporter/authentication_test.go

@@ -0,0 +1,97 @@
+// Copyright  The OpenTelemetry 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 kafkaexporter
+
+import (
+	"testing"
+
+	"github.com/Shopify/sarama"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/config/configtls"
+)
+
+func TestAuthentication(t *testing.T) {
+	saramaPlaintext := &sarama.Config{}
+	saramaPlaintext.Net.SASL.Enable = true
+	saramaPlaintext.Net.SASL.User = "jdoe"
+	saramaPlaintext.Net.SASL.Password = "pass"
+
+	saramaTLSCfg := &sarama.Config{}
+	saramaTLSCfg.Net.TLS.Enable = true
+	tlsClient := configtls.TLSClientSetting{}
+	tlscfg, err := tlsClient.LoadTLSConfig()
+	require.NoError(t, err)
+	saramaTLSCfg.Net.TLS.Config = tlscfg
+
+	saramaKerberosCfg := &sarama.Config{}
+	saramaKerberosCfg.Net.SASL.Mechanism = sarama.SASLTypeGSSAPI
+	saramaKerberosCfg.Net.SASL.Enable = true
+	saramaKerberosCfg.Net.SASL.GSSAPI.ServiceName = "foobar"
+	saramaKerberosCfg.Net.SASL.GSSAPI.AuthType = sarama.KRB5_USER_AUTH
+
+	saramaKerberosKeyTabCfg := &sarama.Config{}
+	saramaKerberosKeyTabCfg.Net.SASL.Mechanism = sarama.SASLTypeGSSAPI
+	saramaKerberosKeyTabCfg.Net.SASL.Enable = true
+	saramaKerberosKeyTabCfg.Net.SASL.GSSAPI.KeyTabPath = "/path"
+	saramaKerberosKeyTabCfg.Net.SASL.GSSAPI.AuthType = sarama.KRB5_KEYTAB_AUTH
+
+	tests := []struct {
+		auth         Authentication
+		saramaConfig *sarama.Config
+		err          string
+	}{
+		{
+			auth: Authentication{Type: AuthType("foo")},
+			err:  "unknown/unsupported authentication method",
+		},
+		{
+			auth:         Authentication{Type: AuthTypePlaintext, PlainText: PlainTextConfig{Username: "jdoe", Password: "pass"}},
+			saramaConfig: saramaPlaintext,
+		},
+		{
+			auth:         Authentication{Type: AuthTypeTLS, TLS: configtls.TLSClientSetting{}},
+			saramaConfig: saramaTLSCfg,
+		},
+		{
+			auth: Authentication{Type: AuthTypeTLS, TLS: configtls.TLSClientSetting{
+				TLSSetting: configtls.TLSSetting{CAFile: "/doesnotexists"},
+			}},
+			saramaConfig: saramaTLSCfg,
+			err:          "failed to load TLS config",
+		},
+		{
+			auth:         Authentication{Type: AuthTypeKerberos, Kerberos: KerberosConfig{ServiceName: "foobar"}},
+			saramaConfig: saramaKerberosCfg,
+		},
+		{
+			auth:         Authentication{Type: AuthTypeKerberos, Kerberos: KerberosConfig{UseKeyTab: true, KeyTabPath: "/path"}},
+			saramaConfig: saramaKerberosKeyTabCfg,
+		},
+	}
+	for _, test := range tests {
+		t.Run(string(test.auth.Type), func(t *testing.T) {
+			config := &sarama.Config{}
+			err := ConfigureAuthentication(test.auth, config)
+			if test.err != "" {
+				require.Error(t, err)
+				assert.Contains(t, err.Error(), test.err)
+			} else {
+				assert.Equal(t, test.saramaConfig, config)
+			}
+		})
+	}
+}

exporter/kafkaexporter/config.go

@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
 )
 
@@ -41,8 +42,47 @@ type Config struct {
 	// Client, and shared by the Producer/Consumer.
 	Metadata Metadata `mapstructure:"metadata"`
 
-	// TODO authentication
-	// TODO batch settings
+	// Authentication defines used authentication mechanism.
+	Authentication Authentication `mapstructure:"authentication"`
+}
+
+// AuthType defines authentication type
+type AuthType string
+
+const (
+	// No authentication
+	AuthTypeNone AuthType = "none"
+	// Plain text authentication
+	AuthTypePlaintext AuthType = "plain_text"
+	// TLS authentication
+	AuthTypeTLS AuthType = "tls"
+	// Kerberos authentication
+	AuthTypeKerberos AuthType = "kerberos"
+)
+
+// Authentication defines authentication.
+type Authentication struct {
+	Type      AuthType                   `mapstructure:"type"`
+	PlainText PlainTextConfig            `mapstructure:"plain_text"`
+	TLS       configtls.TLSClientSetting `mapstructure:"tls"`
+	Kerberos  KerberosConfig             `mapstructure:"kerberos"`
+}
+
+// PlainTextConfig defines plaintext authentication.
+type PlainTextConfig struct {
+	Username string `mapstructure:"username"`
+	Password string `mapstructure:"password"`
+}
+
+// KerberosConfig defines kereros configuration.
+type KerberosConfig struct {
+	ServiceName string `mapstructure:"service_name"`
+	Realm       string `mapstructure:"realm"`
+	UseKeyTab   bool   `mapstructure:"use_keytab"`
+	Username    string `mapstructure:"username"`
+	Password    string `mapstructure:"password" json:"-"`
+	ConfigPath  string `mapstructure:"config_file"`
+	KeyTabPath  string `mapstructure:"keytab_file"`
 }
 
 // Metadata defines configuration for retrieving metadata from the broker.

exporter/kafkaexporter/config_test.go

@@ -61,6 +61,13 @@ func TestLoadConfig(t *testing.T) {
 		Topic:    "spans",
 		Encoding: "otlp_proto",
 		Brokers:  []string{"foo:123", "bar:456"},
+		Authentication: Authentication{
+			Type: "plain_text",
+			PlainText: PlainTextConfig{
+				Username: "jdoe",
+				Password: "pass",
+			},
+		},
 		Metadata: Metadata{
 			Full: false,
 			Retry: MetadataRetry{

exporter/kafkaexporter/factory.go

@@ -77,6 +77,9 @@ func createDefaultConfig() configmodels.Exporter {
 		Brokers:         []string{defaultBroker},
 		Topic:           defaultTopic,
 		Encoding:        defaultEncoding,
+		Authentication: Authentication{
+			Type: AuthTypeNone,
+		},
 		Metadata: Metadata{
 			Full: defaultMetadataFull,
 			Retry: MetadataRetry{

exporter/kafkaexporter/factory_test.go

@@ -32,6 +32,7 @@ func TestCreateDefaultConfig(t *testing.T) {
 	assert.NoError(t, configcheck.ValidateConfig(cfg))
 	assert.Equal(t, []string{defaultBroker}, cfg.Brokers)
 	assert.Equal(t, defaultTopic, cfg.Topic)
+	assert.Equal(t, AuthTypeNone, cfg.Authentication.Type)
 }
 
 func TestCreateTracesExporter(t *testing.T) {

exporter/kafkaexporter/kafka_exporter.go

@@ -61,6 +61,9 @@ func newExporter(config Config, params component.ExporterCreateParams, marshalle
 		}
 		c.Version = version
 	}
+	if err := ConfigureAuthentication(config.Authentication, c); err != nil {
+		return nil, err
+	}
 	producer, err := sarama.NewSyncProducer(config.Brokers, c)
 	if err != nil {
 		return nil, err

exporter/kafkaexporter/kafka_exporter_test.go

@@ -39,11 +39,26 @@ func TestNewExporter_err_version(t *testing.T) {
 
 func TestNewExporter_err_encoding(t *testing.T) {
 	c := Config{Encoding: "foo"}
-	exp, err := newExporter(c, component.ExporterCreateParams{}, map[string]Marshaller{})
+	exp, err := newExporter(c, component.ExporterCreateParams{}, defaultMarshallers())
 	assert.EqualError(t, err, errUnrecognizedEncoding.Error())
 	assert.Nil(t, exp)
 }
 
+func TestNewExporter_err_auth_type(t *testing.T) {
+	c := Config{
+		ProtocolVersion: "2.0.0",
+		Authentication:  Authentication{Type: AuthType("foo")},
+		Encoding:        defaultEncoding,
+		Metadata: Metadata{
+			Full: false,
+		},
+	}
+	exp, err := newExporter(c, component.ExporterCreateParams{}, defaultMarshallers())
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "unknown/unsupported authentication method")
+	assert.Nil(t, exp)
+}
+
 func TestTraceDataPusher(t *testing.T) {
 	c := sarama.NewConfig()
 	producer := mocks.NewSyncProducer(t, c)

exporter/kafkaexporter/testdata/config.yaml

@@ -9,6 +9,11 @@ exporters:
       retry:
         max: 15
     timeout: 10s
+    authentication:
+      type: plain_text
+      plain_text:
+        username: jdoe
+        password: pass
     sending_queue:
       enabled: true
       num_consumers: 2

receiver/kafkareceiver/README.md

@@ -17,6 +17,22 @@ The following settings can be optionally configured:
   - `zipkin_thrift`: the payload is deserialized into Zipkin Thrift spans.
 - `group_id` (default = otel-collector):  The consumer group that receiver will be consuming messages from
 - `client_id` (default = otel-collector): The consumer client ID that receiver will use
+- `authentication`
+  - `type` (default = none): The authentication type. Supported types are `plain_text`, `tls`, `kerberos`.
+  - `plain_text`
+    - `username`: The username to use.
+    - `password`: The password to use
+  - `tls`
+    - `ca_file` path to the CA cert. For a client this verifies the server certificate. Should
+      only be used if `insecure` is set to true.
+    - `cert_file` path to the TLS cert to use for TLS required connections. Should
+      only be used if `insecure` is set to true.
+    - `key_file` path to the TLS key to use for TLS required connections. Should
+      only be used if `insecure` is set to true.
+    - `insecure`  (default = false): Disable verifying the server's certificate chain and host 
+      name (`InsecureSkipVerify` in the tls config)
+    - `server_name_override`: ServerName indicates the name of the server requested by the client
+      in order to support virtual hosting.
 - `metadata`
   - `full` (default = true): Whether to maintain a full set of metadata. 
            When disabled the client does not make the initial request to broker at the startup.

receiver/kafkareceiver/config.go

@@ -39,5 +39,5 @@ type Config struct {
 	// Client, and shared by the Producer/Consumer.
 	Metadata kafkaexporter.Metadata `mapstructure:"metadata"`
 
-	// TODO authentication
+	Authentication kafkaexporter.Authentication `mapstructure:"authentication"`
 }

receiver/kafkareceiver/config_test.go

@@ -25,6 +25,7 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/config/configtest"
+	"go.opentelemetry.io/collector/config/configtls"
 	"go.opentelemetry.io/collector/exporter/kafkaexporter"
 )
 
@@ -49,6 +50,16 @@ func TestLoadConfig(t *testing.T) {
 		Brokers:  []string{"foo:123", "bar:456"},
 		ClientID: "otel-collector",
 		GroupID:  "otel-collector",
+		Authentication: kafkaexporter.Authentication{
+			Type: kafkaexporter.AuthTypeTLS,
+			TLS: configtls.TLSClientSetting{
+				TLSSetting: configtls.TLSSetting{
+					CAFile:   "ca.pem",
+					CertFile: "cert.pem",
+					KeyFile:  "key.pem",
+				},
+			},
+		},
 		Metadata: kafkaexporter.Metadata{
 			Full: true,
 			Retry: kafkaexporter.MetadataRetry{