Add BoundServiceAccountToken trigger authentication type

Signed-off-by: Max Cao <[email protected]>

apis/keda/v1alpha1/triggerauthentication_types.go

@@ -95,6 +95,9 @@ type TriggerAuthenticationSpec struct {
 
 	// +optional
 	AwsSecretManager *AwsSecretManager `json:"awsSecretManager,omitempty"`
+
+	// +optional
+	BoundServiceAccountToken []BoundServiceAccountToken `json:"boundServiceAccountToken,omitempty"`
 }
 
 // TriggerAuthenticationStatus defines the observed state of TriggerAuthentication
@@ -378,6 +381,12 @@ type AwsSecretManagerSecret struct {
 	VersionStage string `json:"versionStage,omitempty"`
 }
 
+type BoundServiceAccountToken struct {
+	Parameter          string `json:"parameter"`
+	ServiceAccountName string `json:"serviceAccountName"`
+	Expiry             string `json:"expiry"`
+}
+
 func init() {
 	SchemeBuilder.Register(&ClusterTriggerAuthentication{}, &ClusterTriggerAuthenticationList{})
 	SchemeBuilder.Register(&TriggerAuthentication{}, &TriggerAuthenticationList{})

cmd/operator/main.go

@@ -47,6 +47,7 @@ import (
 	"github.com/kedacore/keda/v2/pkg/k8s"
 	"github.com/kedacore/keda/v2/pkg/metricscollector"
 	"github.com/kedacore/keda/v2/pkg/metricsservice"
+	"github.com/kedacore/keda/v2/pkg/scalers/authentication"
 	"github.com/kedacore/keda/v2/pkg/scaling"
 	kedautil "github.com/kedacore/keda/v2/pkg/util"
 	//+kubebuilder:scaffold:imports
@@ -217,8 +218,14 @@ func main() {
 		os.Exit(1)
 	}
 
-	scaledHandler := scaling.NewScaleHandler(mgr.GetClient(), scaleClient, mgr.GetScheme(), globalHTTPTimeout, eventRecorder, secretInformer.Lister())
-	eventEmitter := eventemitter.NewEventEmitter(mgr.GetClient(), eventRecorder, k8sClusterName, secretInformer.Lister())
+	authClientSet := &authentication.AuthClientSet{
+		TokenReviewInterface: kubeClientset.AuthenticationV1().TokenReviews(),
+		CoreV1Interface:      kubeClientset.CoreV1(),
+		SecretLister:         secretInformer.Lister(),
+	}
+
+	scaledHandler := scaling.NewScaleHandler(mgr.GetClient(), scaleClient, mgr.GetScheme(), globalHTTPTimeout, eventRecorder, authClientSet)
+	eventEmitter := eventemitter.NewEventEmitter(mgr.GetClient(), eventRecorder, k8sClusterName, authClientSet)
 
 	if err = (&kedacontrollers.ScaledObjectReconciler{
 		Client:       mgr.GetClient(),
@@ -237,8 +244,7 @@ func main() {
 		Scheme:            mgr.GetScheme(),
 		GlobalHTTPTimeout: globalHTTPTimeout,
 		EventEmitter:      eventEmitter,
-		SecretsLister:     secretInformer.Lister(),
-		SecretsSynced:     secretInformer.Informer().HasSynced,
+		AuthClientSet:     authClientSet,
 	}).SetupWithManager(mgr, controller.Options{
 		MaxConcurrentReconciles: scaledJobMaxReconciles,
 	}); err != nil {

config/crd/bases/keda.sh_clustertriggerauthentications.yaml

@@ -302,6 +302,21 @@ spec:
                 - secrets
                 - vaultUri
                 type: object
+              boundServiceAccountToken:
+                items:
+                  properties:
+                    expiry:
+                      type: string
+                    parameter:
+                      type: string
+                    serviceAccountName:
+                      type: string
+                  required:
+                  - expiry
+                  - parameter
+                  - serviceAccountName
+                  type: object
+                type: array
               configMapTargetRef:
                 items:
                   description: AuthConfigMapTargetRef is used to authenticate using

config/crd/bases/keda.sh_triggerauthentications.yaml

@@ -301,6 +301,21 @@ spec:
                 - secrets
                 - vaultUri
                 type: object
+              boundServiceAccountToken:
+                items:
+                  properties:
+                    expiry:
+                      type: string
+                    parameter:
+                      type: string
+                    serviceAccountName:
+                      type: string
+                  required:
+                  - expiry
+                  - parameter
+                  - serviceAccountName
+                  type: object
+                type: array
               configMapTargetRef:
                 items:
                   description: AuthConfigMapTargetRef is used to authenticate using

config/rbac/role.yaml

@@ -44,6 +44,13 @@ rules:
   verbs:
   - list
   - watch
+- apiGroups:
+  - ""
+  resources:
+  - serviceaccounts/token
+  verbs:
+  - create
+  - get
 - apiGroups:
   - '*'
   resources:

controllers/keda/scaledjob_controller.go

@@ -29,7 +29,6 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
@@ -45,6 +44,7 @@ import (
 	"github.com/kedacore/keda/v2/pkg/eventemitter"
 	"github.com/kedacore/keda/v2/pkg/eventreason"
 	"github.com/kedacore/keda/v2/pkg/metricscollector"
+	"github.com/kedacore/keda/v2/pkg/scalers/authentication"
 	"github.com/kedacore/keda/v2/pkg/scaling"
 	kedastatus "github.com/kedacore/keda/v2/pkg/status"
 	"github.com/kedacore/keda/v2/pkg/util"
@@ -59,11 +59,10 @@ type ScaledJobReconciler struct {
 	Scheme            *runtime.Scheme
 	GlobalHTTPTimeout time.Duration
 	EventEmitter      eventemitter.EventHandler
+	AuthClientSet     *authentication.AuthClientSet
 
 	scaledJobGenerations *sync.Map
 	scaleHandler         scaling.ScaleHandler
-	SecretsLister        corev1listers.SecretLister
-	SecretsSynced        cache.InformerSynced
 }
 
 type scaledJobMetricsData struct {
@@ -83,7 +82,7 @@ func init() {
 
 // SetupWithManager initializes the ScaledJobReconciler instance and starts a new controller managed by the passed Manager instance.
 func (r *ScaledJobReconciler) SetupWithManager(mgr ctrl.Manager, options controller.Options) error {
-	r.scaleHandler = scaling.NewScaleHandler(mgr.GetClient(), nil, mgr.GetScheme(), r.GlobalHTTPTimeout, mgr.GetEventRecorderFor("scale-handler"), r.SecretsLister)
+	r.scaleHandler = scaling.NewScaleHandler(mgr.GetClient(), nil, mgr.GetScheme(), r.GlobalHTTPTimeout, mgr.GetEventRecorderFor("scale-handler"), r.AuthClientSet)
 	r.scaledJobGenerations = &sync.Map{}
 	return ctrl.NewControllerManagedBy(mgr).
 		WithOptions(options).

controllers/keda/scaledobject_controller.go

@@ -58,6 +58,7 @@ import (
 // +kubebuilder:rbac:groups=autoscaling,resources=horizontalpodautoscalers,verbs="*"
 // +kubebuilder:rbac:groups="",resources=configmaps;configmaps/status,verbs=get;list;watch
 // +kubebuilder:rbac:groups="",resources=events,verbs="*"
+// +kubebuilder:rbac:groups="",resources=serviceaccounts/token,verbs=create;get
 // +kubebuilder:rbac:groups="",resources=pods;services;services;secrets;external,verbs=get;list;watch
 // +kubebuilder:rbac:groups="*",resources="*/scale",verbs=get;list;watch;update;patch
 // +kubebuilder:rbac:groups="",resources="serviceaccounts",verbs=list;watch

go.mod

@@ -200,7 +200,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.17 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.22.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.26.4 // indirect
-	github.com/aws/smithy-go v1.20.3 // indirect
+	github.com/aws/smithy-go v1.20.3
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/blang/semver/v4 v4.0.0 // indirect
 	github.com/cenkalti/backoff/v3 v3.2.2 // indirect
@@ -240,7 +240,7 @@ require (
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang-jwt/jwt v3.2.2+incompatible // indirect
 	github.com/golang-jwt/jwt/v4 v4.5.0 // indirect
-	github.com/golang-jwt/jwt/v5 v5.2.1 // indirect
+	github.com/golang-jwt/jwt/v5 v5.2.1
 	github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9 // indirect
 	github.com/golang-sql/sqlexp v0.1.0 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect

pkg/eventemitter/eventemitter.go

@@ -36,14 +36,14 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
-	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 
 	eventingv1alpha1 "github.com/kedacore/keda/v2/apis/eventing/v1alpha1"
 	"github.com/kedacore/keda/v2/pkg/eventemitter/eventdata"
 	"github.com/kedacore/keda/v2/pkg/metricscollector"
+	"github.com/kedacore/keda/v2/pkg/scalers/authentication"
 	"github.com/kedacore/keda/v2/pkg/scaling/resolver"
 	kedastatus "github.com/kedacore/keda/v2/pkg/status"
 )
@@ -66,7 +66,7 @@ type EventEmitter struct {
 	eventFilterCacheLock     *sync.RWMutex
 	eventLoopContexts        *sync.Map
 	cloudEventProcessingChan chan eventdata.EventData
-	secretsLister            corev1listers.SecretLister
+	authClientSet            *authentication.AuthClientSet
 }
 
 // EventHandler defines the behavior for EventEmitter clients
@@ -96,7 +96,7 @@ const (
 )
 
 // NewEventEmitter creates a new EventEmitter
-func NewEventEmitter(client client.Client, recorder record.EventRecorder, clusterName string, secretsLister corev1listers.SecretLister) EventHandler {
+func NewEventEmitter(client client.Client, recorder record.EventRecorder, clusterName string, authClientSet *authentication.AuthClientSet) EventHandler {
 	return &EventEmitter{
 		log:                      logf.Log.WithName("event_emitter"),
 		client:                   client,
@@ -108,7 +108,7 @@ func NewEventEmitter(client client.Client, recorder record.EventRecorder, cluste
 		eventFilterCacheLock:     &sync.RWMutex{},
 		eventLoopContexts:        &sync.Map{},
 		cloudEventProcessingChan: make(chan eventdata.EventData, maxChannelBuffer),
-		secretsLister:            secretsLister,
+		authClientSet:            authClientSet,
 	}
 }
 
@@ -188,7 +188,7 @@ func (e *EventEmitter) createEventHandlers(ctx context.Context, cloudEventSource
 	}
 
 	// Resolve auth related
-	authParams, podIdentity, err := resolver.ResolveAuthRefAndPodIdentity(ctx, e.client, e.log, spec.AuthenticationRef, nil, cloudEventSourceI.GetNamespace(), e.secretsLister)
+	authParams, podIdentity, err := resolver.ResolveAuthRefAndPodIdentity(ctx, e.client, e.log, spec.AuthenticationRef, nil, cloudEventSourceI.GetNamespace(), e.authClientSet)
 	if err != nil {
 		e.log.Error(err, "error resolving auth params", "cloudEventSource", cloudEventSourceI)
 		return

pkg/scalers/authentication/authentication_helpers.go

@@ -13,10 +13,19 @@ import (
 	libs "github.com/dysnix/predictkube-libs/external/configs"
 	"github.com/dysnix/predictkube-libs/external/http_transport"
 	pConfig "github.com/prometheus/common/config"
+	authenticationv1client "k8s.io/client-go/kubernetes/typed/authentication/v1"
+	corev1client "k8s.io/client-go/kubernetes/typed/core/v1"
+	corev1listers "k8s.io/client-go/listers/core/v1"
 
 	kedautil "github.com/kedacore/keda/v2/pkg/util"
 )
 
+type AuthClientSet struct {
+	authenticationv1client.TokenReviewInterface
+	corev1client.CoreV1Interface
+	corev1listers.SecretLister
+}
+
 const (
 	AuthModesKey = "authModes"
 )