AWS MSK (Kafka) Support

[jlambatl]

Hello,

I have seen several issues related to AWS's MSK support not being fully supported or tested. References: #1516 and #1833

I have considered contributing to this, but I wanted to check in to see what the maintainers would prefer to implement it.

tl;dr AWS has some nuances with Authenticating to their managed Kafka product; they have created a library/reference implementation that is available at https://github.com/aws/aws-msk-iam-auth/tree/main, and there are Rust implementations available too https://github.com/yuhao-su/aws-msk-iam-sasl-signer-rs/tree/main

Most of the Rust implementation's functionality already exists within the vector codebase, so I don't see it as useful to implement by pulling in yet another dependency. I would be keen to get a maintainer's thoughts on this, just in case I am off track.

Implementation Idea

IAMS support leverages existing OAUTHBEARER support in rdkafka, which is handled by implementing a method generate_oauth_token, that generates and maintains the authentication credentials transparently. The existing Kafka implementation uses this purely for statistics. By updating this to include OAUTH, it could work for source and sink configurations. An example from the Rust implementation can be found here.

If the client's sasl configuration is not using OAUTHBEARER, it shouldn't be used, so it should be a non-breaking change for existing users. What I wasn't clear about from reading the source was what the idiomatic way to refresh AWS credentials is. This would be required in the ClientContext method so that credentials continue to work as AWS expires them. The OAUTH token needs to be signed. It looks like this needs to be implemented directly; the existing functions appear to handle signing requests. Is that correct?

Now this poses some additional questions that are more vector/business logic than technical....

Introducing MSK support introduces the framework to support OAUTHBEARER support more generically when using upstream Kafka. OAUTHBEARER support is built into the library out of the box; the use cases I've tested seem to work fine with the various providers that I tested with. The credential refresh appears to work flawlessly in all cases except AWS's MSK, which would suggest that the ClientContext would only need to refresh the tokens IF it's MSK. This could be written to be extended in the future with an enum or similar to match on the type. Similar to how HTTP Auth works with strategy: Basic, ... AWS. I'm keen to understand how that would be received in a PR Do we need to implement everything OAUTH or create the foundation to be extensible in the future?

** Summary **

We would like to contribute MSK support to Vector using OAUTHBEARER and using as much of the existing codebase as possible, rather than bringing in additional dependencies and duplicated capabilities; however, we have the following questions

1. What's the idiomatic/preferred way to refresh AWS credentials in the ClientContext? Are there any examples of where that happens in the codebase as it stands today?

2. Should we aim to support extensibility in OAUTH patterns for Kafka or create a much smaller patch set that's more focused on MSK with a strong focus on superb documentation?