Repository: vllm-project/aibrix
File: pkg/controller/rayclusterfleet/recreate.go (Go)

Code before:
			continue
		}
		scaledRS, updatedRS, err := r.scaleReplicaSetAndRecordEvent(ctx, rs, 0, deployment)
		if err != nil {
			return false, err
		}
		if scaledRS {
			oldRSs[i] = updatedRS
			scaled = true
		}
	}
	return scaled, nil
}

// oldPodsRunning returns whether there are old pods running or any of the old ReplicaSets thinks that it runs pods.
func oldPodsRunning(newRS *orchestrationv1alpha1.RayClusterReplicaSet, oldRSs []*orchestrationv1alpha1.RayClusterReplicaSet, podMap map[types.UID][]*rayclusterv1.RayCluster) bool {
	if oldPods := util.GetActualReplicaCountForReplicaSets(oldRSs); oldPods > 0 {
		return true
	}
	for rsUID, podList := range podMap {
		// If the pods belong to the new ReplicaSet, ignore.
		if newRS != nil && newRS.UID == rsUID {
			continue
		}
		for _, pod := range podList {
			if !pod.DeletionTimestamp.IsZero() {
				return true
			}
		}
	}
	return false
}

// scaleUpNewReplicaSetForRecreate scales up new replica set when deployment strategy is "Recreate".
func (r *RayClusterFleetReconciler) scaleUpNewReplicaSetForRecreate(ctx context.Context, newRS *orchestrationv1alpha1.RayClusterReplicaSet, deployment *orchestrationv1alpha1.RayClusterFleet) (bool, error) {
	scaled, _, err := r.scaleReplicaSetAndRecordEvent(ctx, newRS, *(deployment.Spec.Replicas), deployment)
	return scaled, err
}


Review comment (bug):
In a Deployment, `oldPodsRunning` is used to check if all the Pods from the old ReplicaSet have already released their resources. If so, we can create a new ReplicaSet to update, following the recreate strategy.

The code below is copied from `https://github.com/kubernetes/kubernetes`(we can see that Pod in status `v1.PodFailed` and `v1.PodSucceeded` do not occupy any resources and thus will pass the check): 
```go
for _, pod := range podList {
    switch pod.Status.Phase {
    case v1.PodFailed, v1.PodSucceeded:
        // Don't count pods in terminal state.
        continue
    case v1.PodUnknown:
        // v1.PodUnknown is a deprecated status.
        // This logic is kept for backward compatibility.
        // This used to happen in situations like when the node is temporarily disconnected from the cluster.
        // If we can't be sure that the pod is not running, we have to count it.
        return true
    default:
        // Pod is not in terminal phase.
        return true
    }
}
```

Similarly, here, we check whether the old RayClusterReplicaSet still has Rayclusters occupying resources.

Unlike Pods, Ray clusters do not have a terminal status to indicate they have finished and will no longer occupy resources. Therefore, the best approach is to detect if the Ray cluster has been marked for deletion (`DeletionTimestamp`). If so, the KubeRay operator will eventually delete it.

Another tricky issue is "unhealthy" or "failed" Ray clusters will not automatically deleted and will continue to occupy resources. However, that is a separate issue. For now, we just ensure that all countable (healthy) Rayclusters have been marked as deleted for the old ReplicaSet.

Code after:
		if *(rs.Spec.Replicas) == 0 {
			continue
		}
		scaledRS, updatedRS, err := r.scaleReplicaSetAndRecordEvent(ctx, rs, 0, deployment)
		if err != nil {
			return false, err
		}
		if scaledRS {
			oldRSs[i] = updatedRS
			scaled = true
		}
	}
	return scaled, nil
}

// oldPodsRunning returns whether there are old pods running or any of the old ReplicaSets thinks that it runs pods.
func oldPodsRunning(newRS *orchestrationv1alpha1.RayClusterReplicaSet, oldRSs []*orchestrationv1alpha1.RayClusterReplicaSet, podMap map[types.UID][]*rayclusterv1.RayCluster) bool {
	if oldPods := util.GetActualReplicaCountForReplicaSets(oldRSs); oldPods > 0 {
		return true
	}
	for rsUID, podList := range podMap {
		// If the pods belong to the new ReplicaSet, ignore.
		if newRS != nil && newRS.UID == rsUID {
			continue
		}
		for _, pod := range podList {
			if !pod.DeletionTimestamp.IsZero() && rayclusterutil.IsRayClusterReady(pod) {
				return true
			}
		}
	}
	return false
}

// scaleUpNewReplicaSetForRecreate scales up new replica set when deployment strategy is "Recreate".
func (r *RayClusterFleetReconciler) scaleUpNewReplicaSetForRecreate(ctx context.Context, newRS *orchestrationv1alpha1.RayClusterReplicaSet, deployment *orchestrationv1alpha1.RayClusterFleet) (bool, error) {
	scaled, _, err := r.scaleReplicaSetAndRecordEvent(ctx, newRS, *(deployment.Spec.Replicas), deployment)
	return scaled, err
}