[internal/kafka] define a common validation function for kafka authentication

[sakulali]

Component(s)

internal/kafka

Is your feature request related to a problem? Please describe.

          Speaking of validation, how about defining a common validation function for `kafka.Authentication` in `internal/kafka` package, which can be reused by `kafkametricsreceiver`, `kafkareceiver` and `kafkaexporter`.

_Originally posted by @fatsheep9146 in https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/27289#discussion_r1343303663_

Currently, we have validation logic inside kafka configure authentication refer to configureSASL. In addition, there is similar validation logic between the validateSASLConfig and configureSASL functions. We can define a common validation function for kafka authentication which can be reused by kafkametricsreceiver, kafkareceiver and kafkaexporter, to make the validation and configuration semantics clearer, and to remove redundant validation logic.

Describe the solution you'd like

1. Define a common validation function ValidateAuthentication , validateTLS and validateSASL for kafka authentication

   
   // ValidateAuthentication validates authentication.
   func ValidateAuthentication(config Authentication) error {
   var errs error
   if config.TLS != nil {
       _, err := validateTLS(*config.TLS)
       errs = multierr.Append(errs, err)
   }
   if config.SASL != nil {
       err := validateSASL(*config.SASL)
       errs = multierr.Append(errs, err)
   }
   return errs
   }

func validateTLS(c configtls.TLSClientSetting) (*tls.Config, error) { tlsConfig, err := c.LoadTLSConfig() if err != nil { return nil, fmt.Errorf("error loading tls config: %w", err) }

return tlsConfig, nil

}

func validateSASL(c SASLConfig) error { if c.Username == "" { return fmt.Errorf("auth.sasl.username is required") }

if c.Password == "" {
    return fmt.Errorf("auth.sasl.password is required")
}

switch c.Mechanism {
case "PLAIN", "AWS_MSK_IAM", "SCRAM-SHA-256", "SCRAM-SHA-512":
    // Do nothing, valid mechanism
default:
    return fmt.Errorf("auth.sasl.mechanism should be one of 'PLAIN', 'AWS_MSK_IAM', 'SCRAM-SHA-256' or 'SCRAM-SHA-512'. configured value %v", c.Mechanism)
}

if c.Version < 0 || c.Version > 1 {
    return fmt.Errorf("auth.sasl.version has to be either 0 or 1. configured value %v", c.Version)
}

return nil

}

2. Chore [`configureSASL`](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/96d53a20663a0f593565fb84abbad99db3ce88da/internal/kafka/authentication.go#L93) and [`configureTLS`](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/96d53a20663a0f593565fb84abbad99db3ce88da/internal/kafka/authentication.go#L137) functions as shown below, first validation, then configuration:

func configureSASL(config SASLConfig, saramaConfig *sarama.Config) error { if err := validateSASL(config); err != nil { return err }

// configuration
    ......
return nil

}

func configureTLS(config configtls.TLSClientSetting, saramaConfig *sarama.Config) error { tlsConfig, err := validateTLS(config) if err != nil { return err }

    // configuration
    ......
return nil

}

3. Replace `validateSASLConfig` with `ValidateAuthentication` in [`Validate`](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/96d53a20663a0f593565fb84abbad99db3ce88da/exporter/kafkaexporter/config.go#L95C5-L95C5)

// Validate checks if the exporter configuration is valid func (cfg *Config) Validate() error { // kafka exporter config validation ...... return ValidateAuthentication(cfg.Authentication) }

### Describe alternatives you've considered

_No response_

### Additional context

_No response_

[github-actions[bot]]

Pinging code owners:

• internal/kafka: @pavolloffay @MovieStoreGuy

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[bryan-aguilar]

edit: I see why the PR is linked :)

[github-actions[bot]]

This issue has been inactive for 60 days. It will be closed in 60 days if there is no activity. To ping code owners by adding a component label, see Adding Labels via Comments, or if you are unsure of which component this issue relates to, please ping @open-telemetry/collector-contrib-triagers. If this issue is still relevant, please ping the code owners or leave a comment explaining why it is still relevant. Otherwise, please close it.

Pinging code owners:

• internal/kafka: @pavolloffay @MovieStoreGuy

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[github-actions[bot]]

This issue has been inactive for 60 days. It will be closed in 60 days if there is no activity. To ping code owners by adding a component label, see Adding Labels via Comments, or if you are unsure of which component this issue relates to, please ping @open-telemetry/collector-contrib-triagers. If this issue is still relevant, please ping the code owners or leave a comment explaining why it is still relevant. Otherwise, please close it.

Pinging code owners:

• internal/kafka: @pavolloffay @MovieStoreGuy

See Adding Labels via Comments if you do not have permissions to add labels yourself.

[github-actions[bot]]

This issue has been closed as inactive because it has been stale for 120 days with no activity.