updated per review

Signed-off-by: Ming Meng <ming.meng@collibra.com>

pkg/scalers/authentication/authentication_helpers.go

@@ -93,6 +93,10 @@ func GetAuthConfigs(triggerMetadata, authParams map[string]string) (out *AuthMet
 				return nil, errors.New("both oauth and bearer authentication can not be set")
 			}
 			out.EnableOAuth = true
+			out.OauthTokenURI = authParams["oauthTokenURI"]
+			if authParams["scope"] != "" {
+				out.Scopes = strings.Split(authParams["scope"], " ")
+			}
 			out.ClientID = authParams["clientID"]
 			out.ClientSecret = authParams["clientSecret"]
 		default:

pkg/scalers/authentication/authentication_types.go

@@ -45,9 +45,11 @@ type AuthMeta struct {
 	CA        string
 
 	// oAuth2
-	EnableOAuth  bool
-	ClientID     string
-	ClientSecret string
+	EnableOAuth   bool
+	OauthTokenURI string
+	Scopes        []string
+	ClientID      string
+	ClientSecret  string
 
 	// custom auth header
 	EnableCustomAuth bool

pkg/scalers/pulsar_scaler.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"reflect"
 	"strconv"
 	"strings"
 
@@ -33,11 +34,6 @@ type pulsarMetadata struct {
 	msgBacklogThreshold           int64
 	activationMsgBacklogThreshold int64
 
-	oauthTokenURI string
-	grantType     string
-	scopes        []string
-	clientID      string
-
 	pulsarAuth *authentication.AuthMeta
 
 	statsURL    string
@@ -171,16 +167,6 @@ func parsePulsarMetadata(config *ScalerConfig) (pulsarMetadata, error) {
 		return meta, errors.New("no subscription given")
 	}
 
-	if config.TriggerMetadata["oauthTokenURI"] != "" {
-		meta.oauthTokenURI = config.TriggerMetadata["oauthTokenURI"]
-	}
-	if config.TriggerMetadata["grantType"] != "" {
-		meta.grantType = config.TriggerMetadata["grantType"]
-	}
-	if config.TriggerMetadata["scope"] != "" {
-		meta.scopes = strings.Split(config.TriggerMetadata["scope"], " ")
-	}
-
 	meta.metricName = fmt.Sprintf("%s-%s-%s", "pulsar", meta.topic, meta.subscription)
 
 	meta.activationMsgBacklogThreshold = 0
@@ -217,19 +203,36 @@ func parsePulsarMetadata(config *ScalerConfig) (pulsarMetadata, error) {
 	}
 
 	if auth != nil && auth.EnableOAuth {
-		// use clientID from authenticationRef if provided
-		// otherwise from the metadata
-		if auth.ClientID != "" {
-			meta.clientID = auth.ClientID
-		} else {
-			meta.clientID = config.TriggerMetadata["clientID"]
+		auth.OauthTokenURI = readOAuthConfig(auth, config.TriggerMetadata, "OauthTokenURI")
+		auth.ClientID = readOAuthConfig(auth, config.TriggerMetadata, "ClientID")
+		if auth.Scopes == nil && config.TriggerMetadata["scope"] != "" {
+			auth.Scopes = strings.Split(config.TriggerMetadata["scope"], " ")
 		}
 	}
 	meta.pulsarAuth = auth
 	meta.scalerIndex = config.ScalerIndex
 	return meta, nil
 }
 
+// use values from authenticationRef if provided, otherwise try the metadata
+func readOAuthConfig(auth *authentication.AuthMeta, TriggerMetadata map[string]string, key string) string {
+	authValue := reflect.ValueOf(auth).Elem()
+	value := authValue.FieldByName(key)
+	if value.IsValid() {
+		val := value.Interface().(string)
+		if val != "" {
+			return val
+		}
+	}
+
+	jsonKey := strings.ToLower(string(key[0])) + key[1:]
+	if value, ok := TriggerMetadata[jsonKey]; ok {
+		return fmt.Sprintf("%v", value)
+	}
+
+	return ""
+}
+
 func (s *pulsarScaler) GetStats(ctx context.Context) (*pulsarStats, error) {
 	stats := new(pulsarStats)
 
@@ -241,10 +244,10 @@ func (s *pulsarScaler) GetStats(ctx context.Context) (*pulsarStats, error) {
 	client := s.client
 	if s.metadata.pulsarAuth.EnableOAuth {
 		config := clientcredentials.Config{
-			ClientID:     s.metadata.clientID,
+			ClientID:     s.metadata.pulsarAuth.ClientID,
 			ClientSecret: s.metadata.pulsarAuth.ClientSecret,
-			TokenURL:     s.metadata.oauthTokenURI,
-			Scopes:       s.metadata.scopes,
+			TokenURL:     s.metadata.pulsarAuth.OauthTokenURI,
+			Scopes:       s.metadata.pulsarAuth.Scopes,
 		}
 		client = config.Client(context.Background())
 	}