Move TLS config for Kafka components to top level (open-telemetry#39115)

#### Description

Deprecates `auth::tls` and introduce `tls` to be consistent with other
components, such as the OTLP receiver and exporter. Until we remove it
altogether, `auth::tls` will continue to be honoured only if `tls` is
unspecified.

Why? Two reasons:
- TLS is primarily used for encryption. It might also be for client
authentication, but that's less common.
- Consistency with the majority of other components using
configtls.ClientConfig. Almost all of them put it at the top-level.

Put together, I think this makes for a clearer structure. Having
consistency makes it easier to understand the configuration,
particularly if you're already familiar with other components. Moving
out of "auth" will also make it easier to understand if you're not using
TLS for auth.

#### Link to tracking issue

Fixes open-telemetry#37776

#### Testing

Unit tests pass. Added tests covering the new deprecated config
handling.

#### Documentation

Simplified READMEs by just pointing to the configtls docs, reducing
opportunity for drift.

Author: axw

.chloggen/kafka-clientconfig-tls-kafkaexporter.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kafkaexporter
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Deprecate `auth::tls` and introduce `tls` config
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37776]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

.chloggen/kafka-clientconfig-tls-kafkametricsreceiver.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kafkametricsreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Deprecate `auth::tls` and introduce `tls` config
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37776]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

.chloggen/kafka-clientconfig-tls-kafkareceiver.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kafkareceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Deprecate `auth::tls` and introduce `tls` config
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37776]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

.chloggen/kafka-clientconfig-tls-kafkatopicsobserver.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: deprecation
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: kafkatopicsobserverextension
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Deprecate `auth::tls` and introduce `tls` config
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [37776]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/kafkaexporter/README.md

@@ -40,6 +40,7 @@ The following settings can be optionally configured:
 - `partition_traces_by_id` (default = false): configures the exporter to include the trace ID as the message key in trace messages sent to kafka. *Please note:* this setting does not have any effect on Jaeger encoding exporters since Jaeger exporters include trace ID as the message key by default.
 - `partition_metrics_by_resource_attributes` (default = false)  configures the exporter to include the hash of sorted resource attributes as the message partitioning key in metric messages sent to kafka.
 - `partition_logs_by_resource_attributes` (default = false)  configures the exporter to include the hash of sorted resource attributes as the message partitioning key in log messages sent to kafka.
+- `tls`: see [TLS Configuration Settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configtls/README.md) for the full set of available options.
 - `auth`
   - `plain_text` (Deprecated in v0.123.0: use sasl with mechanism set to PLAIN instead.)
     - `username`: The username to use.
@@ -51,17 +52,7 @@ The following settings can be optionally configured:
     - `version` (default = 0): The SASL protocol version to use (0 or 1)
     - `aws_msk.region`: AWS Region in case of AWS_MSK_IAM or AWS_MSK_IAM_OAUTHBEARER mechanism
     - `aws_msk.broker_addr`: MSK Broker address in case of AWS_MSK_IAM mechanism
-  - `tls`: see [TLS Configuration Settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configtls/README.md) for the full set of available options.
-    - `ca_file`: path to the CA cert. For a client this verifies the server certificate. Should
-      only be used if `insecure` is set to false.
-    - `cert_file`: path to the TLS cert to use for TLS required connections. Should
-      only be used if `insecure` is set to false.
-    - `key_file`: path to the TLS key to use for TLS required connections. Should
-      only be used if `insecure` is set to false.
-    - `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.
+  - `tls` (Deprecated in v0.124.0: configure tls at the top level): this is an alias for tls at the top level.
   - `kerberos`
     - `service_name`: Kerberos service name
     - `realm`: Kerberos realm

extension/observer/kafkatopicsobserver/README.md

@@ -28,6 +28,7 @@ The following settings can be optionally configured:
 - `protocol_version` (default = 2.1.0): Kafka protocol version e.g. 2.0.0
 - `client_id` (default = "otel-collector"): The client ID to configure the Kafka client with.
 - `topics_sync_interval` (default 5s)
+- `tls`: see [TLS Configuration Settings](https://github.com/open-telemetry/opentelemetry-collector/blob/main/config/configtls/README.md) for the full set of available options.
 - `auth`
     - `plain_text` (Deprecated in v0.123.0: use sasl with mechanism set to PLAIN instead.)
         - `username`: The username to use.
@@ -38,17 +39,7 @@ The following settings can be optionally configured:
         - `mechanism`: The sasl mechanism to use (SCRAM-SHA-256, SCRAM-SHA-512, AWS_MSK_IAM, AWS_MSK_IAM_OAUTHBEARER or PLAIN)
         - `aws_msk.region`: AWS Region in case of AWS_MSK_IAM or AWS_MSK_IAM_OAUTHBEARER mechanism
         - `aws_msk.broker_addr`: MSK Broker address in case of AWS_MSK_IAM mechanism
-    - `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 false.
-        - `cert_file`: path to the TLS cert to use for TLS required connections. Should
-          only be used if `insecure` is set to false.
-        - `key_file`: path to the TLS key to use for TLS required connections. Should
-          only be used if `insecure` is set to false.
-        - `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.
+    - `tls` (Deprecated in v0.124.0: configure tls at the top level): this is an alias for tls at the top level.
     - `kerberos`
         - `service_name`: Kerberos service name
         - `realm`: Kerberos realm

internal/kafka/authentication.go

@@ -7,42 +7,31 @@ import (
 	"context"
 	"crypto/sha256"
 	"crypto/sha512"
-	"crypto/tls"
-	"fmt"
 
 	"github.com/IBM/sarama"
 	"github.com/aws/aws-msk-iam-sasl-signer-go/signer"
-	"go.opentelemetry.io/collector/config/configtls"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka/awsmsk"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka/configkafka"
 )
 
-// ConfigureSaramaAuthentication configures authentication in sarama.Config.
+// configureSaramaAuthentication configures authentication in sarama.Config.
 //
 // The provided config is assumed to have been validated.
-func ConfigureSaramaAuthentication(
+func configureSaramaAuthentication(
 	ctx context.Context,
 	config configkafka.AuthenticationConfig,
 	saramaConfig *sarama.Config,
-) error {
+) {
 	if config.PlainText != nil {
 		configurePlaintext(*config.PlainText, saramaConfig)
 	}
-	if config.TLS != nil {
-		if err := configureTLS(ctx, *config.TLS, saramaConfig); err != nil {
-			return err
-		}
-	}
 	if config.SASL != nil {
-		if err := configureSASL(ctx, *config.SASL, saramaConfig); err != nil {
-			return err
-		}
+		configureSASL(ctx, *config.SASL, saramaConfig)
 	}
 	if config.Kerberos != nil {
 		configureKerberos(*config.Kerberos, saramaConfig)
 	}
-	return nil
 }
 
 func configurePlaintext(config configkafka.PlainTextConfig, saramaConfig *sarama.Config) {
@@ -51,7 +40,7 @@ func configurePlaintext(config configkafka.PlainTextConfig, saramaConfig *sarama
 	saramaConfig.Net.SASL.Password = config.Password
 }
 
-func configureSASL(ctx context.Context, config configkafka.SASLConfig, saramaConfig *sarama.Config) error {
+func configureSASL(ctx context.Context, config configkafka.SASLConfig, saramaConfig *sarama.Config) {
 	saramaConfig.Net.SASL.Enable = true
 	saramaConfig.Net.SASL.User = config.Username
 	saramaConfig.Net.SASL.Password = config.Password
@@ -74,21 +63,7 @@ func configureSASL(ctx context.Context, config configkafka.SASLConfig, saramaCon
 	case "AWS_MSK_IAM_OAUTHBEARER":
 		saramaConfig.Net.SASL.Mechanism = sarama.SASLTypeOAuth
 		saramaConfig.Net.SASL.TokenProvider = &awsMSKTokenProvider{ctx: ctx, region: config.AWSMSK.Region}
-		tlsConfig := tls.Config{}
-		saramaConfig.Net.TLS.Enable = true
-		saramaConfig.Net.TLS.Config = &tlsConfig
-	}
-	return nil
-}
-
-func configureTLS(ctx context.Context, config configtls.ClientConfig, saramaConfig *sarama.Config) error {
-	tlsConfig, err := config.LoadTLSConfig(ctx)
-	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 configkafka.KerberosConfig, saramaConfig *sarama.Config) {

internal/kafka/authentication_test.go

@@ -5,13 +5,10 @@ package kafka
 
 import (
 	"context"
-	"crypto/tls"
 	"testing"
 
 	"github.com/IBM/sarama"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"go.opentelemetry.io/collector/config/configtls"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/kafka/configkafka"
 )
@@ -47,13 +44,6 @@ func TestAuthentication(t *testing.T) {
 	saramaSASLPLAINConfig.Net.SASL.Password = "pass"
 	saramaSASLPLAINConfig.Net.SASL.Mechanism = sarama.SASLTypePlaintext
 
-	saramaTLSCfg := &sarama.Config{}
-	saramaTLSCfg.Net.TLS.Enable = true
-	tlsClient := configtls.ClientConfig{}
-	tlscfg, err := tlsClient.LoadTLSConfig(context.Background())
-	require.NoError(t, err)
-	saramaTLSCfg.Net.TLS.Config = tlscfg
-
 	saramaSASLAWSIAMOAUTHConfig := &sarama.Config{}
 	saramaSASLAWSIAMOAUTHConfig.Net.SASL.Enable = true
 	saramaSASLAWSIAMOAUTHConfig.Net.SASL.Mechanism = sarama.SASLTypeOAuth
@@ -62,10 +52,6 @@ func TestAuthentication(t *testing.T) {
 		region: "region",
 	}
 
-	tlsConfig := tls.Config{}
-	saramaSASLAWSIAMOAUTHConfig.Net.TLS.Enable = true
-	saramaSASLAWSIAMOAUTHConfig.Net.TLS.Config = &tlsConfig
-
 	saramaKerberosCfg := &sarama.Config{}
 	saramaKerberosCfg.Net.SASL.Mechanism = sarama.SASLTypeGSSAPI
 	saramaKerberosCfg.Net.SASL.Enable = true
@@ -103,17 +89,6 @@ func TestAuthentication(t *testing.T) {
 			},
 			saramaConfig: saramaPlaintext,
 		},
-		{
-			auth:         configkafka.AuthenticationConfig{TLS: &configtls.ClientConfig{}},
-			saramaConfig: saramaTLSCfg,
-		},
-		{
-			auth: configkafka.AuthenticationConfig{TLS: &configtls.ClientConfig{
-				Config: configtls.Config{CAFile: "/doesnotexists"},
-			}},
-			saramaConfig: saramaTLSCfg,
-			err:          "failed to load TLS config",
-		},
 		{
 			auth: configkafka.AuthenticationConfig{
 				Kerberos: &configkafka.KerberosConfig{ServiceName: "foobar"},
@@ -168,26 +143,11 @@ func TestAuthentication(t *testing.T) {
 	for _, test := range tests {
 		t.Run("", func(t *testing.T) {
 			config := &sarama.Config{}
-			err := ConfigureSaramaAuthentication(context.Background(), test.auth, config)
-			if test.err != "" {
-				assert.ErrorContains(t, err, test.err)
-			} else {
-				// equalizes SCRAMClientGeneratorFunc to do assertion with the same reference.
-				config.Net.SASL.SCRAMClientGeneratorFunc = test.saramaConfig.Net.SASL.SCRAMClientGeneratorFunc
-				assert.Equal(t, test.saramaConfig, config)
-			}
-		})
-	}
-}
+			configureSaramaAuthentication(context.Background(), test.auth, config)
 
-func TestConfigureSaramaAuthentication_TLS(t *testing.T) {
-	auth := configkafka.AuthenticationConfig{
-		TLS: &configtls.ClientConfig{
-			Config: configtls.Config{
-				CAFile: "/nonexistent",
-			},
-		},
+			// equalizes SCRAMClientGeneratorFunc to do assertion with the same reference.
+			config.Net.SASL.SCRAMClientGeneratorFunc = test.saramaConfig.Net.SASL.SCRAMClientGeneratorFunc
+			assert.Equal(t, test.saramaConfig, config)
+		})
 	}
-	err := ConfigureSaramaAuthentication(context.Background(), auth, &sarama.Config{})
-	require.ErrorContains(t, err, "failed to load TLS config")
 }