Add ScalersCache to reuse scales unless they need changing

Closes #1121

Signed-off-by: Ahmed ElSayed <ahmels@microsoft.com>

CREATE-NEW-SCALER.md

@@ -93,7 +93,7 @@ The constructor should have the following parameters:
 
 ## Lifecycle of a scaler
 
-The scaler is created and closed everytime KEDA or HPA wants to call `GetMetrics`, and everytime a new ScaledObject is created or updated that has a trigger for that scaler. Thus, a developer of a scaler should not assume that the scaler will maintain any state between these calls.
+Scalers are created and cached until the ScaledObject is modified, or `.IsActive()`/`GetMetrics()` result in an error. The cached scaler is then invalidated and a new scaler is created. `Close()` is called on all scalers when disposed.
 
 ## Note
 The scaler code is embedded into the two separate binaries comprising KEDA, the operator and the custom metrics server component. The metrics server must be occasionally rebuilt published and deployed to k8s for it to have the same code as your operator.

adapter/main.go

@@ -17,16 +17,18 @@ limitations under the License.
 package main
 
 import (
+	"context"
 	"flag"
 	"fmt"
 	"os"
 	"runtime"
 	"strconv"
 	"time"
 
+	"github.com/go-logr/logr"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/util/wait"
+	k8sruntime "k8s.io/apimachinery/pkg/runtime"
 	openapinamer "k8s.io/apiserver/pkg/endpoints/openapi"
 	genericapiserver "k8s.io/apiserver/pkg/server"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -36,12 +38,14 @@ import (
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 
 	basecmd "sigs.k8s.io/custom-metrics-apiserver/pkg/cmd"
 	"sigs.k8s.io/custom-metrics-apiserver/pkg/provider"
 
 	generatedopenapi "github.com/kedacore/keda/v2/adapter/generated/openapi"
 	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
+	kedacontrollers "github.com/kedacore/keda/v2/controllers/keda"
 	prommetrics "github.com/kedacore/keda/v2/pkg/metrics"
 	kedaprovider "github.com/kedacore/keda/v2/pkg/provider"
 	"github.com/kedacore/keda/v2/pkg/scaling"
@@ -65,7 +69,7 @@ var (
 	adapterClientRequestBurst int
 )
 
-func (a *Adapter) makeProvider(globalHTTPTimeout time.Duration) (provider.MetricsProvider, error) {
+func (a *Adapter) makeProvider(globalHTTPTimeout time.Duration) (provider.MetricsProvider, <-chan struct{}, error) {
 	// Get a config to talk to the apiserver
 	cfg, err := config.GetConfig()
 	if cfg != nil {
@@ -75,25 +79,25 @@ func (a *Adapter) makeProvider(globalHTTPTimeout time.Duration) (provider.Metric
 
 	if err != nil {
 		logger.Error(err, "failed to get the config")
-		return nil, fmt.Errorf("failed to get the config (%s)", err)
+		return nil, nil, fmt.Errorf("failed to get the config (%s)", err)
 	}
 
 	scheme := scheme.Scheme
 	if err := appsv1.SchemeBuilder.AddToScheme(scheme); err != nil {
 		logger.Error(err, "failed to add apps/v1 scheme to runtime scheme")
-		return nil, fmt.Errorf("failed to add apps/v1 scheme to runtime scheme (%s)", err)
+		return nil, nil, fmt.Errorf("failed to add apps/v1 scheme to runtime scheme (%s)", err)
 	}
 	if err := kedav1alpha1.SchemeBuilder.AddToScheme(scheme); err != nil {
 		logger.Error(err, "failed to add keda scheme to runtime scheme")
-		return nil, fmt.Errorf("failed to add keda scheme to runtime scheme (%s)", err)
+		return nil, nil, fmt.Errorf("failed to add keda scheme to runtime scheme (%s)", err)
 	}
 
 	kubeclient, err := client.New(cfg, client.Options{
 		Scheme: scheme,
 	})
 	if err != nil {
 		logger.Error(err, "unable to construct new client")
-		return nil, fmt.Errorf("unable to construct new client (%s)", err)
+		return nil, nil, fmt.Errorf("unable to construct new client (%s)", err)
 	}
 
 	broadcaster := record.NewBroadcaster()
@@ -103,13 +107,43 @@ func (a *Adapter) makeProvider(globalHTTPTimeout time.Duration) (provider.Metric
 	namespace, err := getWatchNamespace()
 	if err != nil {
 		logger.Error(err, "failed to get watch namespace")
-		return nil, fmt.Errorf("failed to get watch namespace (%s)", err)
+		return nil, nil, fmt.Errorf("failed to get watch namespace (%s)", err)
 	}
 
 	prometheusServer := &prommetrics.PrometheusMetricServer{}
 	go func() { prometheusServer.NewServer(fmt.Sprintf(":%v", prometheusMetricsPort), prometheusMetricsPath) }()
+	stopCh := make(chan struct{})
+	if err := runScaledObjectController(scheme, namespace, handler, logger, stopCh); err != nil {
+		return nil, nil, err
+	}
+
+	return kedaprovider.NewProvider(logger, handler, kubeclient, namespace), stopCh, nil
+}
+
+func runScaledObjectController(scheme *k8sruntime.Scheme, namespace string, scaleHandler scaling.ScaleHandler, logger logr.Logger, stopCh chan<- struct{}) error {
+	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
+		Scheme:    scheme,
+		Namespace: namespace,
+	})
+	if err != nil {
+		return err
+	}
+
+	if err := (&kedacontrollers.MetricsScaledObjectReconciler{
+		ScaleHandler: scaleHandler,
+	}).SetupWithManager(mgr, controller.Options{}); err != nil {
+		return err
+	}
+
+	go func() {
+		if err := mgr.Start(context.Background()); err != nil {
+			logger.Error(err, "controller-runtime encountered an error")
+			stopCh <- struct{}{}
+			close(stopCh)
+		}
+	}()
 
-	return kedaprovider.NewProvider(logger, handler, kubeclient, namespace), nil
+	return nil
 }
 
 func printVersion() {
@@ -171,15 +205,15 @@ func main() {
 		return
 	}
 
-	kedaProvider, err := cmd.makeProvider(time.Duration(globalHTTPTimeoutMS) * time.Millisecond)
+	kedaProvider, stopCh, err := cmd.makeProvider(time.Duration(globalHTTPTimeoutMS) * time.Millisecond)
 	if err != nil {
 		logger.Error(err, "making provider")
 		return
 	}
 	cmd.WithExternalMetrics(kedaProvider)
 
 	logger.Info(cmd.Message)
-	if err = cmd.Run(wait.NeverStop); err != nil {
+	if err = cmd.Run(stopCh); err != nil {
 		return
 	}
 }

config/crd/bases/keda.sh_clustertriggerauthentications.yaml

@@ -4,7 +4,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
-    controller-gen.kubebuilder.io/version: v0.4.1
+    controller-gen.kubebuilder.io/version: v0.6.1
   creationTimestamp: null
   name: clustertriggerauthentications.keda.sh
 spec:
@@ -90,6 +90,8 @@ spec:
                     type: object
                   mount:
                     type: string
+                  namespace:
+                    type: string
                   role:
                     type: string
                   secrets:

config/crd/bases/keda.sh_scaledobjects.yaml

@@ -4,7 +4,7 @@ apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
-    controller-gen.kubebuilder.io/version: v0.4.1
+    controller-gen.kubebuilder.io/version: v0.6.1
   creationTimestamp: null
   name: scaledobjects.keda.sh
 spec:

controllers/keda/hpa.go

@@ -160,35 +160,32 @@ func (r *ScaledObjectReconciler) getScaledObjectMetricSpecs(ctx context.Context,
 	var externalMetricNames []string
 	var resourceMetricNames []string
 
-	scalers, err := r.scaleHandler.GetScalers(ctx, scaledObject)
+	cache, err := r.scaleHandler.GetScalersCache(ctx, scaledObject)
 	if err != nil {
 		logger.Error(err, "Error getting scalers")
 		return nil, err
 	}
 
-	for _, scaler := range scalers {
-		metricSpecs := scaler.GetMetricSpecForScaling(ctx)
+	metricSpecs := cache.GetMetricSpecForScaling(ctx)
 
-		for _, metricSpec := range metricSpecs {
-			if metricSpec.Resource != nil {
-				resourceMetricNames = append(resourceMetricNames, string(metricSpec.Resource.Name))
-			}
-
-			if metricSpec.External != nil {
-				externalMetricName := metricSpec.External.Metric.Name
-				if kedacontrollerutil.Contains(externalMetricNames, externalMetricName) {
-					return nil, fmt.Errorf("metricName %s defined multiple times in ScaledObject %s, please refer the documentation how to define metricName manually", externalMetricName, scaledObject.Name)
-				}
+	for _, metricSpec := range metricSpecs {
+		if metricSpec.Resource != nil {
+			resourceMetricNames = append(resourceMetricNames, string(metricSpec.Resource.Name))
+		}
 
-				// add the scaledobject.keda.sh/name label. This is how the MetricsAdapter will know which scaledobject a metric is for when the HPA queries it.
-				metricSpec.External.Metric.Selector = &metav1.LabelSelector{MatchLabels: make(map[string]string)}
-				metricSpec.External.Metric.Selector.MatchLabels["scaledobject.keda.sh/name"] = scaledObject.Name
-				externalMetricNames = append(externalMetricNames, externalMetricName)
+		if metricSpec.External != nil {
+			externalMetricName := metricSpec.External.Metric.Name
+			if kedacontrollerutil.Contains(externalMetricNames, externalMetricName) {
+				return nil, fmt.Errorf("metricName %s defined multiple times in ScaledObject %s, please refer the documentation how to define metricName manually", externalMetricName, scaledObject.Name)
 			}
+
+			// add the scaledobject.keda.sh/name label. This is how the MetricsAdapter will know which scaledobject a metric is for when the HPA queries it.
+			metricSpec.External.Metric.Selector = &metav1.LabelSelector{MatchLabels: make(map[string]string)}
+			metricSpec.External.Metric.Selector.MatchLabels["scaledobject.keda.sh/name"] = scaledObject.Name
+			externalMetricNames = append(externalMetricNames, externalMetricName)
 		}
-		scaledObjectMetricSpecs = append(scaledObjectMetricSpecs, metricSpecs...)
-		scaler.Close(ctx)
 	}
+	scaledObjectMetricSpecs = append(scaledObjectMetricSpecs, metricSpecs...)
 
 	// sort metrics in ScaledObject, this way we always check the same resource in Reconcile loop and we can prevent unnecessary HPA updates,
 	// see https://github.com/kedacore/keda/issues/1531 for details

controllers/keda/metrics_adapter_controller.go

@@ -0,0 +1,45 @@
+/*
+Copyright 2021 The KEDA Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package keda
+
+import (
+	"context"
+
+	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
+	"github.com/kedacore/keda/v2/pkg/scaling"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+)
+
+type MetricsScaledObjectReconciler struct {
+	ScaleHandler scaling.ScaleHandler
+}
+
+func (r *MetricsScaledObjectReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	r.ScaleHandler.ClearScalersCache(ctx, req.Name, req.Namespace)
+	return ctrl.Result{}, nil
+}
+
+func (r *MetricsScaledObjectReconciler) SetupWithManager(mgr ctrl.Manager, options controller.Options) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&kedav1alpha1.ScaledObject{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})).
+		Owns(&kedav1alpha1.ScaledObject{}).
+		WithOptions(options).
+		Complete(r)
+}

controllers/keda/scaledjob_controller.go

@@ -140,7 +140,7 @@ func (r *ScaledJobReconciler) reconcileScaledJob(ctx context.Context, logger log
 	}
 
 	// Check ScaledJob is Ready or not
-	_, err = r.scaleHandler.GetScalers(ctx, scaledJob)
+	_, err = r.scaleHandler.GetScalersCache(ctx, scaledJob)
 	if err != nil {
 		logger.Error(err, "Error getting scalers")
 		return "Failed to ensure ScaledJob is correctly created", err

pkg/provider/fallback.go

@@ -23,26 +23,23 @@ import (
 	"k8s.io/api/autoscaling/v2beta2"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/metrics/pkg/apis/external_metrics"
 	runtimeclient "sigs.k8s.io/controller-runtime/pkg/client"
 
 	kedav1alpha1 "github.com/kedacore/keda/v2/apis/keda/v1alpha1"
-	"github.com/kedacore/keda/v2/pkg/scalers"
 )
 
 func isFallbackEnabled(scaledObject *kedav1alpha1.ScaledObject, metricSpec v2beta2.MetricSpec) bool {
 	return scaledObject.Spec.Fallback != nil && metricSpec.External.Target.Type == v2beta2.AverageValueMetricType
 }
 
-func (p *KedaProvider) getMetricsWithFallback(ctx context.Context, scaler scalers.Scaler, metricName string, metricSelector labels.Selector, scaledObject *kedav1alpha1.ScaledObject, metricSpec v2beta2.MetricSpec) ([]external_metrics.ExternalMetricValue, error) {
+func (p *KedaProvider) getMetricsWithFallback(ctx context.Context, metrics []external_metrics.ExternalMetricValue, suppressedError error, metricName string, scaledObject *kedav1alpha1.ScaledObject, metricSpec v2beta2.MetricSpec) ([]external_metrics.ExternalMetricValue, error) {
 	status := scaledObject.Status.DeepCopy()
 
 	initHealthStatus(status)
-	metrics, err := scaler.GetMetrics(ctx, metricName, metricSelector)
 	healthStatus := getHealthStatus(status, metricName)
 
-	if err == nil {
+	if suppressedError == nil {
 		zero := int32(0)
 		healthStatus.NumberOfFailures = &zero
 		healthStatus.Status = kedav1alpha1.HealthStatusHappy
@@ -60,14 +57,14 @@ func (p *KedaProvider) getMetricsWithFallback(ctx context.Context, scaler scaler
 
 	switch {
 	case !isFallbackEnabled(scaledObject, metricSpec):
-		return nil, err
+		return nil, suppressedError
 	case !validateFallback(scaledObject):
 		logger.Info("Failed to validate ScaledObject Spec. Please check that parameters are positive integers")
-		return nil, err
+		return nil, suppressedError
 	case *healthStatus.NumberOfFailures > scaledObject.Spec.Fallback.FailureThreshold:
-		return doFallback(scaledObject, metricSpec, metricName, err), nil
+		return doFallback(scaledObject, metricSpec, metricName, suppressedError), nil
 	default:
-		return nil, err
+		return nil, suppressedError
 	}
 }