Support csv-format for WATCH_NAMESPACE env var (#5631)

Signed-off-by: Jirka Kremser <jiri.kremser@gmail.com> Signed-off-by: Jirka Kremser <535866+jkremser@users.noreply.github.com>
kedacore · Apr 10, 2024 · 241c2e0 · 241c2e0
1 parent 08aeb57
commit 241c2e0
Show file tree

Hide file tree

Showing 9 changed files with 37 additions and 7 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -70,6 +70,7 @@ Here is an overview of all new **experimental** features:
 - **General**: Add GRPC Healthchecks ([#5590](https://github.com/kedacore/keda/issues/5590))
 - **General**: Add OPENTELEMETRY flag in e2e test YAML ([#5375](https://github.com/kedacore/keda/issues/5375))
 - **General**: Add support for cross tenant/cloud authentication when using Azure Workload Identity for TriggerAuthentication ([#5441](https://github.com/kedacore/keda/issues/5441))
+- **General**: Support csv-format for WATCH_NAMESPACE env var ([#5670](https://github.com/kedacore/keda/issues/5670))
 - **Azure Event Hub Scaler**: Remove usage of checkpoint offsets to account for SDK checkpointing implementation changes ([#5574](https://github.com/kedacore/keda/issues/5574))
 - **GCP Stackdriver Scaler**: Add missing parameters 'rate' and 'count' for GCP Stackdriver Scaler alignment ([#5633](https://github.com/kedacore/keda/issues/5633))
 - **Metrics API Scaler**: Add support for various formats: json, xml, yaml, prometheus ([#2633](https://github.com/kedacore/keda/issues/2633))

diff --git a/controllers/eventing/cloudeventsource_controller.go b/controllers/eventing/cloudeventsource_controller.go
@@ -33,6 +33,7 @@ import (
 	"github.com/kedacore/keda/v2/pkg/eventemitter"
 	"github.com/kedacore/keda/v2/pkg/metricscollector"
 	kedastatus "github.com/kedacore/keda/v2/pkg/status"
+	"github.com/kedacore/keda/v2/pkg/util"
 )
 
 // CloudEventSourceReconciler reconciles a EventSource object
@@ -115,6 +116,7 @@ func (r *CloudEventSourceReconciler) Reconcile(ctx context.Context, req ctrl.Req
 func (r *CloudEventSourceReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&eventingv1alpha1.CloudEventSource{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
+		WithEventFilter(util.IgnoreOtherNamespaces()).
 		Complete(r)
 }
 

diff --git a/controllers/keda/scaledjob_controller.go b/controllers/keda/scaledjob_controller.go
@@ -45,6 +45,7 @@ import (
 	"github.com/kedacore/keda/v2/pkg/metricscollector"
 	"github.com/kedacore/keda/v2/pkg/scaling"
 	kedastatus "github.com/kedacore/keda/v2/pkg/status"
+	"github.com/kedacore/keda/v2/pkg/util"
 )
 
 // +kubebuilder:rbac:groups=keda.sh,resources=scaledjobs;scaledjobs/finalizers;scaledjobs/status,verbs="*"
@@ -91,6 +92,7 @@ func (r *ScaledJobReconciler) SetupWithManager(mgr ctrl.Manager, options control
 				kedacontrollerutil.PausedPredicate{},
 				predicate.GenerationChangedPredicate{},
 			))).
+		WithEventFilter(util.IgnoreOtherNamespaces()).
 		Complete(r)
 }
 

diff --git a/controllers/keda/scaledobject_controller.go b/controllers/keda/scaledobject_controller.go
@@ -50,6 +50,7 @@ import (
 	"github.com/kedacore/keda/v2/pkg/metricscollector"
 	"github.com/kedacore/keda/v2/pkg/scaling"
 	kedastatus "github.com/kedacore/keda/v2/pkg/status"
+	"github.com/kedacore/keda/v2/pkg/util"
 )
 
 // +kubebuilder:rbac:groups=keda.sh,resources=scaledobjects;scaledobjects/finalizers;scaledobjects/status,verbs="*"
@@ -134,6 +135,7 @@ func (r *ScaledObjectReconciler) SetupWithManager(mgr ctrl.Manager, options cont
 				predicate.GenerationChangedPredicate{},
 			),
 		)).
+		WithEventFilter(util.IgnoreOtherNamespaces()).
 		// Trigger a reconcile only when the HPA spec,label or annotation changes.
 		// Ignore updates to HPA status
 		Owns(&autoscalingv2.HorizontalPodAutoscaler{}, builder.WithPredicates(

diff --git a/controllers/keda/triggerauthentication_controller.go b/controllers/keda/triggerauthentication_controller.go
@@ -32,6 +32,7 @@ import (
 	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
 	"github.com/kedacore/keda/v2/pkg/eventreason"
 	"github.com/kedacore/keda/v2/pkg/metricscollector"
+	"github.com/kedacore/keda/v2/pkg/util"
 )
 
 // TriggerAuthenticationReconciler reconciles a TriggerAuthentication object
@@ -90,6 +91,7 @@ func (r *TriggerAuthenticationReconciler) Reconcile(ctx context.Context, req ctr
 func (r *TriggerAuthenticationReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&kedav1alpha1.TriggerAuthentication{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
+		WithEventFilter(util.IgnoreOtherNamespaces()).
 		Complete(r)
 }
 

diff --git a/pkg/metricscollector/opentelemetry.go b/pkg/metricscollector/opentelemetry.go
@@ -268,7 +268,7 @@ func (o *OtelMetrics) RecordScaledObjectPaused(namespace string, scaledObject st
 		attribute.Key("scaledObject").String(scaledObject),
 	)
 
-	cback := func(ctx context.Context, obsrv api.Float64Observer) error {
+	cback := func(_ context.Context, obsrv api.Float64Observer) error {
 		obsrv.Observe(float64(activeVal), opt)
 		return nil
 	}

diff --git a/pkg/scalers/aws/aws_common.go b/pkg/scalers/aws/aws_common.go
@@ -75,7 +75,7 @@ func GetAwsConfig(ctx context.Context, awsRegion string, awsAuthorization Author
 
 	if metadata.awsAuthorization.AwsRoleArn != "" {
 		stsSvc := sts.NewFromConfig(cfg)
-		stsCredentialProvider := stscreds.NewAssumeRoleProvider(stsSvc, metadata.awsAuthorization.AwsRoleArn, func(options *stscreds.AssumeRoleOptions) {})
+		stsCredentialProvider := stscreds.NewAssumeRoleProvider(stsSvc, metadata.awsAuthorization.AwsRoleArn, func(_ *stscreds.AssumeRoleOptions) {})
 		cfg.Credentials = aws.NewCredentialsCache(stsCredentialProvider)
 	}
 	return &cfg, err

diff --git a/pkg/scalers/aws/aws_sigv4.go b/pkg/scalers/aws/aws_sigv4.go
@@ -109,7 +109,7 @@ func NewSigV4RoundTripper(config *scalersconfig.ScalerConfig) (http.RoundTripper
 		return nil, err
 	}
 
-	client := amp.NewFromConfig(*awsCfg, func(o *amp.Options) {})
+	client := amp.NewFromConfig(*awsCfg, func(_ *amp.Options) {})
 	rt := &roundTripper{
 		client: client,
 	}

diff --git a/pkg/util/watch.go b/pkg/util/watch.go
@@ -3,8 +3,11 @@ package util
 import (
 	"fmt"
 	"os"
+	"strings"
 
 	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/event"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 )
 
 // GetWatchNamespaces returns the namespaces the operator should be watching for changes
@@ -15,11 +18,29 @@ func GetWatchNamespaces() (map[string]cache.Config, error) {
 		return map[string]cache.Config{}, fmt.Errorf("%s must be set", WatchNamespaceEnvVar)
 	}
 
-	if ns == "" {
+	if ns == "" || ns == "\"\"" {
 		return map[string]cache.Config{}, nil
 	}
+	nss := strings.Split(ns, ",")
+	nssMap := make(map[string]cache.Config)
+	for _, n := range nss {
+		nssMap[n] = cache.Config{}
+	}
+
+	return nssMap, nil
+}
 
-	return map[string]cache.Config{
-		ns: {},
-	}, nil
+// IgnoreOtherNamespaces returns the predicate for watched events that will filter out those that are not coming
+// from a watched namespace (empty namespace or unset env var denotes all)
+func IgnoreOtherNamespaces() predicate.Predicate {
+	nss, e := GetWatchNamespaces()
+	if len(nss) == 0 || e != nil {
+		return predicate.And() // no-op predicate that returns always true
+	}
+	return predicate.Funcs{
+		GenericFunc: func(e event.GenericEvent) bool {
+			_, ok := nss[e.Object.GetNamespace()]
+			return ok
+		},
+	}
 }