RealtimeAPI CRD

cli/cmd/lib_apis.go

@@ -24,6 +24,11 @@ import (
 func replicaCountTable(counts *status.ReplicaCounts) table.Table {
 	var rows [][]interface{}
 	for _, replicaCountType := range status.ReplicaCountTypes {
+		// skip up-to-date count
+		if replicaCountType == status.ReplicaCountUpToDate {
+			continue
+		}
+
 		count := counts.GetCountBy(replicaCountType)
 		canBeHiddenIfZero := false
 		switch replicaCountType {

cli/cmd/lib_async_apis.go

@@ -60,8 +60,8 @@ func asyncDescribeAPITable(asyncAPI schema.APIResponse, env cliconfig.Environmen
 		return "", errors.ErrorUnexpected("missing metadata from operator response")
 	}
 
-	if asyncAPI.Status == nil {
-		return "", errors.ErrorUnexpected(fmt.Sprintf("missing status for %s api", asyncAPI.Metadata.Name))
+	if asyncAPI.ReplicaCounts == nil {
+		return "", errors.ErrorUnexpected(fmt.Sprintf("missing replica counts for %s api", asyncAPI.Metadata.Name))
 	}
 
 	t := asyncAPIsTable([]schema.APIResponse{asyncAPI}, []string{env.Name})
@@ -75,7 +75,7 @@ func asyncDescribeAPITable(asyncAPI schema.APIResponse, env cliconfig.Environmen
 		out += "\n" + console.Bold("endpoint: ") + *asyncAPI.Endpoint + "\n"
 	}
 
-	t = replicaCountTable(asyncAPI.Status.ReplicaCounts)
+	t = replicaCountTable(asyncAPI.ReplicaCounts)
 	out += "\n" + t.MustFormat()
 
 	return out, nil
@@ -85,15 +85,27 @@ func asyncAPIsTable(asyncAPIs []schema.APIResponse, envNames []string) table.Tab
 	rows := make([][]interface{}, 0, len(asyncAPIs))
 
 	for i, asyncAPI := range asyncAPIs {
-		if asyncAPI.Metadata == nil || asyncAPI.Status == nil {
+		if asyncAPI.Metadata == nil || (asyncAPI.Status == nil && asyncAPI.ReplicaCounts == nil) {
 			continue
 		}
+
+		var ready, requested, upToDate int32
+		if asyncAPI.Status != nil {
+			ready = asyncAPI.Status.Ready
+			requested = asyncAPI.Status.Requested
+			upToDate = asyncAPI.Status.UpToDate
+		} else {
+			ready = asyncAPI.ReplicaCounts.Ready
+			requested = asyncAPI.ReplicaCounts.Requested
+			upToDate = asyncAPI.ReplicaCounts.UpToDate
+		}
+
 		lastUpdated := time.Unix(asyncAPI.Metadata.LastUpdated, 0)
 		rows = append(rows, []interface{}{
 			envNames[i],
 			asyncAPI.Metadata.Name,
-			fmt.Sprintf("%d/%d", asyncAPI.Status.Ready, asyncAPI.Status.Requested),
-			asyncAPI.Status.UpToDate,
+			fmt.Sprintf("%d/%d", ready, requested),
+			upToDate,
 			libtime.SinceStr(&lastUpdated),
 		})
 	}

cli/cmd/lib_realtime_apis.go

@@ -59,8 +59,8 @@ func realtimeDescribeAPITable(realtimeAPI schema.APIResponse, env cliconfig.Envi
 		return "", errors.ErrorUnexpected("missing metadata from operator response")
 	}
 
-	if realtimeAPI.Status == nil {
-		return "", errors.ErrorUnexpected(fmt.Sprintf("missing status for %s api", realtimeAPI.Metadata.Name))
+	if realtimeAPI.ReplicaCounts == nil {
+		return "", errors.ErrorUnexpected(fmt.Sprintf("missing replica counts for %s api", realtimeAPI.Metadata.Name))
 	}
 
 	t := realtimeAPIsTable([]schema.APIResponse{realtimeAPI}, []string{env.Name})
@@ -74,7 +74,7 @@ func realtimeDescribeAPITable(realtimeAPI schema.APIResponse, env cliconfig.Envi
 		out += "\n" + console.Bold("endpoint: ") + *realtimeAPI.Endpoint + "\n"
 	}
 
-	t = replicaCountTable(realtimeAPI.Status.ReplicaCounts)
+	t = replicaCountTable(realtimeAPI.ReplicaCounts)
 	out += "\n" + t.MustFormat()
 
 	return out, nil
@@ -84,15 +84,27 @@ func realtimeAPIsTable(realtimeAPIs []schema.APIResponse, envNames []string) tab
 	rows := make([][]interface{}, 0, len(realtimeAPIs))
 
 	for i, realtimeAPI := range realtimeAPIs {
-		if realtimeAPI.Metadata == nil || realtimeAPI.Status == nil {
+		if realtimeAPI.Metadata == nil || (realtimeAPI.Status == nil && realtimeAPI.ReplicaCounts == nil) {
 			continue
 		}
+
+		var ready, requested, upToDate int32
+		if realtimeAPI.Status != nil {
+			ready = realtimeAPI.Status.Ready
+			requested = realtimeAPI.Status.Requested
+			upToDate = realtimeAPI.Status.UpToDate
+		} else {
+			ready = realtimeAPI.ReplicaCounts.Ready
+			requested = realtimeAPI.ReplicaCounts.Requested
+			upToDate = realtimeAPI.ReplicaCounts.UpToDate
+		}
+
 		lastUpdated := time.Unix(realtimeAPI.Metadata.LastUpdated, 0)
 		rows = append(rows, []interface{}{
 			envNames[i],
 			realtimeAPI.Metadata.Name,
-			fmt.Sprintf("%d/%d", realtimeAPI.Status.Ready, realtimeAPI.Status.Requested),
-			realtimeAPI.Status.UpToDate,
+			fmt.Sprintf("%d/%d", ready, requested),
+			upToDate,
 			libtime.SinceStr(&lastUpdated),
 		})
 	}

cmd/autoscaler/main.go

@@ -28,6 +28,7 @@ import (
 	"time"
 
 	"github.com/cortexlabs/cortex/pkg/autoscaler"
+	serverless "github.com/cortexlabs/cortex/pkg/crds/apis/serverless/v1alpha1"
 	"github.com/cortexlabs/cortex/pkg/lib/aws"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	"github.com/cortexlabs/cortex/pkg/lib/k8s"
@@ -107,6 +108,9 @@ func main() {
 	defer telemetry.Close()
 
 	scheme := runtime.NewScheme()
+	if err := serverless.AddToScheme(scheme); err != nil {
+		exit(log, err, "failed to add k8s serverless to scheme")
+	}
 	if err := clientgoscheme.AddToScheme(scheme); err != nil {
 		exit(log, err, "failed to add k8s client-go-scheme to scheme")
 	}

go.mod

@@ -26,7 +26,7 @@ require (
 	github.com/go-ole/go-ole v1.2.5 // indirect
 	github.com/gobwas/glob v0.2.3
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
-	github.com/google/go-cmp v0.5.6 // indirect
+	github.com/google/go-cmp v0.5.6
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/google/uuid v1.2.0
 	github.com/googleapis/gnostic v0.5.5 // indirect

manager/manifests/autoscaler.yaml.j2

@@ -42,6 +42,13 @@ rules:
   - get
   - update
   - watch
+- apiGroups:
+  - "serverless.cortex.dev"
+  resources:
+  - realtimeapis
+  verbs:
+  - get
+  - update
 
 ---
 

pkg/activator/activator.go

@@ -158,7 +158,7 @@ func (a *activator) getOrCreateReadinessTracker(apiName string) *readinessTracke
 }
 
 func (a *activator) addAPI(obj interface{}) {
-	apiMetadata, err := getAPIMeta(obj)
+	apiMetadata, err := getAPIMeta(obj, true)
 	if err != nil {
 		a.logger.Errorw("error during virtual service informer add callback", zap.Error(err))
 		telemetry.Error(err)
@@ -182,7 +182,7 @@ func (a *activator) addAPI(obj interface{}) {
 }
 
 func (a *activator) updateAPI(oldObj interface{}, newObj interface{}) {
-	apiMetadata, err := getAPIMeta(newObj)
+	apiMetadata, err := getAPIMeta(newObj, true)
 	if err != nil {
 		a.logger.Errorw("error during virtual service informer update callback", zap.Error(err))
 		telemetry.Error(err)
@@ -195,7 +195,7 @@ func (a *activator) updateAPI(oldObj interface{}, newObj interface{}) {
 
 	apiName := apiMetadata.apiName
 
-	oldAPIMetatada, err := getAPIMeta(oldObj)
+	oldAPIMetatada, err := getAPIMeta(oldObj, true)
 	if err != nil {
 		a.logger.Errorw("error during virtual service informer update callback", zap.Error(err))
 		telemetry.Error(err)
@@ -212,7 +212,7 @@ func (a *activator) updateAPI(oldObj interface{}, newObj interface{}) {
 }
 
 func (a *activator) removeAPI(obj interface{}) {
-	apiMetadata, err := getAPIMeta(obj)
+	apiMetadata, err := getAPIMeta(obj, false)
 	if err != nil {
 		a.logger.Errorw("error during virtual service informer delete callback", zap.Error(err))
 		telemetry.Error(err)
@@ -250,7 +250,7 @@ func (a *activator) updateReadinessTracker(obj interface{}) {
 		return
 	}
 
-	api, err := getAPIMeta(obj)
+	api, err := getAPIMeta(obj, false)
 	if err != nil {
 		a.logger.Errorw("error during deployment informer callback", zap.Error(err))
 		telemetry.Error(err)
@@ -271,7 +271,7 @@ func (a *activator) updateReadinessTracker(obj interface{}) {
 }
 
 func (a *activator) removeReadinessTracker(obj interface{}) {
-	api, err := getAPIMeta(obj)
+	api, err := getAPIMeta(obj, false)
 	if err != nil {
 		a.logger.Errorw("error during deployment informer callback", zap.Error(err))
 		telemetry.Error(err)

pkg/activator/helpers.go

@@ -31,7 +31,7 @@ type apiMeta struct {
 	maxQueueLength int
 }
 
-func getAPIMeta(obj interface{}) (apiMeta, error) {
+func getAPIMeta(obj interface{}, includeAnnotations bool) (apiMeta, error) {
 	resource, err := meta.Accessor(obj)
 	if err != nil {
 		return apiMeta{}, err
@@ -48,16 +48,22 @@ func getAPIMeta(obj interface{}) (apiMeta, error) {
 		return apiMeta{}, errors.ErrorUnexpected("got a virtual service without apiName label")
 	}
 
-	maxQueueLength, maxConcurrency, err := userconfig.ConcurrencyFromAnnotations(resource)
-	if err != nil {
-		return apiMeta{}, err
+	var maxQueueLength, maxConcurrency int
+	var annotations map[string]string
+
+	if includeAnnotations {
+		maxQueueLength, maxConcurrency, err = userconfig.ConcurrencyFromAnnotations(resource)
+		if err != nil {
+			return apiMeta{}, err
+		}
+		annotations = resource.GetAnnotations()
 	}
 
 	return apiMeta{
 		apiName:        apiName,
 		apiKind:        userconfig.KindFromString(apiKind),
 		labels:         labels,
-		annotations:    resource.GetAnnotations(),
+		annotations:    annotations,
 		maxConcurrency: maxConcurrency,
 		maxQueueLength: maxQueueLength,
 	}, nil

pkg/autoscaler/autoscaler_test.go

@@ -297,12 +297,9 @@ func TestAutoscaler_Awake(t *testing.T) {
 
 	ticker := time.NewTicker(250 * time.Millisecond)
 	go func() {
-		for {
-			select {
-			case <-ticker.C:
-				err := autoscaleFn()
-				require.NoError(t, err)
-			}
+		for range ticker.C {
+			err := autoscaleFn()
+			require.NoError(t, err)
 		}
 	}()
 
@@ -372,12 +369,9 @@ func TestAutoscaler_MinReplicas(t *testing.T) {
 
 	ticker := time.NewTicker(250 * time.Millisecond)
 	go func() {
-		for {
-			select {
-			case <-ticker.C:
-				err := autoscaleFn()
-				require.NoError(t, err)
-			}
+		for range ticker.C {
+			err := autoscaleFn()
+			require.NoError(t, err)
 		}
 	}()
 
@@ -444,12 +438,9 @@ func TestAutoscaler_MaxReplicas(t *testing.T) {
 
 	ticker := time.NewTicker(250 * time.Millisecond)
 	go func() {
-		for {
-			select {
-			case <-ticker.C:
-				err := autoscaleFn()
-				require.NoError(t, err)
-			}
+		for range ticker.C {
+			err := autoscaleFn()
+			require.NoError(t, err)
 		}
 	}()
 

pkg/autoscaler/helpers.go

@@ -0,0 +1,65 @@
+/*
+Copyright 2021 Cortex Labs, Inc.
+
+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 autoscaler
+
+import (
+	serverless "github.com/cortexlabs/cortex/pkg/crds/apis/serverless/v1alpha1"
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	libstrings "github.com/cortexlabs/cortex/pkg/lib/strings"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+)
+
+func generateAutoscalingFromServerlessRealtimeAPI(realtimeAPI serverless.RealtimeAPI) (*userconfig.Autoscaling, error) {
+	targetInFlight, ok := libstrings.ParseFloat64(realtimeAPI.Spec.Autoscaling.TargetInFlight)
+	if !ok {
+		return nil, errors.ErrorUnexpected("failed to parse target-in-flight requests from autoscaling spec")
+	}
+
+	maxDownscaleFactor, ok := libstrings.ParseFloat64(realtimeAPI.Spec.Autoscaling.MaxDownscaleFactor)
+	if !ok {
+		return nil, errors.ErrorUnexpected("failed to parse max downscale factor from autoscaling spec")
+	}
+
+	maxUpscaleFactor, ok := libstrings.ParseFloat64(realtimeAPI.Spec.Autoscaling.MaxUpscaleFactor)
+	if !ok {
+		return nil, errors.ErrorUnexpected("failed to parse max upscale factor from autoscaling spec")
+	}
+
+	downscaleTolerance, ok := libstrings.ParseFloat64(realtimeAPI.Spec.Autoscaling.DownscaleTolerance)
+	if !ok {
+		return nil, errors.ErrorUnexpected("failed to parse downscale tolerance from autoscaling spec")
+	}
+
+	upscaleTolerance, ok := libstrings.ParseFloat64(realtimeAPI.Spec.Autoscaling.UpscaleTolerance)
+	if !ok {
+		return nil, errors.ErrorUnexpected("failed to parse upscale tolerance from autoscaling spec")
+	}
+
+	return &userconfig.Autoscaling{
+		MinReplicas:                  realtimeAPI.Spec.Autoscaling.MinReplicas,
+		MaxReplicas:                  realtimeAPI.Spec.Autoscaling.MaxReplicas,
+		InitReplicas:                 realtimeAPI.Spec.Autoscaling.InitReplicas,
+		TargetInFlight:               &targetInFlight,
+		Window:                       realtimeAPI.Spec.Autoscaling.Window.Duration,
+		DownscaleStabilizationPeriod: realtimeAPI.Spec.Autoscaling.DownscaleStabilizationPeriod.Duration,
+		UpscaleStabilizationPeriod:   realtimeAPI.Spec.Autoscaling.UpscaleStabilizationPeriod.Duration,
+		MaxDownscaleFactor:           maxDownscaleFactor,
+		MaxUpscaleFactor:             maxUpscaleFactor,
+		DownscaleTolerance:           downscaleTolerance,
+		UpscaleTolerance:             upscaleTolerance,
+	}, nil
+}