Unable to create Scaled Object with Oauthbearer SASL type with Confluent Cloud cluster #5757
Comments
|
@dttung2905 @zroubalik , you are the kafka experts 😄 |
|
Hi all, I experienced the same issue reported by @acartag7 |
|
sorry I missed this. Let me take a look into this in the next few days.Probably need to create a local strimzi kafka cluster with SASL/OAUTHBEARER authentication first 😆 |
|
I just spend some times looking through the code but can't find anything yet. Partly I think its because I had not been able to set up a local strimzi cluster with SASL/OAUTHBEARER authentication. 🤦 Hi @adrien-f , I see you have recently contributed with this PR #5692. I think you might have an already kafka cluster setup with SASL/OAUTHBEARER and has much better understanding of this part of the code than me 😄 Do you mind help us take a look at this too ? |
|
Greetings ! I'd gladly help as much as possible! So as a reference, if not configured for IAM MSK, it will use the Kafka OAuthBearerTokenProvider: keda/pkg/scalers/kafka_scaler.go Line 659 in d63bc7a The errors mentioning the use of OAUth, we can assume that the configuration was parsed properly. We can also add a unit test to verify that later. What's left is to try out the Token Provider, you could try this manually (code written by hand): package main
import (
"context"
"sync"
"time"
"github.com/IBM/sarama"
"github.com/aws/aws-msk-iam-sasl-signer-go/signer"
"github.com/aws/aws-sdk-go-v2/aws"
"golang.org/x/oauth2"
"golang.org/x/oauth2/clientcredentials"
)
type TokenProvider interface {
sarama.AccessTokenProvider
String() string
}
type oauthBearerTokenProvider struct {
tokenSource oauth2.TokenSource
extensions map[string]string
}
func OAuthBearerTokenProvider(clientID, clientSecret, tokenURL string, scopes []string, extensions map[string]string) TokenProvider {
cfg := clientcredentials.Config{
ClientID: clientID,
ClientSecret: clientSecret,
TokenURL: tokenURL,
Scopes: scopes,
}
return &oauthBearerTokenProvider{
tokenSource: cfg.TokenSource(context.Background()),
extensions: extensions,
}
}
func (o *oauthBearerTokenProvider) Token() (*sarama.AccessToken, error) {
token, err := o.tokenSource.Token()
if err != nil {
return nil, err
}
return &sarama.AccessToken{Token: token.AccessToken, Extensions: o.extensions}, nil
}
func (o *oauthBearerTokenProvider) String() string {
return "OAuthBearer"
}
func main() {
tp := OAuthBearerTokenProvider("xxx", "xxx", ....)
token, err := tp.Token()
if err != nil {
fmt.Println("could not get token", err)
} else {
fmt.Println("got token", token)
}
}If we do indeed have a Token, we'll have to dig deeper 😅 |
|
Hello, we are encountering the same scenario trying to set up KEDA scaler using oauthbearer to Confluent Cloud. saslplaintext also works as the original poster described. Are there any updates on this issue? |
|
Hello @djmacken557 ! Would that be possible to run the debugging script like I mentioned to ensure the token is indeed fetched? The idea is to validate that it's possible to fetch the token before going further into the library. |
|
@adrien-f @dttung2905 getting rid of the extension_ prefix solved the issue for us. For instance try with this: |
|
@rasifmahmud thanks for the confirmation :) Are you willing to update docs? @adrien-f do you think it makes sense to add an unit test for this? |
|
@zroubalik sure, I will raise a PR |
|
This issue has been automatically marked as stale because it has not had recent activity. It will be closed in 7 days if no further activity occurs. Thank you for your contributions. |
|
This issue has been automatically closed due to inactivity. |
Report
We are trying to setup Kafka Scaled object with our confluent cloud dedicated cluster using SASL/OAUTHBEARER authentication and we are getting authentication failures.
I've tried following the documentation here, but seems like there is a problem with the extensions part as during my tests, I saw the following:
If I set the
oauthExtensions: invalid=nothingI get the usual logicalCluster is missing a cluster_idNow if I set up
oauthExtensions: extension_identityPoolId=pool-ebYjwithout the cluster id I get the authentication failed message (the same if I put the the pool id and cluster id as in the manifests below):I know its not the credentials as when I input incorrect credentials I get the following message in the operator directly from microsoft Entra ID:
The Scaledobject using sasl plaintext and the api keys is working without issues but we can't use this auth method in our setup.
I think this issue hasn't been reported before, any ideas on what I could try?
Expected Behavior
The Kafka scaler is active, in ready status and the deployments scale properly.
Actual Behavior
Authentication Fails with sasl/oauthbearer.
Steps to Reproduce the Problem
To test this you will need a kafka cluster and you should authenticate with SASL/OAUTHBEARER
Deploy the manifests below:
Logs from KEDA operator
KEDA Version
2.14.0
Kubernetes Version
1.27
Platform
Any
Scaler Details
Kafka
Anything else?
No response
The text was updated successfully, but these errors were encountered: