ScaledJob: introduce MultipleScalersCalculation

CHANGELOG.md

@@ -70,6 +70,7 @@
 - Fix READY and ACTIVE fields of ScaledJob to show status when we run `kubectl get sj` ([#1855](https://github.com/kedacore/keda/pull/1855))
 - Show HashiCorp Vault Address when using `kubectl get ta` or `kubectl get cta` ([#1862](https://github.com/kedacore/keda/pull/1862))
 - Don't panic when HashiCorp Vault path doesn't exist ([#1864](https://github.com/kedacore/keda/pull/1864))
+- Fix the issue when Scaled Jobs has multiple triggers ([#2016](https://github.com/kedacore/keda/pull/2016))
 
 ### Breaking Changes
 

apis/keda/v1alpha1/scaledjob_types.go

@@ -87,6 +87,8 @@ type ScalingStrategy struct {
 	CustomScalingRunningJobPercentage string `json:"customScalingRunningJobPercentage,omitempty"`
 	// +optional
 	PendingPodConditions []string `json:"pendingPodConditions,omitempty"`
+	// +optional
+	MultipleScalersCalculation string `json:"multipleScalersCalculation,omitempty"`
 }
 
 func init() {

config/crd/bases/keda.sh_scaledjobs.yaml

@@ -7362,6 +7362,8 @@ spec:
                     type: integer
                   customScalingRunningJobPercentage:
                     type: string
+                  multipleScalersCalculation:
+                    type: string
                   pendingPodConditions:
                     items:
                       type: string

pkg/metrics/scaledjob_prometheus_metrics.go

@@ -0,0 +1,124 @@
+package metrics
+
+import (
+	"log"
+	"net/http"
+	"strconv"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
+
+var (
+	scaledJobMetricLabels      = []string{"namespace", "metric", "scaledJob", "scaler", "scalerIndex"}
+	scaledJobScalerErrorsTotal = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "keda_operator",
+			Subsystem: "scaler",
+			Name:      "errors_total",
+			Help:      "Total number of errors for all scalers",
+		},
+		[]string{},
+	)
+	scaledJobScalerMetricsValue = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "keda_operator",
+			Subsystem: "scaler",
+			Name:      "metrics_value",
+			Help:      "Metric Value used for ScaledJob",
+		},
+		scaledJobMetricLabels,
+	)
+	scaledJobScalerErrors = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "keda_operator",
+			Subsystem: "scaler",
+			Name:      "errors",
+			Help:      "Number of scaler errors",
+		},
+		scaledJobMetricLabels,
+	)
+	scaledJobErrors = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "keda_operator",
+			Subsystem: "scaled_job",
+			Name:      "errors",
+			Help:      "Number of scaled object errors",
+		},
+		[]string{"namespace", "scaledObject"},
+	)
+)
+
+// PrometheusMetricServer the type of MetricsServer
+type ScaledJobPrometheusMetricServer struct{}
+
+var scaledJobRegistry *prometheus.Registry
+
+func init() {
+	scaledJobRegistry = prometheus.NewRegistry()
+	scaledJobRegistry.MustRegister(scaledJobScalerErrorsTotal)
+	scaledJobRegistry.MustRegister(scaledJobScalerMetricsValue)
+	scaledJobRegistry.MustRegister(scaledJobScalerErrors)
+	scaledJobRegistry.MustRegister(scaledJobErrors)
+}
+
+// NewServer creates a new http serving instance of prometheus metrics
+func (metricsServer ScaledJobPrometheusMetricServer) NewServer(address string, pattern string) {
+	http.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		_, err := w.Write([]byte("OK"))
+		if err != nil {
+			log.Fatalf("Unable to write to serve custom metrics for scaledJob: %v", err)
+		}
+	})
+	log.Printf("Starting ScaledJob metrics server at %v", address)
+	http.Handle(pattern, promhttp.HandlerFor(registry, promhttp.HandlerOpts{}))
+
+	// initialize the total error metric
+	_, errscaler := scaledJobScalerErrorsTotal.GetMetricWith(prometheus.Labels{})
+	if errscaler != nil {
+		log.Fatalf("Unable to initialize scaledJob total error metrics as : %v", errscaler)
+	}
+
+	log.Fatal(http.ListenAndServe(address, nil))
+}
+
+// RecordScaledJobScalerMetric create a measurement of the external metric used by the ScaledJob
+func (metricsServer ScaledJobPrometheusMetricServer) RecordScaledJobScalerMetric(namespace string, scaledJob string, scaler string, scalerIndex int, metric string, value int64) {
+	scaledJobScalerMetricsValue.With(metricsServer.getLabels(namespace, scaledJob, scaler, scalerIndex, metric)).Set(float64(value))
+}
+
+// RecordScaledJobScalerError counts the number of errors occurred in trying get an external metric used by the ScaledJob
+func (metricsServer ScaledJobPrometheusMetricServer) RecordScaledJobScalerError(namespace string, scaledJob string, scaler string, scalerIndex int, metric string, err error) {
+	if err != nil {
+		scaledJobScalerErrors.With(metricsServer.getLabels(namespace, scaledJob, scaler, scalerIndex, metric)).Inc()
+		// scaledJobErrors.With(prometheus.Labels{"namespace": namespace, "scaledObject": scaledObject}).Inc()
+		metricsServer.RecordScalerObjectError(namespace, scaledJob, err)
+		scaledJobScalerErrorsTotal.With(prometheus.Labels{}).Inc()
+		return
+	}
+	// initialize metric with 0 if not already set
+	_, errscaler := scaledJobScalerErrors.GetMetricWith(metricsServer.getLabels(namespace, scaledJob, scaler, scalerIndex, metric))
+	if errscaler != nil {
+		log.Fatalf("Unable to write to serve custom metrics for scaledJob: %v", errscaler)
+	}
+}
+
+// RecordScalerObjectError counts the number of errors with the scaled job
+func (metricsServer ScaledJobPrometheusMetricServer) RecordScalerObjectError(namespace string, scaledJob string, err error) {
+	labels := prometheus.Labels{"namespace": namespace, "scaledJob": scaledJob}
+	if err != nil {
+		scaledJobErrors.With(labels).Inc()
+		return
+	}
+	// initialize metric with 0 if not already set
+	_, errscaledjob := scaledJobErrors.GetMetricWith(labels)
+	if errscaledjob != nil {
+		log.Fatalf("Unable to write to serve custom metrics for scaledJob: %v", errscaledjob)
+		return
+	}
+}
+
+func (metricsServer ScaledJobPrometheusMetricServer) getLabels(namespace string, scaledJob string, scaler string, scalerIndex int, metric string) prometheus.Labels {
+	return prometheus.Labels{"namespace": namespace, "scaledJob": scaledJob, "scaler": scaler, "scalerIndex": strconv.Itoa(scalerIndex), "metric": metric}
+}

pkg/scaling/scale_handler.go

@@ -23,7 +23,6 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
-	"k8s.io/api/autoscaling/v2beta2"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -37,6 +36,7 @@ import (
 	"github.com/kedacore/keda/v2/pkg/scalers"
 	"github.com/kedacore/keda/v2/pkg/scaling/executor"
 	"github.com/kedacore/keda/v2/pkg/scaling/resolver"
+	"github.com/kedacore/keda/v2/pkg/scaling/scaledjob"
 )
 
 // ScaleHandler encapsulates the logic of calling the right scalers for
@@ -264,99 +264,7 @@ func (h *scaleHandler) isScaledObjectActive(ctx context.Context, scalers []scale
 }
 
 func (h *scaleHandler) isScaledJobActive(ctx context.Context, scalers []scalers.Scaler, scaledJob *kedav1alpha1.ScaledJob) (bool, int64, int64) {
-	var queueLength int64
-	var targetAverageValue int64
-	var maxValue int64
-	isActive := false
-
-	// TODO refactor this, do chores, reduce the verbosity ie: V(1) and frequency of logs
-	// move relevant funcs getTargetAverageValue(), min() and divideWithCeil() out of scaler_handler.go
-	for _, scaler := range scalers {
-		scalerLogger := h.logger.WithValues("Scaler", scaler)
-
-		metricSpecs := scaler.GetMetricSpecForScaling()
-
-		// skip scaler that doesn't return any metric specs (usually External scaler with incorrect metadata)
-		// or skip cpu/memory resource scaler
-		if len(metricSpecs) < 1 || metricSpecs[0].External == nil {
-			continue
-		}
-
-		isTriggerActive, err := scaler.IsActive(ctx)
-
-		scalerLogger.Info("Active trigger", "isTriggerActive", isTriggerActive)
-
-		targetAverageValue = getTargetAverageValue(metricSpecs)
-
-		scalerLogger.Info("Scaler targetAverageValue", "targetAverageValue", targetAverageValue)
-
-		metrics, _ := scaler.GetMetrics(ctx, "queueLength", nil)
-
-		var metricValue int64
-
-		for _, m := range metrics {
-			if m.MetricName == "queueLength" {
-				metricValue, _ = m.Value.AsInt64()
-				queueLength += metricValue
-			}
-		}
-		scalerLogger.Info("QueueLength Metric value", "queueLength", queueLength)
-
-		scaler.Close()
-		if err != nil {
-			scalerLogger.V(1).Info("Error getting scale decision, but continue", "Error", err)
-			h.recorder.Event(scaledJob, corev1.EventTypeWarning, eventreason.KEDAScalerFailed, err.Error())
-			continue
-		} else if isTriggerActive {
-			isActive = true
-			scalerLogger.Info("Scaler is active")
-		}
-	}
-	if targetAverageValue != 0 {
-		maxValue = min(scaledJob.MaxReplicaCount(), divideWithCeil(queueLength, targetAverageValue))
-	}
-	h.logger.Info("Scaler maxValue", "maxValue", maxValue)
-	return isActive, queueLength, maxValue
-}
-
-func getTargetAverageValue(metricSpecs []v2beta2.MetricSpec) int64 {
-	var targetAverageValue int64
-	var metricValue int64
-	var flag bool
-	for _, metric := range metricSpecs {
-		if metric.External.Target.AverageValue == nil {
-			metricValue = 0
-		} else {
-			metricValue, flag = metric.External.Target.AverageValue.AsInt64()
-			if !flag {
-				metricValue = 0
-			}
-		}
-
-		targetAverageValue += metricValue
-	}
-	count := int64(len(metricSpecs))
-	if count != 0 {
-		return targetAverageValue / count
-	}
-	return 0
-}
-
-func divideWithCeil(x, y int64) int64 {
-	ans := x / y
-	reminder := x % y
-	if reminder != 0 {
-		return ans + 1
-	}
-	return ans
-}
-
-// Min function for int64
-func min(x, y int64) int64 {
-	if x > y {
-		return y
-	}
-	return x
+	return scaledjob.GetScaleMetrics(ctx, scalers, scaledJob, h.recorder)
 }
 
 // buildScalers returns list of Scalers for the specified triggers

pkg/scaling/scale_handler_test.go

@@ -37,37 +37,6 @@ import (
 	"github.com/kedacore/keda/v2/pkg/scaling/executor"
 )
 
-func TestTargetAverageValue(t *testing.T) {
-	// count = 0
-	specs := []v2beta2.MetricSpec{}
-	targetAverageValue := getTargetAverageValue(specs)
-	assert.Equal(t, int64(0), targetAverageValue)
-	// 1 1
-	specs = []v2beta2.MetricSpec{
-		createMetricSpec(1),
-		createMetricSpec(1),
-	}
-	targetAverageValue = getTargetAverageValue(specs)
-	assert.Equal(t, int64(1), targetAverageValue)
-	// 5 5 3
-	specs = []v2beta2.MetricSpec{
-		createMetricSpec(5),
-		createMetricSpec(5),
-		createMetricSpec(3),
-	}
-	targetAverageValue = getTargetAverageValue(specs)
-	assert.Equal(t, int64(4), targetAverageValue)
-
-	// 5 5 4
-	specs = []v2beta2.MetricSpec{
-		createMetricSpec(5),
-		createMetricSpec(5),
-		createMetricSpec(3),
-	}
-	targetAverageValue = getTargetAverageValue(specs)
-	assert.Equal(t, int64(4), targetAverageValue)
-}
-
 func TestCheckScaledObjectScalersWithError(t *testing.T) {
 	ctrl := gomock.NewController(t)
 	client := mock_client.NewMockClient(ctrl)