Fix deployment status propagation when scaling from zero

pkg/apis/autoscaling/v1alpha1/pa_lifecycle.go

@@ -35,6 +35,7 @@ var podCondSet = apis.NewLivingConditionSet(
 	PodAutoscalerConditionActive,
 	PodAutoscalerConditionScaleTargetInitialized,
 	PodAutoscalerConditionSKSReady,
+	PodAutoscalerConditionScaleTargetScaled,
 )
 
 // GetConditionSet retrieves the condition set for this resource. Implements the KRShaped interface.
@@ -215,6 +216,12 @@ func (pas *PodAutoscalerStatus) MarkScaleTargetInitialized() {
 	podCondSet.Manage(pas).MarkTrue(PodAutoscalerConditionScaleTargetInitialized)
 }
 
+// ScaleTargetNotScaled returns true if the PodAutoscaler's scale target has failed
+// to scaled.
+func (pas *PodAutoscalerStatus) ScaleTargetNotScaled() bool {
+	return pas.GetCondition(PodAutoscalerConditionScaleTargetScaled).IsFalse()
+}
+
 // MarkSKSReady marks the PA condition denoting that SKS is ready.
 func (pas *PodAutoscalerStatus) MarkSKSReady() {
 	podCondSet.Manage(pas).MarkTrue(PodAutoscalerConditionSKSReady)
@@ -250,6 +257,16 @@ func (pas *PodAutoscalerStatus) MarkInactive(reason, message string) {
 	podCondSet.Manage(pas).MarkFalse(PodAutoscalerConditionActive, reason, message)
 }
 
+// MarkWithNoScaleFailures marks the PA as free of scale failures.
+func (pas *PodAutoscalerStatus) MarkWithNoScaleFailures() {
+	podCondSet.Manage(pas).MarkTrue(PodAutoscalerConditionScaleTargetScaled)
+}
+
+// MarkWithScaleFailures marks the PA as failed due to scale failures.
+func (pas *PodAutoscalerStatus) MarkWithScaleFailures(reason, message string) {
+	podCondSet.Manage(pas).MarkFalse(PodAutoscalerConditionScaleTargetScaled, reason, message)
+}
+
 // MarkResourceNotOwned changes the "Active" condition to false to reflect that the
 // resource of the given kind and name has already been created, and we do not own it.
 func (pas *PodAutoscalerStatus) MarkResourceNotOwned(kind, name string) {

pkg/apis/autoscaling/v1alpha1/pa_lifecycle_test.go

@@ -1056,6 +1056,7 @@ func TestTypicalFlow(t *testing.T) {
 	// When we see traffic, mark ourselves active.
 	r.MarkActive()
 	r.MarkScaleTargetInitialized()
+	r.MarkWithNoScaleFailures()
 	apistest.CheckConditionSucceeded(r, PodAutoscalerConditionScaleTargetInitialized, t)
 	apistest.CheckConditionSucceeded(r, PodAutoscalerConditionSKSReady, t)
 	apistest.CheckConditionSucceeded(r, PodAutoscalerConditionActive, t)
@@ -1064,6 +1065,7 @@ func TestTypicalFlow(t *testing.T) {
 	// Check idempotency.
 	r.MarkActive()
 	r.MarkScaleTargetInitialized()
+	r.MarkWithNoScaleFailures()
 	r.MarkSKSReady()
 	apistest.CheckConditionSucceeded(r, PodAutoscalerConditionScaleTargetInitialized, t)
 	apistest.CheckConditionSucceeded(r, PodAutoscalerConditionActive, t)

pkg/apis/autoscaling/v1alpha1/pa_types.go

@@ -108,6 +108,11 @@ const (
 	PodAutoscalerConditionActive apis.ConditionType = "Active"
 	// PodAutoscalerConditionSKSReady is set when SKS is ready.
 	PodAutoscalerConditionSKSReady = "SKSReady"
+	// PodAutoscalerConditionScaleTargetScaled is set when scaling the revision is successful.
+	// There are cases where there are pod failures during scaling eg. from zero but
+	// K8s does not properly propagate the failure to the deployment. In those cases
+	// progressdeadline is not enough.
+	PodAutoscalerConditionScaleTargetScaled apis.ConditionType = "ScaleTargetScaled"
 )
 
 // PodAutoscalerStatus communicates the observed state of the PodAutoscaler (from the controller).

pkg/apis/serving/v1/revision_lifecycle.go

@@ -200,6 +200,13 @@ func (rs *RevisionStatus) PropagateAutoscalerStatus(ps *autoscalingv1alpha1.PodA
 		// that implies that |service.endpoints| > 0.
 		rs.MarkResourcesAvailableTrue()
 		rs.MarkContainerHealthyTrue()
+
+		// Mark resource unavailable if we are scaling back to zero, but we never achieved the required scale
+		// and deployment status was not updated properly by K8s. For example due to an image pull error.
+		if ps.ScaleTargetNotScaled() {
+			condScaled := ps.GetCondition(autoscalingv1alpha1.PodAutoscalerConditionScaleTargetScaled)
+			rs.MarkResourcesAvailableFalse(condScaled.Reason, condScaled.Message)
+		}
 	}
 
 	// Mark resource unavailable if we don't have a Service Name and the deployment is ready

pkg/reconciler/autoscaling/kpa/kpa.go

@@ -114,10 +114,15 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, pa *autoscalingv1alpha1.
 	if err := c.ReconcileMetric(ctx, pa, resolveScrapeTarget(ctx, pa)); err != nil {
 		return fmt.Errorf("error reconciling Metric: %w", err)
 	}
+	podCounter := resourceutil.NewPodAccessor(c.podsLister, pa.Namespace, pa.Labels[serving.RevisionLabelKey])
+
+	if err != nil {
+		return fmt.Errorf("error getting a pod for the revision: %w", err)
+	}
 
 	// Get the appropriate current scale from the metric, and right size
 	// the scaleTargetRef based on it.
-	want, err := c.scaler.scale(ctx, pa, sks, decider.Status.DesiredScale)
+	want, err := c.scaler.scale(ctx, pa, sks, decider.Status.DesiredScale, &podCounter)
 	if err != nil {
 		return fmt.Errorf("error scaling target: %w", err)
 	}
@@ -145,7 +150,6 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, pa *autoscalingv1alpha1.
 	}
 
 	// Compare the desired and observed resources to determine our situation.
-	podCounter := resourceutil.NewPodAccessor(c.podsLister, pa.Namespace, pa.Labels[serving.RevisionLabelKey])
 	ready, notReady, pending, terminating, err := podCounter.PodCountsByState()
 	if err != nil {
 		return fmt.Errorf("error getting pod counts: %w", err)
@@ -272,6 +276,7 @@ func computeActiveCondition(ctx context.Context, pa *autoscalingv1alpha1.PodAuto
 	minReady := activeThreshold(ctx, pa)
 	if pc.ready >= minReady && pa.Status.ServiceName != "" {
 		pa.Status.MarkScaleTargetInitialized()
+		pa.Status.MarkWithNoScaleFailures()
 	}
 
 	switch {

pkg/reconciler/autoscaling/kpa/kpa_test.go

@@ -294,7 +294,7 @@ func TestReconcile(t *testing.T) {
 	}
 	activeKPAMinScale := func(g, w int32) *autoscalingv1alpha1.PodAutoscaler {
 		return kpa(
-			testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized,
+			testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 			withScales(g, w), WithReachabilityReachable,
 			withMinScale(defaultScale), WithPAStatusService(testRevision), WithPAMetricsService(privateSvc),
 			WithObservedGeneration(1),
@@ -332,7 +332,7 @@ func TestReconcile(t *testing.T) {
 		Name: "steady state",
 		Key:  key,
 		Objects: []runtime.Object{
-			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic,
+			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, WithScaleTargetNoScaleFailures,
 				markScaleTargetInitialized, WithPAMetricsService(privateSvc),
 				withScales(1, defaultScale), WithPAStatusService(testRevision), WithObservedGeneration(1)),
 			defaultSKS,
@@ -360,12 +360,12 @@ func TestReconcile(t *testing.T) {
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: kpa(testNamespace, testRevision, WithTraffic,
-				markScaleTargetInitialized, WithPASKSReady,
+				markScaleTargetInitialized, WithPASKSReady, WithScaleTargetNoScaleFailures,
 				WithPAMetricsService(privateSvc), withScales(1, defaultScale),
 				WithPAStatusService(testRevision), WithObservedGeneration(1)),
 		}, {
 			Object: kpa(testNamespace, testRevision, WithTraffic,
-				markScaleTargetInitialized, WithPASKSReady,
+				markScaleTargetInitialized, WithPASKSReady, WithScaleTargetNoScaleFailures,
 				WithPAMetricsService(privateSvc), withScales(1, defaultScale),
 				WithPAStatusService(testRevision), WithObservedGeneration(1)),
 		}},
@@ -414,12 +414,12 @@ func TestReconcile(t *testing.T) {
 		Name: "create metric",
 		Key:  key,
 		Objects: []runtime.Object{
-			kpa(testNamespace, testRevision, WithTraffic, markScaleTargetInitialized,
+			kpa(testNamespace, testRevision, WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				withScales(1, defaultScale), WithPAStatusService(testRevision)),
 			defaultSKS, defaultDeployment, defaultReady,
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: kpa(testNamespace, testRevision, WithTraffic, markScaleTargetInitialized,
+			Object: kpa(testNamespace, testRevision, WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				withScales(1, defaultScale), WithPASKSReady, WithPAStatusService(testRevision),
 				WithPAMetricsService(privateSvc), WithObservedGeneration(1)),
 		}},
@@ -430,7 +430,7 @@ func TestReconcile(t *testing.T) {
 		Name: "scale up deployment",
 		Key:  key,
 		Objects: []runtime.Object{
-			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized,
+			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				WithPAMetricsService(privateSvc), withScales(1, defaultScale), WithPAStatusService(testRevision),
 				WithObservedGeneration(1)),
 			defaultSKS,
@@ -503,7 +503,7 @@ func TestReconcile(t *testing.T) {
 			preciseReady...),
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: kpa(testNamespace, testRevision, withScales(defaultScale, defaultScale),
-				WithPASKSNotReady(""), WithTraffic, markScaleTargetInitialized,
+				WithPASKSNotReady(""), WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				WithPAMetricsService(privateSvc), WithPAStatusService(testRevision), WithObservedGeneration(1)),
 		}},
 	}, {
@@ -566,7 +566,7 @@ func TestReconcile(t *testing.T) {
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: kpa(testNamespace, testRevision, WithPASKSReady,
-				WithTraffic, markScaleTargetInitialized, withMinScale(2), WithPAMetricsService(privateSvc),
+				WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures, withMinScale(2), WithPAMetricsService(privateSvc),
 				withScales(1, defaultScale), WithPAStatusService(testRevision), WithReachabilityUnreachable,
 				WithObservedGeneration(1)),
 		}},
@@ -582,7 +582,7 @@ func TestReconcile(t *testing.T) {
 		}, makeReadyPods(2, testNamespace, testRevision)...),
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: kpa(testNamespace, testRevision, WithPASKSReady,
-				WithTraffic, markScaleTargetInitialized, withMinScale(2), WithPAMetricsService(privateSvc),
+				WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures, withMinScale(2), WithPAMetricsService(privateSvc),
 				withScales(2, defaultScale), WithPAStatusService(testRevision), WithReachabilityReachable,
 				WithObservedGeneration(1)),
 		}},
@@ -598,7 +598,7 @@ func TestReconcile(t *testing.T) {
 		}, makeReadyPods(2, testNamespace, testRevision)...),
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: kpa(testNamespace, testRevision, WithPASKSReady,
-				WithTraffic, markScaleTargetInitialized, withMinScale(2), WithPAMetricsService(privateSvc),
+				WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures, withMinScale(2), WithPAMetricsService(privateSvc),
 				withScales(2, defaultScale), WithPAStatusService(testRevision), WithReachabilityUnknown,
 				WithObservedGeneration(1)),
 		}},
@@ -761,7 +761,7 @@ func TestReconcile(t *testing.T) {
 			deploy(testNamespace, testRevision), defaultReady,
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: kpa(testNamespace, testRevision, markScaleTargetInitialized, withScales(1, 0),
+			Object: kpa(testNamespace, testRevision, markScaleTargetInitialized, WithScaleTargetNoScaleFailures, withScales(1, 0),
 				WithPASKSReady, WithPAMetricsService(privateSvc),
 				WithNoTraffic(noTrafficReason, "The target is not receiving traffic."),
 				WithPAStatusService(testRevision), WithPAMetricsService(privateSvc),
@@ -777,7 +777,7 @@ func TestReconcile(t *testing.T) {
 			decider(testNamespace, testRevision, 0 /* desiredScale */, 0 /* ebc */)),
 		Objects: []runtime.Object{
 			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic,
-				markScaleTargetInitialized, withScales(1, 1),
+				markScaleTargetInitialized, WithScaleTargetNoScaleFailures, withScales(1, 1),
 				WithPAStatusService(testRevision), WithPAMetricsService(privateSvc), WithObservedGeneration(1)),
 			defaultSKS,
 			metric(testNamespace, testRevision),
@@ -797,7 +797,7 @@ func TestReconcile(t *testing.T) {
 			deploy(testNamespace, testRevision), defaultReady,
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: kpa(testNamespace, testRevision, markScaleTargetInitialized, WithPASKSReady, WithPAMetricsService(privateSvc),
+			Object: kpa(testNamespace, testRevision, markScaleTargetInitialized, WithScaleTargetNoScaleFailures, WithPASKSReady, WithPAMetricsService(privateSvc),
 				WithNoTraffic("TimedOut", "The target could not be activated."), withScales(1, 0),
 				WithPAStatusService(testRevision), WithPAMetricsService(privateSvc),
 				WithObservedGeneration(1)),
@@ -873,7 +873,7 @@ func TestReconcile(t *testing.T) {
 			minScalePatch,
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: activatingKPAMinScale(underscale, markScaleTargetInitialized, WithPASKSReady),
+			Object: activatingKPAMinScale(underscale, markScaleTargetInitialized, WithScaleTargetNoScaleFailures, WithPASKSReady),
 		}},
 	}, {
 		// Scale to `minScale` and mark PA "active"
@@ -968,7 +968,7 @@ func TestReconcile(t *testing.T) {
 				-42 /* ebc */)),
 		Objects: []runtime.Object{
 			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic,
-				markScaleTargetInitialized, WithPAMetricsService(privateSvc),
+				markScaleTargetInitialized, WithScaleTargetNoScaleFailures, WithPAMetricsService(privateSvc),
 				withScales(1, defaultScale), WithPAStatusService(testRevision), WithObservedGeneration(1)),
 			defaultProxySKS,
 			metric(testNamespace, testRevision),
@@ -981,7 +981,7 @@ func TestReconcile(t *testing.T) {
 			decider(testNamespace, testRevision, defaultScale, /* desiredScale */
 				-18 /* ebc */)),
 		Objects: []runtime.Object{
-			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized,
+			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				WithPAMetricsService(privateSvc), withScales(1, defaultScale),
 				WithPAStatusService(testRevision), WithObservedGeneration(1)),
 			sks(testNamespace, testRevision, WithDeployRef(deployName), WithSKSReady,
@@ -1000,7 +1000,7 @@ func TestReconcile(t *testing.T) {
 			decider(testNamespace, testRevision, defaultScale, /* desiredScale */
 				1 /* ebc */)),
 		Objects: []runtime.Object{
-			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized,
+			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				WithPAMetricsService(privateSvc), withScales(1, defaultScale),
 				WithPAStatusService(testRevision), WithObservedGeneration(1)),
 			defaultProxySKS,
@@ -1055,7 +1055,7 @@ func TestReconcile(t *testing.T) {
 		}, makeReadyPods(20, testNamespace, testRevision)...),
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
 			Object: kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic,
-				markScaleTargetInitialized, withScales(20, 20), withInitialScale(20), WithReachabilityReachable,
+				markScaleTargetInitialized, WithScaleTargetNoScaleFailures, withScales(20, 20), withInitialScale(20), WithReachabilityReachable,
 				WithPAStatusService(testRevision), WithPAMetricsService(privateSvc), WithObservedGeneration(1),
 			),
 		}},
@@ -1081,7 +1081,7 @@ func TestReconcile(t *testing.T) {
 			}),
 		},
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: kpa(testNamespace, testRevision, markScaleTargetInitialized,
+			Object: kpa(testNamespace, testRevision, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				WithNoTraffic(noTrafficReason, "The target is not receiving traffic."),
 				withScales(0, -1), WithReachabilityReachable,
 				WithPAMetricsService(privateSvc), WithObservedGeneration(1),
@@ -1151,7 +1151,7 @@ func TestReconcile(t *testing.T) {
 			}),
 		}, makeReadyPods(2, testNamespace, testRevision)...),
 		WantStatusUpdates: []clientgotesting.UpdateActionImpl{{
-			Object: kpa(testNamespace, testRevision, WithTraffic, WithPASKSReady, markScaleTargetInitialized,
+			Object: kpa(testNamespace, testRevision, WithTraffic, WithPASKSReady, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				withScales(2, 2), WithReachabilityReachable, WithPAStatusService(testRevision),
 				WithPAMetricsService(privateSvc), WithObservedGeneration(1),
 			),
@@ -1163,7 +1163,7 @@ func TestReconcile(t *testing.T) {
 			decider(testNamespace, testRevision, defaultScale, /* desiredScale */
 				1 /* ebc */)),
 		Objects: []runtime.Object{
-			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized,
+			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				WithPAMetricsService(privateSvc), withScales(1, defaultScale),
 				WithPAStatusService(testRevision), WithObservedGeneration(1)),
 			defaultProxySKS,
@@ -1179,7 +1179,7 @@ func TestReconcile(t *testing.T) {
 			decider(testNamespace, testRevision, defaultScale, /* desiredScale */
 				1 /* ebc */)),
 		Objects: []runtime.Object{
-			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized,
+			kpa(testNamespace, testRevision, WithPASKSReady, WithTraffic, markScaleTargetInitialized, WithScaleTargetNoScaleFailures,
 				WithPAMetricsService(privateSvc), withScales(1, defaultScale),
 				WithPAStatusService(testRevision), WithObservedGeneration(1)),
 			defaultSKS,