executor/util: wait for the deployment controller to observe the scale request

WaitForDeployment returned as soon as AvailableReplicas was high enough, without checking that the status belonged to the current spec. UpdateScale only writes the spec, so a Get issued right after it can still carry the previous generation's status, counting the pod that is being torn down.

Gate AvailableReplicas behind ObservedGeneration, the same condition WaitForDeploymentRollout already uses in the integration framework.

pkg/executor/util/deployment.go

@@ -62,9 +62,25 @@ func nextWaitPollDelay(cur time.Duration) time.Duration {
 	return min(cur*3/2, waitPollMaxDelay)
 }
 
-// WaitForDeployment polls the deployment until it has at least replicas
-// AvailableReplicas, or until specializationTimeout seconds elapse. Shared by
-// the newdeploy and container managers. specializationTimeout is floored at
+// deploymentAvailable reports whether depl has at least replicas available,
+// as observed by the deployment controller for the spec currently stored.
+//
+// The generation gate matters on the scale-from-zero path. UpdateScale only
+// writes the spec and bumps metadata.generation; status is written later, by
+// the deployment controller. A Get issued right after ScaleDeployment can
+// therefore still carry the status of the previous generation, which counts
+// the pod the idle reaper is tearing down, so AvailableReplicas alone would
+// report a serving deployment while the Service has no endpoint. Until
+// ObservedGeneration catches up, the status says nothing about this scale
+// request. Same gate as WaitForDeploymentRollout in the integration framework.
+func deploymentAvailable(depl *appsv1.Deployment, replicas int32) bool {
+	return depl.Status.ObservedGeneration >= depl.Generation && depl.Status.AvailableReplicas >= replicas
+}
+
+// WaitForDeployment polls the deployment until the deployment controller has
+// observed the current spec and reports at least replicas AvailableReplicas,
+// or until specializationTimeout seconds elapse. Shared by the newdeploy and
+// container managers. specializationTimeout is floored at
 // fv1.DefaultSpecializationTimeOut. AvailableReplicas is used in preference to
 // ReadyReplicas since the pods may not be able to serve network traffic yet.
 func WaitForDeployment(ctx context.Context, kubeClient kubernetes.Interface, logger logr.Logger, depl *appsv1.Deployment, replicas int32, specializationTimeout int) (latestDepl *appsv1.Deployment, err error) {
@@ -92,7 +108,7 @@ func WaitForDeployment(ctx context.Context, kubeClient kubernetes.Interface, log
 			return nil, err
 		}
 		// TODO check for imagePullerror
-		if latestDepl.Status.AvailableReplicas >= replicas {
+		if deploymentAvailable(latestDepl, replicas) {
 			otelUtils.SpanTrackEvent(ctx, "deploymentAvailable", otelUtils.GetAttributesForDeployment(latestDepl)...)
 			return latestDepl, err
 		}

pkg/executor/util/deployment_test.go

@@ -194,6 +194,35 @@ func TestWaitForDeployment(t *testing.T) {
 		assert.GreaterOrEqual(t, got.Status.AvailableReplicas, int32(1))
 	})
 
+	t.Run("ignores a status the controller has not caught up with", func(t *testing.T) {
+		t.Parallel()
+		// Scaled in from one replica to zero and back up: the spec is at
+		// generation 2, while the status still describes generation 1 and
+		// counts the pod the idle reaper is tearing down.
+		one := int32(1)
+		scaled := &appsv1.Deployment{
+			Namespace: ns, Name: "scaled-up", Generation: 2,
+			Spec:   appsv1.DeploymentSpec{Replicas: &one},
+			Status: appsv1.DeploymentStatus{ObservedGeneration: 1, AvailableReplicas: 1},
+		}
+		client := fake.NewClientset(scaled)
+		var gets int
+		client.PrependReactor("get", "deployments", func(k8stesting.Action) (bool, runtime.Object, error) {
+			latest := scaled.DeepCopy()
+			gets++
+			if gets > 1 {
+				latest.Status.ObservedGeneration = latest.Generation
+			}
+			return true, latest, nil
+		})
+
+		got, err := WaitForDeployment(t.Context(), client, loggerfactory.GetLogger(), scaled, 1, 1)
+		require.NoError(t, err)
+		require.NotNil(t, got)
+		assert.Equal(t, got.Generation, got.Status.ObservedGeneration,
+			"returned on a status that predates the scale request")
+	})
+
 	t.Run("surfaces a get error", func(t *testing.T) {
 		t.Parallel()
 		client := fake.NewClientset()