Adding support for AWS MSK IAM authentication for rdkafka2 - #537
Adding support for AWS MSK IAM authentication for rdkafka2#537madebydna wants to merge 1 commit into
Conversation
Signed-off-by: Andrea Singh <info@madebydna.com>
|
This PR has been automatically marked as stale because it has been open 90 days with no activity. Remove stale label or comment or this PR will be closed in 30 days |
|
This PR has been automatically marked as stale because it has been open 90 days with no activity. Remove stale label or comment or this PR will be closed in 30 days |
|
This PR was automatically closed because of stale in 30 days |
|
This PR has been automatically marked as stale because it has been open 90 days with no activity. Remove stale label or comment or this PR will be closed in 30 days |
| if Gem::Version.new(RUBY_VERSION) >= Gem::Version.new('3.0') | ||
| gem.add_dependency 'aws-msk-iam-sasl-signer', '~> 0.1.1' | ||
| end | ||
| end |
There was a problem hiding this comment.
Reading ENV and RUBY_VERSION in the gemspec makes the published gem's dependencies depend on how the release was built.
If USE_RDKAFKA is not set at release time, aws-msk-iam-sasl-signer is not declared, but out_rdkafka2.rb requires it unconditionally, so the plugin fails with LoadError for every user on Ruby 3.0+.
If it is set, rdkafka becomes a hard runtime dependency for everyone, which contradicts the README ("You need to install rdkafka gem") since it is an optional C extension.
Could we keep the rdkafka line in the Gemfile and declare neither gem in the gemspec?
Wrapping the require in begin/rescue PLoadErrorand raising aConfigError` only when MSK IAM is actually configured would match how rdkafka itself is already handled.
| if Gem::Version.create(RUBY_VERSION) >= Gem::Version.create('3.0') | ||
| require 'aws-msk-iam-sasl-signer' | ||
| end |
There was a problem hiding this comment.
This require is not guarded, and aws-msk-iam-sasl-signer is not a dependency of the released gem, so out_rdkafka2 cannot be loaded at all on Ruby 3.0+ unless the user happens to have that gem installed.
On the PR HEAD, ruby -Ilib -e 'require "fluent/plugin/out_rdkafka2"' fails with LoadError: cannot load such file -- aws-msk-iam-sasl-signer.
This breaks users who do not use MSK IAM at all, so it is a regression for existing rdkafka2 users.
CI does not catch it because have_rdkafka in test_out_rdkafka2.rb rescues LoadError and omits the whole suite.
Could we wrap it in begin/rescue LoadError and raise a ConfigError only when MSK IAM is actually configured, the same way the optional rdkafka gem is already handled?
begin
require 'aws-msk-iam-sasl-signer'
rescue LoadError
end| def refresh_token(_config, _client_name) | ||
| log.info("+--- Refreshing token") | ||
| client = get_producer | ||
| # This will happen once upon initialization and is expected to fail, as the producer isnt set yet | ||
| # We will set the token manually after creation and after that this refresh method will work |
There was a problem hiding this comment.
refresh_token calls get_producer, which takes @producers_mutex, but rdkafka's NativeKafka#start runs an initial rd_kafka_poll on the calling thread, so the token refresh callback fires while that same thread already holds the mutex.
With the default share_producer false, Ruby's Mutex is not reentrant, so every write fails with ThreadError: deadlock; recursive locking.
The README says share_producer true is required, but nothing enforces it and the default is false.
Could configure() raise a ConfigError when OAUTHBEARER is configured without share_producer true?
| log.info("Could not get shared client handle, unable to set/refresh token (this is expected one time on startup)") | ||
| return | ||
| end | ||
| signer = AwsMskIamSaslSigner::MSKTokenProvider.new(region: @aws_msk_region) |
There was a problem hiding this comment.
generate_auth_token raises on transient AWS failures (missing credentials, STS throttling, network errors) instead of returning nil, and refresh_token has no rescue.
After startup the callback runs on rdkafka's polling thread, which sets abort_on_exception = true, so one failed refresh takes the process down.
Because the signer raises rather than returning nil, the else branch that calls oauthbearer_set_token_failure is never reached.
Could we wrap the body in begin/rescue and report the error via client.oauthbearer_set_token_failure(e.message) so librdkafka can retry?
2 participants
This adds support for using MSK IAM authentication with the rdkafka2 output type. Authentication and authorization with an MSK cluster are facilitated through a base64-encoded signed URL, which is generated by the aws-msk-iam-sasl-signer-ruby library.
Fixes #522