Sync unbound launcher-based server-providing pods #362
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -85,6 +85,25 @@ 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) { | ||
| 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) | ||
| if err != nil { | ||
| if apierrors.IsNotFound(err) { | ||
| logger.V(2).Info("Launcher pod deleted, cleaning up launcher data") | ||
| ctl.clearLauncherData(nodeDat, item.LauncherPodName) | ||
| ctl.enqueueUnboundInfSvrItemsOnNode(ctx, item.NodeName, fmt.Sprintf("launcher pod %s deleted", item.LauncherPodName)) | ||
| return nil, false | ||
| } | ||
| return err, true | ||
| } | ||
|
|
||
| ctl.enqueueUnboundInfSvrItemsOnNode(ctx, item.NodeName, fmt.Sprintf("launcher pod %s changed", item.LauncherPodName)) | ||
| return nil, false | ||
| } | ||
|
|
||
| func (item infSvrItem) process(urCtx context.Context, ctl *controller, nodeDat *nodeData) (error, bool) { | ||
| logger := klog.FromContext(urCtx).WithValues("serverUID", item.UID, "requesterName", item.RequesterName) | ||
| ctx := klog.NewContext(urCtx, logger) | ||
|
|
@@ -350,8 +369,13 @@ func (item infSvrItem) process(urCtx context.Context, ctl *controller, nodeDat * | |
| if err != nil { | ||
| return err, true | ||
| } | ||
| // Relay readiness if not already done. | ||
| // For launcher-based providers, readiness follows the bound instance's | ||
| // sleeping state rather than the launcher's Pod readiness. | ||
| ready := utils.IsPodReady(providingPod) | ||
| if launcherBased { | ||
| ready = !*serverDat.Sleeping | ||
| } | ||
| if serverDat.ReadinessRelayed == nil || ready != *serverDat.ReadinessRelayed { | ||
| url, readiness := fmt.Sprintf("http://%s:%s", requestingPod.Status.PodIP, adminPort), "" | ||
| if ready { | ||
|
|
@@ -588,24 +612,21 @@ func (ctl *controller) selectBestLauncherPod( | |
| continue | ||
| } | ||
|
|
||
| insts, err, retry := ctl.syncLauncherInstances(ctx, nodeDat, launcherPod) | ||
MikeSpreitzer marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| if err != nil || retry { | ||
| somePodsNotReady = true | ||
| continue | ||
| } | ||
|
|
||
| // Check if this launcher has a sleeping instance matching the iscHash | ||
| hasSleepingInstance := false | ||
| for _, inst := range insts.Instances { | ||
| if inst.InstanceID == iscHash { | ||
|
Collaborator
There was a problem hiding this comment. launcherPodAnys contains all launchers made from the right LauncherConfig object, right? Including ones with an awake child whose InstanceID == iscHash, right?
Collaborator
Author
There was a problem hiding this comment. Not including an awake child because of the same logic as llm-d-fast-model-actuation/pkg/controller/dual-pods/inference-server.go Lines 430 to 431 in 0583aea
Collaborator
There was a problem hiding this comment. Not exactly analogous. You cited logic for the direct (milestone 2) case, in which the index value is a hash that takes the node name and the GPU list into account. In that case it is reasonable to expect the mentioned Kube Pod scheduler behavior. But here in the launcher-based case, the index is on the hash of the LauncherConfig augmented by the node name and "gpus=all". So this index is not so discriminating, and could include other existing launcher Pods with the right LauncherConfig and node but being used for an awake vllm instance using different GPU(s).
Collaborator
Author
There was a problem hiding this comment. Info for GPUs are hashed into iscHash, right?
Collaborator
There was a problem hiding this comment.
Collaborator
Author
There was a problem hiding this comment. Yes. The reason that I mentioned iscHash as well is that the two hashes work here together.
Collaborator
There was a problem hiding this comment. Oh, right. If the iscHash matches then the GPU sets are the same and so the instance being considered must be sleeping.
Collaborator
There was a problem hiding this comment. Maybe this is tricky enough to warrant a comment too. |
||
| hasSleepingInstance = true | ||
| break | ||
| } | ||
| } | ||
| if hasSleepingInstance { | ||
| // Priority 1: Found a sleeping instance | ||
| logger.V(5).Info("Found launcher with sleeping instance (fastest path)", | ||
| "name", launcherPod.Name, | ||
|
|
@@ -1237,6 +1258,52 @@ var coreScheme *k8sruntime.Scheme | |
| var codecFactory k8sserializer.CodecFactory | ||
| var podDecoder k8sruntime.Decoder | ||
|
|
||
| // 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) (*AllInstancesStatus, error, bool) { | ||
| logger := klog.FromContext(ctx) | ||
|
|
||
| if launcherPod.Status.PodIP == "" || !utils.IsPodReady(launcherPod) { | ||
| logger.V(5).Info("Launcher pod not ready yet, waiting for another Pod event", "name", launcherPod.Name) | ||
| return nil, nil, true | ||
| } | ||
|
|
||
| launcherBaseURL := fmt.Sprintf("http://%s:%d", launcherPod.Status.PodIP, ctlrcommon.LauncherServicePort) | ||
| lClient, err := NewLauncherClient(launcherBaseURL) | ||
| if err != nil { | ||
| logger.Error(err, "Failed to create launcher client") | ||
| return nil, err, true | ||
| } | ||
|
|
||
| insts, err := lClient.ListInstances(ctx) | ||
| if err != nil { | ||
| logger.Error(err, "Failed to list instances from launcher") | ||
| return nil, err, true | ||
| } | ||
|
|
||
| launcherDat := ctl.getLauncherData(nodeDat, launcherPod.Name) | ||
| newInstances := make(map[string]time.Time) | ||
| for _, inst := range insts.Instances { | ||
| if lastUsed, exists := launcherDat.Instances[inst.InstanceID]; exists { | ||
| newInstances[inst.InstanceID] = lastUsed | ||
| } else { | ||
| newInstances[inst.InstanceID] = time.Now() | ||
| } | ||
| } | ||
|
|
||
|
Collaborator
There was a problem hiding this comment. If an awake instance evaporated then this should cause the relevant |
||
| launcherDat.Instances = newInstances | ||
| launcherDat.Accurate = true | ||
|
|
||
| logger.V(2).Info("Synced launcher instances", | ||
| "launcherPod", launcherPod.Name, | ||
| "totalInstances", insts.TotalInstances, | ||
| "runningInstances", insts.RunningInstances, | ||
| "instanceCount", len(newInstances)) | ||
|
|
||
| return insts, nil, false | ||
| } | ||
|
|
||
| func init() { | ||
| coreScheme = k8sruntime.NewScheme() | ||
| err := corev1.AddToScheme(coreScheme) | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Include a statement about what does synchronize access. Perhaps something like the following.
Similarly, every func (below
nodeItem.process()) that (directly or in a called func) accesses this field should have a comment stating the restriction. Perhaps something like the following.