Repository: vllm-project/aibrix
File: pkg/controller/podset/podset_controller.go (Go)

Code before:
}

func (r *PodSetReconciler) reconcilePods(ctx context.Context, podSet *orchestrationv1alpha1.PodSet) error {
	// Get existing pods
	podList, err := r.getPodsForPodSet(ctx, podSet)
	if err != nil {
		return fmt.Errorf("failed to get pods for PodSet: %w", err)
	}

	activePods := filterActivePods(podList.Items)
	currentPodCount := len(activePods)
	desiredPodCount := int(podSet.Spec.PodGroupSize)

	if currentPodCount < desiredPodCount {
		// Need Create missing pods
		switch podSet.Spec.RecoveryPolicy {
		case orchestrationv1alpha1.ReplaceUnhealthy:
			// Proactively find and delete unhealthy pods based on restart count.
			var podsToDelete []*corev1.Pod
			for i := range activePods {
				pod := activePods[i] // Create a local copy for safe referencing
				isUnhealthy := false

				// Check for container restarts.
				for _, containerStatus := range pod.Status.ContainerStatuses {
					if containerStatus.RestartCount > 0 {
						isUnhealthy = true
						klog.InfoS("Marking pod as unhealthy due to container restarts", "pod", pod.Name, "restarts", containerStatus.RestartCount)
						break // One unhealthy container is enough.
					}
				}

				if isUnhealthy {
					podsToDelete = append(podsToDelete, &pod)
				}
			}

			// If any unhealthy pods are found, delete them and return.
			// Replacement creation will occur in the next reconciliation loop.
			if len(podsToDelete) > 0 {
				klog.InfoS("Found unhealthy pods to be deleted", "count", len(podsToDelete), "podset", podSet.Name)
				for _, pod := range podsToDelete {
					if err := r.Delete(ctx, pod); err != nil {
						klog.ErrorS(err, "Failed to delete unhealthy pod", "pod", pod.Name)
						r.EventRecorder.Eventf(podSet, corev1.EventTypeWarning, "FailedDelete", "Failed to delete unhealthy pod %s: %v", pod.Name, err)
						continue
					}
					r.EventRecorder.Eventf(podSet, corev1.EventTypeNormal, "DeletingUnhealthyPod", "Deleting unhealthy pod %s due to container restarts.", pod.Name)
				}
				// Return nil to trigger a requeue. The next reconcile will handle the missing pods.
				return nil

Review comment (bug):
I'm curious about the logic `containerStatus.RestartCount > 0`.
The activePods are already in a Ready status, meaning they are running and can serve. 

beside, restart count could result from:

- Temporary env issues (e.g., network,)
- Sidecar containers restarting
- Past crashes that have already been recovered


What happens if a vllm pod is restarted but is already running?

Code after:
}

func (r *PodSetReconciler) reconcilePods(ctx context.Context, podSet *orchestrationv1alpha1.PodSet) error {
	// Get existing pods
	podList, err := r.getPodsForPodSet(ctx, podSet)
	if err != nil {
		return fmt.Errorf("failed to get pods for PodSet: %w", err)
	}

	activePods := filterActivePods(podList.Items)
	currentPodCount := len(activePods)
	desiredPodCount := int(podSet.Spec.PodGroupSize)

	if currentPodCount < desiredPodCount {
		switch podSet.Spec.RecoveryPolicy {
		case orchestrationv1alpha1.ReplaceUnhealthy:
			return r.handleReplaceUnhealthy(ctx, podSet, activePods, currentPodCount, desiredPodCount)
		case orchestrationv1alpha1.RecreatePodRecreateStrategy:
			return r.handleRecreateStrategy(ctx, podSet, activePods, desiredPodCount)
		}
	} else if currentPodCount > desiredPodCount {
		return r.handleScaleDown(ctx, podSet, activePods, currentPodCount, desiredPodCount)
	}

	return nil
}

func (r *PodSetReconciler) handleReplaceUnhealthy(ctx context.Context, podSet *orchestrationv1alpha1.PodSet,
	activePods []corev1.Pod, currentPodCount, desiredPodCount int) error {

	// Proactively find and delete unhealthy pods
	var podsToDelete []*corev1.Pod
	for i := range activePods {
		pod := activePods[i]
		for _, cs := range pod.Status.ContainerStatuses {
			if cs.RestartCount > 0 {
				klog.InfoS("Marking pod as unhealthy due to container restarts", "pod", pod.Name, "restarts", cs.RestartCount)
				podsToDelete = append(podsToDelete, &pod)
				break
			}
		}
	}
	if len(podsToDelete) > 0 {
		klog.InfoS("Found unhealthy pods to be deleted", "count", len(podsToDelete), "podset", podSet.Name)
		for _, pod := range podsToDelete {
			if err := r.Delete(ctx, pod); err != nil {
				klog.ErrorS(err, "Failed to delete unhealthy pod", "pod", pod.Name)
				continue
			}
		}
		r.EventRecorder.Eventf(podSet, corev1.EventTypeNormal, "DeletingUnhealthyPods", "Deleting %d unhealthy pods due to container restarts.", len(podsToDelete))