Merge pull request llm-d-incubation#218 from osswangxining/refine-launcher-pod-deletion-logic

Refine launcher pod deletion logic

Author: osswangxining

pkg/controller/common/interface.go

@@ -0,0 +1,5 @@
+package common
+
+const (
+	RequesterAnnotationKey = "dual-pods.llm-d.ai/requester"
+)

pkg/controller/dual-pods/controller.go

@@ -29,7 +29,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/utils"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -44,6 +43,7 @@ import (
 	fmav1alpha1 "github.com/llm-d-incubation/llm-d-fast-model-actuation/api/fma/v1alpha1"
 	"github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/api"
 	genctlr "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/generic"
+	"github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/utils"
 	fmainformers "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/generated/informers/externalversions"
 	fmalisters "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/generated/listers/fma/v1alpha1"
 )

pkg/controller/generic/interface.go

@@ -26,4 +26,6 @@ const (
 	LauncherConfigNameLabelKey = "dual-pods.llm-d.ai/launcher-config-name"
 
 	NodeNameLabelKey = "dual-pods.llm-d.ai/node-name"
+
+	NominalHashAnnotationKey = "dual-pods.llm-d.ai/nominal"
 )

pkg/controller/launcher-populator/interface.go

@@ -22,10 +22,9 @@ import (
 	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"strings"
 
-	"github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/api"
-	dualpods "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/dual-pods"
-	"github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/utils"
+	"github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/common"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -41,7 +40,10 @@ import (
 	"k8s.io/klog/v2"
 
 	fmav1alpha1 "github.com/llm-d-incubation/llm-d-fast-model-actuation/api/fma/v1alpha1"
+	"github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/api"
+	dualpods "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/dual-pods"
 	genctlr "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/generic"
+	"github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/utils"
 	fmainformers "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/generated/informers/externalversions"
 	fmalisters "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/generated/listers/fma/v1alpha1"
 )
@@ -322,7 +324,7 @@ func (ctl *controller) reconcileLaunchersOnNode(ctx context.Context, key NodeLau
 }
 
 // getCurrentLaunchersOnNode returns launcher pods for a specific config on a specific node
-func (ctl *controller) getCurrentLaunchersOnNode(ctx context.Context, key NodeLauncherKey) ([]corev1.Pod, error) {
+func (ctl *controller) getCurrentLaunchersOnNode(ctx context.Context, key NodeLauncherKey) ([]*corev1.Pod, error) {
 	launcherLabels := map[string]string{
 		ComponentLabelKey:          LauncherComponentLabelValue,
 		LauncherConfigNameLabelKey: key.LauncherConfigName,
@@ -334,13 +336,7 @@ func (ctl *controller) getCurrentLaunchersOnNode(ctx context.Context, key NodeLa
 		return nil, fmt.Errorf("failed to list pods with launcher labels: %w", err)
 	}
 
-	// Filter pods that are on the specified node
-	var filteredPods []corev1.Pod
-	for _, pod := range pods {
-		filteredPods = append(filteredPods, *pod)
-	}
-
-	return filteredPods, nil
+	return pods, nil
 }
 
 // createLaunchers creates the specified number of launcher pods on a node
@@ -377,19 +373,73 @@ func (ctl *controller) createLaunchers(ctx context.Context, node corev1.Node, ke
 }
 
 // deleteExcessLaunchers deletes the specified number of launcher pods
-func (ctl *controller) deleteExcessLaunchers(ctx context.Context, launchers []corev1.Pod, count int) error {
+func (ctl *controller) deleteExcessLaunchers(ctx context.Context, launchers []*corev1.Pod, count int) error {
 	logger := klog.FromContext(ctx)
-	// Delete the specified number of launcher pods (starting from the end)
+
+	deletedCount := 0
 	for i := 0; i < count && i < len(launchers); i++ {
 		pod := launchers[len(launchers)-1-i]
-		if err := ctl.coreclient.Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{}); err != nil {
+		isBound, requesterPodName := ctl.isLauncherBoundToServerRequestingPod(pod)
+		if isBound {
+			logger.V(5).Info("Skipping deletion of launcher pod as it is bound to a server-requesting pod",
+				"pod", pod.Name, "server-requesting pod", requesterPodName)
+			continue
+		}
+
+		if err := ctl.coreclient.Pods(pod.Namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{
+			Preconditions: &metav1.Preconditions{
+				ResourceVersion: &pod.ResourceVersion,
+			},
+		}); err != nil {
+			if apierrors.IsNotFound(err) {
+				logger.Info("Launcher pod already deleted", "pod", pod.Name)
+				deletedCount++ // Count as deletion target achieved
+				continue
+			}
+			if apierrors.IsConflict(err) {
+				logger.Info("Launcher pod version conflict, skipping deletion",
+					"pod", pod.Name, "error", err)
+				continue
+			}
 			return fmt.Errorf("failed to delete launcher pod %s: %w", pod.Name, err)
 		}
 		logger.Info("Deleted launcher pod", "pod", pod.Name)
+		deletedCount++
+	}
+
+	if deletedCount < count {
+		logger.Info("Fewer launcher pods were deleted than requested due to bound pods or concurrent changes",
+			"requested", count,
+			"deleted", deletedCount,
+			"skipped", count-deletedCount)
+	} else {
+		logger.Info("Deleted unbound launcher pods",
+			"deleted", deletedCount)
 	}
+
 	return nil
 }
 
+// isLauncherBoundToServerRequestingPod checks if the launcher pod is bound to any server-requesting pod
+func (ctl *controller) isLauncherBoundToServerRequestingPod(launcherPod *corev1.Pod) (bool, string) {
+	// Check if the launcher pod has annotations indicating assignment to a server-requesting pod
+	requesterAnnotationValue, exists := launcherPod.Annotations[common.RequesterAnnotationKey]
+	if !exists {
+		return false, ""
+	}
+
+	// Verify the format of the annotation value: should be "UID name"
+	parts := strings.Split(requesterAnnotationValue, " ")
+	if len(parts) != 2 {
+		return false, "" // Invalid format
+	}
+
+	// Optionally verify that the referenced pod actually exists
+	// @TODO if need, we can append the check logic in further PR
+
+	return true, parts[1]
+}
+
 // buildPodFromTemplate creates a pod from a template and assigns it to a node
 func (ctl *controller) buildPodFromTemplate(template corev1.PodTemplateSpec, key NodeLauncherKey) (*corev1.Pod, error) {
 	pod := &corev1.Pod{
@@ -421,7 +471,7 @@ func (ctl *controller) buildPodFromTemplate(template corev1.PodTemplateSpec, key
 	if pod.Annotations == nil {
 		pod.Annotations = make(map[string]string)
 	}
-	pod.Annotations = dualpods.MapSet(pod.Annotations, genctlr.NominalHashAnnotationKey, nominalHash)
+	pod.Annotations = dualpods.MapSet(pod.Annotations, NominalHashAnnotationKey, nominalHash)
 
 	cIdx, serverPort, err := utils.GetInferenceServerPort(pod)
 	if err != nil {