Scale to zero for Async API (#2224)

Author: RobertLucian

docs/workloads/async/configuration.md

@@ -42,7 +42,7 @@
           success_threshold: <int>  # minimum consecutive successes for the probe to be considered successful after having failed (default: 1)
           failure_threshold: <int>  # minimum consecutive failures for the probe to be considered failed after having succeeded (default: 3)
   autoscaling:  # autoscaling configuration (default: see below)
-    min_replicas: <int>  # minimum number of replicas (default: 1)
+    min_replicas: <int>  # minimum number of replicas (default: 1; min value: 0)
     max_replicas: <int>  # maximum number of replicas (default: 100)
     init_replicas: <int>  # initial number of replicas (default: <min_replicas>)
     target_in_flight: <int>  # desired number of in-flight requests per replica (including requests actively being processed as well as queued), which the autoscaler tries to maintain (default: 1)

pkg/operator/lib/autoscaler/autoscaler.go

@@ -156,10 +156,6 @@ func AutoscaleFn(initialDeployment *kapps.Deployment, apiSpec *spec.API, getInFl
 			recommendation = upscaleFactorCeil
 		}
 
-		if recommendation < 1 {
-			recommendation = 1
-		}
-
 		if recommendation < autoscalingSpec.MinReplicas {
 			recommendation = autoscalingSpec.MinReplicas
 		}

pkg/operator/resources/asyncapi/autoscaler.go

@@ -26,6 +26,7 @@ import (
 	"github.com/aws/aws-sdk-go/service/sqs"
 	"github.com/cortexlabs/cortex/pkg/config"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/lib/pointer"
 	"github.com/cortexlabs/cortex/pkg/types/userconfig"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -111,12 +112,8 @@ func getMessagesInQueue(apiName string, window time.Duration) (*float64, error)
 		return nil, errors.ErrorUnexpected("failed to convert prometheus metric to vector")
 	}
 
-	// no values available
-	if values.Len() == 0 {
-		return nil, nil
+	if values.Len() != 0 {
+		return pointer.Float64(float64(values[0].Value)), nil
 	}
-
-	avgMessagesInQueue := float64(values[0].Value)
-
-	return &avgMessagesInQueue, nil
+	return nil, nil
 }

pkg/operator/resources/asyncapi/k8s_specs.go

@@ -250,7 +250,7 @@ func deploymentSpec(api spec.API, prevDeployment *kapps.Deployment, queueURL str
 func getRequestedReplicasFromDeployment(api spec.API, deployment *kapps.Deployment) int32 {
 	requestedReplicas := api.Autoscaling.InitReplicas
 
-	if deployment != nil && deployment.Spec.Replicas != nil && *deployment.Spec.Replicas > 0 {
+	if deployment != nil && deployment.Spec.Replicas != nil {
 		requestedReplicas = *deployment.Spec.Replicas
 	}
 

pkg/types/spec/validations.go

@@ -54,14 +54,14 @@ func apiValidation(resource userconfig.Resource) *cr.StructValidation {
 		structFieldValidations = append(resourceStructValidations,
 			podValidation(userconfig.RealtimeAPIKind),
 			networkingValidation(),
-			autoscalingValidation(),
+			autoscalingValidation(resource.Kind),
 			updateStrategyValidation(),
 		)
 	case userconfig.AsyncAPIKind:
 		structFieldValidations = append(resourceStructValidations,
 			podValidation(userconfig.AsyncAPIKind),
 			networkingValidation(),
-			autoscalingValidation(),
+			autoscalingValidation(resource.Kind),
 			updateStrategyValidation(),
 		)
 	case userconfig.BatchAPIKind:
@@ -465,16 +465,21 @@ func computeValidation() *cr.StructFieldValidation {
 	}
 }
 
-func autoscalingValidation() *cr.StructFieldValidation {
+func autoscalingValidation(kind userconfig.Kind) *cr.StructFieldValidation {
+	minReplicas := int32(1)
+	if kind == userconfig.AsyncAPIKind {
+		minReplicas = int32(0)
+	}
+
 	return &cr.StructFieldValidation{
 		StructField: "Autoscaling",
 		StructValidation: &cr.StructValidation{
 			StructFieldValidations: []*cr.StructFieldValidation{
 				{
 					StructField: "MinReplicas",
 					Int32Validation: &cr.Int32Validation{
-						Default:     1,
-						GreaterThan: pointer.Int32(0),
+						Default:              1,
+						GreaterThanOrEqualTo: pointer.Int32(minReplicas),
 					},
 				},
 				{
@@ -488,7 +493,7 @@ func autoscalingValidation() *cr.StructFieldValidation {
 					StructField:  "InitReplicas",
 					DefaultField: "MinReplicas",
 					Int32Validation: &cr.Int32Validation{
-						GreaterThan: pointer.Int32(0),
+						GreaterThanOrEqualTo: pointer.Int32(minReplicas),
 					},
 				},
 				{