Improve the Datadog scaler, including a new optional parameter metricUnavailableValue to fill data when no Datadog metric was returned

CHANGELOG.md

@@ -38,6 +38,7 @@
 
 - **Azure Queue:** Don't call Azure queue GetProperties API unnecessarily ([#2613](https://github.com/kedacore/keda/pull/2613))
 - **Datadog Scaler:** Validate query to contain `{` to prevent panic on invalid query ([#2625](https://github.com/kedacore/keda/issues/2625))
+- **Datadog Scaler:** Several improvements, including a new optional parameter `metricUnavailableValue` to fill data when no Datadog metric was returned ([#2657](https://github.com/kedacore/keda/issues/2657))
 - **GCP Pubsub Scaler** Adding e2e test for GCP PubSub scaler ([#1528](https://github.com/kedacore/keda/issues/1528))
 - **Kafka Scaler** Make "disable" a valid value for tls auth parameter ([#2608](https://github.com/kedacore/keda/issues/2608))
 - **RabbitMQ Scaler:** Include `vhost` for RabbitMQ when retrieving queue info with `useRegex` ([#2498](https://github.com/kedacore/keda/issues/2498))

pkg/scalers/datadog_scaler.go

@@ -3,6 +3,7 @@ package scalers
 import (
  "context"
  "fmt"
+ "regexp"
  "strconv"
  "strings"
  "time"
@@ -39,10 +40,20 @@ type datadogMetadata struct {
  vType valueType
  metricName string
  age int
+ useFiller bool
+ fillValue float64
 }
 
 var datadogLog = logf.Log.WithName("datadog_scaler")
 
+var aggregator, filter, rollup *regexp.Regexp
+
+func init() {
+ aggregator = regexp.MustCompile(`^(avg|sum|min|max):.*`)
+ filter = regexp.MustCompile(`.*\{.*\}.*`)
+ rollup = regexp.MustCompile(`.*\.rollup\(([a-z]+,)?\s*(.+)\)`)
+}
+
 // NewDatadogScaler creates a new Datadog scaler
 func NewDatadogScaler(ctx context.Context, config *ScalerConfig) (Scaler, error) {
  meta, err := parseDatadogMetadata(config)
@@ -60,17 +71,65 @@ func NewDatadogScaler(ctx context.Context, config *ScalerConfig) (Scaler, error)
  }, nil
 }
 
+// parseAndTransformDatadogQuery checks correctness of the user query and adds rollup if not available
+func parseAndTransformDatadogQuery(q string, age int) (string, error) {
+ // Queries should start with a valid aggregator. If not found, prepend avg as default
+ if !aggregator.MatchString(q) {
+ q = "avg:" + q
+ }
+
+ // Wellformed Datadog queries require a filter (between curly brackets)
+ if !filter.MatchString(q) {
+ return "", fmt.Errorf("malformed Datadog query")
+ }
+
+ // Queries can contain rollup functions.
+ match := rollup.FindStringSubmatch(q)
+ if match != nil {
+ // If added, check that the number of seconds is an int
+ rollupAgg, err := strconv.Atoi(match[2])
+ if err != nil {
+ return "", fmt.Errorf("malformed Datadog query")
+ }
+
+ if rollupAgg > age {
+ return "", fmt.Errorf("rollup cannot be bigger than time window")
+ }
+ } else { // If not added, use a default rollup based on the time window size
+ s := fmt.Sprintf(".rollup(avg, %d)", age/5)
+ q += s
+ }
+
+ return q, nil
+}
+
 func parseDatadogMetadata(config *ScalerConfig) (*datadogMetadata, error) {
  meta := datadogMetadata{}
 
+ if val, ok := config.TriggerMetadata["age"]; ok {
+ age, err := strconv.Atoi(val)
+ if err != nil {
+ return nil, fmt.Errorf("age parsing error %s", err.Error())
+ }
+ meta.age = age
+
+ if age < 60 {
+ datadogLog.Info("selecting a window smaller than 60 seconds can cause Datadog not finding a metric value for the query")
+ }
+ } else {
+ meta.age = 90 // Default window 90 seconds
+ }
+
  if val, ok := config.TriggerMetadata["query"]; ok {
- meta.query = val
+ query, err := parseAndTransformDatadogQuery(val, meta.age)
+
+ if err != nil {
+ return nil, fmt.Errorf("error in query: %s", err.Error())
+ }
+ meta.query = query
  } else {
  return nil, fmt.Errorf("no query given")
  }
- if !strings.Contains(meta.query, "{") {
- return nil, fmt.Errorf("expecting query to contain `{`, got %s", meta.query)
- }
 
  if val, ok := config.TriggerMetadata["queryValue"]; ok {
  queryValue, err := strconv.Atoi(val)
@@ -82,20 +141,15 @@ func parseDatadogMetadata(config *ScalerConfig) (*datadogMetadata, error) {
  return nil, fmt.Errorf("no queryValue given")
  }
 
- if val, ok := config.TriggerMetadata["age"]; ok {
- age, err := strconv.Atoi(val)
+ if val, ok := config.TriggerMetadata["metricUnavailableValue"]; ok {
+ fillValue, err := strconv.ParseFloat(val, 64)
  if err != nil {
- return nil, fmt.Errorf("age parsing error %s", err.Error())
+ return nil, fmt.Errorf("metricUnavailableValue parsing error %s", err.Error())
  }
- meta.age = age
- } else {
- meta.age = 90 // Default window 90 seconds
+ meta.fillValue = fillValue
+ meta.useFiller = true
  }
 
- // For all the points in a given window, we take the rollup to the window size
- rollup := fmt.Sprintf(".rollup(avg, %d)", meta.age)
- meta.query += rollup
-
  if val, ok := config.TriggerMetadata["type"]; ok {
  val = strings.ToLower(val)
  switch val {
@@ -174,50 +228,19 @@ func (s *datadogScaler) Close(context.Context) error {
  return nil
 }
 
-// IsActive returns true if we are able to get metrics from Datadog
+// IsActive checks whether the value returned by Datadog is higher than the target value
 func (s *datadogScaler) IsActive(ctx context.Context) (bool, error) {
- ctx = context.WithValue(
- ctx,
- datadog.ContextAPIKeys,
- map[string]datadog.APIKey{
- "apiKeyAuth": {
- Key: s.metadata.apiKey,
- },
- "appKeyAuth": {
- Key: s.metadata.appKey,
- },
- },
- )
-
- ctx = context.WithValue(ctx,
- datadog.ContextServerVariables,
- map[string]string{
- "site": s.metadata.datadogSite,
- })
-
- resp, _, err := s.apiClient.MetricsApi.QueryMetrics(ctx, time.Now().Unix()-int64(s.metadata.age), time.Now().Unix(), s.metadata.query)
+ num, err := s.getQueryResult(ctx)
 
  if err != nil {
  return false, err
  }
 
- series := resp.GetSeries()
-
- if len(series) == 0 {
- return false, nil
- }
-
- points := series[0].GetPointlist()
-
- if len(points) == 0 {
- return false, nil
- }
-
- return true, nil
+ return num > float64(s.metadata.queryValue), nil
 }
 
 // getQueryResult returns result of the scaler query
-func (s *datadogScaler) getQueryResult(ctx context.Context) (int, error) {
+func (s *datadogScaler) getQueryResult(ctx context.Context) (float64, error) {
  ctx = context.WithValue(
  ctx,
  datadog.ContextAPIKeys,
@@ -237,24 +260,48 @@ func (s *datadogScaler) getQueryResult(ctx context.Context) (int, error) {
  "site": s.metadata.datadogSite,
  })
 
- resp, _, err := s.apiClient.MetricsApi.QueryMetrics(ctx, time.Now().Unix()-int64(s.metadata.age), time.Now().Unix(), s.metadata.query)
+ resp, r, err := s.apiClient.MetricsApi.QueryMetrics(ctx, time.Now().Unix()-int64(s.metadata.age), time.Now().Unix(), s.metadata.query)
+
+ if r.StatusCode == 429 {
+ rateLimit := r.Header.Get("X-Ratelimit-Limit")
+ rateLimitReset := r.Header.Get("X-Ratelimit-Reset")
+
+ return -1, fmt.Errorf("your Datadog account reached the %s queries per hour rate limit, next limit reset will happen in %s seconds: %s", rateLimit, rateLimitReset, err)
+ }
+
+ if r.StatusCode != 200 {
+ return -1, fmt.Errorf("error when retrieving Datadog metrics: %s", err)
+ }
+
  if err != nil {
  return -1, fmt.Errorf("error when retrieving Datadog metrics: %s", err)
  }
 
  series := resp.GetSeries()
 
+ if len(series) > 1 {
+ return 0, fmt.Errorf("query returned more than 1 series; modify the query to return only 1 series")
+ }
+
  if len(series) == 0 {
- return 0, fmt.Errorf("no Datadog metrics returned")
+ if !s.metadata.useFiller {
+ return 0, fmt.Errorf("no Datadog metrics returned for the given time window")
+ }
+ return s.metadata.fillValue, nil
  }
 
  points := series[0].GetPointlist()
 
  if len(points) == 0 || len(points[0]) < 2 {
- return 0, fmt.Errorf("no Datadog metrics returned")
+ if !s.metadata.useFiller {
+ return 0, fmt.Errorf("no Datadog metrics returned for the given time window")
+ }
+ return s.metadata.fillValue, nil
  }
 
- return int(*points[0][1]), nil
+ // Return the last point from the series
+ index := len(points) - 1
+ return *points[index][1], nil
 }
 
 // GetMetricSpecForScaling returns the MetricSpec for the Horizontal Pod Autoscaler
@@ -301,7 +348,7 @@ func (s *datadogScaler) GetMetrics(ctx context.Context, metricName string, metri
 
  metric := external_metrics.ExternalMetricValue{
  MetricName: s.metadata.metricName,
- Value: *resource.NewQuantity(int64(num), resource.DecimalSI),
+ Value: *resource.NewMilliQuantity(int64(num*1000), resource.DecimalSI),
  Timestamp: metav1.Now(),
  }
 

pkg/scalers/datadog_scaler_test.go

@@ -5,6 +5,13 @@ import (
  "testing"
 )
 
+type datadogQueries struct {
+ input string
+ age int
+ output string
+ isError bool
+}
+
 type datadogMetricIdentifier struct {
  metadataTestData *datadogAuthMetadataTestData
  scalerIndex int
@@ -17,11 +24,48 @@ type datadogAuthMetadataTestData struct {
  isError bool
 }
 
+var testParseQueries = []datadogQueries{
+ {"", 0, "", true},
+ // All properly formed
+ {"avg:system.cpu.user{*}.rollup(sum, 30)", 120, "avg:system.cpu.user{*}.rollup(sum, 30)", false},
+ {"sum:system.cpu.user{*}.rollup(30)", 30, "sum:system.cpu.user{*}.rollup(30)", false},
+ {"avg:system.cpu.user{automatic-restart:false,bosh_address:192.168.101.12}.rollup(avg, 30)", 120, "avg:system.cpu.user{automatic-restart:false,bosh_address:192.168.101.12}.rollup(avg, 30)", false},
+
+ // Default aggregator
+ {"system.cpu.user{*}.rollup(sum, 30)", 120, "avg:system.cpu.user{*}.rollup(sum, 30)", false},
+
+ // Default rollup
+ {"min:system.cpu.user{*}", 120, "min:system.cpu.user{*}.rollup(avg, 24)", false},
+
+ // Missing filter
+ {"min:system.cpu.user", 120, "", true},
+
+ // Malformed rollup
+ {"min:system.cpu.user{*}.rollup(avg)", 120, "", true},
+}
+
+func TestDatadogScalerParseQueries(t *testing.T) {
+ for _, testData := range testParseQueries {
+ output, err := parseAndTransformDatadogQuery(testData.input, testData.age)
+
+ if err != nil && !testData.isError {
+ t.Error("Expected success but got error", err)
+ }
+ if testData.isError && err == nil {
+ t.Error("Expected error but got success")
+ }
+
+ if output != testData.output {
+ t.Errorf("Expected %s, got %s", testData.output, output)
+ }
+ }
+}
+
 var testDatadogMetadata = []datadogAuthMetadataTestData{
  {map[string]string{}, map[string]string{}, true},
 
  // all properly formed
- {map[string]string{"query": "sum:trace.redis.command.hits{env:none,service:redis}.as_count()", "queryValue": "7", "type": "average", "age": "60"}, map[string]string{"apiKey": "apiKey", "appKey": "appKey", "datadogSite": "datadogSite"}, false},
+ {map[string]string{"query": "sum:trace.redis.command.hits{env:none,service:redis}.as_count()", "queryValue": "7", "metricUnavailableValue": "1.5", "type": "average", "age": "60"}, map[string]string{"apiKey": "apiKey", "appKey": "appKey", "datadogSite": "datadogSite"}, false},
  // default age
  {map[string]string{"query": "sum:trace.redis.command.hits{env:none,service:redis}.as_count()", "queryValue": "7", "type": "average"}, map[string]string{"apiKey": "apiKey", "appKey": "appKey", "datadogSite": "datadogSite"}, false},
  // default type
@@ -32,6 +76,10 @@ var testDatadogMetadata = []datadogAuthMetadataTestData{
  {map[string]string{"query": "sum:trace.redis.command.hits{env:none,service:redis}.as_count()", "type": "average", "age": "60"}, map[string]string{"apiKey": "apiKey", "appKey": "appKey", "datadogSite": "datadogSite"}, true},
  // wrong query value type
  {map[string]string{"query": "sum:trace.redis.command.hits{env:none,service:redis}.as_count()", "queryValue": "notanint", "type": "average", "age": "60"}, map[string]string{"apiKey": "apiKey", "appKey": "appKey", "datadogSite": "datadogSite"}, true},
+ // malformed query
+ {map[string]string{"query": "sum:trace.redis.command.hits", "queryValue": "7", "type": "average", "age": "60"}, map[string]string{"apiKey": "apiKey", "appKey": "appKey", "datadogSite": "datadogSite"}, true},
+ // wrong unavailableMetricValue type
+ {map[string]string{"query": "sum:trace.redis.command.hits{env:none,service:redis}.as_count()", "queryValue": "7", "metricUnavailableValue": "notafloat", "type": "average", "age": "60"}, map[string]string{"apiKey": "apiKey", "appKey": "appKey", "datadogSite": "datadogSite"}, true},
 
  // success api/app keys
  {map[string]string{"query": "sum:trace.redis.command.hits{env:none,service:redis}.as_count()", "queryValue": "7"}, map[string]string{"apiKey": "apiKey", "appKey": "appKey", "datadogSite": "datadogSite"}, false},

tests/scalers/datadog.test.ts

@@ -130,7 +130,7 @@ test.serial(`NGINX deployment should scale to 3 (the max) when getting too many
  // keda based deployment should start scaling up with http requests issued
  let replicaCount = '1'
  for (let i = 0; i < 60 && replicaCount !== '3'; i++) {
- t.log(`Waited ${5 * i} seconds for nginx deployment to scale up`)
+ t.log(`Waited ${15 * i} seconds for nginx deployment to scale up`)
 
  replicaCount = sh.exec(
  `kubectl get deployment nginx --namespace ${testNamespace} -o jsonpath="{.spec.replicas}"`
@@ -151,11 +151,12 @@ test.serial(`NGINX deployment should scale to 3 (the max) when getting too many
  )
 
  for (let i = 0; i < 50 && replicaCount !== '1'; i++) {
+ t.log(`Waited ${15 * i} seconds for nginx deployment to scale down`)
  replicaCount = sh.exec(
  `kubectl get deployment nginx --namespace ${testNamespace} -o jsonpath="{.spec.replicas}"`
  ).stdout
  if (replicaCount !== '1') {
- sh.exec('sleep 5')
+ sh.exec('sleep 15')
  }
  }
 
@@ -164,6 +165,7 @@ test.serial(`NGINX deployment should scale to 3 (the max) when getting too many
 })
 
 test.after.always.cb('clean up datadog resources', t => {
+ sh.exec(`kubectl delete scaledobject -n ${testNamespace} --all`)
  sh.exec(`helm repo rm datadog`)
  sh.exec(`kubectl delete namespace ${datadogNamespace} --force`)
  sh.exec(`kubectl delete namespace ${testNamespace} --force`)
@@ -282,7 +284,6 @@ spec:
  name: nginx
  minReplicaCount: 1
  maxReplicaCount: 3
- pollingInterval: 5
  cooldownPeriod: 10
  advanced:
  horizontalPodAutoscalerConfig:
@@ -295,7 +296,7 @@ spec:
  query: "avg:nginx.net.request_per_s{cluster_name:keda-datadog-cluster}"
  queryValue: "2"
  type: "global"
- age: "60"
+ age: "120"
  authenticationRef:
  name: keda-trigger-auth-datadog-secret
 `