Revisions should wait for minScale replicas to report ready

pkg/reconciler/autoscaling/kpa/kpa.go

@@ -19,6 +19,7 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strconv"
 
 	perrors "github.com/pkg/errors"
 	"go.uber.org/zap"
@@ -448,19 +449,22 @@ func reportMetrics(pa *pav1alpha1.PodAutoscaler, want int32, got int) error {
 // computeActiveCondition updates the status of PA, depending on scales desired and present.
 // computeActiveCondition returns true if it thinks SKS needs an update.
 func computeActiveCondition(pa *pav1alpha1.PodAutoscaler, want int32, got int) (ret bool) {
+	minReady := activeThreshold(pa)
+
 	switch {
 	case want == 0:
 		ret = !pa.Status.IsInactive() // Any state but inactive should change SKS.
 		pa.Status.MarkInactive("NoTraffic", "The target is not receiving traffic.")
 
-	case got == 0 && want > 0:
+	case got < minReady && want > 0:
 		ret = pa.Status.IsInactive() // If we were inactive and became activating.
 		pa.Status.MarkActivating(
 			"Queued", "Requests to the target are being buffered as resources are provisioned.")
 
-	case got > 0:
+	case got >= minReady:
 		// SKS should already be active.
 		pa.Status.MarkActive()
+
 	case want == scaleUnknown:
 		// We don't know what scale we want, so don't touch PA at all.
 	}
@@ -469,6 +473,19 @@ func computeActiveCondition(pa *pav1alpha1.PodAutoscaler, want int32, got int) (
 	return
 }
 
+// activeThreshold returns the scale required for the kpa to be marked Active
+func activeThreshold(pa *pav1alpha1.PodAutoscaler) int {
+	if min, ok := pa.Annotations[autoscaling.MinScaleAnnotationKey]; ok {
+		if ms, err := strconv.Atoi(min); err == nil {
+			if ms > 1 {
+				return ms
+			}
+		}
+	}
+
+	return 1
+}
+
 func (c *Reconciler) updateStatus(desired *pav1alpha1.PodAutoscaler) (*pav1alpha1.PodAutoscaler, error) {
 	pa, err := c.paLister.PodAutoscalers(desired.Namespace).Get(desired.Name)
 	if err != nil {

pkg/reconciler/autoscaling/kpa/kpa_test.go

@@ -19,6 +19,7 @@ package kpa
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -50,6 +51,7 @@ import (
 	"github.com/knative/serving/pkg/reconciler/autoscaling/kpa/resources"
 	aresources "github.com/knative/serving/pkg/reconciler/autoscaling/resources"
 	revisionresources "github.com/knative/serving/pkg/reconciler/revision/resources"
+	presources "github.com/knative/serving/pkg/resources"
 	perrors "github.com/pkg/errors"
 	fakedynamic "k8s.io/client-go/dynamic/fake"
 
@@ -541,6 +543,32 @@ func TestReconcile(t *testing.T) {
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: kpa(testNamespace, testRevision, markActive, WithPAStatusService(testRevision)),
 		}},
+	}, {
+		Name: "kpa does not become ready without minScale endpoints",
+		Key:  key,
+		Objects: []runtime.Object{
+			kpa(testNamespace, testRevision, withMinScale(2)),
+			sks(testNamespace, testRevision, WithDeployRef(deployName), WithSKSReady),
+			metricsSvc(testNamespace, testRevision, withSvcSelector(usualSelector)),
+			expectedDeploy,
+			makeSKSPrivateEndpoints(1, testNamespace, testRevision),
+		},
+		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+			Object: kpa(testNamespace, testRevision, markActivating, withMinScale(2), WithPAStatusService(testRevision)),
+		}},
+	}, {
+		Name: "kpa becomes ready with minScale endpoints",
+		Key:  key,
+		Objects: []runtime.Object{
+			kpa(testNamespace, testRevision, markActivating, withMinScale(2), WithPAStatusService(testRevision)),
+			sks(testNamespace, testRevision, WithDeployRef(deployName), WithSKSReady),
+			metricsSvc(testNamespace, testRevision, withSvcSelector(usualSelector)),
+			expectedDeploy,
+			makeSKSPrivateEndpoints(2, testNamespace, testRevision),
+		},
+		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
+			Object: kpa(testNamespace, testRevision, markActive, withMinScale(2), WithPAStatusService(testRevision)),
+		}},
 	}, {
 		Name: "sks does not exist",
 		Key:  key,
@@ -1512,12 +1540,25 @@ func makeSKSPrivateEndpoints(num int, ns, n string) *corev1.Endpoints {
 }
 
 func addEndpoint(ep *corev1.Endpoints) *corev1.Endpoints {
-	ep.Subsets = []corev1.EndpointSubset{{
-		Addresses: []corev1.EndpointAddress{{IP: "127.0.0.1"}},
-	}}
+	if ep.Subsets == nil {
+		ep.Subsets = []corev1.EndpointSubset{{
+			Addresses: []corev1.EndpointAddress{},
+		}}
+	}
+
+	ep.Subsets[0].Addresses = append(ep.Subsets[0].Addresses, corev1.EndpointAddress{IP: "127.0.0.1"})
 	return ep
 }
 
+func withMinScale(minScale int) PodAutoscalerOption {
+	return func(pa *asv1a1.PodAutoscaler) {
+		pa.Annotations = presources.UnionMaps(
+			pa.Annotations,
+			map[string]string{autoscaling.MinScaleAnnotationKey: strconv.Itoa(minScale)},
+		)
+	}
+}
+
 type testConfigStore struct {
 	config *config.Config
 }

pkg/reconciler/autoscaling/kpa/scaler.go

@@ -138,6 +138,7 @@ func (ks *scaler) handleScaleToZero(pa *pav1alpha1.PodAutoscaler, desiredScale i
 	if desiredScale != 0 {
 		return desiredScale, true
 	}
+
 	// We should only scale to zero when three of the following conditions are true:
 	//   a) enable-scale-to-zero from configmap is true
 	//   b) The PA has been active for at least the stable window, after which it gets marked inactive
@@ -149,7 +150,9 @@ func (ks *scaler) handleScaleToZero(pa *pav1alpha1.PodAutoscaler, desiredScale i
 
 	if pa.Status.IsActivating() { // Active=Unknown
 		// Don't scale-to-zero during activation
-		desiredScale = scaleUnknown
+		if min, _ := pa.ScaleBounds(); min == 0 {
+			return scaleUnknown, false
+		}
 	} else if pa.Status.IsReady() { // Active=True
 		// Don't scale-to-zero if the PA is active
 
@@ -176,6 +179,7 @@ func (ks *scaler) handleScaleToZero(pa *pav1alpha1.PodAutoscaler, desiredScale i
 			return desiredScale, false
 		}
 	}
+
 	return desiredScale, true
 }
 
@@ -208,20 +212,19 @@ func (ks *scaler) applyScale(ctx context.Context, pa *pav1alpha1.PodAutoscaler,
 
 	logger.Debug("Successfully scaled.")
 	return desiredScale, nil
-
 }
 
 // Scale attempts to scale the given PA's target reference to the desired scale.
 func (ks *scaler) Scale(ctx context.Context, pa *pav1alpha1.PodAutoscaler, desiredScale int32) (int32, error) {
 	logger := logging.FromContext(ctx)
 
-	desiredScale, shouldApplyScale := ks.handleScaleToZero(pa, desiredScale, config.FromContext(ctx).Autoscaler)
-	if !shouldApplyScale {
+	if desiredScale < 0 {
+		logger.Debug("Metrics are not yet being collected.")
 		return desiredScale, nil
 	}
 
-	if desiredScale < 0 {
-		logger.Debug("Metrics are not yet being collected.")
+	desiredScale, shouldApplyScale := ks.handleScaleToZero(pa, desiredScale, config.FromContext(ctx).Autoscaler)
+	if !shouldApplyScale {
 		return desiredScale, nil
 	}
 
@@ -236,11 +239,11 @@ func (ks *scaler) Scale(ctx context.Context, pa *pav1alpha1.PodAutoscaler, desir
 		logger.Errorw(fmt.Sprintf("Resource %q not found", pa.Name), zap.Error(err))
 		return desiredScale, err
 	}
+
 	currentScale := int32(1)
 	if ps.Spec.Replicas != nil {
 		currentScale = *ps.Spec.Replicas
 	}
-
 	if desiredScale == currentScale {
 		return desiredScale, nil
 	}

test/e2e/minscale_readiness_test.go

@@ -0,0 +1,79 @@
+// +build e2e
+
+/*
+Copyright 2019 The Knative Authors
+
+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 e2e
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/knative/serving/pkg/apis/autoscaling"
+	"github.com/knative/serving/pkg/apis/serving/v1alpha1"
+	"github.com/knative/serving/test"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+func TestMinScale(t *testing.T) {
+	const minScale = 4
+
+	clients := Setup(t)
+
+	names := test.ResourceNames{
+		Config: test.ObjectNameForTest(t),
+		Image:  "helloworld",
+	}
+
+	if _, err := test.CreateConfiguration(t, clients, names, &test.Options{}, func(cfg *v1alpha1.Configuration) {
+		if cfg.Spec.Template.Annotations == nil {
+			cfg.Spec.Template.Annotations = make(map[string]string)
+		}
+
+		cfg.Spec.Template.Annotations[autoscaling.MinScaleAnnotationKey] = strconv.Itoa(minScale)
+
+	}); err != nil {
+		t.Fatalf("Failed to create Configuration: %v", err)
+	}
+
+	test.CleanupOnInterrupt(func() { test.TearDown(clients, names) })
+	defer test.TearDown(clients, names)
+
+	// Wait for the Config have a LatestCreatedRevisionName
+	if err := test.WaitForConfigurationState(clients.ServingClient, names.Config, test.ConfigurationHasCreatedRevision, "ConfigurationHasCreatedRevision"); err != nil {
+		t.Fatalf("The Configuration %q does not have a LatestCreatedRevisionName: %v", names.Config, err)
+	}
+
+	config, err := clients.ServingClient.Configs.Get(names.Config, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to get Configuration after it was seen to be live: %v", err)
+	}
+
+	revName := config.Status.LatestCreatedRevisionName
+
+	if err = test.WaitForRevisionState(clients.ServingClient, revName, test.IsRevisionReady, "RevisionIsReady"); err != nil {
+		t.Fatal("Revision did not become ready.")
+	}
+
+	deployment, err := clients.KubeClient.Kube.ExtensionsV1beta1().Deployments(test.ServingNamespace).Get(revName+"-deployment", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Failed to get Deployment for Revision %s, err: %v", revName, err)
+	}
+
+	if deployment.Status.AvailableReplicas < int32(minScale) {
+		t.Fatalf("Reported ready with %d replicas when minScale was %d", deployment.Status.AvailableReplicas, minScale)
+	}
+}