fix: do not update status.replicas until pod operation succeeds (#620)

Signed-off-by: Derek Wang <whynowy@gmail.com>
numaproj · Apr 3, 2023 · 90ca505 · 90ca505
1 parent 99db568
commit 90ca505
Showing 1 changed file with 9 additions and 9 deletions.
diff --git a/pkg/reconciler/vertex/controller.go b/pkg/reconciler/vertex/controller.go
@@ -156,15 +156,6 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 		}
 	}
 
-	currentReplicas := int(vertex.Status.Replicas)
-	if currentReplicas != desiredReplicas || vertex.Status.Selector == "" {
-		log.Infow("Replicas changed", "currentReplicas", currentReplicas, "desiredReplicas", desiredReplicas)
-		vertex.Status.Replicas = uint32(desiredReplicas)
-		vertex.Status.LastScaledAt = metav1.Time{Time: time.Now()}
-	}
-	selector, _ := labels.Parse(dfv1.KeyPipelineName + "=" + vertex.Spec.PipelineName + "," + dfv1.KeyVertexName + "=" + vertex.Spec.Name)
-	vertex.Status.Selector = selector.String()
-
 	pipeline := &dfv1.Pipeline{}
 	if err := r.client.Get(ctx, types.NamespacedName{Namespace: vertex.Namespace, Name: vertex.Spec.PipelineName}, pipeline); err != nil {
 		log.Errorw("Failed to get pipeline object", zap.Error(err))
@@ -250,6 +241,15 @@ func (r *vertexReconciler) reconcile(ctx context.Context, vertex *dfv1.Vertex) (
 		}
 	}
 
+	currentReplicas := int(vertex.Status.Replicas)
+	if currentReplicas != desiredReplicas || vertex.Status.Selector == "" {
+		log.Infow("Replicas changed", "currentReplicas", currentReplicas, "desiredReplicas", desiredReplicas)
+		vertex.Status.Replicas = uint32(desiredReplicas)
+		vertex.Status.LastScaledAt = metav1.Time{Time: time.Now()}
+	}
+	selector, _ := labels.Parse(dfv1.KeyPipelineName + "=" + vertex.Spec.PipelineName + "," + dfv1.KeyVertexName + "=" + vertex.Spec.Name)
+	vertex.Status.Selector = selector.String()
+
 	// create services
 	existingSvcs, err := r.findExistingServices(ctx, vertex)
 	if err != nil {