React on externalized instance state change (#414)

* React on externalized instance state change

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* Rename type launcherPodItem

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* Don't delete requester if deletion of a stopped instance fails

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* Unified stopped-instance cleanup in syncLauncherInstances

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* One TODO is done, another to be done by launcher populator

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* Minor simplifications; some changes to comments

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* Improve error handling for unboundLauncherPodItem.process

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* Introduce IsInstanceNotFoundError

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* Clean up logging for infSvrItem.process

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* Retry when failed to delete a stopped instance

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

* Return retry rather than true when err is outstanding from syncLauncherInstances

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

---------

Signed-off-by: Jun Duan <jun.duan.phd@outlook.com>

Author: waltforme

pkg/controller/dual-pods/controller.go

@@ -332,7 +332,7 @@ type infSvrItem struct {
 	RequesterName string
 }
 
-type launcherPodItem struct {
+type unboundLauncherPodItem struct {
 	LauncherPodName string
 	NodeName        string
 }
@@ -347,6 +347,7 @@ const (
 	infSvrItemBoundProvider infSvrItemType = "bound_provider"
 	// infSvrItemUnboundLauncherBasedProvider is for a server-providing Pod that
 	// is launcher-based and not bound to any server-requesting Pods.
+	// Note that technically an unbound launcher-based server-providing Pod is not part of any inference server (yet/anymore).
 	infSvrItemUnboundLauncherBasedProvider infSvrItemType = "unbound_launcher_based_provider"
 	// infSvrItemDontCare is not a real infSvrItemType but only a placeholder
 	// saying the corresponding infSvrItem is not relevant to the controller.
@@ -431,10 +432,10 @@ func (ctl *controller) OnAdd(obj any, isInInitialList bool) {
 				return
 			}
 			nd := ctl.getNodeData(nodeName)
-			launcherPodItem := launcherPodItem{LauncherPodName: typed.Name, NodeName: nodeName}
+			unboundLauncher := unboundLauncherPodItem{LauncherPodName: typed.Name, NodeName: nodeName}
 			ctl.enqueueLogger.V(5).Info("Enqueuing launcher reference due to notification of add",
 				"nodeName", nodeName, "launcherPod", typed.Name, "isInInitialList", isInInitialList, "resourceVersion", typed.ResourceVersion)
-			nd.add(launcherPodItem)
+			nd.add(unboundLauncher)
 			ctl.Queue.Add(nodeItem{nodeName})
 		} else {
 			nodeName := typed.Spec.NodeName
@@ -484,10 +485,10 @@ func (ctl *controller) OnUpdate(prev, obj any) {
 				return
 			}
 			nd := ctl.getNodeData(nodeName)
-			launcherPodItem := launcherPodItem{LauncherPodName: typed.Name, NodeName: nodeName}
+			unboundLauncher := unboundLauncherPodItem{LauncherPodName: typed.Name, NodeName: nodeName}
 			ctl.enqueueLogger.V(5).Info("Enqueuing launcher reference due to notification of update",
 				"nodeName", nodeName, "launcherPod", typed.Name, "resourceVersion", typed.ResourceVersion)
-			nd.add(launcherPodItem)
+			nd.add(unboundLauncher)
 			ctl.Queue.Add(nodeItem{nodeName})
 		} else {
 			nodeName := typed.Spec.NodeName
@@ -540,10 +541,10 @@ func (ctl *controller) OnDelete(obj any) {
 				return
 			}
 			nd := ctl.getNodeData(nodeName)
-			launcherPodItem := launcherPodItem{LauncherPodName: typed.Name, NodeName: nodeName}
+			unboundLauncher := unboundLauncherPodItem{LauncherPodName: typed.Name, NodeName: nodeName}
 			ctl.enqueueLogger.V(5).Info("Enqueuing launcher reference due to notification of delete",
 				"nodeName", nodeName, "launcherPod", typed.Name, "resourceVersion", typed.ResourceVersion)
-			nd.add(launcherPodItem)
+			nd.add(unboundLauncher)
 			ctl.Queue.Add(nodeItem{nodeName})
 		} else {
 			nodeName := typed.Spec.NodeName

pkg/controller/dual-pods/inference-server.go

@@ -57,6 +57,12 @@ type nodeItem struct {
 	NodeName string
 }
 
+type launcherSyncResult struct {
+	instances                 *AllInstancesState
+	deletedStoppedInstanceIDs sets.Set[string]
+	failedStoppedInstanceErrs map[string]error
+}
+
 func (ni nodeItem) process(ctx context.Context, ctl *controller) (error, bool) {
 	logger := klog.FromContext(ctx).WithValues("node", ni.NodeName)
 	ctx = klog.NewContext(ctx, logger)
@@ -85,11 +91,11 @@ func (ni nodeItem) process(ctx context.Context, ctl *controller) (error, bool) {
 	return nil, retries > 0
 }
 
-func (item launcherPodItem) process(ctx context.Context, ctl *controller, nodeDat *nodeData) (error, bool) {
+func (item unboundLauncherPodItem) process(ctx context.Context, ctl *controller, nodeDat *nodeData) (error, bool) {
 	logger := klog.FromContext(ctx).WithValues("launcherPod", item.LauncherPodName, "node", item.NodeName)
 	ctx = klog.NewContext(ctx, logger)
 
-	_, err := ctl.podLister.Pods(ctl.namespace).Get(item.LauncherPodName)
+	launcherPod, err := ctl.podLister.Pods(ctl.namespace).Get(item.LauncherPodName)
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			logger.V(2).Info("Launcher pod deleted, cleaning up launcher data")
@@ -100,8 +106,15 @@ func (item launcherPodItem) process(ctx context.Context, ctl *controller, nodeDa
 		return err, true
 	}
 
+	// Sync launcher instances to keep internal state fresh and clean up stopped instances.
+	_, syncErr, syncRetry := ctl.syncLauncherInstances(ctx, nodeDat, launcherPod)
+
 	ctl.enqueueUnboundInfSvrItemsOnNode(ctx, item.NodeName, fmt.Sprintf("launcher pod %s changed", item.LauncherPodName))
-	return nil, false
+
+	if syncErr != nil {
+		return fmt.Errorf("failed to sync launcher instances: %w", syncErr), syncRetry
+	}
+	return nil, syncRetry
 }
 
 func (item infSvrItem) process(urCtx context.Context, ctl *controller, nodeDat *nodeData) (error, bool) {
@@ -350,6 +363,51 @@ func (item infSvrItem) process(urCtx context.Context, ctl *controller, nodeDat *
 				return fmt.Errorf("unable to wake up server because port not known: %w", err), true
 			}
 		}
+		// For launcher-based providers, check whether the bound instance is still alive.
+		// The sidecar notifier updates the Pod annotation when instance status changes,
+		// which triggers reconciliation through the informer.
+		if launcherBased && serverDat.InstanceID != "" && providingPod.Status.PodIP != "" {
+			syncResult, err, retry := ctl.syncLauncherInstances(ctx, nodeDat, providingPod)
+			if err != nil || retry {
+				if err != nil {
+					return fmt.Errorf("failed to sync launcher instances for bound launcher Pod: %w", err), retry
+				}
+				return nil, true
+			}
+
+			_, instancePresent := findInstanceState(syncResult.instances.Instances, serverDat.InstanceID)
+			if delErr, failedCleanup := syncResult.failedStoppedInstanceErrs[serverDat.InstanceID]; failedCleanup {
+				return fmt.Errorf("failed to delete stopped instance %q from launcher: %w", serverDat.InstanceID, delErr), true
+			}
+			if _, deletedStopped := syncResult.deletedStoppedInstanceIDs[serverDat.InstanceID]; deletedStopped || !instancePresent {
+				if deletedStopped {
+					logger.V(2).Info("Deleted stopped bound instance from launcher during sync")
+				} else {
+					logger.V(2).Info("Bound instance not found in launcher after sync, treating as deleted")
+				}
+				// Mark as sleeping so that ensureUnbound (called during requester deletion)
+				// does not attempt to POST /sleep on the dead instance.
+				// The instance process is dead — this is not a real sleeping state,
+				// but it prevents ensureUnbound from hitting a dead endpoint and retrying forever.
+				serverDat.Sleeping = ptr.To(true)
+				// Delete the server-requesting Pod.
+				// This is analogous to the direct-provider case where a providing Pod's
+				// deletion is reflected to deletion of the server-requesting Pod.
+				// The ReplicaSet will recreate the requesting Pod, triggering a fresh bind.
+				err = podOps.Delete(ctx, requestingPod.Name, metav1.DeleteOptions{
+					PropagationPolicy: ptr.To(metav1.DeletePropagationBackground),
+					Preconditions:     &metav1.Preconditions{UID: &item.UID, ResourceVersion: &requestingPod.ResourceVersion}})
+				if err == nil {
+					logger.V(2).Info("Requested deletion of server-requesting Pod because bound instance stopped")
+				} else if apierrors.IsGone(err) || apierrors.IsNotFound(err) {
+					logger.V(5).Info("The server-requesting Pod is already gone")
+				} else {
+					return fmt.Errorf("failed to delete server-requesting Pod for stopped instance: %w", err), true
+				}
+				serverDat.RequesterDeleteRequested = true
+				return nil, false
+			}
+		}
 		if serverDat.Sleeping == nil {
 			sleeping, err := ctl.querySleeping(ctx, providingPod, serverPort)
 			if err != nil {
@@ -469,7 +527,6 @@ func (item infSvrItem) process(urCtx context.Context, ctl *controller, nodeDat *
 	lcName := isc.Spec.LauncherConfigName
 	lc, err := ctl.lcLister.LauncherConfigs(ctl.namespace).Get(lcName)
 	if err != nil {
-		// TODO(waltforme): introduce the 'enqueue requesters by launcherconfigs' logic to the controller
 		return ctl.ensureReqStatus(ctx, requestingPod, serverDat,
 			fmt.Sprintf("failed to get LauncherConfig %q: %v", lcName, err),
 		)
@@ -561,7 +618,6 @@ func (item infSvrItem) process(urCtx context.Context, ctl *controller, nodeDat *
 	}
 	// Sleeper budget is met. Make a new launcher Pod.
 
-	// TODO(waltforme): introduce the 'enqueue requesters by launcher pods' logic to the controller.
 	echo, err := podOps.Create(ctx, desiredLauncherPod, metav1.CreateOptions{})
 	if err != nil {
 		errMsg := err.Error()
@@ -621,11 +677,12 @@ func (ctl *controller) selectBestLauncherPod(
 			continue
 		}
 
-		insts, err, retry := ctl.syncLauncherInstances(ctx, nodeDat, launcherPod)
+		syncResult, err, retry := ctl.syncLauncherInstances(ctx, nodeDat, launcherPod)
 		if err != nil || retry {
 			somePodsNotReady = true
 			continue
 		}
+		insts := syncResult.instances
 
 		// Check if this launcher has a sleeping instance matching the iscHash
 		hasSleepingInstance := false
@@ -650,7 +707,7 @@ func (ctl *controller) selectBestLauncherPod(
 				hasPortConflict = true
 				break
 			}
-			if inst.InstanceID == iscHash {
+			if inst.InstanceID == iscHash && inst.Status != InstanceStatusStopped {
 				hasSleepingInstance = true
 			}
 		}
@@ -1324,10 +1381,19 @@ var coreScheme *k8sruntime.Scheme
 var codecFactory k8sserializer.CodecFactory
 var podDecoder k8sruntime.Decoder
 
+func findInstanceState(insts []InstanceState, instanceID string) (*InstanceState, bool) {
+	for idx := range insts {
+		if insts[idx].InstanceID == instanceID {
+			return &insts[idx], true
+		}
+	}
+	return nil, false
+}
+
 // syncLauncherInstances queries the launcher pod for its current instances,
 // updates the controller's internal launcherData state, and returns the fresh
 // launcher response used for the update.
-func (ctl *controller) syncLauncherInstances(ctx context.Context, nodeDat *nodeData, launcherPod *corev1.Pod) (*AllInstancesState, error, bool) {
+func (ctl *controller) syncLauncherInstances(ctx context.Context, nodeDat *nodeData, launcherPod *corev1.Pod) (*launcherSyncResult, error, bool) {
 	logger := klog.FromContext(ctx)
 
 	if launcherPod.Status.PodIP == "" || !utils.IsPodReady(launcherPod) {
@@ -1350,14 +1416,43 @@ func (ctl *controller) syncLauncherInstances(ctx context.Context, nodeDat *nodeD
 
 	launcherDat := ctl.getLauncherData(nodeDat, launcherPod.Name)
 	newInstances := make(map[string]time.Time)
+	remainingInstances := make([]InstanceState, 0, len(insts.Instances))
+	deletedStoppedInstanceIDs := sets.New[string]()
+	failedStoppedInstanceErrs := map[string]error{}
+	runningCount := 0
 	for _, inst := range insts.Instances {
+		if inst.Status == InstanceStatusStopped {
+			// Clean up stopped instances from the launcher.
+			_, delErr := lClient.DeleteInstance(ctx, inst.InstanceID)
+			if delErr != nil && !IsInstanceNotFoundError(delErr) {
+				logger.V(3).Info("Failed to delete stopped instance from launcher during sync",
+					"instanceID", inst.InstanceID, "err", delErr)
+				// Deletion failed — the instance still occupies a slot in the launcher.
+				failedStoppedInstanceErrs[inst.InstanceID] = delErr
+			} else {
+				logger.V(2).Info("Deleted stopped instance from launcher during sync",
+					"instanceID", inst.InstanceID)
+				deletedStoppedInstanceIDs.Insert(inst.InstanceID)
+				continue
+			}
+		}
+		remainingInstances = append(remainingInstances, inst)
+		if inst.Status == "running" {
+			runningCount++
+		}
 		if lastUsed, exists := launcherDat.Instances[inst.InstanceID]; exists {
 			newInstances[inst.InstanceID] = lastUsed
 		} else {
 			newInstances[inst.InstanceID] = time.Now()
 		}
 	}
 
+	// Replace the returned instance list and counts with the filtered view
+	// so that callers (e.g. selectBestLauncherPod) see accurate capacity.
+	insts.Instances = remainingInstances
+	insts.TotalInstances = len(remainingInstances)
+	insts.RunningInstances = runningCount
+
 	launcherDat.Instances = newInstances
 	launcherDat.Accurate = true
 
@@ -1367,7 +1462,11 @@ func (ctl *controller) syncLauncherInstances(ctx context.Context, nodeDat *nodeD
 		"runningInstances", insts.RunningInstances,
 		"instanceCount", len(newInstances))
 
-	return insts, nil, false
+	return &launcherSyncResult{
+		instances:                 insts,
+		deletedStoppedInstanceIDs: deletedStoppedInstanceIDs,
+		failedStoppedInstanceErrs: failedStoppedInstanceErrs,
+	}, nil, false
 }
 
 func init() {

pkg/controller/dual-pods/launcherclient.go

@@ -20,6 +20,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -34,9 +35,22 @@ type LauncherClient struct {
 	httpClient *http.Client
 }
 
+type launcherError struct {
+	StatusCode int
+	Body       string
+}
+
+func (e *launcherError) Error() string {
+	return fmt.Sprintf("launcher error %d: %s", e.StatusCode, e.Body)
+}
+
 const (
 	VllmConfigISCNameAnnotationKey       = "isc-name"
 	VllmConfigInferencePortAnnotationKey = "inference-port"
+
+	// InstanceStatusStopped is the status value reported by the launcher
+	// when a vLLM instance's process has terminated.
+	InstanceStatusStopped = "stopped"
 )
 
 func NewLauncherClient(baseURL string) (*LauncherClient, error) {
@@ -184,6 +198,13 @@ func (c *LauncherClient) create(
 	return &out, nil
 }
 
+// IsInstanceNotFoundError returns true when the launcher reports the instance
+// does not exist.
+func IsInstanceNotFoundError(err error) bool {
+	var launcherErr *launcherError
+	return errors.As(err, &launcherErr) && launcherErr.StatusCode == http.StatusNotFound
+}
+
 func (c *LauncherClient) do(
 	ctx context.Context,
 	method string,
@@ -218,7 +239,7 @@ func (c *LauncherClient) do(
 
 	if resp.StatusCode >= 300 {
 		b, _ := io.ReadAll(resp.Body)
-		return fmt.Errorf("launcher error %d: %s", resp.StatusCode, string(b))
+		return &launcherError{StatusCode: resp.StatusCode, Body: string(b)}
 	}
 
 	if out != nil {

test/e2e/test-cases.sh

@@ -499,4 +499,71 @@ if [ "$E2E_PLATFORM" = "openshift" ]; then check_gpu_pin $req_after_delete; fi
 
 cheer Successful unbound launcher deletion cleanup
 
+# ---------------------------------------------------------------------------
+# Stopped Instance Recovery
+# ---------------------------------------------------------------------------
+
+intro_case Stopped Instance Recovery
+
+# This test verifies that the dual-pods controller detects a stopped vLLM
+# instance (via the sidecar notifier annotation) and deletes the server-
+# requesting Pod so that the ReplicaSet recreates it with a fresh instance.
+#
+# Starting state: $req_after_delete is bound to $launcher_after_delete, both Ready.
+
+echo "Bound requester: $req_after_delete, launcher: $launcher_after_delete"
+req_uid_before=$(kubectl get pod $req_after_delete -n "$NS" -o jsonpath='{.metadata.uid}')
+
+# Get the running instance ID from the launcher
+instance_id=$(kubectl exec -n "$NS" $launcher_after_delete -c inference-server -- python3 -c '
+import json, urllib.request
+resp = json.load(urllib.request.urlopen("http://127.0.0.1:8001/v2/vllm/instances"))
+for inst in resp["instances"]:
+    if inst["status"] == "running":
+        print(inst["instance_id"])
+        break
+')
+echo "Running instance ID: $instance_id"
+[ -n "$instance_id" ]
+
+# Delete the running instance from the launcher to simulate a crash.
+# The notifier sidecar will detect the change and update the Pod annotation.
+# The dual-pods controller will then query the instance, get 404, and delete the requester.
+kubectl exec -n "$NS" $launcher_after_delete -c inference-server -- python3 -c '
+import urllib.request
+req = urllib.request.Request(
+    "http://127.0.0.1:8001/v2/vllm/instances/'"$instance_id"'",
+    method="DELETE",
+)
+urllib.request.urlopen(req)
+print("Instance deleted from launcher")
+'
+
+# Wait for the old requester Pod to be deleted (the dual-pods controller should do this)
+expect '[ "$(kubectl get pod -n '"$NS"' $req_after_delete -o jsonpath={.metadata.uid} 2>/dev/null)" != "$req_uid_before" ]'
+echo "Old requester $req_after_delete was deleted by the controller"
+
+# Wait for the ReplicaSet to recreate a new requester Pod
+expect "kubectl get pods -n $NS -o name -l app=dp-example,instance=$inst | wc -l | grep -w 1"
+req_recovered=$(kubectl get pods -n "$NS" -o name -l app=dp-example,instance=$inst | sed s%pod/%%)
+echo "Recovered server-requesting Pod: $req_recovered"
+
+# Wait for the new requester to be bound to the same launcher
+expect "kubectl get pods -n $NS -o name -l dual-pods.llm-d.ai/dual=$req_recovered | wc -l | grep -w 1"
+launcher_recovered=$(kubectl get pods -n "$NS" -o name -l dual-pods.llm-d.ai/dual=$req_recovered | sed s%pod/%%)
+echo "Recovered launcher: $launcher_recovered"
+[ "$launcher_recovered" == "$launcher_after_delete" ]
+
+# Verify bidirectional binding
+expect '[ "$(kubectl get pod -n '"$NS"' $req_recovered -o jsonpath={.metadata.labels.dual-pods\\.llm-d\\.ai/dual})" == "$launcher_after_delete" ]'
+
+# Wait for both to be ready
+date
+kubectl wait --for condition=Ready pod/$req_recovered -n "$NS" --timeout=120s
+[ "$(kubectl get pod $launcher_after_delete -n "$NS" -o jsonpath='{.status.conditions[?(@.type=="Ready")].status}')" = "True" ]
+
+if [ "$E2E_PLATFORM" = "openshift" ]; then check_gpu_pin $req_recovered; fi
+
+cheer Successful stopped instance recovery
+
 cheer All launcher-based tests passed