Repository: cortexlabs/cortex (language: python)

Issue:
Enable Replica Auto Scaling down to zero
It would be really useful especially for smaller applications to be able to scale GPU's down to 0 when there is no traffic.

## Possible approach

* To trigger scaling 1 -> 0, check CloudWatch metrics for no requests for a certain amount of time (user-configurable?).
* Scale 1 -> 0 by setting `deployment.spec.replicas` to 0.
* When scaling 1 -> 0, also update the Istio Virtual Service to route requests to that API to a new deployment running in the Cortex node (or use the existing operator)
* 0 -> 1 scaling is triggered when a request comes in to that service
* Scale 0 -> 1 by setting deployment.spec.replicas to 0
* Either the service holds onto the request until the pod is ready, forwards it, and replies with the response, or responds immediately with a message saying e.g. "0 -> 1 scaling has been triggered, please try again in a few minutes"


Discussion / hints:
Definitely interested in this. Would make Cortext a no-brainer for ML projects which don't yet have users enough to need 1 consistent GPU. Then we'd benefit from the rest of the scaling/infra when the users _do_ come.
To be clear, it's the unused running GPU instance we're concerned about. If it's the case that _something_ has to run, that's fine. Eg, a nano/micro that serves as the orchestrator and must always be on - if that makes anything easier. (Forgive my Cortex newbishness if that's kinda how it already works)
@lefnire Thanks for reaching out; yes, that makes sense, and is exactly how we'd implement it!

We haven't decided yet on our priority for implementing this feature. One thing that can render it less useful (or at least "awkward") is how long it takes to spin up a GPU instance and install the dependencies on it; we'd have to hold on to the request for 5+ minutes before forwarding it along. A more intuitive approach might be to support an asynchronous API instead, where you make the API request and it responds immediately with an execution ID, and then you can make an additional request to another API to query the status/results for the execution ID (we have https://github.com/cortexlabs/cortex/issues/1610 to track this).

In the meantime, in case it's helpful, it is possible to create/delete APIs programmatically via the Cortex CLI or Python client. So if you know you are expecting traffic, or it happens on a regular schedule, you could create/delete APIs accordingly.

Also, we do currently support batch jobs, which is a bit like the asynchronous approach I described, except that autoscaling behaves differently: for batch jobs, you submit a job and indicate how many containers you want to run it on, and then once the job is done, the containers spin down. So it does "scale to 0", but is not designed to handle real-time traffic where each individual request is fairly lightweight, and can come at any time from any source.
> In the meantime, in case it's helpful, it is possible to create/delete APIs programmatically via the Cortex CLI or Python client. So if you know you are expecting traffic, or it happens on a regular schedule, you could create/delete APIs accordingly.

That is indeed helpful. With how valuable Cortex is, any leg-work on our part is worth it for this use-case.

I'll look into the batch jobs. I'm currently using AWS Batch anyway, where the container `exit(0)`s when it hasn't received traffic for a while. So it's a hybrid of real batch jobs vs hosted model, which I might be able to pull off with Cortex batch jobs. AWS Batch has given me ulcers, so if Cortex works well even just replacing Batch could be worth it. 

Thanks for the suggestions! I'll keep subscribed to this issue in case scale->0 ever comes.
@lefnire Sounds good, feel free to reach out on our [gitter](https://gitter.im/cortexlabs/cortex) if you have any questions!

Fix patch:
diff --git a/build/images.sh b/build/images.sh
index a41c24df65..037978a22e 100644
--- a/build/images.sh
+++ b/build/images.sh
@@ -25,6 +25,8 @@ dev_images=(
   "async-gateway"
   "enqueuer"
   "dequeuer"
+  "autoscaler"
+  "activator"
 )
 
 non_dev_images=(
diff --git a/cmd/activator/main.go b/cmd/activator/main.go
new file mode 100644
index 0000000000..4460c865b1
--- /dev/null
+++ b/cmd/activator/main.go
@@ -0,0 +1,199 @@
+/*
+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 main
+
+import (
+	"context"
+	"flag"
+	"net/http"
+	"os"
+	"os/signal"
+	"strconv"
+	"time"
+
+	"github.com/cortexlabs/cortex/pkg/activator"
+	"github.com/cortexlabs/cortex/pkg/autoscaler"
+	"github.com/cortexlabs/cortex/pkg/lib/aws"
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/lib/k8s"
+	"github.com/cortexlabs/cortex/pkg/lib/logging"
+	"github.com/cortexlabs/cortex/pkg/lib/telemetry"
+	"github.com/cortexlabs/cortex/pkg/types/clusterconfig"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"go.uber.org/zap"
+	istioinformers "istio.io/client-go/pkg/informers/externalversions"
+	kmeta "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	kinformers "k8s.io/client-go/informers"
+)
+
+func main() {
+	var (
+		port              int
+		inCluster         bool
+		autoscalerURL     string
+		namespace         string
+		clusterConfigPath string
+	)
+
+	flag.IntVar(&port, "port", 8000, "port where the activator server will be exposed")
+	flag.BoolVar(&inCluster, "in-cluster", false, "use when autoscaler runs in-cluster")
+	flag.StringVar(&autoscalerURL, "autoscaler-url", "", "the URL for the cortex autoscaler endpoint")
+	flag.StringVar(&namespace, "namespace", os.Getenv("CORTEX_NAMESPACE"),
+		"kubernetes namespace where the cortex APIs are deployed "+
+			"(can be set through the CORTEX_NAMESPACE env variable)",
+	)
+	flag.StringVar(&clusterConfigPath, "cluster-config", "", "cluster config path")
+	flag.Parse()
+
+	log := logging.GetLogger()
+	defer func() {
+		_ = log.Sync()
+	}()
+
+	switch {
+	case autoscalerURL == "":
+		log.Fatal("--autoscaler-url is a required option")
+	case namespace == "":
+		log.Fatal("--namespace is a required option")
+	case clusterConfigPath == "":
+		log.Fatal("--cluster-config flag is required")
+	}
+
+	clusterConfig, err := clusterconfig.NewForFile(clusterConfigPath)
+	if err != nil {
+		exit(log, err)
+	}
+
+	awsClient, err := aws.NewForRegion(clusterConfig.Region)
+	if err != nil {
+		exit(log, err)
+	}
+
+	_, userID, err := awsClient.CheckCredentials()
+	if err != nil {
+		exit(log, err)
+	}
+
+	err = telemetry.Init(telemetry.Config{
+		Enabled: clusterConfig.Telemetry,
+		UserID:  userID,
+		Properties: map[string]string{
+			"kind":       userconfig.RealtimeAPIKind.String(),
+			"image_type": "activator",
+		},
+		Environment: "operator",
+		LogErrors:   true,
+		BackoffMode: telemetry.BackoffDuplicateMessages,
+	})
+	if err != nil {
+		log.Fatalw("failed to initialize telemetry", zap.Error(err))
+	}
+	defer telemetry.Close()
+
+	k8sClient, err := k8s.New(namespace, inCluster, nil, runtime.NewScheme())
+	if err != nil {
+		exit(log, err, "failed to initialize kubernetes client")
+	}
+
+	istioClient := k8sClient.IstioClientSet()
+	kubeClient := k8sClient.ClientSet()
+	autoscalerClient := autoscaler.NewClient(autoscalerURL)
+
+	istioInformerFactory := istioinformers.NewSharedInformerFactoryWithOptions(
+		istioClient, 10*time.Second, // TODO: check how much makes sense
+		istioinformers.WithNamespace(namespace),
+		istioinformers.WithTweakListOptions(informerFilter),
+	)
+	virtualServiceInformer := istioInformerFactory.Networking().V1beta1().VirtualServices().Informer()
+	virtualServiceClient := istioClient.NetworkingV1beta1().VirtualServices(namespace)
+
+	kubeInformerFactory := kinformers.NewSharedInformerFactoryWithOptions(
+		kubeClient, 2*time.Second, // TODO: check how much makes sense
+		kinformers.WithNamespace(namespace),
+		kinformers.WithTweakListOptions(informerFilter),
+	)
+	deploymentInformer := kubeInformerFactory.Apps().V1().Deployments().Informer()
+
+	act := activator.New(virtualServiceClient, deploymentInformer, virtualServiceInformer, autoscalerClient, log)
+
+	handler := activator.NewHandler(act, log)
+	server := &http.Server{
+		Addr:    ":" + strconv.Itoa(port),
+		Handler: handler,
+	}
+
+	stopCh := make(chan struct{})
+	go virtualServiceInformer.Run(stopCh)
+	go deploymentInformer.Run(stopCh)
+	defer func() {
+		stopCh <- struct{}{}
+	}()
+
+	errCh := make(chan error)
+	go func() {
+		log.Infof("Starting activator server on %s", server.Addr)
+		errCh <- server.ListenAndServe()
+	}()
+
+	sigint := make(chan os.Signal, 1)
+	signal.Notify(sigint, os.Interrupt)
+
+	select {
+	case err = <-errCh:
+		exit(log, err, "failed to start activator server")
+	case <-sigint:
+		// We received an interrupt signal, shut down.
+		log.Info("Received TERM signal, handling a graceful shutdown...")
+		log.Info("Shutting down server")
+		if err = server.Shutdown(context.Background()); err != nil {
+			// Error from closing listeners, or context timeout:
+			log.Warnw("HTTP server Shutdown Error", zap.Error(err))
+		}
+		log.Info("Shutdown complete, exiting...")
+	}
+}
+
+func informerFilter(listOptions *kmeta.ListOptions) {
+	listOptions.LabelSelector = kmeta.FormatLabelSelector(&kmeta.LabelSelector{
+		MatchLabels: map[string]string{
+			"apiKind": userconfig.RealtimeAPIKind.String(),
+		},
+		MatchExpressions: []kmeta.LabelSelectorRequirement{
+			{
+				Key:      "apiName",
+				Operator: kmeta.LabelSelectorOpExists,
+			},
+		},
+	})
+}
+
+func exit(log *zap.SugaredLogger, err error, wrapStrs ...string) {
+	for _, str := range wrapStrs {
+		err = errors.Wrap(err, str)
+	}
+
+	if err != nil && !errors.IsNoTelemetry(err) {
+		telemetry.Error(err)
+	}
+
+	if err != nil && !errors.IsNoPrint(err) {
+		log.Error(err)
+	}
+
+	os.Exit(1)
+}
diff --git a/cmd/autoscaler/main.go b/cmd/autoscaler/main.go
new file mode 100644
index 0000000000..64aca47208
--- /dev/null
+++ b/cmd/autoscaler/main.go
@@ -0,0 +1,296 @@
+/*
+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 main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"net/http"
+	"os"
+	"os/signal"
+	"strconv"
+	"time"
+
+	"github.com/cortexlabs/cortex/pkg/autoscaler"
+	"github.com/cortexlabs/cortex/pkg/lib/aws"
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/lib/k8s"
+	"github.com/cortexlabs/cortex/pkg/lib/logging"
+	"github.com/cortexlabs/cortex/pkg/lib/telemetry"
+	"github.com/cortexlabs/cortex/pkg/types/clusterconfig"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"github.com/gorilla/mux"
+	promapi "github.com/prometheus/client_golang/api"
+	promv1 "github.com/prometheus/client_golang/api/prometheus/v1"
+	"go.uber.org/zap"
+	istioclient "istio.io/client-go/pkg/clientset/versioned"
+	istioinformers "istio.io/client-go/pkg/informers/externalversions"
+	"k8s.io/apimachinery/pkg/api/meta"
+	kmeta "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/tools/cache"
+)
+
+func main() {
+	var (
+		port              int
+		inCluster         bool
+		prometheusURL     string
+		namespace         string
+		clusterConfigPath string
+	)
+
+	flag.IntVar(&port, "port", 8000, "port where the autoscaler server will be exposed")
+	flag.BoolVar(&inCluster, "in-cluster", false, "use when autoscaler runs in-cluster")
+	flag.StringVar(&prometheusURL, "prometheus-url", os.Getenv("CORTEX_PROMETHEUS_URL"),
+		"prometheus url (can be set through the CORTEX_PROMETHEUS_URL env variable)",
+	)
+	flag.StringVar(&namespace, "namespace", os.Getenv("CORTEX_NAMESPACE"),
+		"kubernetes namespace where the cortex APIs are deployed "+
+			"(can be set through the CORTEX_NAMESPACE env variable)",
+	)
+	flag.StringVar(&clusterConfigPath, "cluster-config", "", "cluster config path")
+	flag.Parse()
+
+	log := logging.GetLogger()
+	defer func() {
+		_ = log.Sync()
+	}()
+
+	switch {
+	case prometheusURL == "":
+		log.Fatal("--prometheus-url is a required option")
+	case namespace == "":
+		log.Fatal("--namespace is a required option")
+	case clusterConfigPath == "":
+		log.Fatal("--cluster-config flag is required")
+	}
+
+	clusterConfig, err := clusterconfig.NewForFile(clusterConfigPath)
+	if err != nil {
+		exit(log, err)
+	}
+
+	awsClient, err := aws.NewForRegion(clusterConfig.Region)
+	if err != nil {
+		exit(log, err)
+	}
+
+	_, userID, err := awsClient.CheckCredentials()
+	if err != nil {
+		exit(log, err)
+	}
+
+	err = telemetry.Init(telemetry.Config{
+		Enabled: clusterConfig.Telemetry,
+		UserID:  userID,
+		Properties: map[string]string{
+			"kind":       userconfig.RealtimeAPIKind.String(),
+			"image_type": "autoscaler",
+		},
+		Environment: "operator",
+		LogErrors:   true,
+		BackoffMode: telemetry.BackoffDuplicateMessages,
+	})
+	if err != nil {
+		log.Fatalw("failed to initialize telemetry", zap.Error(err))
+	}
+	defer telemetry.Close()
+
+	scheme := runtime.NewScheme()
+	if err := clientgoscheme.AddToScheme(scheme); err != nil {
+		exit(log, err, "failed to add k8s client-go-scheme to scheme")
+	}
+
+	k8sClient, err := k8s.New(namespace, inCluster, nil, scheme)
+	if err != nil {
+		exit(log, err, "failed to initialize kubernetes client")
+	}
+
+	//goland:noinspection GoNilness
+	istioClient, err := istioclient.NewForConfig(k8sClient.RestConfig)
+	if err != nil {
+		exit(log, err, "failed to initialize istio client")
+	}
+
+	promClient, err := promapi.NewClient(
+		promapi.Config{
+			Address: prometheusURL,
+		},
+	)
+	if err != nil {
+		exit(log, err, "failed to initialize prometheus client")
+	}
+
+	promAPIClient := promv1.NewAPI(promClient)
+
+	realtimeScaler := autoscaler.NewRealtimeScaler(k8sClient, promAPIClient, log)
+	asyncScaler := autoscaler.NewAsyncScaler(k8sClient, promAPIClient)
+
+	autoScaler := autoscaler.New(log)
+	autoScaler.AddScaler(realtimeScaler, userconfig.RealtimeAPIKind)
+	autoScaler.AddScaler(asyncScaler, userconfig.AsyncAPIKind)
+	defer autoScaler.Stop()
+
+	istioInformerFactory := istioinformers.NewSharedInformerFactoryWithOptions(
+		istioClient, 10*time.Second, // TODO: check how much makes sense
+		istioinformers.WithNamespace(namespace),
+		istioinformers.WithTweakListOptions(informerFilter),
+	)
+	virtualServiceInformer := istioInformerFactory.Networking().V1beta1().VirtualServices().Informer()
+	virtualServiceInformer.AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				resource, err := meta.Accessor(obj)
+				if err != nil {
+					log.Errorw("failed to access resource metadata", zap.Error(err))
+					return
+				}
+
+				if resource.GetNamespace() != namespace {
+					// filter out virtual services that are not in the cortex namespace
+					return
+				}
+
+				api, err := apiResourceFromLabels(resource.GetLabels())
+				if err != nil {
+					// filter out non-cortex apis
+					return
+				}
+
+				if err := autoScaler.AddAPI(api); err != nil {
+					log.Errorw("failed to add API to autoscaler",
+						zap.Error(err),
+						zap.String("apiName", api.Name),
+						zap.String("apiKind", api.Kind.String()),
+					)
+					return
+				}
+			},
+			DeleteFunc: func(obj interface{}) {
+				resource, err := meta.Accessor(obj)
+				if err != nil {
+					log.Errorw("failed to access resource metadata", zap.Error(err))
+				}
+
+				if resource.GetNamespace() != namespace {
+					// filter out virtual services that are not in the cortex namespace
+					return
+				}
+
+				api, err := apiResourceFromLabels(resource.GetLabels())
+				if err != nil {
+					// filter out non-cortex apis
+					return
+				}
+
+				autoScaler.RemoveAPI(api)
+			},
+		},
+	)
+
+	handler := autoscaler.NewHandler(autoScaler)
+	router := mux.NewRouter()
+	router.HandleFunc("/awaken", handler.Awaken).Methods(http.MethodPost)
+	router.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
+		_, _ = w.Write([]byte("ok"))
+	}).Methods(http.MethodGet)
+
+	server := &http.Server{
+		Addr:    ":" + strconv.Itoa(port),
+		Handler: router,
+	}
+
+	stopCh := make(chan struct{})
+	go virtualServiceInformer.Run(stopCh)
+	defer func() { stopCh <- struct{}{} }()
+
+	errCh := make(chan error)
+	go func() {
+		log.Infof("Starting autoscaler server on %s", server.Addr)
+		errCh <- server.ListenAndServe()
+	}()
+
+	sigint := make(chan os.Signal, 1)
+	signal.Notify(sigint, os.Interrupt)
+
+	select {
+	case err = <-errCh:
+		exit(log, err, "failed to start autoscaler server")
+	case <-sigint:
+		// We received an interrupt signal, shut down.
+		log.Info("Received TERM signal, handling a graceful shutdown...")
+		log.Info("Shutting down server")
+		if err = server.Shutdown(context.Background()); err != nil {
+			// Error from closing listeners, or context timeout:
+			log.Warnw("HTTP server Shutdown Error", zap.Error(err))
+		}
+		log.Info("Shutdown complete, exiting...")
+	}
+}
+
+func apiResourceFromLabels(labels map[string]string) (userconfig.Resource, error) {
+	apiName, ok := labels["apiName"]
+	if !ok {
+		return userconfig.Resource{}, fmt.Errorf("apiName key does not exist")
+	}
+
+	apiKind, ok := labels["apiKind"]
+	if !ok {
+		return userconfig.Resource{}, fmt.Errorf("apiKind key does not exist")
+	}
+
+	return userconfig.Resource{
+		Name: apiName,
+		Kind: userconfig.KindFromString(apiKind),
+	}, nil
+}
+
+func informerFilter(listOptions *kmeta.ListOptions) {
+	listOptions.LabelSelector = kmeta.FormatLabelSelector(&kmeta.LabelSelector{
+		MatchExpressions: []kmeta.LabelSelectorRequirement{
+			{
+				Key:      "apiName",
+				Operator: kmeta.LabelSelectorOpExists,
+			},
+			{
+				Key:      "apiKind",
+				Operator: kmeta.LabelSelectorOpExists,
+			},
+		},
+	})
+}
+
+func exit(log *zap.SugaredLogger, err error, wrapStrs ...string) {
+	for _, str := range wrapStrs {
+		err = errors.Wrap(err, str)
+	}
+
+	if err != nil && !errors.IsNoTelemetry(err) {
+		telemetry.Error(err)
+	}
+
+	if err != nil && !errors.IsNoPrint(err) {
+		log.Error(err)
+	}
+
+	os.Exit(1)
+}
diff --git a/cmd/operator/main.go b/cmd/operator/main.go
index 3ed2ca4615..bf5a50d33b 100644
--- a/cmd/operator/main.go
+++ b/cmd/operator/main.go
@@ -30,7 +30,6 @@ import (
 	"github.com/cortexlabs/cortex/pkg/operator/operator"
 	"github.com/cortexlabs/cortex/pkg/operator/resources/asyncapi"
 	"github.com/cortexlabs/cortex/pkg/operator/resources/job/taskapi"
-	"github.com/cortexlabs/cortex/pkg/operator/resources/realtimeapi"
 	"github.com/cortexlabs/cortex/pkg/types/userconfig"
 	"github.com/gorilla/handlers"
 	"github.com/gorilla/mux"
@@ -65,30 +64,13 @@ func main() {
 
 	for i := range deployments {
 		deployment := deployments[i]
-		apiKind := userconfig.KindFromString(deployment.Labels["apiKind"])
-		if apiKind == userconfig.RealtimeAPIKind ||
-			(apiKind == userconfig.AsyncAPIKind && deployment.Labels["cortex.dev/async"] != "gateway") {
-
-			apiID := deployment.Labels["apiID"]
-			apiName := deployment.Labels["apiName"]
-			api, err := operator.DownloadAPISpec(apiName, apiID)
-			if err != nil {
-				exit.Error(errors.Wrap(err, "init"))
-			}
-
-			switch apiKind {
-			case userconfig.RealtimeAPIKind:
-				if err := realtimeapi.UpdateAutoscalerCron(&deployment, api); err != nil {
-					operatorLogger.Fatal(errors.Wrap(err, "init"))
-				}
-			case userconfig.AsyncAPIKind:
+		apiKind := deployment.Labels["apiKind"]
+		switch apiKind {
+		case userconfig.AsyncAPIKind.String():
+			if deployment.Labels["cortex.dev/async"] == "api" {
 				if err := asyncapi.UpdateAPIMetricsCron(&deployment); err != nil {
 					operatorLogger.Fatal(errors.Wrap(err, "init"))
 				}
-
-				if err := asyncapi.UpdateAPIAutoscalerCron(&deployment, *api); err != nil {
-					operatorLogger.Fatal(errors.Wrap(err, "init"))
-				}
 			}
 		}
 	}
diff --git a/cmd/proxy/main.go b/cmd/proxy/main.go
index f162be3d67..e58d1e319a 100644
--- a/cmd/proxy/main.go
+++ b/cmd/proxy/main.go
@@ -176,7 +176,7 @@ func main() {
 			log.Infof("Shutting down %s server", name)
 			if err := server.Shutdown(context.Background()); err != nil {
 				// Error from closing listeners, or context timeout:
-				log.Warn("HTTP server Shutdown Error", zap.Error(err))
+				log.Warnw("HTTP server Shutdown Error", zap.Error(err))
 				telemetry.Error(errors.Wrap(err, "HTTP server Shutdown Error"))
 			}
 		}
diff --git a/images/activator/Dockerfile b/images/activator/Dockerfile
new file mode 100644
index 0000000000..4c221c86cc
--- /dev/null
+++ b/images/activator/Dockerfile
@@ -0,0 +1,26 @@
+FROM golang:1.16.5 as builder
+
+WORKDIR /workspace
+# Copy the Go Modules manifests
+COPY go.mod go.mod
+COPY go.sum go.sum
+# cache deps before building and copying source so that we don't need to re-download as much
+# and so that source changes don't invalidate our downloaded layer
+RUN go mod download
+
+# Copy the go source
+COPY pkg pkg
+COPY cmd/activator cmd/activator
+WORKDIR /workspace/cmd/activator
+
+# Build
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 GO111MODULE=on go build -a -o /workspace/bin/activator main.go
+
+# Use distroless as minimal base image to package the manager binary
+# Refer to https://github.com/GoogleContainerTools/distroless for more details
+FROM gcr.io/distroless/static:nonroot
+WORKDIR /
+COPY --from=builder /workspace/bin/activator .
+USER 65532:65532
+
+ENTRYPOINT ["/activator"]
diff --git a/images/autoscaler/Dockerfile b/images/autoscaler/Dockerfile
new file mode 100644
index 0000000000..ed1e623e82
--- /dev/null
+++ b/images/autoscaler/Dockerfile
@@ -0,0 +1,26 @@
+FROM golang:1.16.5 as builder
+
+WORKDIR /workspace
+# Copy the Go Modules manifests
+COPY go.mod go.mod
+COPY go.sum go.sum
+# cache deps before building and copying source so that we don't need to re-download as much
+# and so that source changes don't invalidate our downloaded layer
+RUN go mod download
+
+# Copy the go source
+COPY pkg pkg
+COPY cmd/autoscaler cmd/autoscaler
+WORKDIR /workspace/cmd/autoscaler
+
+# Build
+RUN CGO_ENABLED=0 GOOS=linux GOARCH=amd64 GO111MODULE=on go build -a -o /workspace/bin/autoscaler main.go
+
+# Use distroless as minimal base image to package the manager binary
+# Refer to https://github.com/GoogleContainerTools/distroless for more details
+FROM gcr.io/distroless/static:nonroot
+WORKDIR /
+COPY --from=builder /workspace/bin/autoscaler .
+USER 65532:65532
+
+ENTRYPOINT ["/autoscaler"]
diff --git a/images/proxy/Dockerfile b/images/proxy/Dockerfile
index c32c2f61b0..d7489b13ad 100644
--- a/images/proxy/Dockerfile
+++ b/images/proxy/Dockerfile
@@ -1,6 +1,5 @@
 ARG TARGETARCH, TARGETOS
 
-# Build the manager binary
 FROM golang:1.16.5 as builder
 
 WORKDIR /workspace
diff --git a/manager/install.sh b/manager/install.sh
index 21dba663aa..c7d6cd99bd 100755
--- a/manager/install.sh
+++ b/manager/install.sh
@@ -41,13 +41,13 @@ function cluster_up() {
 
   echo -n "￮ configuring networking (this will take a few minutes) "
   setup_istio
-  python render_template.py $CORTEX_CLUSTER_CONFIG_FILE manifests/apis.yaml.j2 > /workspace/apis.yaml
-  kubectl apply -f /workspace/apis.yaml >/dev/null
+  python render_template.py $CORTEX_CLUSTER_CONFIG_FILE manifests/apis.yaml.j2 | kubectl apply -f - >/dev/null
   echo "✓"
 
   echo -n "￮ configuring autoscaling "
-  python render_template.py $CORTEX_CLUSTER_CONFIG_FILE manifests/cluster-autoscaler.yaml.j2 > /workspace/cluster-autoscaler.yaml
-  kubectl apply -f /workspace/cluster-autoscaler.yaml >/dev/null
+  python render_template.py $CORTEX_CLUSTER_CONFIG_FILE manifests/autoscaler.yaml.j2 | kubectl apply -f - >/dev/null
+  python render_template.py $CORTEX_CLUSTER_CONFIG_FILE manifests/activator.yaml.j2 | kubectl apply -f - >/dev/null
+  python render_template.py $CORTEX_CLUSTER_CONFIG_FILE manifests/cluster-autoscaler.yaml.j2 | kubectl apply -f - >/dev/null
   echo "✓"
 
   echo -n "￮ configuring logging "
@@ -98,8 +98,7 @@ function cluster_configure() {
 
   # this is necessary since max_instances may have been updated
   echo -n "￮ configuring autoscaling "
-  python render_template.py $CORTEX_CLUSTER_CONFIG_FILE manifests/cluster-autoscaler.yaml.j2 > /workspace/cluster-autoscaler.yaml
-  kubectl apply -f /workspace/cluster-autoscaler.yaml >/dev/null
+  python render_template.py $CORTEX_CLUSTER_CONFIG_FILE manifests/cluster-autoscaler.yaml.j2 | kubectl apply -f - >/dev/null
   echo "✓"
 
   restart_operator
diff --git a/manager/manifests/activator.yaml.j2 b/manager/manifests/activator.yaml.j2
new file mode 100644
index 0000000000..6901c25b1d
--- /dev/null
+++ b/manager/manifests/activator.yaml.j2
@@ -0,0 +1,158 @@
+# 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.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: activator
+  namespace: default
+
+---
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: activator-role
+rules:
+- apiGroups:
+  - "networking.istio.io"
+  resources:
+  - virtualservices
+  verbs:
+  - list
+  - watch
+- apiGroups:
+  - "apps"
+  resources:
+  - deployments
+  verbs:
+  - list
+  - watch
+
+---
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: activator-rolebinding
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: activator-role
+subjects:
+- kind: ServiceAccount
+  name: activator
+  namespace: default
+
+---
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: activator
+spec:
+  selector:
+    matchLabels:
+      app: activator
+  template:
+    metadata:
+      name: activator
+      labels:
+        app: activator
+    spec:
+      serviceAccountName: activator
+      containers:
+        - name: activator
+          imagePullPolicy: Always
+          image: {{ config['image_activator'] }}
+          args:
+            - "--in-cluster"
+            - "--port=8000"
+            - "--autoscaler-url=http://autoscaler.default:8000"
+            - "--namespace=default"
+            - "--cluster-config=/configs/cluster/cluster.yaml"
+          ports:
+            - containerPort: 8000
+          livenessProbe:
+            httpGet:
+              port: 8000
+              path: /
+              httpHeaders:
+                - name: X-Cortex-Probe
+                  value: "true"
+          readinessProbe:
+            httpGet:
+              port: 8000
+              path: /
+              httpHeaders:
+                - name: X-Cortex-Probe
+                  value: "true"
+          resources:
+            requests:
+              cpu: 100m
+              memory: 100Mi
+            limits:
+              cpu: 250m
+              memory: 300Mi
+          envFrom:
+          - configMapRef:
+              name: env-vars
+          volumeMounts:
+          - mountPath: /configs/cluster/cluster.yaml
+            name: cluster-config
+            subPath: cluster.yaml
+      volumes:
+      - configMap:
+          defaultMode: 420
+          name: cluster-config
+        name: cluster-config
+---
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: activator
+spec:
+  type: ClusterIP
+  selector:
+    app: activator
+  ports:
+    - port: 8000
+
+---
+
+apiVersion: autoscaling/v2beta2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: activator-hpa
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: activator
+  minReplicas: 1
+  maxReplicas: 10
+  metrics:
+    - type: Resource
+      resource:
+        name: cpu
+        target:
+          type: Utilization
+          averageUtilization: 70
+    - type: Resource
+      resource:
+        name: memory
+        target:
+          type: Utilization
+          averageUtilization: 70
diff --git a/manager/manifests/autoscaler.yaml.j2 b/manager/manifests/autoscaler.yaml.j2
new file mode 100644
index 0000000000..df42a90965
--- /dev/null
+++ b/manager/manifests/autoscaler.yaml.j2
@@ -0,0 +1,126 @@
+# 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.
+
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: autoscaler
+  namespace: default
+
+---
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: autoscaler-role
+rules:
+- apiGroups:
+  - "networking.istio.io"
+  resources:
+  - virtualservices
+  verbs:
+  - get
+  - list
+  - watch
+  - update
+- apiGroups:
+  - "apps"
+  resources:
+  - deployments
+  verbs:
+  - get
+  - update
+  - watch
+
+---
+
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: autoscaler-rolebinding
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: autoscaler-role
+subjects:
+- kind: ServiceAccount
+  name: autoscaler
+  namespace: default
+
+---
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: autoscaler
+spec:
+  selector:
+    matchLabels:
+      app: autoscaler
+  template:
+    metadata:
+      name: autoscaler
+      labels:
+        app: autoscaler
+    spec:
+      serviceAccountName: autoscaler
+      containers:
+        - name: autoscaler
+          imagePullPolicy: Always
+          image: {{ config['image_autoscaler'] }}
+          args:
+            - "--in-cluster"
+            - "--port=8000"
+            - "--prometheus-url=http://prometheus.default:9090"
+            - "--namespace=default"
+            - "--cluster-config=/configs/cluster/cluster.yaml"
+          ports:
+            - containerPort: 8000
+          livenessProbe:
+            httpGet:
+              port: 8000
+              path: /healthz
+          readinessProbe:
+            httpGet:
+              port: 8000
+              path: /healthz
+          resources:
+            requests:
+              cpu: 100m
+              memory: 100Mi
+          envFrom:
+          - configMapRef:
+              name: env-vars
+          volumeMounts:
+          - mountPath: /configs/cluster/cluster.yaml
+            name: cluster-config
+            subPath: cluster.yaml
+      volumes:
+      - configMap:
+          defaultMode: 420
+          name: cluster-config
+        name: cluster-config
+
+---
+
+apiVersion: v1
+kind: Service
+metadata:
+  name: autoscaler
+spec:
+  type: ClusterIP
+  selector:
+    app: autoscaler
+  ports:
+    - port: 8000
diff --git a/pkg/activator/activator.go b/pkg/activator/activator.go
new file mode 100644
index 0000000000..f43410199e
--- /dev/null
+++ b/pkg/activator/activator.go
@@ -0,0 +1,269 @@
+/*
+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 activator
+
+import (
+	"context"
+	"sync"
+
+	"github.com/cortexlabs/cortex/pkg/autoscaler"
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"github.com/cortexlabs/cortex/pkg/workloads"
+	"go.uber.org/zap"
+	istionetworkingclient "istio.io/client-go/pkg/clientset/versioned/typed/networking/v1beta1"
+	kapps "k8s.io/api/apps/v1"
+	kmeta "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
+)
+
+type ctxValue string
+
+const APINameCtxKey ctxValue = "apiName"
+
+type Activator interface {
+	Try(ctx context.Context, fn func() error) error
+}
+
+type activator struct {
+	activatorsMux     sync.Mutex
+	trackersMux       sync.Mutex
+	autoscalerClient  autoscaler.Client
+	apiActivators     map[string]*apiActivator
+	readinessTrackers map[string]*readinessTracker
+	istioClient       istionetworkingclient.VirtualServiceInterface
+	logger            *zap.SugaredLogger
+}
+
+func New(
+	istioClient istionetworkingclient.VirtualServiceInterface,
+	deploymentInformer cache.SharedIndexInformer,
+	virtualServiceInformer cache.SharedIndexInformer,
+	autoscalerClient autoscaler.Client,
+	logger *zap.SugaredLogger,
+) Activator {
+	log := logger.With(zap.String("apiKind", userconfig.RealtimeAPIKind.String()))
+
+	act := &activator{
+		apiActivators:     make(map[string]*apiActivator),
+		readinessTrackers: make(map[string]*readinessTracker),
+		istioClient:       istioClient,
+		logger:            log,
+		autoscalerClient:  autoscalerClient,
+	}
+
+	virtualServiceInformer.AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc:    act.addAPI,
+			UpdateFunc: act.updateAPI,
+			DeleteFunc: act.removeAPI,
+		},
+	)
+
+	deploymentInformer.AddEventHandler(
+		cache.ResourceEventHandlerFuncs{
+			AddFunc: act.updateReadinessTracker,
+			UpdateFunc: func(_, newObj interface{}) {
+				act.updateReadinessTracker(newObj)
+			},
+			DeleteFunc: act.removeReadinessTracker,
+		},
+	)
+
+	return act
+}
+
+func (a *activator) Try(ctx context.Context, fn func() error) error {
+	apiNameValue := ctx.Value(APINameCtxKey)
+	apiName, ok := apiNameValue.(string)
+	if !ok || apiName == "" {
+		return errors.ErrorUnexpected("failed to get the api name from context")
+	}
+
+	act, err := a.getOrCreateAPIActivator(ctx, apiName)
+	if err != nil {
+		return err
+	}
+
+	tracker := a.getOrCreateReadinessTracker(apiName)
+
+	if act.inFlight() == 0 {
+		go a.awakenAPI(apiName)
+	}
+
+	return act.try(ctx, fn, tracker)
+}
+
+func (a *activator) getOrCreateAPIActivator(ctx context.Context, apiName string) (*apiActivator, error) {
+	a.activatorsMux.Lock()
+	defer a.activatorsMux.Unlock()
+
+	act, ok := a.apiActivators[apiName]
+	if ok {
+		return act, nil
+	}
+
+	vs, err := a.istioClient.Get(ctx, workloads.K8sName(apiName), kmeta.GetOptions{})
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	maxQueueLength, maxConcurrency, err := concurrencyFromAnnotations(vs.Annotations)
+	if err != nil {
+		return nil, err
+	}
+
+	apiAct := newAPIActivator(maxQueueLength, maxConcurrency)
+
+	a.apiActivators[apiName] = apiAct
+
+	return apiAct, nil
+}
+
+func (a *activator) getOrCreateReadinessTracker(apiName string) *readinessTracker {
+	a.trackersMux.Lock()
+	defer a.trackersMux.Unlock()
+	tracker, ok := a.readinessTrackers[apiName]
+	if ok {
+		return tracker
+	}
+
+	tracker = newReadinessTracker()
+	a.readinessTrackers[apiName] = tracker
+
+	return tracker
+}
+
+func (a *activator) addAPI(obj interface{}) {
+	apiMetadata, err := getAPIMeta(obj)
+	if err != nil {
+		a.logger.Errorw("error during virtual service informer add callback", zap.Error(err))
+		return
+	}
+
+	if apiMetadata.apiKind != userconfig.RealtimeAPIKind {
+		return
+	}
+
+	apiName := apiMetadata.apiName
+
+	a.activatorsMux.Lock()
+	if a.apiActivators[apiName] == nil {
+		a.logger.Debugw("adding new api activator", zap.String("apiName", apiName))
+		a.apiActivators[apiName] = newAPIActivator(apiMetadata.maxQueueLength, apiMetadata.maxConcurrency)
+	}
+	a.activatorsMux.Unlock()
+}
+
+func (a *activator) updateAPI(oldObj interface{}, newObj interface{}) {
+	apiMetadata, err := getAPIMeta(newObj)
+	if err != nil {
+		a.logger.Errorw("error during virtual service informer update callback", zap.Error(err))
+		return
+	}
+
+	if apiMetadata.apiKind != userconfig.RealtimeAPIKind {
+		return
+	}
+
+	apiName := apiMetadata.apiName
+
+	oldAPIMetatada, err := getAPIMeta(oldObj)
+	if err != nil {
+		a.logger.Errorw("error during virtual service informer update callback", zap.Error(err))
+		return
+	}
+
+	if oldAPIMetatada.maxConcurrency != apiMetadata.maxConcurrency || oldAPIMetatada.maxQueueLength != apiMetadata.maxQueueLength {
+		a.logger.Debugw("updating api activator", zap.String("apiName", apiName))
+
+		a.activatorsMux.Lock()
+		a.apiActivators[apiName].updateQueueParams(apiMetadata.maxQueueLength, apiMetadata.maxConcurrency)
+		a.activatorsMux.Unlock()
+	}
+}
+
+func (a *activator) removeAPI(obj interface{}) {
+	apiMetadata, err := getAPIMeta(obj)
+	if err != nil {
+		a.logger.Errorw("error during virtual service informer delete callback", zap.Error(err))
+		return
+	}
+
+	if apiMetadata.apiKind != userconfig.RealtimeAPIKind {
+		return
+	}
+
+	a.logger.Debugw("deleting api activator", zap.String("apiName", apiMetadata.apiName))
+
+	a.activatorsMux.Lock()
+	delete(a.apiActivators, apiMetadata.apiName)
+	a.activatorsMux.Unlock()
+}
+
+func (a *activator) awakenAPI(apiName string) {
+	err := a.autoscalerClient.Awaken(
+		userconfig.Resource{
+			Name: apiName,
+			Kind: userconfig.RealtimeAPIKind, // only realtime apis are supported as of now, so we can assume the kind
+		},
+	)
+	if err != nil {
+		a.logger.Errorw("failed to awake api", zap.Error(err), zap.String("apiName", apiName))
+	}
+}
+
+func (a *activator) updateReadinessTracker(obj interface{}) {
+	deployment, ok := obj.(*kapps.Deployment)
+	if !ok {
+		return
+	}
+
+	api, err := getAPIMeta(obj)
+	if err != nil {
+		a.logger.Errorw("error during deployment informer callback", zap.Error(err))
+		return
+	}
+
+	if api.apiKind != userconfig.RealtimeAPIKind {
+		return
+	}
+
+	tracker := a.getOrCreateReadinessTracker(api.apiName)
+	tracker.Update(deployment)
+
+	a.logger.Debugw("updated readiness tracker",
+		zap.Bool("ready", deployment.Status.ReadyReplicas > 0),
+		zap.String("apiName", api.apiName),
+	)
+}
+
+func (a *activator) removeReadinessTracker(obj interface{}) {
+	api, err := getAPIMeta(obj)
+	if err != nil {
+		a.logger.Errorw("error during deployment informer callback", zap.Error(err))
+		return
+	}
+
+	if api.apiKind != userconfig.RealtimeAPIKind {
+		return
+	}
+
+	a.trackersMux.Lock()
+	defer a.trackersMux.Unlock()
+	delete(a.readinessTrackers, api.apiName)
+}
diff --git a/pkg/activator/api_activator.go b/pkg/activator/api_activator.go
new file mode 100644
index 0000000000..7e01d304bc
--- /dev/null
+++ b/pkg/activator/api_activator.go
@@ -0,0 +1,117 @@
+/*
+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 activator
+
+import (
+	"context"
+	"sync"
+
+	"github.com/cortexlabs/cortex/pkg/consts"
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/proxy"
+	kapps "k8s.io/api/apps/v1"
+)
+
+type apiActivator struct {
+	breaker *proxy.Breaker
+}
+
+func newAPIActivator(maxQueueLength, maxConcurrency int) *apiActivator {
+	breaker := proxy.NewBreaker(proxy.BreakerParams{
+		QueueDepth:      maxQueueLength,
+		MaxConcurrency:  maxConcurrency,
+		InitialCapacity: maxConcurrency,
+	})
+
+	return &apiActivator{breaker: breaker}
+}
+
+// try waits for the readinessTracker to be ready and then attempts to execute the passed callback.
+// If the readinessTracker does not reach a ready state, it will timeout.
+func (a *apiActivator) try(ctx context.Context, fn func() error, tracker *readinessTracker) error {
+	var execErr error
+
+	if err := a.breaker.Maybe(ctx, func() {
+		ctx, cancel := context.WithTimeout(ctx, consts.WaitForReadyReplicasTimeout)
+		defer cancel()
+
+		if !tracker.IsReady() {
+		loop:
+			for {
+				select {
+				case <-ctx.Done():
+					execErr = errors.Wrap(ctx.Err(), "no ready replicas available")
+					return
+				case <-tracker.Wait():
+					break loop
+				}
+			}
+		}
+		execErr = fn()
+	}); err != nil {
+		return err
+	}
+
+	return execErr
+}
+
+// updateQueueParams updates the breaker queue parameters (not thread safe)
+func (a *apiActivator) updateQueueParams(maxQueueLength, maxConcurrency int) {
+	a.breaker.UpdateConcurrency(maxConcurrency)
+	a.breaker.UpdateQueueLength(maxQueueLength)
+}
+
+// inFlight returns the amount of in-flight requests of the breaker
+func (a *apiActivator) inFlight() int64 {
+	return a.breaker.InFlight()
+}
+
+type readinessTracker struct {
+	mux   sync.RWMutex
+	c     chan bool
+	ready bool
+}
+
+func newReadinessTracker() *readinessTracker {
+	return &readinessTracker{
+		c: make(chan bool),
+	}
+}
+
+func (t *readinessTracker) Update(deployment *kapps.Deployment) {
+	t.mux.Lock()
+	defer t.mux.Unlock()
+
+	// if changing from unready to ready state, wake up waiting goroutines
+	if !t.ready && deployment.Status.ReadyReplicas > 0 {
+		close(t.c)            // wake up all of the goroutines waiting on this channel
+		t.c = make(chan bool) // make a new channel ready to be used
+	}
+	t.ready = deployment.Status.ReadyReplicas > 0
+}
+
+func (t *readinessTracker) IsReady() bool {
+	t.mux.RLock()
+	defer t.mux.RUnlock()
+	return t.ready
+}
+
+func (t *readinessTracker) Wait() chan bool {
+	t.mux.RLock()
+	defer t.mux.RUnlock()
+	return t.c
+}
diff --git a/pkg/activator/handler.go b/pkg/activator/handler.go
new file mode 100644
index 0000000000..2811c5ea6f
--- /dev/null
+++ b/pkg/activator/handler.go
@@ -0,0 +1,96 @@
+/*
+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 activator
+
+import (
+	"context"
+	stderrors "errors"
+	"fmt"
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+
+	"github.com/cortexlabs/cortex/pkg/consts"
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/proxy"
+	"go.uber.org/zap"
+)
+
+type Handler struct {
+	activator Activator
+	logger    *zap.SugaredLogger
+}
+
+func NewHandler(act Activator, logger *zap.SugaredLogger) *Handler {
+	return &Handler{
+		activator: act,
+		logger:    logger,
+	}
+}
+
+func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if hasCortexProbeHeader(r) {
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+
+	apiName := r.Header.Get(consts.CortexAPINameHeader)
+	if apiName == "" {
+		http.Error(w, fmt.Sprintf("missing %s", consts.CortexAPINameHeader), http.StatusInternalServerError)
+		return
+	}
+
+	ctx := r.Context()
+	ctx = context.WithValue(ctx, APINameCtxKey, apiName)
+
+	if err := h.activator.Try(ctx, func() error {
+		return h.proxyRequest(w, r)
+	}); err != nil {
+		h.logger.Errorw("activator try error", zap.Error(err))
+
+		if stderrors.Is(err, context.DeadlineExceeded) || stderrors.Is(err, proxy.ErrRequestQueueFull) {
+			http.Error(w, err.Error(), http.StatusServiceUnavailable)
+		} else {
+			w.WriteHeader(http.StatusInternalServerError)
+		}
+	}
+}
+
+func (h *Handler) proxyRequest(w http.ResponseWriter, r *http.Request) error {
+	target := r.Header.Get(consts.CortexTargetServiceHeader)
+	if target == "" {
+		return errors.ErrorUnexpected("missing header", consts.CortexTargetServiceHeader)
+	}
+
+	targetURL, err := url.Parse(target)
+	if err != nil {
+		return errors.WithStack(err)
+	}
+
+	// delete activator specific headers
+	r.Header.Del(consts.CortexAPINameHeader)
+	r.Header.Del(consts.CortexTargetServiceHeader)
+
+	reverseProxy := httputil.NewSingleHostReverseProxy(targetURL)
+	reverseProxy.ServeHTTP(w, r)
+
+	return nil
+}
+
+func hasCortexProbeHeader(r *http.Request) bool {
+	return r.Header.Get(consts.CortexProbeHeader) != ""
+}
diff --git a/pkg/activator/helpers.go b/pkg/activator/helpers.go
new file mode 100644
index 0000000000..f32c7e54f2
--- /dev/null
+++ b/pkg/activator/helpers.go
@@ -0,0 +1,81 @@
+/*
+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 activator
+
+import (
+	"strconv"
+
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"k8s.io/apimachinery/pkg/api/meta"
+)
+
+type apiMeta struct {
+	apiName        string
+	apiKind        userconfig.Kind
+	labels         map[string]string
+	annotations    map[string]string
+	maxConcurrency int
+	maxQueueLength int
+}
+
+func getAPIMeta(obj interface{}) (apiMeta, error) {
+	resource, err := meta.Accessor(obj)
+	if err != nil {
+		return apiMeta{}, err
+	}
+
+	labels := resource.GetLabels()
+	apiKind, ok := labels["apiKind"]
+	if !ok {
+		return apiMeta{}, err
+	}
+
+	apiName, ok := labels["apiName"]
+	if !ok {
+		return apiMeta{}, errors.ErrorUnexpected("got a virtual service without apiName label")
+	}
+
+	annotations := resource.GetAnnotations()
+	maxQueueLength, maxConcurrency, err := concurrencyFromAnnotations(annotations)
+	if err != nil {
+		return apiMeta{}, err
+	}
+
+	return apiMeta{
+		apiName:        apiName,
+		apiKind:        userconfig.KindFromString(apiKind),
+		labels:         labels,
+		annotations:    annotations,
+		maxConcurrency: maxConcurrency,
+		maxQueueLength: maxQueueLength,
+	}, nil
+}
+
+func concurrencyFromAnnotations(annotations map[string]string) (int, int, error) {
+	maxQueueLength, err := strconv.Atoi(annotations[userconfig.MaxQueueLengthAnnotationKey])
+	if err != nil {
+		return 0, 0, errors.ErrorUnexpected("failed to parse annotation", userconfig.MaxQueueLengthAnnotationKey)
+	}
+
+	maxConcurrency, err := strconv.Atoi(annotations[userconfig.MaxConcurrencyAnnotationKey])
+	if err != nil {
+		return 0, 0, errors.ErrorUnexpected("failed to parse annotation", userconfig.MaxConcurrencyAnnotationKey)
+	}
+
+	return maxQueueLength, maxConcurrency, err
+}
diff --git a/pkg/autoscaler/async_scaler.go b/pkg/autoscaler/async_scaler.go
new file mode 100644
index 0000000000..47973b3490
--- /dev/null
+++ b/pkg/autoscaler/async_scaler.go
@@ -0,0 +1,130 @@
+/*
+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 (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/lib/k8s"
+	"github.com/cortexlabs/cortex/pkg/lib/pointer"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"github.com/cortexlabs/cortex/pkg/workloads"
+	promv1 "github.com/prometheus/client_golang/api/prometheus/v1"
+	"github.com/prometheus/common/model"
+)
+
+type asyncScaler struct {
+	k8s        *k8s.Client
+	prometheus promv1.API
+}
+
+func NewAsyncScaler(k8sClient *k8s.Client, promClient promv1.API) Scaler {
+	return &asyncScaler{
+		k8s:        k8sClient,
+		prometheus: promClient,
+	}
+}
+
+func (s *asyncScaler) Scale(apiName string, request int32) error {
+	deployment, err := s.k8s.GetDeployment(workloads.K8sName(apiName))
+	if err != nil {
+		return err
+	}
+
+	if deployment == nil {
+		return errors.ErrorUnexpected("unable to find k8s deployment", apiName)
+	}
+
+	current := deployment.Status.Replicas
+	if current == request {
+		return nil
+	}
+
+	deployment.Spec.Replicas = pointer.Int32(request)
+
+	if _, err = s.k8s.UpdateDeployment(deployment); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *asyncScaler) GetInFlightRequests(apiName string, window time.Duration) (*float64, error) {
+	windowSeconds := int64(window.Seconds())
+
+	// PromQL query:
+	// 	sum(sum_over_time(cortex_async_queue_length{api_name="<apiName>"}[60s])) /
+	//	sum(count_over_time(cortex_async_queue_length{api_name="<apiName>"}[60s]))
+	query := fmt.Sprintf(
+		"sum(sum_over_time(cortex_async_queue_length{api_name=\"%s\"}[%ds])) / "+
+			"max(count_over_time(cortex_async_queue_length{api_name=\"%s\"}[%ds]))",
+		apiName, windowSeconds,
+		apiName, windowSeconds,
+	)
+
+	ctx, cancel := context.WithTimeout(context.Background(), _prometheusQueryTimeoutSeconds*time.Second)
+	defer cancel()
+
+	valuesQuery, _, err := s.prometheus.Query(ctx, query, time.Now())
+	if err != nil {
+		return nil, err
+	}
+
+	values, ok := valuesQuery.(model.Vector)
+	if !ok {
+		return nil, errors.ErrorUnexpected("failed to convert prometheus metric to vector")
+	}
+
+	if values.Len() != 0 {
+		return pointer.Float64(float64(values[0].Value)), nil
+	}
+	return nil, nil
+}
+
+func (s *asyncScaler) GetAutoscalingSpec(apiName string) (*userconfig.Autoscaling, error) {
+	deployment, err := s.k8s.GetDeployment(workloads.K8sName(apiName))
+	if err != nil {
+		return nil, err
+	}
+
+	if deployment == nil {
+		return nil, errors.ErrorUnexpected("unable to find k8s deployment", apiName)
+	}
+
+	autoscalingSpec, err := userconfig.AutoscalingFromAnnotations(deployment)
+	if err != nil {
+		return nil, err
+	}
+
+	return autoscalingSpec, nil
+}
+
+func (s *asyncScaler) CurrentReplicas(apiName string) (int32, error) {
+	deployment, err := s.k8s.GetDeployment(workloads.K8sName(apiName))
+	if err != nil {
+		return 0, err
+	}
+
+	if deployment == nil {
+		return 0, errors.ErrorUnexpected("unable to find k8s deployment", apiName)
+	}
+
+	return deployment.Status.Replicas, nil
+}
diff --git a/pkg/autoscaler/autoscaler.go b/pkg/autoscaler/autoscaler.go
new file mode 100644
index 0000000000..93e283380f
--- /dev/null
+++ b/pkg/autoscaler/autoscaler.go
@@ -0,0 +1,294 @@
+/*
+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 (
+	"fmt"
+	"math"
+	"sync"
+	"time"
+
+	"github.com/cortexlabs/cortex/pkg/lib/cron"
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	libmath "github.com/cortexlabs/cortex/pkg/lib/math"
+	"github.com/cortexlabs/cortex/pkg/lib/telemetry"
+	libtime "github.com/cortexlabs/cortex/pkg/lib/time"
+	"github.com/cortexlabs/cortex/pkg/types/spec"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"go.uber.org/zap"
+)
+
+const (
+	_prometheusQueryTimeoutSeconds = 10
+)
+
+type Autoscaler struct {
+	sync.Mutex
+	logger              *zap.SugaredLogger
+	crons               map[string]cron.Cron
+	scalers             map[userconfig.Kind]Scaler
+	lastAwakenTimestamp map[string]time.Time
+}
+
+func New(logger *zap.SugaredLogger) *Autoscaler {
+	return &Autoscaler{
+		logger:              logger,
+		crons:               make(map[string]cron.Cron),
+		scalers:             make(map[userconfig.Kind]Scaler),
+		lastAwakenTimestamp: make(map[string]time.Time),
+	}
+}
+
+func (a *Autoscaler) AddScaler(scaler Scaler, kind userconfig.Kind) {
+	a.scalers[kind] = scaler
+}
+
+func (a *Autoscaler) Awaken(api userconfig.Resource) error {
+	a.Lock()
+	defer a.Unlock()
+
+	scaler, ok := a.scalers[api.Kind]
+	if !ok {
+		return errors.ErrorUnexpected(
+			fmt.Sprintf("autoscaler does not have a scaler for the %s kind", api.Kind),
+		)
+	}
+
+	log := a.logger.With(
+		zap.String("apiName", api.Name),
+		zap.String("apiKind", api.Kind.String()),
+	)
+
+	currentReplicas, err := scaler.CurrentReplicas(api.Name)
+	if err != nil {
+		return errors.Wrap(err, "failed to get current replicas")
+	}
+
+	if currentReplicas > 0 {
+		return nil
+	}
+
+	log.Infof("autoscaling awake event")
+	if err := scaler.Scale(api.Name, 1); err != nil {
+		return errors.Wrap(err, "failed to scale api to one")
+	}
+
+	a.lastAwakenTimestamp[api.Name] = time.Now()
+
+	return nil
+}
+
+func (a *Autoscaler) AddAPI(api userconfig.Resource) error {
+	if _, ok := a.crons[api.Name]; ok {
+		return nil
+	}
+
+	autoscaleFn, err := a.autoscaleFn(api)
+	if err != nil {
+		return err
+	}
+
+	errorHandler := func(err error) {
+		log := a.logger.With(
+			zap.String("apiName", api.Name),
+			zap.String("apiKind", api.Kind.String()),
+		)
+
+		log.Error(err)
+		telemetry.Error(err)
+	}
+
+	a.crons[api.Name] = cron.Run(autoscaleFn, errorHandler, spec.AutoscalingTickInterval)
+
+	// make sure there is no awaken call registered to an older API with the same name
+	a.Lock()
+	delete(a.lastAwakenTimestamp, api.Name)
+	a.Unlock()
+
+	return nil
+}
+
+func (a *Autoscaler) RemoveAPI(api userconfig.Resource) {
+	log := a.logger.With(
+		zap.String("apiName", api.Name),
+		zap.String("apiKind", api.Kind.String()),
+	)
+
+	if autoscalerCron, ok := a.crons[api.Name]; ok {
+		autoscalerCron.Cancel()
+		delete(a.crons, api.Name)
+	}
+
+	a.Lock()
+	delete(a.lastAwakenTimestamp, api.Name)
+	a.Unlock()
+
+	log.Info("autoscaler stop")
+}
+
+func (a *Autoscaler) Stop() {
+	for apiName, apiCron := range a.crons {
+		apiCron.Cancel()
+		delete(a.crons, apiName)
+	}
+}
+
+func (a *Autoscaler) autoscaleFn(api userconfig.Resource) (func() error, error) {
+	log := a.logger.With(
+		zap.String("apiName", api.Name),
+		zap.String("apiKind", api.Kind.String()),
+	)
+
+	scaler, ok := a.scalers[api.Kind]
+	if !ok {
+		return nil, errors.ErrorUnexpected(
+			fmt.Sprintf("autoscaler does not have a scaler for the %s kind", api.Kind),
+		)
+	}
+
+	autoscalingSpec, err := scaler.GetAutoscalingSpec(api.Name)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get autoscaling spec")
+	}
+
+	log.Info("autoscaler init")
+
+	var startTime time.Time
+	recs := make(recommendations)
+
+	return func() error {
+		currentReplicas, err := scaler.CurrentReplicas(api.Name)
+		if err != nil {
+			return errors.Wrap(err, "failed to get current replicas")
+		}
+
+		if startTime.IsZero() {
+			startTime = time.Now()
+		}
+
+		avgInFlight, err := scaler.GetInFlightRequests(api.Name, autoscalingSpec.Window)
+		if err != nil {
+			return errors.Wrap(err, "failed to get in-flight requests")
+		}
+		if avgInFlight == nil {
+			log.Debug("autoscaler tick: metrics not available yet")
+			return nil
+		}
+
+		rawRecommendation := *avgInFlight / *autoscalingSpec.TargetInFlight
+		recommendation := int32(math.Ceil(rawRecommendation))
+
+		if rawRecommendation < float64(currentReplicas) && rawRecommendation > float64(currentReplicas)*(1-autoscalingSpec.DownscaleTolerance) {
+			recommendation = currentReplicas
+		}
+
+		if rawRecommendation > float64(currentReplicas) && rawRecommendation < float64(currentReplicas)*(1+autoscalingSpec.UpscaleTolerance) {
+			recommendation = currentReplicas
+		}
+
+		// always allow subtraction of 1
+		downscaleFactorFloor := libmath.MinInt32(currentReplicas-1, int32(math.Ceil(float64(currentReplicas)*autoscalingSpec.MaxDownscaleFactor)))
+		if recommendation < downscaleFactorFloor {
+			recommendation = downscaleFactorFloor
+		}
+
+		// always allow addition of 1
+		upscaleFactorCeil := libmath.MaxInt32(currentReplicas+1, int32(math.Ceil(float64(currentReplicas)*autoscalingSpec.MaxUpscaleFactor)))
+		if recommendation > upscaleFactorCeil {
+			recommendation = upscaleFactorCeil
+		}
+
+		if recommendation < autoscalingSpec.MinReplicas {
+			recommendation = autoscalingSpec.MinReplicas
+		}
+
+		if recommendation > autoscalingSpec.MaxReplicas {
+			recommendation = autoscalingSpec.MaxReplicas
+		}
+
+		// Rule of thumb: any modifications that don't consider historical recommendations should be performed before
+		// recording the recommendation, any modifications that use historical recommendations should be performed after
+		recs.add(recommendation)
+
+		// This is just for garbage collection
+		recs.deleteOlderThan(libtime.MaxDuration(autoscalingSpec.DownscaleStabilizationPeriod, autoscalingSpec.UpscaleStabilizationPeriod))
+
+		request := recommendation
+		var downscaleStabilizationFloor *int32
+		var upscaleStabilizationCeil *int32
+
+		if request < currentReplicas {
+			downscaleStabilizationFloor = recs.maxSince(autoscalingSpec.DownscaleStabilizationPeriod)
+			if time.Since(startTime) < autoscalingSpec.DownscaleStabilizationPeriod {
+				request = currentReplicas
+			} else if downscaleStabilizationFloor != nil && request < *downscaleStabilizationFloor {
+				request = *downscaleStabilizationFloor
+			}
+
+			// awaken state: was scaled from zero
+			// This needs to be protected by a Mutex because an Awaken call will also modify it
+			a.Lock()
+			lastAwakenTimestamp := a.lastAwakenTimestamp[api.Name]
+
+			// Make sure we don't scale below zero if API was recently awaken
+			if time.Since(lastAwakenTimestamp) < autoscalingSpec.DownscaleStabilizationPeriod {
+				request = libmath.MaxInt32(request, 1)
+			}
+			a.Unlock()
+		}
+		if request > currentReplicas {
+			upscaleStabilizationCeil = recs.minSince(autoscalingSpec.UpscaleStabilizationPeriod)
+			if time.Since(startTime) < autoscalingSpec.UpscaleStabilizationPeriod {
+				request = currentReplicas
+			} else if upscaleStabilizationCeil != nil && request > *upscaleStabilizationCeil {
+				request = *upscaleStabilizationCeil
+			}
+		}
+
+		log.Debugw("autoscaler tick",
+			"autoscaling", map[string]interface{}{
+				"avg_in_flight":                  *avgInFlight,
+				"target_in_flight":               *autoscalingSpec.TargetInFlight,
+				"raw_recommendation":             rawRecommendation,
+				"current_replicas":               currentReplicas,
+				"downscale_tolerance":            autoscalingSpec.DownscaleTolerance,
+				"upscale_tolerance":              autoscalingSpec.UpscaleTolerance,
+				"max_downscale_factor":           autoscalingSpec.MaxDownscaleFactor,
+				"downscale_factor_floor":         downscaleFactorFloor,
+				"max_upscale_factor":             autoscalingSpec.MaxUpscaleFactor,
+				"upscale_factor_ceil":            upscaleFactorCeil,
+				"min_replicas":                   autoscalingSpec.MinReplicas,
+				"max_replicas":                   autoscalingSpec.MaxReplicas,
+				"recommendation":                 recommendation,
+				"downscale_stabilization_period": autoscalingSpec.DownscaleStabilizationPeriod.Seconds(),
+				"downscale_stabilization_floor":  downscaleStabilizationFloor,
+				"upscale_stabilization_period":   autoscalingSpec.UpscaleStabilizationPeriod.Seconds(),
+				"upscale_stabilization_ceil":     upscaleStabilizationCeil,
+				"request":                        request,
+			},
+		)
+
+		if currentReplicas != request {
+			log.Infof("autoscaling event: %d -> %d", currentReplicas, request)
+			if err = scaler.Scale(api.Name, request); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	}, nil
+}
diff --git a/pkg/autoscaler/client.go b/pkg/autoscaler/client.go
new file mode 100644
index 0000000000..3a61f5e57f
--- /dev/null
+++ b/pkg/autoscaler/client.go
@@ -0,0 +1,71 @@
+/*
+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 (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+
+	"github.com/cortexlabs/cortex/pkg/lib/json"
+	"github.com/cortexlabs/cortex/pkg/lib/urls"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+)
+
+type Client interface {
+	Awaken(api userconfig.Resource) error
+}
+
+type client struct {
+	httpClient *http.Client
+	endpoint   string
+}
+
+func NewClient(endpoint string) Client {
+	return &client{
+		httpClient: &http.Client{},
+		endpoint:   endpoint,
+	}
+}
+
+func (c *client) Awaken(api userconfig.Resource) error {
+	payload, err := json.Marshal(api)
+	if err != nil {
+		return err
+	}
+
+	response, err := c.httpClient.Post(
+		urls.Join(c.endpoint, "/awaken"), "application/json", bytes.NewBuffer(payload),
+	)
+	if err != nil {
+		return err
+	}
+	defer func() { _ = response.Body.Close() }()
+
+	if response.StatusCode != http.StatusOK {
+		bodyBytes, _ := ioutil.ReadAll(response.Body)
+		errMsg := fmt.Sprintf("failed to awake api (status code %d)", response.StatusCode)
+		if bodyBytes != nil {
+			errMsg = errMsg + fmt.Sprintf(": %s", string(bodyBytes))
+		}
+
+		return fmt.Errorf(errMsg)
+	}
+
+	return nil
+}
diff --git a/pkg/autoscaler/handler.go b/pkg/autoscaler/handler.go
new file mode 100644
index 0000000000..6ff83caf2b
--- /dev/null
+++ b/pkg/autoscaler/handler.go
@@ -0,0 +1,49 @@
+/*
+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 (
+	"encoding/json"
+	"net/http"
+
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+)
+
+type Handler struct {
+	autoscaler *Autoscaler
+}
+
+func NewHandler(autoscaler *Autoscaler) *Handler {
+	return &Handler{autoscaler: autoscaler}
+}
+
+func (h *Handler) Awaken(w http.ResponseWriter, r *http.Request) {
+	var api userconfig.Resource
+	if err := json.NewDecoder(r.Body).Decode(&api); err != nil {
+		http.Error(w, "failed to json decode request body", http.StatusBadRequest)
+		return
+	}
+	defer func() { _ = r.Body.Close() }()
+
+	if err := h.autoscaler.Awaken(api); err != nil {
+		http.Error(w, errors.Wrap(err, "failed to awaken api").Error(), http.StatusInternalServerError)
+		return
+	}
+
+	w.WriteHeader(http.StatusOK)
+}
diff --git a/pkg/autoscaler/realtime_scaler.go b/pkg/autoscaler/realtime_scaler.go
new file mode 100644
index 0000000000..6e28fb95e2
--- /dev/null
+++ b/pkg/autoscaler/realtime_scaler.go
@@ -0,0 +1,261 @@
+/*
+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 (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/cortexlabs/cortex/pkg/consts"
+	"github.com/cortexlabs/cortex/pkg/lib/errors"
+	"github.com/cortexlabs/cortex/pkg/lib/k8s"
+	"github.com/cortexlabs/cortex/pkg/lib/pointer"
+	"github.com/cortexlabs/cortex/pkg/lib/telemetry"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"github.com/cortexlabs/cortex/pkg/workloads"
+	promv1 "github.com/prometheus/client_golang/api/prometheus/v1"
+	"github.com/prometheus/common/model"
+	"go.uber.org/zap"
+	kapps "k8s.io/api/apps/v1"
+	kmeta "k8s.io/apimachinery/pkg/apis/meta/v1"
+	ctrlclient "sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type realtimeScaler struct {
+	k8s        *k8s.Client
+	prometheus promv1.API
+	logger     *zap.SugaredLogger
+}
+
+func NewRealtimeScaler(k8sClient *k8s.Client, promClient promv1.API, logger *zap.SugaredLogger) Scaler {
+	return &realtimeScaler{
+		k8s:        k8sClient,
+		prometheus: promClient,
+		logger:     logger,
+	}
+}
+
+func (s *realtimeScaler) Scale(apiName string, request int32) error {
+	ctx := context.Background()
+
+	// we use the controller-runtime client to make use of the cache mechanism
+	var deployment kapps.Deployment
+	err := s.k8s.Get(ctx, ctrlclient.ObjectKey{
+		Namespace: s.k8s.Namespace,
+		Name:      workloads.K8sName(apiName),
+	}, &deployment)
+	if err != nil {
+		return errors.Wrap(err, "failed to get deployment")
+	}
+
+	current := deployment.Status.Replicas
+
+	if current == request {
+		return nil
+	}
+
+	if request == 0 {
+		if err = s.routeToActivator(&deployment); err != nil {
+			return errors.Wrap(err, "failed to re-route traffic to activator")
+		}
+	}
+
+	deployment.Spec.Replicas = pointer.Int32(request)
+
+	if err = s.k8s.Update(ctx, &deployment); err != nil {
+		return errors.Wrap(err, "failed to update deployment")
+	}
+
+	if current == 0 && request > 0 {
+		go func() {
+			if err := s.routeToService(&deployment); err != nil {
+				s.logger.Errorw("failed to re-route traffic to API",
+					zap.Error(err), zap.String("apiName", apiName),
+				)
+				telemetry.Error(err)
+			}
+		}()
+
+	}
+
+	return nil
+}
+
+func (s *realtimeScaler) GetInFlightRequests(apiName string, window time.Duration) (*float64, error) {
+	windowSeconds := int64(window.Seconds())
+
+	// PromQL query:
+	// 	sum(sum_over_time(cortex_in_flight_requests{api_name="<apiName>"}[60s])) /
+	//	sum(count_over_time(cortex_in_flight_requests{api_name="<apiName>"}[60s]))
+	query := fmt.Sprintf(
+		"sum(sum_over_time(cortex_in_flight_requests{api_name=\"%s\"}[%ds])) / "+
+			"max(count_over_time(cortex_in_flight_requests{api_name=\"%s\"}[%ds]))",
+		apiName, windowSeconds,
+		apiName, windowSeconds,
+	)
+
+	ctx, cancel := context.WithTimeout(context.Background(), _prometheusQueryTimeoutSeconds*time.Second)
+	defer cancel()
+
+	valuesQuery, _, err := s.prometheus.Query(ctx, query, time.Now())
+	if err != nil {
+		return nil, err
+	}
+
+	values, ok := valuesQuery.(model.Vector)
+	if !ok {
+		return nil, errors.ErrorUnexpected("failed to convert prometheus metric to vector")
+	}
+
+	// no values available
+	if values.Len() == 0 {
+		return nil, nil
+	}
+
+	avgInflightRequests := float64(values[0].Value)
+
+	return &avgInflightRequests, nil
+}
+
+func (s *realtimeScaler) GetAutoscalingSpec(apiName string) (*userconfig.Autoscaling, error) {
+	deployment, err := s.k8s.GetDeployment(workloads.K8sName(apiName))
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to get deployment")
+	}
+
+	if deployment == nil {
+		return nil, errors.ErrorUnexpected("unable to find k8s deployment", apiName)
+	}
+
+	autoscalingSpec, err := userconfig.AutoscalingFromAnnotations(deployment)
+	if err != nil {
+		return nil, err
+	}
+
+	return autoscalingSpec, nil
+}
+
+func (s *realtimeScaler) CurrentReplicas(apiName string) (int32, error) {
+	ctx := context.Background()
+
+	// we use the controller-runtime client to make use of the cache mechanism
+	var deployment kapps.Deployment
+	err := s.k8s.Get(ctx, ctrlclient.ObjectKey{
+		Namespace: s.k8s.Namespace,
+		Name:      workloads.K8sName(apiName),
+	}, &deployment)
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to get deployment")
+	}
+
+	return deployment.Status.Replicas, nil
+}
+
+func (s *realtimeScaler) routeToService(deployment *kapps.Deployment) error {
+	ctx := context.Background()
+	vs, err := s.k8s.GetVirtualService(deployment.Name)
+	if err != nil {
+		return errors.Wrap(err, "failed to get virtual service")
+	}
+
+	if len(vs.Spec.Http) < 1 {
+		return errors.ErrorUnexpected("virtual service does not have any http entries")
+	}
+
+	if err = s.waitForReadyReplicas(ctx, deployment); err != nil {
+		return errors.Wrap(err, "no ready replicas available")
+	}
+
+	for i := range vs.Spec.Http {
+		if len(vs.Spec.Http[i].Route) != 2 {
+			return errors.ErrorUnexpected("virtual service does not have the required number of 2 http routes")
+		}
+
+		vs.Spec.Http[i].Route[0].Weight = 100 // service traffic
+		vs.Spec.Http[i].Route[1].Weight = 0   // activator traffic
+	}
+
+	vsClient := s.k8s.IstioClientSet().NetworkingV1beta1().VirtualServices(s.k8s.Namespace)
+	if _, err = vsClient.Update(ctx, vs, kmeta.UpdateOptions{}); err != nil {
+		return errors.Wrap(err, "failed to update virtual service")
+	}
+
+	return nil
+}
+
+func (s *realtimeScaler) routeToActivator(deployment *kapps.Deployment) error {
+	ctx := context.Background()
+	vs, err := s.k8s.GetVirtualService(deployment.Name)
+	if err != nil {
+		return errors.Wrap(err, "failed to get virtual service")
+	}
+
+	if len(vs.Spec.Http) < 1 {
+		return errors.ErrorUnexpected("virtual service does not have any http entries")
+	}
+
+	for i := range vs.Spec.Http {
+		if len(vs.Spec.Http[i].Route) != 2 {
+			return errors.ErrorUnexpected("virtual service does not have the required number of 2 http routes")
+		}
+
+		vs.Spec.Http[i].Route[0].Weight = 0   // service traffic
+		vs.Spec.Http[i].Route[1].Weight = 100 // activator traffic
+	}
+
+	vsClient := s.k8s.IstioClientSet().NetworkingV1beta1().VirtualServices(s.k8s.Namespace)
+	if _, err = vsClient.Update(ctx, vs, kmeta.UpdateOptions{}); err != nil {
+		return errors.Wrap(err, "failed to update virtual service")
+	}
+
+	return nil
+}
+
+func (s *realtimeScaler) waitForReadyReplicas(ctx context.Context, deployment *kapps.Deployment) error {
+	watcher, err := s.k8s.ClientSet().AppsV1().Deployments(s.k8s.Namespace).Watch(
+		ctx,
+		kmeta.ListOptions{
+			FieldSelector: fmt.Sprintf("metadata.name=%s", deployment.Name),
+			Watch:         true,
+		},
+	)
+	if err != nil {
+		return errors.Wrap(err, "could not create deployment watcher")
+	}
+
+	defer watcher.Stop()
+
+	ctx, cancel := context.WithTimeout(ctx, consts.WaitForReadyReplicasTimeout)
+	defer cancel()
+
+	for {
+		select {
+		case event := <-watcher.ResultChan():
+			deploy, ok := event.Object.(*kapps.Deployment)
+			if !ok {
+				continue
+			}
+
+			if deploy.Status.ReadyReplicas > 0 {
+				return nil
+			}
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+}
diff --git a/pkg/autoscaler/recommendations.go b/pkg/autoscaler/recommendations.go
new file mode 100644
index 0000000000..300dd93563
--- /dev/null
+++ b/pkg/autoscaler/recommendations.go
@@ -0,0 +1,74 @@
+/*
+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 (
+	"math"
+	"time"
+)
+
+type recommendations map[time.Time]int32
+
+func (r recommendations) add(rec int32) {
+	r[time.Now()] = rec
+}
+
+func (r recommendations) deleteOlderThan(period time.Duration) {
+	for t := range r {
+		if time.Since(t) > period {
+			delete(r, t)
+		}
+	}
+}
+
+// Returns nil if no recommendations in the period
+func (r recommendations) maxSince(period time.Duration) *int32 {
+	max := int32(math.MinInt32)
+	foundRecommendation := false
+
+	for t, rec := range r {
+		if time.Since(t) <= period && rec > max {
+			max = rec
+			foundRecommendation = true
+		}
+	}
+
+	if !foundRecommendation {
+		return nil
+	}
+
+	return &max
+}
+
+// Returns nil if no recommendations in the period
+func (r recommendations) minSince(period time.Duration) *int32 {
+	min := int32(math.MaxInt32)
+	foundRecommendation := false
+
+	for t, rec := range r {
+		if time.Since(t) <= period && rec < min {
+			min = rec
+			foundRecommendation = true
+		}
+	}
+
+	if !foundRecommendation {
+		return nil
+	}
+
+	return &min
+}
diff --git a/pkg/autoscaler/scaler.go b/pkg/autoscaler/scaler.go
new file mode 100644
index 0000000000..96229b7fe2
--- /dev/null
+++ b/pkg/autoscaler/scaler.go
@@ -0,0 +1,69 @@
+/*
+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 (
+	"time"
+
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+)
+
+type Scaler interface {
+	Scale(apiName string, request int32) error
+	GetInFlightRequests(apiName string, window time.Duration) (*float64, error)
+	GetAutoscalingSpec(apiName string) (*userconfig.Autoscaling, error)
+	CurrentReplicas(apiName string) (int32, error)
+}
+
+type ScalerFunc struct {
+	ScaleFunc               func(apiName string, request int32) error
+	GetInFlightRequestsFunc func(apiName string, window time.Duration) (*float64, error)
+	GetAutoscalingSpecFunc  func(apiName string) (*userconfig.Autoscaling, error)
+	CurrentReplicasFunc     func(apiName string) (int32, error)
+}
+
+func (s *ScalerFunc) Scale(apiName string, request int32) error {
+	if s.ScaleFunc == nil {
+		return nil
+	}
+
+	return s.ScaleFunc(apiName, request)
+}
+
+func (s *ScalerFunc) GetInFlightRequests(apiName string, window time.Duration) (*float64, error) {
+	if s.GetInFlightRequestsFunc == nil {
+		return nil, nil
+	}
+
+	return s.GetInFlightRequestsFunc(apiName, window)
+}
+
+func (s *ScalerFunc) GetAutoscalingSpec(apiName string) (*userconfig.Autoscaling, error) {
+	if s.GetAutoscalingSpecFunc == nil {
+		return nil, nil
+	}
+
+	return s.GetAutoscalingSpecFunc(apiName)
+}
+
+func (s *ScalerFunc) CurrentReplicas(apiName string) (int32, error) {
+	if s.CurrentReplicasFunc == nil {
+		return 0, nil
+	}
+
+	return s.CurrentReplicasFunc(apiName)
+}
diff --git a/pkg/consts/consts.go b/pkg/consts/consts.go
index 4c2af2d49d..4dddd258ad 100644
--- a/pkg/consts/consts.go
+++ b/pkg/consts/consts.go
@@ -18,6 +18,7 @@ package consts
 
 import (
 	"os"
+	"time"
 
 	kresource "k8s.io/apimachinery/pkg/api/resource"
 )
@@ -31,13 +32,17 @@ var (
 
 	DefaultUserPodPortInt32 = int32(8080)
 
-	ProxyListeningPortStr   = "8888"
-	ProxyListeningPortInt32 = int32(8888)
+	ProxyPortStr   = "8888"
+	ProxyPortInt32 = int32(8888)
+
+	ActivatorName      = "activator"
+	ActivatorPortInt32 = int32(8000)
 
 	AdminPortName  = "admin"
 	AdminPortStr   = "15000"
 	AdminPortInt32 = int32(15000)
-	AuthHeader     = "X-Cortex-Authorization"
+
+	AuthHeader = "X-Cortex-Authorization"
 
 	CortexProxyCPU    = kresource.MustParse("100m")
 	CortexProxyMem    = kresource.MustParse("100Mi")
@@ -49,13 +54,24 @@ var (
 	AsyncWorkloadsExpirationDays = int64(7)
 
 	ReservedContainerPorts = []int32{
-		ProxyListeningPortInt32,
+		ProxyPortInt32,
 		AdminPortInt32,
 	}
 	ReservedContainerNames = []string{
 		"dequeuer",
 		"proxy",
 	}
+
+	UserAgentKey             = "User-Agent"
+	KubeProbeUserAgentPrefix = "kube-probe/"
+
+	CortexAPINameHeader       = "X-Cortex-API-Name"
+	CortexTargetServiceHeader = "X-Cortex-Target-Service"
+	CortexProbeHeader         = "X-Cortex-Probe"
+	CortexOriginHeader        = "X-Cortex-Origin"
+
+	WaitForInitializingReplicasTimeout = 15 * time.Minute
+	WaitForReadyReplicasTimeout        = 20 * time.Minute
 )
 
 func DefaultRegistry() string {
diff --git a/pkg/crds/config/manager/manager.yaml b/pkg/crds/config/manager/manager.yaml
index 48fb8ad0e1..28520fd891 100644
--- a/pkg/crds/config/manager/manager.yaml
+++ b/pkg/crds/config/manager/manager.yaml
@@ -51,6 +51,9 @@ spec:
           requests:
             cpu: 100m
             memory: 80Mi
+        envFrom:
+        - configMapRef:
+            name: env-vars
         volumeMounts:
           - mountPath: /mnt/cluster.yaml
             name: cluster-config
diff --git a/pkg/crds/controllers/batch/batchjob_controller_helpers.go b/pkg/crds/controllers/batch/batchjob_controller_helpers.go
index e8ff9ef072..b48c4b2460 100644
--- a/pkg/crds/controllers/batch/batchjob_controller_helpers.go
+++ b/pkg/crds/controllers/batch/batchjob_controller_helpers.go
@@ -300,6 +300,8 @@ func (r *BatchJobReconciler) desiredEnqueuerJob(batchJob batch.BatchJob, queueUR
 								"-apiName", batchJob.Spec.APIName,
 								"-jobID", batchJob.Name,
 							},
+							Env:             workloads.BaseEnvVars,
+							EnvFrom:         workloads.BaseClusterEnvVars(),
 							ImagePullPolicy: kcore.PullAlways,
 						},
 					},
diff --git a/pkg/lib/k8s/k8s.go b/pkg/lib/k8s/k8s.go
index 56e28174f2..4f6a30f43c 100644
--- a/pkg/lib/k8s/k8s.go
+++ b/pkg/lib/k8s/k8s.go
@@ -53,7 +53,8 @@ var (
 type Client struct {
 	ctrl.Client
 	RestConfig           *kclientrest.Config
-	clientset            *kclientset.Clientset
+	clientSet            *kclientset.Clientset
+	istioClientSet       *istioclient.Clientset
 	dynamicClient        kclientdynamic.Interface
 	podClient            kclientcore.PodInterface
 	nodeClient           kclientcore.NodeInterface
@@ -86,7 +87,7 @@ func New(namespace string, inCluster bool, restConfig *kclientrest.Config, schem
 		return nil, errors.Wrap(err, "kubeconfig")
 	}
 
-	client.clientset, err = kclientset.NewForConfig(client.RestConfig)
+	client.clientSet, err = kclientset.NewForConfig(client.RestConfig)
 	if err != nil {
 		return nil, errors.Wrap(err, "kubeconfig")
 	}
@@ -101,24 +102,32 @@ func New(namespace string, inCluster bool, restConfig *kclientrest.Config, schem
 		return nil, errors.Wrap(err, "kubeconfig")
 	}
 
-	istioClient, err := istioclient.NewForConfig(client.RestConfig)
+	client.istioClientSet, err = istioclient.NewForConfig(client.RestConfig)
 	if err != nil {
 		return nil, errors.Wrap(err, "kubeconfig")
 	}
-	client.virtualServiceClient = istioClient.NetworkingV1beta1().VirtualServices(namespace)
-
-	client.podClient = client.clientset.CoreV1().Pods(namespace)
-	client.nodeClient = client.clientset.CoreV1().Nodes()
-	client.serviceClient = client.clientset.CoreV1().Services(namespace)
-	client.configMapClient = client.clientset.CoreV1().ConfigMaps(namespace)
-	client.secretClient = client.clientset.CoreV1().Secrets(namespace)
-	client.deploymentClient = client.clientset.AppsV1().Deployments(namespace)
-	client.jobClient = client.clientset.BatchV1().Jobs(namespace)
-	client.ingressClient = client.clientset.ExtensionsV1beta1().Ingresses(namespace)
-	client.hpaClient = client.clientset.AutoscalingV2beta2().HorizontalPodAutoscalers(namespace)
+	client.virtualServiceClient = client.istioClientSet.NetworkingV1beta1().VirtualServices(namespace)
+
+	client.podClient = client.clientSet.CoreV1().Pods(namespace)
+	client.nodeClient = client.clientSet.CoreV1().Nodes()
+	client.serviceClient = client.clientSet.CoreV1().Services(namespace)
+	client.configMapClient = client.clientSet.CoreV1().ConfigMaps(namespace)
+	client.secretClient = client.clientSet.CoreV1().Secrets(namespace)
+	client.deploymentClient = client.clientSet.AppsV1().Deployments(namespace)
+	client.jobClient = client.clientSet.BatchV1().Jobs(namespace)
+	client.ingressClient = client.clientSet.ExtensionsV1beta1().Ingresses(namespace)
+	client.hpaClient = client.clientSet.AutoscalingV2beta2().HorizontalPodAutoscalers(namespace)
 	return client, nil
 }
 
+func (c *Client) ClientSet() *kclientset.Clientset {
+	return c.clientSet
+}
+
+func (c *Client) IstioClientSet() *istioclient.Clientset {
+	return c.istioClientSet
+}
+
 // to be safe, k8s sometimes needs all characters to be lower case, and the first to be a letter
 func RandomName() string {
 	return random.LowercaseLetters(1) + random.LowercaseString(62)
diff --git a/pkg/lib/k8s/pod.go b/pkg/lib/k8s/pod.go
index fa4e2f2767..e841a7b8a8 100644
--- a/pkg/lib/k8s/pod.go
+++ b/pkg/lib/k8s/pod.go
@@ -405,7 +405,7 @@ func (c *Client) Exec(podName string, containerName string, command []string) (s
 		TTY:       true,
 	}
 
-	req := c.clientset.CoreV1().RESTClient().Post().Namespace(c.Namespace).Resource("pods").Name(podName).SubResource("exec")
+	req := c.clientSet.CoreV1().RESTClient().Post().Namespace(c.Namespace).Resource("pods").Name(podName).SubResource("exec")
 	req.VersionedParams(options, kscheme.ParameterCodec)
 
 	exec, err := kremotecommand.NewSPDYExecutor(c.RestConfig, "POST", req.URL())
diff --git a/pkg/lib/k8s/virtual_service.go b/pkg/lib/k8s/virtual_service.go
index 4665f00e73..98ce122c47 100644
--- a/pkg/lib/k8s/virtual_service.go
+++ b/pkg/lib/k8s/virtual_service.go
@@ -18,13 +18,13 @@ package k8s
 
 import (
 	"context"
-	"reflect"
 
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	"github.com/cortexlabs/cortex/pkg/lib/sets/strset"
 	"github.com/cortexlabs/cortex/pkg/lib/urls"
 	istionetworking "istio.io/api/networking/v1beta1"
 	istioclientnetworking "istio.io/client-go/pkg/apis/networking/v1beta1"
+	istionetworkingclient "istio.io/client-go/pkg/clientset/versioned/typed/networking/v1beta1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	kmeta "k8s.io/apimachinery/pkg/apis/meta/v1"
 	klabels "k8s.io/apimachinery/pkg/labels"
@@ -44,6 +44,7 @@ type VirtualServiceSpec struct {
 	Rewrite      *string
 	Labels       map[string]string
 	Annotations  map[string]string
+	Headers      *istionetworking.Headers
 }
 
 type Destination struct {
@@ -51,6 +52,7 @@ type Destination struct {
 	Weight      int32
 	Port        uint32
 	Shadow      bool
+	Headers     *istionetworking.Headers
 }
 
 func VirtualService(spec *VirtualServiceSpec) *istioclientnetworking.VirtualService {
@@ -75,7 +77,8 @@ func VirtualService(spec *VirtualServiceSpec) *istioclientnetworking.VirtualServ
 						Number: destination.Port,
 					},
 				},
-				Weight: destination.Weight,
+				Weight:  destination.Weight,
+				Headers: destination.Headers,
 			})
 		}
 	}
@@ -96,6 +99,7 @@ func VirtualService(spec *VirtualServiceSpec) *istioclientnetworking.VirtualServ
 			Route:            destinations,
 			Mirror:           mirror,
 			MirrorPercentage: mirrorWeight,
+			Headers:          spec.Headers,
 		})
 
 		if spec.Rewrite != nil {
@@ -117,6 +121,7 @@ func VirtualService(spec *VirtualServiceSpec) *istioclientnetworking.VirtualServ
 			Route:            destinations,
 			Mirror:           mirror,
 			MirrorPercentage: mirrorWeight,
+			Headers:          spec.Headers,
 		}
 
 		prefixMatch := &istionetworking.HTTPRoute{
@@ -132,6 +137,7 @@ func VirtualService(spec *VirtualServiceSpec) *istioclientnetworking.VirtualServ
 			Route:            destinations,
 			Mirror:           mirror,
 			MirrorPercentage: mirrorWeight,
+			Headers:          spec.Headers,
 		}
 
 		if spec.Rewrite != nil {
@@ -164,6 +170,10 @@ func VirtualService(spec *VirtualServiceSpec) *istioclientnetworking.VirtualServ
 	return virtualService
 }
 
+func (c *Client) VirtualServiceClient() istionetworkingclient.VirtualServiceInterface {
+	return c.virtualServiceClient
+}
+
 func (c *Client) CreateVirtualService(virtualService *istioclientnetworking.VirtualService) (*istioclientnetworking.VirtualService, error) {
 	virtualService.TypeMeta = _virtualServiceTypeMeta
 	virtualService, err := c.virtualServiceClient.Create(context.Background(), virtualService, kmeta.CreateOptions{})
@@ -269,23 +279,3 @@ func ExtractVirtualServiceEndpoints(virtualService *istioclientnetworking.Virtua
 	}
 	return endpoints
 }
-
-func VirtualServicesMatch(vs1, vs2 istionetworking.VirtualService) bool {
-	if !strset.New(vs1.Hosts...).IsEqual(strset.New(vs2.Hosts...)) {
-		return false
-	}
-
-	if !strset.New(vs1.Gateways...).IsEqual(strset.New(vs2.Gateways...)) {
-		return false
-	}
-
-	if !strset.New(vs1.ExportTo...).IsEqual(strset.New(vs2.ExportTo...)) {
-		return false
-	}
-
-	if !reflect.DeepEqual(vs1.Http, vs2.Http) {
-		return false
-	}
-
-	return true
-}
diff --git a/pkg/operator/resources/asyncapi/api.go b/pkg/operator/resources/asyncapi/api.go
index 50203355d4..35848c5997 100644
--- a/pkg/operator/resources/asyncapi/api.go
+++ b/pkg/operator/resources/asyncapi/api.go
@@ -27,7 +27,6 @@ import (
 	"github.com/cortexlabs/cortex/pkg/lib/k8s"
 	"github.com/cortexlabs/cortex/pkg/lib/parallel"
 	"github.com/cortexlabs/cortex/pkg/lib/pointer"
-	autoscalerlib "github.com/cortexlabs/cortex/pkg/operator/lib/autoscaler"
 	"github.com/cortexlabs/cortex/pkg/operator/lib/routines"
 	"github.com/cortexlabs/cortex/pkg/operator/operator"
 	"github.com/cortexlabs/cortex/pkg/operator/schema"
@@ -41,14 +40,12 @@ import (
 )
 
 const (
-	_stalledPodTimeout = 15 * time.Minute
 	_tickPeriodMetrics = 10 * time.Second
 	_asyncDashboardUID = "asyncapi"
 )
 
 var (
-	_autoscalerCrons = make(map[string]cron.Cron)
-	_metricsCrons    = make(map[string]cron.Cron)
+	_metricsCrons = make(map[string]cron.Cron)
 )
 
 type resources struct {
@@ -347,22 +344,6 @@ func UpdateAPIMetricsCron(apiDeployment *kapps.Deployment) error {
 	return nil
 }
 
-func UpdateAPIAutoscalerCron(apiDeployment *kapps.Deployment, apiSpec spec.API) error {
-	apiName := apiDeployment.Labels["apiName"]
-	if prevAutoscalerCron, ok := _autoscalerCrons[apiName]; ok {
-		prevAutoscalerCron.Cancel()
-	}
-
-	autoscaler, err := autoscalerlib.AutoscaleFn(apiDeployment, &apiSpec, getMessagesInQueue)
-	if err != nil {
-		return err
-	}
-
-	_autoscalerCrons[apiName] = cron.Run(autoscaler, operator.ErrorHandler(apiName+" autoscaler"), spec.AutoscalingTickInterval)
-
-	return nil
-}
-
 func getK8sResources(apiName string) (resources, error) {
 	var deployment *kapps.Deployment
 	var apiConfigMap *kcore.ConfigMap
@@ -445,10 +426,6 @@ func applyK8sResources(api spec.API, prevK8sResources resources, queueURL string
 				return err
 			}
 
-			if err := UpdateAPIAutoscalerCron(&apiDeployment, api); err != nil {
-				return err
-			}
-
 			return nil
 		},
 		func() error {
@@ -552,10 +529,6 @@ func deleteK8sResources(apiName string) error {
 				delete(_metricsCrons, apiName)
 			}
 
-			if autoscalerCron, ok := _autoscalerCrons[apiName]; ok {
-				autoscalerCron.Cancel()
-				delete(_autoscalerCrons, apiName)
-			}
 			_, err := config.K8s.DeleteDeployment(apiK8sName)
 			return err
 		},
diff --git a/pkg/operator/resources/asyncapi/k8s_specs.go b/pkg/operator/resources/asyncapi/k8s_specs.go
index a6e0c900b6..dac6c0971f 100644
--- a/pkg/operator/resources/asyncapi/k8s_specs.go
+++ b/pkg/operator/resources/asyncapi/k8s_specs.go
@@ -121,8 +121,8 @@ func gatewayServiceSpec(api spec.API) kcore.Service {
 	return *k8s.Service(&k8s.ServiceSpec{
 		Name:        workloads.K8sName(api.Name),
 		PortName:    "http",
-		Port:        consts.ProxyListeningPortInt32,
-		TargetPort:  consts.ProxyListeningPortInt32,
+		Port:        consts.ProxyPortInt32,
+		TargetPort:  consts.ProxyPortInt32,
 		Annotations: api.ToK8sAnnotations(),
 		Labels: map[string]string{
 			"apiName":          api.Name,
@@ -145,7 +145,7 @@ func gatewayVirtualServiceSpec(api spec.API) v1beta1.VirtualService {
 		Destinations: []k8s.Destination{{
 			ServiceName: workloads.K8sName(api.Name),
 			Weight:      100,
-			Port:        uint32(consts.ProxyListeningPortInt32),
+			Port:        uint32(consts.ProxyPortInt32),
 		}},
 		PrefixPath:  api.Networking.Endpoint,
 		Rewrite:     pointer.String("/"),
diff --git a/pkg/operator/resources/asyncapi/autoscaler.go b/pkg/operator/resources/asyncapi/queue_metrics.go
similarity index 65%
rename from pkg/operator/resources/asyncapi/autoscaler.go
rename to pkg/operator/resources/asyncapi/queue_metrics.go
index cafe648831..eca0f0d2de 100644
--- a/pkg/operator/resources/asyncapi/autoscaler.go
+++ b/pkg/operator/resources/asyncapi/queue_metrics.go
@@ -18,24 +18,19 @@ package asyncapi
 
 import (
 	"context"
-	"fmt"
 	"strconv"
 	"time"
 
 	"github.com/aws/aws-sdk-go/aws"
 	"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"
-	"github.com/prometheus/common/model"
 )
 
 const (
-	_sqsQueryTimeoutSeconds        = 10
-	_prometheusQueryTimeoutSeconds = 10
+	_sqsQueryTimeoutSeconds = 10
 )
 
 var queueLengthGauge = promauto.NewGaugeVec(
@@ -85,35 +80,3 @@ func updateQueueLengthMetricsFn(apiName, queueURL string) func() error {
 		return nil
 	}
 }
-
-func getMessagesInQueue(apiName string, window time.Duration) (*float64, error) {
-	windowSeconds := int64(window.Seconds())
-
-	// PromQL query:
-	// 	sum(sum_over_time(cortex_async_queue_length{api_name="<apiName>"}[60s])) /
-	//	sum(count_over_time(cortex_async_queue_length{api_name="<apiName>"}[60s]))
-	query := fmt.Sprintf(
-		"sum(sum_over_time(cortex_async_queue_length{api_name=\"%s\"}[%ds])) / "+
-			"max(count_over_time(cortex_async_queue_length{api_name=\"%s\"}[%ds]))",
-		apiName, windowSeconds,
-		apiName, windowSeconds,
-	)
-
-	ctx, cancel := context.WithTimeout(context.Background(), _prometheusQueryTimeoutSeconds*time.Second)
-	defer cancel()
-
-	valuesQuery, _, err := config.Prometheus.Query(ctx, query, time.Now())
-	if err != nil {
-		return nil, err
-	}
-
-	values, ok := valuesQuery.(model.Vector)
-	if !ok {
-		return nil, errors.ErrorUnexpected("failed to convert prometheus metric to vector")
-	}
-
-	if values.Len() != 0 {
-		return pointer.Float64(float64(values[0].Value)), nil
-	}
-	return nil, nil
-}
diff --git a/pkg/operator/resources/asyncapi/status.go b/pkg/operator/resources/asyncapi/status.go
index 3e20c899c2..38e02329d0 100644
--- a/pkg/operator/resources/asyncapi/status.go
+++ b/pkg/operator/resources/asyncapi/status.go
@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/cortexlabs/cortex/pkg/config"
+	"github.com/cortexlabs/cortex/pkg/consts"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	"github.com/cortexlabs/cortex/pkg/lib/k8s"
 	"github.com/cortexlabs/cortex/pkg/lib/parallel"
@@ -272,7 +273,7 @@ func addPodToReplicaCounts(pod *kcore.Pod, deployment *kapps.Deployment, counts
 
 	switch k8s.GetPodStatus(pod) {
 	case k8s.PodStatusPending:
-		if time.Since(pod.CreationTimestamp.Time) > _stalledPodTimeout {
+		if time.Since(pod.CreationTimestamp.Time) > consts.WaitForInitializingReplicasTimeout {
 			subCounts.Stalled++
 		} else {
 			subCounts.Pending++
diff --git a/pkg/operator/resources/job/batchapi/k8s_specs.go b/pkg/operator/resources/job/batchapi/k8s_specs.go
index 1c856273ff..4492a39fae 100644
--- a/pkg/operator/resources/job/batchapi/k8s_specs.go
+++ b/pkg/operator/resources/job/batchapi/k8s_specs.go
@@ -42,7 +42,7 @@ func virtualServiceSpec(api *spec.API) *istioclientnetworking.VirtualService {
 		Destinations: []k8s.Destination{{
 			ServiceName: _operatorService,
 			Weight:      100,
-			Port:        uint32(consts.ProxyListeningPortInt32),
+			Port:        uint32(consts.ProxyPortInt32),
 		}},
 		PrefixPath:  api.Networking.Endpoint,
 		Rewrite:     pointer.String(path.Join("batch", api.Name)),
diff --git a/pkg/operator/resources/job/taskapi/k8s_specs.go b/pkg/operator/resources/job/taskapi/k8s_specs.go
index 84a5b73a8d..dba049b5c2 100644
--- a/pkg/operator/resources/job/taskapi/k8s_specs.go
+++ b/pkg/operator/resources/job/taskapi/k8s_specs.go
@@ -44,7 +44,7 @@ func virtualServiceSpec(api *spec.API) *istioclientnetworking.VirtualService {
 		Destinations: []k8s.Destination{{
 			ServiceName: _operatorService,
 			Weight:      100,
-			Port:        uint32(consts.ProxyListeningPortInt32),
+			Port:        uint32(consts.ProxyPortInt32),
 		}},
 		PrefixPath:  api.Networking.Endpoint,
 		Rewrite:     pointer.String(path.Join("tasks", api.Name)),
diff --git a/pkg/operator/resources/job/worker_stats.go b/pkg/operator/resources/job/worker_stats.go
index 90292be741..07628995e4 100644
--- a/pkg/operator/resources/job/worker_stats.go
+++ b/pkg/operator/resources/job/worker_stats.go
@@ -19,14 +19,13 @@ package job
 import (
 	"time"
 
+	"github.com/cortexlabs/cortex/pkg/consts"
 	"github.com/cortexlabs/cortex/pkg/lib/k8s"
 	"github.com/cortexlabs/cortex/pkg/types/status"
 	kbatch "k8s.io/api/batch/v1"
 	kcore "k8s.io/api/core/v1"
 )
 
-const _stalledPodTimeout = 15 * time.Minute
-
 func GetWorkerCountsForJob(k8sJob kbatch.Job, pods []kcore.Pod) status.WorkerCounts {
 	if k8sJob.Status.Failed > 0 {
 		return status.WorkerCounts{
@@ -50,7 +49,7 @@ func addPodToWorkerCounts(pod *kcore.Pod, workerCounts *status.WorkerCounts) {
 
 	switch k8s.GetPodStatus(pod) {
 	case k8s.PodStatusPending:
-		if time.Since(pod.CreationTimestamp.Time) > _stalledPodTimeout {
+		if time.Since(pod.CreationTimestamp.Time) > consts.WaitForInitializingReplicasTimeout {
 			workerCounts.Stalled++
 		} else {
 			workerCounts.Pending++
diff --git a/pkg/operator/resources/realtimeapi/api.go b/pkg/operator/resources/realtimeapi/api.go
index 90dfc6f576..5604b73bd7 100644
--- a/pkg/operator/resources/realtimeapi/api.go
+++ b/pkg/operator/resources/realtimeapi/api.go
@@ -22,12 +22,10 @@ import (
 	"time"
 
 	"github.com/cortexlabs/cortex/pkg/config"
-	"github.com/cortexlabs/cortex/pkg/lib/cron"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	"github.com/cortexlabs/cortex/pkg/lib/k8s"
 	"github.com/cortexlabs/cortex/pkg/lib/parallel"
 	"github.com/cortexlabs/cortex/pkg/lib/pointer"
-	autoscalerlib "github.com/cortexlabs/cortex/pkg/operator/lib/autoscaler"
 	"github.com/cortexlabs/cortex/pkg/operator/lib/routines"
 	"github.com/cortexlabs/cortex/pkg/operator/operator"
 	"github.com/cortexlabs/cortex/pkg/operator/schema"
@@ -42,8 +40,6 @@ import (
 
 const _realtimeDashboardUID = "realtimeapi"
 
-var _autoscalerCrons = make(map[string]cron.Cron) // apiName -> cron
-
 func generateDeploymentID() string {
 	return k8s.RandomName()[:10]
 }
@@ -74,7 +70,7 @@ func UpdateAPI(apiConfig *userconfig.API, force bool) (*spec.API, string, error)
 
 		if err := applyK8sResources(api, prevDeployment, prevService, prevVirtualService); err != nil {
 			routines.RunWithPanicHandler(func() {
-				deleteK8sResources(api.Name)
+				_ = deleteK8sResources(api.Name)
 			})
 			return nil, "", err
 		}
@@ -167,7 +163,7 @@ func DeleteAPI(apiName string, keepCache bool) error {
 				return nil
 			}
 			// best effort deletion, swallow errors because there could be weird error messages
-			deleteBucketResources(apiName)
+			_ = deleteBucketResources(apiName)
 			return nil
 		},
 	)
@@ -220,21 +216,21 @@ func namesAndIDsFromStatuses(statuses []status.Status) ([]string, []string) {
 	apiNames := make([]string, len(statuses))
 	apiIDs := make([]string, len(statuses))
 
-	for i, status := range statuses {
-		apiNames[i] = status.APIName
-		apiIDs[i] = status.APIID
+	for i, st := range statuses {
+		apiNames[i] = st.APIName
+		apiIDs[i] = st.APIID
 	}
 
 	return apiNames, apiIDs
 }
 
 func GetAPIByName(deployedResource *operator.DeployedResource) ([]schema.APIResponse, error) {
-	status, err := GetStatus(deployedResource.Name)
+	st, err := GetStatus(deployedResource.Name)
 	if err != nil {
 		return nil, err
 	}
 
-	api, err := operator.DownloadAPISpec(status.APIName, status.APIID)
+	api, err := operator.DownloadAPISpec(st.APIName, st.APIID)
 	if err != nil {
 		return nil, err
 	}
@@ -254,7 +250,7 @@ func GetAPIByName(deployedResource *operator.DeployedResource) ([]schema.APIResp
 	return []schema.APIResponse{
 		{
 			Spec:         *api,
-			Status:       status,
+			Status:       st,
 			Metrics:      metrics,
 			Endpoint:     apiEndpoint,
 			DashboardURL: dashboardURL,
@@ -312,7 +308,7 @@ func applyK8sDeployment(api *spec.API, prevDeployment *kapps.Deployment) error {
 		}
 	} else if prevDeployment.Status.ReadyReplicas == 0 {
 		// Delete deployment if it never became ready
-		config.K8s.DeleteDeployment(workloads.K8sName(api.Name))
+		_, _ = config.K8s.DeleteDeployment(workloads.K8sName(api.Name))
 		_, err := config.K8s.CreateDeployment(newDeployment)
 		if err != nil {
 			return err
@@ -324,27 +320,6 @@ func applyK8sDeployment(api *spec.API, prevDeployment *kapps.Deployment) error {
 		}
 	}
 
-	if err := UpdateAutoscalerCron(newDeployment, api); err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func UpdateAutoscalerCron(deployment *kapps.Deployment, apiSpec *spec.API) error {
-	apiName := deployment.Labels["apiName"]
-
-	if prevAutoscalerCron, ok := _autoscalerCrons[apiName]; ok {
-		prevAutoscalerCron.Cancel()
-	}
-
-	autoscaler, err := autoscalerlib.AutoscaleFn(deployment, apiSpec, getInflightRequests)
-	if err != nil {
-		return err
-	}
-
-	_autoscalerCrons[apiName] = cron.Run(autoscaler, operator.ErrorHandler(apiName+" autoscaler"), spec.AutoscalingTickInterval)
-
 	return nil
 }
 
@@ -375,11 +350,6 @@ func applyK8sVirtualService(api *spec.API, prevVirtualService *istioclientnetwor
 func deleteK8sResources(apiName string) error {
 	return parallel.RunFirstErr(
 		func() error {
-			if autoscalerCron, ok := _autoscalerCrons[apiName]; ok {
-				autoscalerCron.Cancel()
-				delete(_autoscalerCrons, apiName)
-			}
-
 			_, err := config.K8s.DeleteDeployment(workloads.K8sName(apiName))
 			return err
 		},
diff --git a/pkg/operator/resources/realtimeapi/autoscaler.go b/pkg/operator/resources/realtimeapi/autoscaler.go
deleted file mode 100644
index 761a229eea..0000000000
--- a/pkg/operator/resources/realtimeapi/autoscaler.go
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
-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 realtimeapi
-
-import (
-	"context"
-	"fmt"
-	"time"
-
-	"github.com/cortexlabs/cortex/pkg/config"
-	"github.com/cortexlabs/cortex/pkg/lib/errors"
-	"github.com/prometheus/common/model"
-)
-
-const (
-	_prometheusQueryTimeoutSeconds = 10
-)
-
-func getInflightRequests(apiName string, window time.Duration) (*float64, error) {
-	windowSeconds := int64(window.Seconds())
-
-	// PromQL query:
-	// 	sum(sum_over_time(cortex_in_flight_requests{api_name="<apiName>"}[60s])) /
-	//	sum(count_over_time(cortex_in_flight_requests{api_name="<apiName>"}[60s]))
-	query := fmt.Sprintf(
-		"sum(sum_over_time(cortex_in_flight_requests{api_name=\"%s\"}[%ds])) / "+
-			"max(count_over_time(cortex_in_flight_requests{api_name=\"%s\"}[%ds]))",
-		apiName, windowSeconds,
-		apiName, windowSeconds,
-	)
-
-	ctx, cancel := context.WithTimeout(context.Background(), _prometheusQueryTimeoutSeconds*time.Second)
-	defer cancel()
-
-	valuesQuery, _, err := config.Prometheus.Query(ctx, query, time.Now())
-	if err != nil {
-		return nil, err
-	}
-
-	values, ok := valuesQuery.(model.Vector)
-	if !ok {
-		return nil, errors.ErrorUnexpected("failed to convert prometheus metric to vector")
-	}
-
-	// no values available
-	if values.Len() == 0 {
-		return nil, nil
-	}
-
-	avgInflightRequests := float64(values[0].Value)
-
-	return &avgInflightRequests, nil
-}
diff --git a/pkg/operator/resources/realtimeapi/k8s_specs.go b/pkg/operator/resources/realtimeapi/k8s_specs.go
index 8742fdfffc..a5dc01f995 100644
--- a/pkg/operator/resources/realtimeapi/k8s_specs.go
+++ b/pkg/operator/resources/realtimeapi/k8s_specs.go
@@ -17,12 +17,16 @@ limitations under the License.
 package realtimeapi
 
 import (
+	"fmt"
+
+	"github.com/cortexlabs/cortex/pkg/config"
 	"github.com/cortexlabs/cortex/pkg/consts"
 	"github.com/cortexlabs/cortex/pkg/lib/k8s"
 	"github.com/cortexlabs/cortex/pkg/lib/pointer"
 	s "github.com/cortexlabs/cortex/pkg/lib/strings"
 	"github.com/cortexlabs/cortex/pkg/types/spec"
 	"github.com/cortexlabs/cortex/pkg/workloads"
+	istionetworking "istio.io/api/networking/v1beta1"
 	istioclientnetworking "istio.io/client-go/pkg/apis/networking/v1beta1"
 	kapps "k8s.io/api/apps/v1"
 	kcore "k8s.io/api/core/v1"
@@ -83,8 +87,8 @@ func serviceSpec(api *spec.API) *kcore.Service {
 	return k8s.Service(&k8s.ServiceSpec{
 		Name:        workloads.K8sName(api.Name),
 		PortName:    "http",
-		Port:        consts.ProxyListeningPortInt32,
-		TargetPort:  consts.ProxyListeningPortInt32,
+		Port:        consts.ProxyPortInt32,
+		TargetPort:  consts.ProxyPortInt32,
 		Annotations: api.ToK8sAnnotations(),
 		Labels: map[string]string{
 			"apiName":        api.Name,
@@ -99,14 +103,51 @@ func serviceSpec(api *spec.API) *kcore.Service {
 }
 
 func virtualServiceSpec(api *spec.API) *istioclientnetworking.VirtualService {
+	var activatorWeight int32
+	if api.Autoscaling.InitReplicas == 0 {
+		activatorWeight = 100
+	}
+
 	return k8s.VirtualService(&k8s.VirtualServiceSpec{
 		Name:     workloads.K8sName(api.Name),
 		Gateways: []string{"apis-gateway"},
-		Destinations: []k8s.Destination{{
-			ServiceName: workloads.K8sName(api.Name),
-			Weight:      100,
-			Port:        uint32(consts.ProxyListeningPortInt32),
-		}},
+		Destinations: []k8s.Destination{
+			{
+				ServiceName: workloads.K8sName(api.Name),
+				Weight:      100 - activatorWeight,
+				Port:        uint32(consts.ProxyPortInt32),
+				Headers: &istionetworking.Headers{
+					Response: &istionetworking.Headers_HeaderOperations{
+						Set: map[string]string{
+							consts.CortexOriginHeader: "api",
+						},
+					},
+				},
+			},
+			{
+				ServiceName: consts.ActivatorName,
+				Weight:      activatorWeight,
+				Port:        uint32(consts.ActivatorPortInt32),
+				Headers: &istionetworking.Headers{
+					Request: &istionetworking.Headers_HeaderOperations{
+						Set: map[string]string{
+							consts.CortexAPINameHeader: api.Name,
+							consts.CortexTargetServiceHeader: fmt.Sprintf(
+								"http://%s.%s:%d",
+								workloads.K8sName(api.Name),
+								config.ClusterConfig.Namespace,
+								consts.ProxyPortInt32,
+							),
+						},
+					},
+					Response: &istionetworking.Headers_HeaderOperations{
+						Set: map[string]string{
+							consts.CortexOriginHeader: consts.ActivatorName,
+						},
+					},
+				},
+			},
+		},
 		PrefixPath:  api.Networking.Endpoint,
 		Rewrite:     pointer.String("/"),
 		Annotations: api.ToK8sAnnotations(),
diff --git a/pkg/operator/resources/realtimeapi/status.go b/pkg/operator/resources/realtimeapi/status.go
index 65eb7698a7..a65716f35c 100644
--- a/pkg/operator/resources/realtimeapi/status.go
+++ b/pkg/operator/resources/realtimeapi/status.go
@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/cortexlabs/cortex/pkg/config"
+	"github.com/cortexlabs/cortex/pkg/consts"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	"github.com/cortexlabs/cortex/pkg/lib/k8s"
 	"github.com/cortexlabs/cortex/pkg/lib/parallel"
@@ -31,8 +32,6 @@ import (
 	kcore "k8s.io/api/core/v1"
 )
 
-const _stalledPodTimeout = 15 * time.Minute
-
 func GetStatus(apiName string) (*status.Status, error) {
 	var deployment *kapps.Deployment
 	var pods []kcore.Pod
@@ -122,7 +121,7 @@ func addPodToReplicaCounts(pod *kcore.Pod, deployment *kapps.Deployment, counts
 
 	switch k8s.GetPodStatus(pod) {
 	case k8s.PodStatusPending:
-		if time.Since(pod.CreationTimestamp.Time) > _stalledPodTimeout {
+		if time.Since(pod.CreationTimestamp.Time) > consts.WaitForInitializingReplicasTimeout {
 			subCounts.Stalled++
 		} else {
 			subCounts.Pending++
diff --git a/pkg/operator/resources/trafficsplitter/api.go b/pkg/operator/resources/trafficsplitter/api.go
index 8e51ce8de2..9d81a17faa 100644
--- a/pkg/operator/resources/trafficsplitter/api.go
+++ b/pkg/operator/resources/trafficsplitter/api.go
@@ -123,7 +123,7 @@ func getTrafficSplitterDestinations(trafficSplitter *spec.API) []k8s.Destination
 		destinations[i] = k8s.Destination{
 			ServiceName: workloads.K8sName(api.Name),
 			Weight:      api.Weight,
-			Port:        uint32(consts.ProxyListeningPortInt32),
+			Port:        uint32(consts.ProxyPortInt32),
 			Shadow:      api.Shadow,
 		}
 	}
diff --git a/pkg/probe/handler.go b/pkg/probe/handler.go
index 80c19ac474..59712772d6 100644
--- a/pkg/probe/handler.go
+++ b/pkg/probe/handler.go
@@ -19,8 +19,10 @@ package probe
 import (
 	"net/http"
 	"strings"
+
+	"github.com/cortexlabs/cortex/pkg/consts"
 )
 
 func IsRequestKubeletProbe(r *http.Request) bool {
-	return strings.HasPrefix(r.Header.Get(_userAgentKey), _kubeProbeUserAgentPrefix)
+	return strings.HasPrefix(r.Header.Get(consts.UserAgentKey), consts.KubeProbeUserAgentPrefix)
 }
diff --git a/pkg/probe/probe.go b/pkg/probe/probe.go
index bb47d55dfc..d3a9697ccd 100644
--- a/pkg/probe/probe.go
+++ b/pkg/probe/probe.go
@@ -26,6 +26,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/cortexlabs/cortex/pkg/consts"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	s "github.com/cortexlabs/cortex/pkg/lib/strings"
 	"go.uber.org/zap"
@@ -33,16 +34,6 @@ import (
 	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
-var (
-	// userAgentKey is the user agent header key
-	_userAgentKey = "User-Agent"
-
-	// kubeProbeUserAgentPrefix is the user agent header prefix used in k8s probes
-	// Since K8s 1.8, prober requests have
-	//   User-Agent = "kube-probe/{major-version}.{minor-version}".
-	_kubeProbeUserAgentPrefix = "kube-probe/"
-)
-
 const (
 	_defaultInitialDelaySeconds = int32(1)
 	_defaultTimeoutSeconds      = int32(1)
@@ -197,7 +188,7 @@ func (p *Probe) httpProbe() error {
 		return err
 	}
 
-	req.Header.Add(_userAgentKey, _kubeProbeUserAgentPrefix)
+	req.Header.Add(consts.UserAgentKey, consts.KubeProbeUserAgentPrefix)
 
 	for _, header := range p.HTTPGet.HTTPHeaders {
 		req.Header.Add(header.Name, header.Value)
diff --git a/pkg/proxy/breaker.go b/pkg/proxy/breaker.go
index aa1e61cba5..ff00f35c38 100644
--- a/pkg/proxy/breaker.go
+++ b/pkg/proxy/breaker.go
@@ -164,11 +164,20 @@ func (b *Breaker) UpdateConcurrency(size int) {
 	b.sem.updateCapacity(size)
 }
 
+// UpdateQueueLength updates the number of allowed requests in-queue
+func (b *Breaker) UpdateQueueLength(size int) {
+	b.totalSlots = int64(b.sem.Capacity() + size)
+}
+
 // Capacity returns the number of allowed in-flight requests on this breaker.
 func (b *Breaker) Capacity() int {
 	return b.sem.Capacity()
 }
 
+func (b *Breaker) QueueLength() int64 {
+	return b.totalSlots - int64(b.sem.Capacity())
+}
+
 // newSemaphore creates a semaphore with the desired initial capacity.
 func newSemaphore(maxCapacity, initialCapacity int) *semaphore {
 	queue := make(chan struct{}, maxCapacity)
diff --git a/pkg/types/clusterconfig/cluster_config.go b/pkg/types/clusterconfig/cluster_config.go
index a985cd62ef..be098efcc8 100644
--- a/pkg/types/clusterconfig/cluster_config.go
+++ b/pkg/types/clusterconfig/cluster_config.go
@@ -34,7 +34,7 @@ import (
 	"github.com/cortexlabs/cortex/pkg/lib/aws"
 	cr "github.com/cortexlabs/cortex/pkg/lib/configreader"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
-	"github.com/cortexlabs/cortex/pkg/lib/hash"
+	libhash "github.com/cortexlabs/cortex/pkg/lib/hash"
 	libmath "github.com/cortexlabs/cortex/pkg/lib/math"
 	"github.com/cortexlabs/cortex/pkg/lib/pointer"
 	"github.com/cortexlabs/cortex/pkg/lib/sets/strset"
@@ -96,6 +96,8 @@ type CoreConfig struct {
 	ImageManager                    string `json:"image_manager" yaml:"image_manager"`
 	ImageKubexit                    string `json:"image_kubexit" yaml:"image_kubexit"`
 	ImageProxy                      string `json:"image_proxy" yaml:"image_proxy"`
+	ImageActivator                  string `json:"image_activator" yaml:"image_activator"`
+	ImageAutoscaler                 string `json:"image_autoscaler" yaml:"image_autoscaler"`
 	ImageAsyncGateway               string `json:"image_async_gateway" yaml:"image_async_gateway"`
 	ImageEnqueuer                   string `json:"image_enqueuer" yaml:"image_enqueuer"`
 	ImageDequeuer                   string `json:"image_dequeuer" yaml:"image_dequeuer"`
@@ -385,6 +387,20 @@ var CoreConfigStructFieldValidations = []*cr.StructFieldValidation{
 			Validator: validateImageVersion,
 		},
 	},
+	{
+		StructField: "ImageActivator",
+		StringValidation: &cr.StringValidation{
+			Default:   consts.DefaultRegistry() + "/activator:" + consts.CortexVersion,
+			Validator: validateImageVersion,
+		},
+	},
+	{
+		StructField: "ImageAutoscaler",
+		StringValidation: &cr.StringValidation{
+			Default:   consts.DefaultRegistry() + "/autoscaler:" + consts.CortexVersion,
+			Validator: validateImageVersion,
+		},
+	},
 	{
 		StructField: "ImageAsyncGateway",
 		StringValidation: &cr.StringValidation{
@@ -902,7 +918,7 @@ func (cc *Config) ToAccessConfig() AccessConfig {
 
 func SQSNamePrefix(clusterName string) string {
 	// 8 was chosen to make sure that other identifiers can be added to the full queue name before reaching the 80 char SQS name limit
-	return "cx" + SQSQueueDelimiter + hash.String(clusterName)[:8] + SQSQueueDelimiter
+	return "cx" + SQSQueueDelimiter + libhash.String(clusterName)[:8] + SQSQueueDelimiter
 }
 
 // SQSNamePrefix returns a string with the hash of cluster name and adds trailing "_" e.g. cx_abcd1234_
@@ -1697,6 +1713,12 @@ func (cc *CoreConfig) TelemetryEvent() map[string]interface{} {
 	if !strings.HasPrefix(cc.ImageProxy, "quay.io/cortexlabs/") {
 		event["image_proxy._is_custom"] = true
 	}
+	if !strings.HasPrefix(cc.ImageActivator, "quay.io/cortexlabs/") {
+		event["image_activator._is_custom"] = true
+	}
+	if !strings.HasPrefix(cc.ImageAutoscaler, "quay.io/cortexlabs/") {
+		event["image_autoscaler._is_custom"] = true
+	}
 	if !strings.HasPrefix(cc.ImageAsyncGateway, "quay.io/cortexlabs/") {
 		event["image_async_gateway._is_custom"] = true
 	}
@@ -1890,7 +1912,7 @@ func (mc *ManagedConfig) GetNodeGroupNames() []string {
 }
 
 func BucketName(accountID, clusterName, region string) string {
-	bucketID := hash.String(accountID + region)[:8] // this is to "guarantee" a globally unique name
+	bucketID := libhash.String(accountID + region)[:8] // this is to "guarantee" a globally unique name
 	return clusterName + "-" + bucketID
 }
 
diff --git a/pkg/types/spec/validations.go b/pkg/types/spec/validations.go
index d203db77c0..71cce373ba 100644
--- a/pkg/types/spec/validations.go
+++ b/pkg/types/spec/validations.go
@@ -54,7 +54,7 @@ func apiValidation(resource userconfig.Resource) *cr.StructValidation {
 			podValidation(userconfig.RealtimeAPIKind),
 			nodegroupsValidation(),
 			networkingValidation(),
-			autoscalingValidation(resource.Kind),
+			autoscalingValidation(),
 			updateStrategyValidation(),
 		)
 	case userconfig.AsyncAPIKind:
@@ -62,7 +62,7 @@ func apiValidation(resource userconfig.Resource) *cr.StructValidation {
 			podValidation(userconfig.AsyncAPIKind),
 			nodegroupsValidation(),
 			networkingValidation(),
-			autoscalingValidation(resource.Kind),
+			autoscalingValidation(),
 			updateStrategyValidation(),
 		)
 	case userconfig.BatchAPIKind:
@@ -471,12 +471,7 @@ func computeValidation() *cr.StructFieldValidation {
 	}
 }
 
-func autoscalingValidation(kind userconfig.Kind) *cr.StructFieldValidation {
-	minReplicas := int32(1)
-	if kind == userconfig.AsyncAPIKind {
-		minReplicas = int32(0)
-	}
-
+func autoscalingValidation() *cr.StructFieldValidation {
 	return &cr.StructFieldValidation{
 		StructField: "Autoscaling",
 		StructValidation: &cr.StructValidation{
@@ -485,7 +480,7 @@ func autoscalingValidation(kind userconfig.Kind) *cr.StructFieldValidation {
 					StructField: "MinReplicas",
 					Int32Validation: &cr.Int32Validation{
 						Default:              1,
-						GreaterThanOrEqualTo: pointer.Int32(minReplicas),
+						GreaterThanOrEqualTo: pointer.Int32(0),
 					},
 				},
 				{
@@ -499,7 +494,7 @@ func autoscalingValidation(kind userconfig.Kind) *cr.StructFieldValidation {
 					StructField:  "InitReplicas",
 					DefaultField: "MinReplicas",
 					Int32Validation: &cr.Int32Validation{
-						GreaterThanOrEqualTo: pointer.Int32(minReplicas),
+						GreaterThanOrEqualTo: pointer.Int32(0),
 					},
 				},
 				{
diff --git a/pkg/workloads/helpers.go b/pkg/workloads/helpers.go
index 505c9c101b..1d0bf847ba 100644
--- a/pkg/workloads/helpers.go
+++ b/pkg/workloads/helpers.go
@@ -92,7 +92,7 @@ func GetReadinessProbesFromContainers(containers []*userconfig.Container) map[st
 	return probes
 }
 
-func baseClusterEnvVars() []kcore.EnvFromSource {
+func BaseClusterEnvVars() []kcore.EnvFromSource {
 	envVars := []kcore.EnvFromSource{
 		{
 			ConfigMapRef: &kcore.ConfigMapEnvSource{
diff --git a/pkg/workloads/k8s.go b/pkg/workloads/k8s.go
index eff75b624e..06e0526472 100644
--- a/pkg/workloads/k8s.go
+++ b/pkg/workloads/k8s.go
@@ -79,14 +79,15 @@ func AsyncGatewayContainer(api spec.API, queueURL string, volumeMounts []kcore.V
 		ImagePullPolicy: kcore.PullAlways,
 		Args: []string{
 			"--cluster-config", consts.DefaultInClusterConfigPath,
-			"--port", s.Int32(consts.ProxyListeningPortInt32),
+			"--port", s.Int32(consts.ProxyPortInt32),
 			"--queue", queueURL,
 			api.Name,
 		},
 		Ports: []kcore.ContainerPort{
-			{ContainerPort: consts.ProxyListeningPortInt32},
+			{ContainerPort: consts.ProxyPortInt32},
 		},
-		Env: baseEnvVars,
+		Env:     BaseEnvVars,
+		EnvFrom: BaseClusterEnvVars(),
 		Resources: kcore.ResourceRequirements{
 			Requests: kcore.ResourceList{
 				kcore.ResourceCPU:    _asyncGatewayCPURequest,
@@ -132,7 +133,8 @@ func asyncDequeuerProxyContainer(api spec.API, queueURL string) (kcore.Container
 			"--user-port", s.Int32(*api.Pod.Port),
 			"--admin-port", consts.AdminPortStr,
 		},
-		Env: baseEnvVars,
+		Env:     BaseEnvVars,
+		EnvFrom: BaseClusterEnvVars(),
 		Ports: []kcore.ContainerPort{
 			{
 				Name:          consts.AdminPortName,
@@ -184,7 +186,8 @@ func batchDequeuerProxyContainer(api spec.API, jobID, queueURL string) (kcore.Co
 			"--user-port", s.Int32(*api.Pod.Port),
 			"--admin-port", consts.AdminPortStr,
 		},
-		Env: baseEnvVars,
+		Env:     BaseEnvVars,
+		EnvFrom: BaseClusterEnvVars(),
 		Resources: kcore.ResourceRequirements{
 			Requests: kcore.ResourceList{
 				kcore.ResourceCPU:    consts.CortexDequeuerCPU,
@@ -220,7 +223,7 @@ func realtimeProxyContainer(api spec.API) (kcore.Container, kcore.Volume) {
 			"--cluster-config",
 			consts.DefaultInClusterConfigPath,
 			"--port",
-			consts.ProxyListeningPortStr,
+			consts.ProxyPortStr,
 			"--admin-port",
 			consts.AdminPortStr,
 			"--user-port",
@@ -232,10 +235,10 @@ func realtimeProxyContainer(api spec.API) (kcore.Container, kcore.Volume) {
 		},
 		Ports: []kcore.ContainerPort{
 			{Name: consts.AdminPortName, ContainerPort: consts.AdminPortInt32},
-			{ContainerPort: consts.ProxyListeningPortInt32},
+			{ContainerPort: consts.ProxyPortInt32},
 		},
-		Env:     baseEnvVars,
-		EnvFrom: baseClusterEnvVars(),
+		Env:     BaseEnvVars,
+		EnvFrom: BaseClusterEnvVars(),
 		VolumeMounts: []kcore.VolumeMount{
 			ClusterConfigMount(),
 		},
@@ -408,7 +411,7 @@ func userPodContainers(api spec.API) ([]kcore.Container, []kcore.Volume) {
 			containerMounts = append(containerMounts, ShmMount("dshm-"+container.Name))
 		}
 
-		containerEnvVars := baseEnvVars
+		containerEnvVars := BaseEnvVars
 
 		containerEnvVars = append(containerEnvVars, kcore.EnvVar{
 			Name:  "CORTEX_CLI_CONFIG_DIR",
@@ -545,7 +548,7 @@ func GenerateNodeAffinities(apiNodeGroups []string) *kcore.Affinity {
 	}
 }
 
-var baseEnvVars = []kcore.EnvVar{
+var BaseEnvVars = []kcore.EnvVar{
 	{
 		Name:  "CORTEX_VERSION",
 		Value: consts.CortexVersion,


Test patch:
diff --git a/pkg/activator/activator_test.go b/pkg/activator/activator_test.go
new file mode 100644
index 0000000000..08e73b857d
--- /dev/null
+++ b/pkg/activator/activator_test.go
@@ -0,0 +1,97 @@
+/*
+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 activator
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/cortexlabs/cortex/pkg/proxy"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+type autoscalerClientMock struct{}
+
+func (m autoscalerClientMock) AddAPI(_ userconfig.Resource) error {
+	return nil
+}
+
+func (m autoscalerClientMock) Awaken(_ userconfig.Resource) error {
+	return nil
+}
+
+func newLogger(t *testing.T) *zap.SugaredLogger {
+	t.Helper()
+
+	config := zap.NewDevelopmentConfig()
+	config.Level = zap.NewAtomicLevelAt(zap.ErrorLevel)
+	logger, err := config.Build()
+	require.NoError(t, err)
+
+	logr := logger.Sugar()
+
+	return logr
+}
+
+func TestActivator_Try(t *testing.T) {
+	t.Parallel()
+
+	log := newLogger(t)
+
+	apiName := "test"
+	act := &activator{
+		autoscalerClient: autoscalerClientMock{},
+		apiActivators: map[string]*apiActivator{
+			apiName: newAPIActivator(1, 1),
+		},
+		readinessTrackers: map[string]*readinessTracker{
+			apiName: {ready: true},
+		},
+		logger: log,
+	}
+
+	ctx := context.Background()
+	ctx = context.WithValue(ctx, APINameCtxKey, apiName)
+
+	errCh := make(chan error)
+	waitCh := make(chan struct{})
+
+	for i := 0; i < 3; i++ {
+		go func() {
+			ctx, cancel := context.WithTimeout(ctx, time.Second)
+			defer cancel()
+
+			errCh <- act.Try(ctx, func() error {
+				<-waitCh
+				return nil
+			})
+		}()
+	}
+
+	err := <-errCh
+	require.Error(t, err)
+	require.True(t, errors.Is(err, proxy.ErrRequestQueueFull))
+
+	for i := 0; i < 2; i++ {
+		waitCh <- struct{}{}
+		require.NoError(t, <-errCh)
+	}
+}
diff --git a/pkg/activator/api_activator_test.go b/pkg/activator/api_activator_test.go
new file mode 100644
index 0000000000..907156892c
--- /dev/null
+++ b/pkg/activator/api_activator_test.go
@@ -0,0 +1,54 @@
+/*
+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 activator
+
+import (
+	"context"
+	"testing"
+
+	"github.com/cortexlabs/cortex/pkg/proxy"
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestApiActivator_Try(t *testing.T) {
+	t.Parallel()
+
+	act := newAPIActivator(1, 1)
+
+	errCh := make(chan error)
+	waitCh := make(chan struct{})
+
+	for i := 0; i < 3; i++ {
+
+		go func() {
+			errCh <- act.try(context.Background(), func() error {
+				<-waitCh
+				return nil
+			}, &readinessTracker{ready: true})
+		}()
+	}
+
+	err := <-errCh
+	require.Error(t, err)
+	require.True(t, errors.Is(err, proxy.ErrRequestQueueFull))
+
+	for i := 0; i < 2; i++ {
+		waitCh <- struct{}{}
+		require.NoError(t, <-errCh)
+	}
+}
diff --git a/pkg/activator/handler_test.go b/pkg/activator/handler_test.go
new file mode 100644
index 0000000000..d896a1240f
--- /dev/null
+++ b/pkg/activator/handler_test.go
@@ -0,0 +1,64 @@
+/*
+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 activator
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/cortexlabs/cortex/pkg/consts"
+	"github.com/stretchr/testify/require"
+)
+
+func TestActivatorHandler_ServeHTTP(t *testing.T) {
+	t.Parallel()
+
+	log := newLogger(t)
+
+	apiName := "test"
+	act := &activator{
+		autoscalerClient: autoscalerClientMock{},
+		apiActivators: map[string]*apiActivator{
+			apiName: newAPIActivator(1, 1),
+		},
+		readinessTrackers: map[string]*readinessTracker{
+			apiName: {ready: true},
+		},
+		logger: log,
+	}
+
+	ah := NewHandler(act, log)
+
+	var callCount int
+	server := httptest.NewServer(
+		http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			callCount++
+			w.WriteHeader(http.StatusOK)
+		}),
+	)
+
+	w := httptest.NewRecorder()
+	r := httptest.NewRequest("POST", "http://fake.cortex.dev/api", nil)
+	r.Header.Set(consts.CortexAPINameHeader, apiName)
+	r.Header.Set(consts.CortexTargetServiceHeader, server.URL)
+
+	ah.ServeHTTP(w, r)
+
+	require.Equal(t, http.StatusOK, w.Code)
+	require.Equal(t, 1, callCount)
+}
diff --git a/pkg/autoscaler/autoscaler_test.go b/pkg/autoscaler/autoscaler_test.go
new file mode 100644
index 0000000000..307b1ca3fa
--- /dev/null
+++ b/pkg/autoscaler/autoscaler_test.go
@@ -0,0 +1,115 @@
+/*
+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 (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/cortexlabs/cortex/pkg/lib/cron"
+	"github.com/cortexlabs/cortex/pkg/lib/pointer"
+	"github.com/cortexlabs/cortex/pkg/types/userconfig"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+func newLogger(t *testing.T) *zap.SugaredLogger {
+	t.Helper()
+
+	config := zap.NewDevelopmentConfig()
+	config.Level = zap.NewAtomicLevelAt(zap.ErrorLevel)
+	logger, err := config.Build()
+	require.NoError(t, err)
+
+	logr := logger.Sugar()
+
+	return logr
+}
+
+func TestAutoscaler_Awake(t *testing.T) {
+	log := newLogger(t)
+
+	mux := sync.RWMutex{}
+	var latestRequest int32
+
+	downscaleStabilizationPeriod := 3 * time.Second
+	scalerMock := &ScalerFunc{
+		ScaleFunc: func(apiName string, request int32) error {
+			mux.Lock()
+			defer mux.Unlock()
+
+			latestRequest = request
+			return nil
+		},
+		GetInFlightRequestsFunc: func(apiName string, window time.Duration) (*float64, error) {
+			return pointer.Float64(0), nil
+		},
+		GetAutoscalingSpecFunc: func(apiName string) (*userconfig.Autoscaling, error) {
+			return &userconfig.Autoscaling{
+				MinReplicas:                  0,
+				MaxReplicas:                  1,
+				InitReplicas:                 1,
+				TargetInFlight:               pointer.Float64(1),
+				Window:                       time.Second,
+				DownscaleStabilizationPeriod: downscaleStabilizationPeriod,
+			}, nil
+		},
+		CurrentReplicasFunc: func(apiName string) (int32, error) {
+			return 0, nil
+		},
+	}
+
+	autoScaler := &Autoscaler{
+		logger:              log,
+		crons:               make(map[string]cron.Cron),
+		scalers:             make(map[userconfig.Kind]Scaler),
+		lastAwakenTimestamp: make(map[string]time.Time),
+	}
+	autoScaler.AddScaler(scalerMock, userconfig.RealtimeAPIKind)
+
+	api := userconfig.Resource{
+		Name: "test",
+		Kind: userconfig.RealtimeAPIKind,
+	}
+
+	autoscaleFn, err := autoScaler.autoscaleFn(api)
+	require.NoError(t, err)
+
+	ticker := time.NewTicker(500 * time.Millisecond)
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				err := autoscaleFn()
+				require.NoError(t, err)
+			}
+		}
+	}()
+
+	err = autoScaler.Awaken(api)
+	require.NoError(t, err)
+
+	_, ok := autoScaler.lastAwakenTimestamp[api.Name]
+	require.True(t, ok)
+
+	require.Never(t, func() bool {
+		mux.RLock()
+		defer mux.RUnlock()
+		return latestRequest != 1
+	}, downscaleStabilizationPeriod, time.Second)
+}
diff --git a/test/apis/realtime/hello-world/cortex_scale_to_zero.yaml b/test/apis/realtime/hello-world/cortex_scale_to_zero.yaml
new file mode 100644
index 0000000000..1c9a0a3bd1
--- /dev/null
+++ b/test/apis/realtime/hello-world/cortex_scale_to_zero.yaml
@@ -0,0 +1,21 @@
+- name: hello-world
+  kind: RealtimeAPI
+  pod:
+    port: 8080
+    max_concurrency: 1
+    max_queue_length: 999
+    containers:
+    - name: api
+      image: quay.io/cortexlabs-test/realtime-hello-world-cpu:latest
+      readiness_probe:
+        http_get:
+          path: "/healthz"
+          port: 8080
+        timeout_seconds: 3
+      compute:
+        cpu: 200m
+        mem: 128Mi
+  autoscaling:
+    min_replicas: 0
+    max_replicas: 1
+    downscale_stabilization_period: 30s
diff --git a/test/e2e/e2e/tests.py b/test/e2e/e2e/tests.py
index 34edcee61c..8c79f3aa10 100644
--- a/test/e2e/e2e/tests.py
+++ b/test/e2e/e2e/tests.py
@@ -52,6 +52,7 @@
     retrieve_results_concurrently,
     stream_api_logs,
     stream_job_logs,
+    wait_for,
 )
 
 TEST_APIS_DIR = Path(__file__).parent.parent.parent / "apis"
@@ -924,3 +925,49 @@ def test_long_running_realtime(
             raise
     finally:
         delete_apis(client, [api_name])
+
+
+def test_realtime_scale_to_zero(
+    client: cx.Client,
+    api: str,
+    timeout: int = None,
+    api_config_name: str = "cortex_cpu.yaml",
+):
+    api_dir = TEST_APIS_DIR / api
+    with open(str(api_dir / api_config_name)) as f:
+        api_specs = yaml.safe_load(f)
+
+    api_name = api_specs[0]["name"]
+    for api_spec in api_specs:
+        client.deploy(api_spec=api_spec)
+
+    try:
+        assert apis_ready(
+            client=client, api_names=[api_name], timeout=timeout
+        ), f"apis {api_name} not ready"
+
+        api_info = client.get_api(api_name)
+        endpoint = api_info["endpoint"]
+
+        response = requests.post(endpoint, json={}, timeout=30)
+
+        # make first request, which should go to activator
+        assert (
+            response.status_code == HTTPStatus.OK
+        ), f"status code: got {response.status_code}, expected {HTTPStatus.OK}"
+
+        assert response.headers.get("x-cortex-origin") == "activator"
+
+        def _make_request() -> bool:
+            res = requests.post(endpoint, json={}, timeout=30)
+
+            # make second request, which should go directly to the api
+            assert (
+                res.status_code == HTTPStatus.OK
+            ), f"status code: got {res.status_code}, expected {HTTPStatus.OK}"
+
+            return res.headers.get("x-cortex-origin") == "api"
+
+        assert wait_for(_make_request, timeout=60)
+    finally:
+        delete_apis(client, [api_name])
diff --git a/test/e2e/tests/aws/test_realtime.py b/test/e2e/tests/aws/test_realtime.py
index 0040f8b26d..6b45e399b0 100644
--- a/test/e2e/tests/aws/test_realtime.py
+++ b/test/e2e/tests/aws/test_realtime.py
@@ -51,10 +51,8 @@
 
 
 @pytest.mark.usefixtures("client")
-@pytest.mark.parametrize("api", TEST_APIS)
+@pytest.mark.parametrize("api", TEST_APIS, ids=[api["name"] for api in TEST_APIS])
 def test_realtime_api(printer: Callable, config: Dict, client: cx.Client, api: Dict[str, str]):
-
-    printer(f"testing {api['name']}")
     e2e.tests.test_realtime_api(
         printer=printer,
         client=client,
@@ -83,13 +81,12 @@ def test_realtime_api_arm(printer: Callable, config: Dict, client: cx.Client, ap
 
 
 @pytest.mark.usefixtures("client")
-@pytest.mark.parametrize("api", TEST_APIS_GPU)
+@pytest.mark.parametrize("api", TEST_APIS_GPU, ids=[api["name"] for api in TEST_APIS_GPU])
 def test_realtime_api_gpu(printer: Callable, config: Dict, client: cx.Client, api: Dict[str, str]):
     skip_gpus = config["global"].get("skip_gpus", False)
     if skip_gpus:
         pytest.skip("--skip-gpus flag detected, skipping GPU tests")
 
-    printer(f"testing {api['name']}")
     e2e.tests.test_realtime_api(
         printer=printer,
         client=client,
diff --git a/test/e2e/tests/aws/test_scale_to_zero.py b/test/e2e/tests/aws/test_scale_to_zero.py
new file mode 100644
index 0000000000..34dcb0f658
--- /dev/null
+++ b/test/e2e/tests/aws/test_scale_to_zero.py
@@ -0,0 +1,33 @@
+# 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.
+
+from typing import Dict
+
+import pytest
+
+import cortex as cx
+import e2e.tests
+
+TEST_APIS = [{"api": "realtime/hello-world", "config": "cortex_scale_to_zero.yaml"}]
+
+
+@pytest.mark.parametrize("api", TEST_APIS, ids=[api["api"] for api in TEST_APIS])
+@pytest.mark.usefixtures("client")
+def test_scale_to_zero_realtime(config: Dict, client: cx.Client, api: Dict[str, str]):
+    e2e.tests.test_realtime_scale_to_zero(
+        client=client,
+        api=api["api"],
+        timeout=config["global"]["realtime_deploy_timeout"],
+        api_config_name=api["config"],
+    )