fix: lastScaledAt not updated during autoscaling down (#2323)

Author: whynowy

pkg/reconciler/monovertex/controller.go

@@ -206,7 +206,8 @@ func (mr *monoVertexReconciler) orchestratePods(ctx context.Context, monoVtx *df
 	if err := mr.cleanUpPodsFromTo(ctx, monoVtx, desiredReplicas, math.MaxInt); err != nil {
 		return fmt.Errorf("failed to clean up mono vertex pods [%v, ∞): %w", desiredReplicas, err)
 	}
-	if currentReplicas := int(monoVtx.Status.Replicas); currentReplicas > desiredReplicas {
+	currentReplicas := int(monoVtx.Status.Replicas)
+	if currentReplicas > desiredReplicas {
 		monoVtx.Status.Replicas = uint32(desiredReplicas)
 	}
 	updatedReplicas := int(monoVtx.Status.UpdatedReplicas)
@@ -286,7 +287,6 @@ func (mr *monoVertexReconciler) orchestratePods(ctx context.Context, monoVtx *df
 		}
 	}
 
-	currentReplicas := int(monoVtx.Status.Replicas)
 	if currentReplicas != desiredReplicas {
 		log.Infow("MonoVertex replicas changed", "currentReplicas", currentReplicas, "desiredReplicas", desiredReplicas)
 		mr.recorder.Eventf(monoVtx, corev1.EventTypeNormal, "ReplicasScaled", "Replicas changed from %d to %d", currentReplicas, desiredReplicas)

pkg/reconciler/vertex/controller.go

@@ -220,7 +220,8 @@ func (r *vertexReconciler) orchestratePods(ctx context.Context, vertex *dfv1.Ver
 	if err := r.cleanUpPodsFromTo(ctx, vertex, desiredReplicas, math.MaxInt); err != nil {
 		return fmt.Errorf("failed to clean up vertex pods [%v, ∞): %w", desiredReplicas, err)
 	}
-	if currentReplicas := int(vertex.Status.Replicas); currentReplicas > desiredReplicas {
+	currentReplicas := int(vertex.Status.Replicas)
+	if currentReplicas > desiredReplicas {
 		vertex.Status.Replicas = uint32(desiredReplicas)
 	}
 	updatedReplicas := int(vertex.Status.UpdatedReplicas)
@@ -300,7 +301,6 @@ func (r *vertexReconciler) orchestratePods(ctx context.Context, vertex *dfv1.Ver
 		}
 	}
 
-	currentReplicas := int(vertex.Status.Replicas)
 	if currentReplicas != desiredReplicas {
 		log.Infow("Pipeline Vertex replicas changed", "currentReplicas", currentReplicas, "desiredReplicas", desiredReplicas)
 		r.recorder.Eventf(vertex, corev1.EventTypeNormal, "ReplicasScaled", "Replicas changed from %d to %d", currentReplicas, desiredReplicas)