Pre creating binding request, delete any pending status updates for the pod (#186)

Author: davidLif

CHANGELOG.md

@@ -4,7 +4,10 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
-## [Unreleased]
+## [v0.4.9]
+
+### Fixed
+- Fixed pod status scheduled race condition between the scheduler and the pod binding
 
 ## [v0.4.8]
 

pkg/scheduler/cache/cache.go

@@ -223,6 +223,7 @@ func (sc *SchedulerCache) WaitForWorkers(stopCh <-chan struct{}) {
 func (sc *SchedulerCache) Bind(taskInfo *pod_info.PodInfo, hostname string) error {
 	startTime := time.Now()
 	defer metrics.UpdateTaskBindDuration(startTime)
+	sc.StatusUpdater.PreBind(taskInfo.Pod)
 
 	log.InfraLogger.V(3).Infof(
 		"Creating bind request for task <%v/%v> to node <%v> gpuGroup: <%v>, requires: <%v> GPUs",

pkg/scheduler/cache/status_updater/concurrency.go

@@ -28,7 +28,7 @@ func (su *defaultStatusUpdater) Run(stopCh <-chan struct{}) {
 func (su *defaultStatusUpdater) SyncPodGroupsWithPendingUpdates(podGroups []*enginev2alpha2.PodGroup) {
 	usedKeys := make(map[updatePayloadKey]bool, len(podGroups))
 	for i := range podGroups {
-		key := su.keyForPayload(podGroups[i].Name, podGroups[i].Namespace, podGroups[i].UID)
+		key := su.keyForPodGroupPayload(podGroups[i].Name, podGroups[i].Namespace, podGroups[i].UID)
 		usedKeys[key] = true
 		inflightUpdateAny, found := su.inFlightPodGroups.Load(key)
 		if !found {
@@ -77,10 +77,18 @@ func (su *defaultStatusUpdater) syncPodGroup(inFlightPodGroup, snapshotPodGroup
 	return staleTimeStampUpdated && updatedSchedulingCondition
 }
 
-func (su *defaultStatusUpdater) keyForPayload(name, namespace string, uid types.UID) updatePayloadKey {
+func (su *defaultStatusUpdater) keyForPodGroupPayload(name, namespace string, uid types.UID) updatePayloadKey {
 	return updatePayloadKey(types.NamespacedName{Name: name, Namespace: namespace}.String() + "_" + string(uid))
 }
 
+func (su *defaultStatusUpdater) keyForPodStatusPayload(name, namespace string, uid types.UID) updatePayloadKey {
+	return updatePayloadKey(types.NamespacedName{Name: name, Namespace: namespace}.String() + "_" + string(uid) + "-Status")
+}
+
+func (su *defaultStatusUpdater) keyForPodLabelsPayload(name, namespace string, uid types.UID) updatePayloadKey {
+	return updatePayloadKey(types.NamespacedName{Name: name, Namespace: namespace}.String() + "_" + string(uid) + "-Labels")
+}
+
 func (su *defaultStatusUpdater) processPayload(ctx context.Context, payload *updatePayload) {
 	updateData, found := su.loadInflighUpdate(payload)
 	if !found {

pkg/scheduler/cache/status_updater/default_status_updater.go

@@ -142,6 +142,11 @@ func (su *defaultStatusUpdater) Bound(
 	return bindError
 }
 
+func (su *defaultStatusUpdater) PreBind(pod *v1.Pod) {
+	// Delete any pending status updates for this pod - after this binding, they will become no longer relevant
+	su.inFlightPods.Delete(su.keyForPodStatusPayload(pod.Name, pod.Namespace, pod.UID))
+}
+
 func (su *defaultStatusUpdater) Pipelined(pod *v1.Pod, message string) {
 	su.recorder.Eventf(pod, v1.EventTypeNormal, "Pipelined", message)
 }
@@ -163,7 +168,7 @@ func (su *defaultStatusUpdater) PatchPodLabels(pod *v1.Pod, labels map[string]an
 
 	su.pushToUpdateQueue(
 		&updatePayload{
-			key:        su.keyForPayload(pod.Name, pod.Namespace, pod.UID) + "-Labels",
+			key:        su.keyForPodLabelsPayload(pod.Name, pod.Namespace, pod.UID),
 			objectType: podType,
 		},
 		&inflightUpdate{
@@ -201,7 +206,7 @@ func (su *defaultStatusUpdater) RecordJobStatusEvent(job *podgroup_info.PodGroup
 	if len(patchData) > 0 || updatePodgroupStatus {
 		su.pushToUpdateQueue(
 			&updatePayload{
-				key:        su.keyForPayload(job.PodGroup.Name, job.PodGroup.Namespace, job.PodGroup.UID),
+				key:        su.keyForPodGroupPayload(job.PodGroup.Name, job.PodGroup.Namespace, job.PodGroup.UID),
 				objectType: podGroupType,
 			},
 			&inflightUpdate{
@@ -278,7 +283,7 @@ func (su *defaultStatusUpdater) updatePodCondition(pod *v1.Pod, condition *v1.Po
 
 		su.pushToUpdateQueue(
 			&updatePayload{
-				key:        su.keyForPayload(pod.Name, pod.Namespace, pod.UID) + "-Status",
+				key:        su.keyForPodStatusPayload(pod.Name, pod.Namespace, pod.UID),
 				objectType: podType,
 			},
 			&inflightUpdate{

pkg/scheduler/cache/status_updater/interface.go

@@ -18,6 +18,7 @@ type PodGroupsSync interface {
 type Interface interface {
 	PodGroupsSync
 	Evicted(evictedPodGroup *enginev2alpha2.PodGroup, evictionMetadata eviction_info.EvictionMetadata, message string)
+	PreBind(pod *v1.Pod)
 	Bound(pod *v1.Pod, hostname string, bindError error, nodePoolName string) error
 	Pipelined(pod *v1.Pod, message string)
 	PatchPodLabels(pod *v1.Pod, labels map[string]interface{})