Kafka Exporter w/ AWS_MSK_IAM Mechanism Not Working

[flyerjayden]

Component(s)

exporter/kafka

What happened?

Description

I am getting following error message when authentication mechanism is set to AWS_MSK_IAM:

Error: failed to build pipelines: failed to create "kafka" exporter, in pipeline "traces/*": kafka: invalid configuration (The SASL mechanism configuration is invalid. Possible values are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`)
2023/03/16 06:28:18 collector server run finished with error: failed to build pipelines: failed to create "kafka" exporter, in pipeline "traces/*": kafka: invalid configuration (The SASL mechanism configuration is invalid. Possible values are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`)

Steps to Reproduce

Deploy OpenTelemetry collector using following config:

exporters:
  kafka:
    topic: {{ kafka_topic_name }}
    brokers:
      - {{ broker1_host }}:9098
      - {{ broker2_host}}:9098
    auth:
      sasl:
        username: {{ aws_access_key }}
        password: {{ aws_secret_key }}
        mechanism: AWS_MSK_IAM
        aws_msk:
          region: us-west-2
          broker_addr: {{ broker1_host }}:9098

Expected Result

OpenTelemetry collector sends traces to MSK topic

Actual Result

Error message:

Error: failed to build pipelines: failed to create "kafka" exporter, in pipeline "traces/*": kafka: invalid configuration (The SASL mechanism configuration is invalid. Possible values are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`)
2023/03/16 06:28:18 collector server run finished with error: failed to build pipelines: failed to create "kafka" exporter, in pipeline "traces/*": kafka: invalid configuration (The SASL mechanism configuration is invalid. Possible values are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`)

Collector version

0.73

Environment information

Environment

OS: EKS 1.24 / Amazon Linux 2
Compiler(if manually compiled): Used Docker image from Docker hub

OpenTelemetry Collector configuration

exporters:
  kafka:
    topic: {{ kafka_topic_name }}
    brokers:
      - {{ broker1_host }}:9098
      - {{ broker2_host}}:9098
    auth:
      sasl:
        username: {{ aws_access_key }}
        password: {{ aws_secret_key }}
        mechanism: AWS_MSK_IAM
        aws_msk:
          region: us-west-2
          broker_addr: {{ broker1_host }}:9098

Log output

Error: failed to build pipelines: failed to create "kafka" exporter, in pipeline "traces/*": kafka: invalid configuration (The SASL mechanism configuration is invalid. Possible values are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`)
2023/03/16 06:28:18 collector server run finished with error: failed to build pipelines: failed to create "kafka" exporter, in pipeline "traces/*": kafka: invalid configuration (The SASL mechanism configuration is invalid. Possible values are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`)

Additional context

It seems Sarama (https://github.com/Shopify/sarama/blob/7dbf0b5d85538641227a30a281f1a54e18a94cb7/broker.go#L64) does not support (and they are not planning to support) AWS_MSK_IAM mechanism.

If anyone was able to configure Kafka exporter with IAM, your advices would be very appreciated.

[github-actions]

Pinging code owners:

• exporter/kafka: @pavolloffay @MovieStoreGuy

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

[atoulme]

I think we have an open PR for this. Would you like to help test it out or review it?

[flyerjayden]

Definitely I would love to help!

[atoulme]

I was wrong sorry, it’s for another component: #19650 (review)

I bet we can reuse the logic though.

[flyerjayden]

Do you know which component has that implementation? Commit that you shared is about README.md update.

[atoulme]

I will look into it for you. Give me a bit.

[atoulme]

Well, I'll be... no this is for the Kafka exporter. Check out the latest README and see how this is now documented. It should work for your use case, or there is a bug.

[flyerjayden]

Error: failed to build pipelines: failed to create "kafka" exporter, in pipeline "traces/*": kafka: invalid configuration (The SASL mechanism configuration is invalid. Possible values are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`)
2023/03/19 02:00:16 collector server run finished with error: failed to build pipelines: failed to create "kafka" exporter, in pipeline "traces/*": kafka: invalid configuration (The SASL mechanism configuration is invalid. Possible values are `OAUTHBEARER`, `PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512` and `GSSAPI`)

It does not work.

[flyerjayden]

AWS_MSK_ISM is not one of mechanism that sarama library supports:

authentication.go line 125

case "AWS_MSK_IAM":
		saramaConfig.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient {
			return awsmsk.NewIAMSASLClient(config.AWSMSK.BrokerAddr, config.AWSMSK.Region, saramaConfig.ClientID)
		}
		saramaConfig.Net.SASL.Mechanism = awsmsk.Mechanism

sarama@v1.38.1/broker.go line 63:

// SASLMechanism specifies the SASL mechanism the client uses to authenticate with the broker
type SASLMechanism string

const (
	// SASLTypeOAuth represents the SASL/OAUTHBEARER mechanism (Kafka 2.0.0+)
	SASLTypeOAuth = "OAUTHBEARER"
	// SASLTypePlaintext represents the SASL/PLAIN mechanism
	SASLTypePlaintext = "PLAIN"
	// SASLTypeSCRAMSHA256 represents the SCRAM-SHA-256 mechanism.
	SASLTypeSCRAMSHA256 = "SCRAM-SHA-256"
	// SASLTypeSCRAMSHA512 represents the SCRAM-SHA-512 mechanism.
	SASLTypeSCRAMSHA512 = "SCRAM-SHA-512"
	SASLTypeGSSAPI      = "GSSAPI"
	// SASLHandshakeV0 is v0 of the Kafka SASL handshake protocol. Client and
	// server negotiate SASL auth using opaque packets.
	SASLHandshakeV0 = int16(0)
	// SASLHandshakeV1 is v1 of the Kafka SASL handshake protocol. Client and
	// server negotiate SASL by wrapping tokens with Kafka protocol headers.
	SASLHandshakeV1 = int16(1)
	// SASLExtKeyAuth is the reserved extension key name sent as part of the
	// SASL/OAUTHBEARER initial client response
	SASLExtKeyAuth = "auth"
)