This is an automated email from the ASF dual-hosted git repository.
manirajv06 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/yunikorn-k8shim.git
The following commit(s) were added to refs/heads/master by this push:
new 893da47a [YUNIKORN-3355] Revert cache node assignment when a pod is
forgotten (#1066)
893da47a is described below
commit 893da47af0ee5a3bd03b8575cb5964eab604dcd7
Author: Tigerquoll <[email protected]>
AuthorDate: Tue Aug 25 11:39:03 2026 +0530
[YUNIKORN-3355] Revert cache node assignment when a pod is forgotten (#1066)
Forgetting a pod passed the cached pod, which carries the node name set
when the pod was assumed, back into updatePod. That re-registered the
assignment, so a failed bind left the pod assigned to a node it never
ran on. Updates from the informer for the still unassigned pod hit the
"use existing assignment" branch and had the stale node stamped back on
them, which also modified the pod owned by the informer cache. Task
re-creation then sent the node ID to the core and a placed allocation
was added for a pod that was never scheduled.
Revert the assignment in forgetPod for a pod that is still assumed, and
only preserve an existing assignment on update while the pod is assumed.
Signed-off-by: Tigerquoll <[email protected]>
Closes: #1066
Signed-off-by: Manikandan R <[email protected]>
---
pkg/cache/external/scheduler_cache.go | 27 +++++--
pkg/cache/external/scheduler_cache_test.go | 122 +++++++++++++++++++++++++++++
2 files changed, 143 insertions(+), 6 deletions(-)
diff --git a/pkg/cache/external/scheduler_cache.go
b/pkg/cache/external/scheduler_cache.go
index 6dddecea..31cea350 100644
--- a/pkg/cache/external/scheduler_cache.go
+++ b/pkg/cache/external/scheduler_cache.go
@@ -335,8 +335,12 @@ func (cache *SchedulerCache) updatePod(pod *v1.Pod) bool {
cache.nodesInfoPodsWithReqAntiAffinity
= nil
}
}
- if pod.Spec.NodeName == "" {
- // new pod wasn't assigned to a node, so use
existing assignment
+ if pod.Spec.NodeName == "" && cache.isAssumedPod(key) {
+ // new pod wasn't assigned to a node, but the
pod is assumed on one, so use the
+ // existing assignment until the result of the
bind shows up. Only assumed pods
+ // keep their assignment: a forgotten pod must
not have it resurrected here.
+ // Copy before updating, the pod passed in can
be owned by the informer cache.
+ pod = pod.DeepCopy()
pod.Spec.NodeName = nodeName
}
}
@@ -472,15 +476,26 @@ func (cache *SchedulerCache) ForgetPod(pod *v1.Pod) {
func (cache *SchedulerCache) forgetPod(pod *v1.Pod) {
key := string(pod.UID)
- // update the pod in cache
- cache.updatePod(pod)
-
- // remove assigned allocation
log.Log(log.ShimCacheExternal).Debug("Removing assumed pod from cache",
zap.String("podName", pod.Name),
zap.String("podKey", key))
+ // a pod that is no longer assumed has been bound already: its
assignment is real and is kept
+ revert := cache.isAssumedPod(key) && pod.Spec.NodeName != ""
+
+ // remove the assumed state first: updatePod preserves the assignment
of an assumed pod
delete(cache.assumedPods, key)
+
+ if revert {
+ // the pod was never bound to the node it was assumed on,
revert the assignment.
+ // Copy before updating, the pod passed in can be shared with
the caller.
+ unassigned := pod.DeepCopy()
+ unassigned.Spec.NodeName = ""
+ pod = unassigned
+ }
+
+ // update the pod in cache
+ cache.updatePod(pod)
}
// Implement k8s.io/client-go/listers/core/v1#PodLister interface
diff --git a/pkg/cache/external/scheduler_cache_test.go
b/pkg/cache/external/scheduler_cache_test.go
index b31a587e..e64d0301 100644
--- a/pkg/cache/external/scheduler_cache_test.go
+++ b/pkg/cache/external/scheduler_cache_test.go
@@ -805,6 +805,128 @@ func TestUpdatePod(t *testing.T) {
assert.Equal(t, pod3Result.Spec.NodeName, "new-node", "node name not
updated")
}
+// this test verifies that an update for an assumed pod which does not carry a
node name keeps
+// the assumed assignment, and that the pod of the update itself is left
untouched as it can be
+// owned by the informer cache
+func TestUpdateAssumedPod(t *testing.T) {
+ cache := NewSchedulerCache(client.NewMockedAPIProvider(false).GetAPIs())
+ cache.UpdateNode(newTestNode())
+
+ pod := newTestPod()
+ cache.UpdatePod(pod)
+
+ assumedPod := pod.DeepCopy()
+ assumedPod.Spec.NodeName = host1
+ cache.AssumePod(assumedPod, true)
+ assert.Check(t, cache.IsAssumedPod(podUID1), "pod is not assumed")
+
+ // the pod is not assigned in the cluster yet, the assumed assignment
must survive the update
+ updatedPod := pod.DeepCopy()
+ cache.UpdatePod(updatedPod)
+ assert.Check(t, cache.IsAssumedPod(podUID1), "pod is not assumed after
update")
+ assert.Equal(t, cache.GetPod(podUID1).Spec.NodeName, host1, "assumed
pod lost its node assignment")
+ assert.Equal(t, cache.assignedPods[podUID1], host1, "assumed pod is not
in the assigned pods")
+ // nolint:staticcheck
+ assert.Equal(t, len(cache.GetNode(host1).Pods), 1, "assumed pod is not
added to the node")
+ assert.Equal(t, updatedPod.Spec.NodeName, "", "pod of the update was
modified")
+}
+
+// this test verifies that forgetting an assumed pod reverts the node
assignment which was made
+// when the pod was assumed: the bind never happened so the pod must be
unassigned in the cache,
+// and an update for the still unassigned pod must not bring the assignment
back
+func TestForgetPod(t *testing.T) {
+ cache := NewSchedulerCache(client.NewMockedAPIProvider(false).GetAPIs())
+ cache.UpdateNode(newTestNode())
+
+ pod := newTestPod()
+ cache.UpdatePod(pod)
+
+ assumedPod := pod.DeepCopy()
+ assumedPod.Spec.NodeName = host1
+ cache.AssumePod(assumedPod, true)
+ assert.Check(t, cache.IsAssumedPod(podUID1), "pod is not assumed")
+ assert.Equal(t, cache.GetPod(podUID1).Spec.NodeName, host1, "assumed
pod is not assigned to the node")
+ // nolint:staticcheck
+ assert.Equal(t, len(cache.GetNode(host1).Pods), 1, "assumed pod is not
added to the node")
+
+ // the bind failed: the assignment must be reverted
+ cache.ForgetPod(cache.GetPod(podUID1))
+ assert.Check(t, !cache.IsAssumedPod(podUID1), "pod is still assumed
after forget")
+ assert.Equal(t, cache.GetPod(podUID1).Spec.NodeName, "", "forgotten pod
is still assigned to a node")
+ assert.Equal(t, len(cache.assignedPods), 0, "forgotten pod is still in
the assigned pods")
+ // nolint:staticcheck
+ assert.Equal(t, len(cache.GetNode(host1).Pods), 0, "forgotten pod is
still added to the node")
+
+ // an update for the pod, which is still unassigned in the cluster,
must not restore the
+ // assignment: the pod would be recovered as an existing allocation on
the node
+ updatedPod := pod.DeepCopy()
+ cache.UpdatePod(updatedPod)
+ assert.Equal(t, cache.GetPod(podUID1).Spec.NodeName, "", "update
restored the node assignment")
+ assert.Equal(t, len(cache.assignedPods), 0, "update restored the
assigned pods entry")
+ // nolint:staticcheck
+ assert.Equal(t, len(cache.GetNode(host1).Pods), 0, "update added the
pod back to the node")
+
+ // forget of a pod that is not assumed must not change anything
+ cache.ForgetPod(cache.GetPod(podUID1))
+ assert.Equal(t, cache.GetPod(podUID1).Spec.NodeName, "", "forgotten pod
is assigned to a node")
+ assert.Equal(t, len(cache.podsMap), 1, "wrong pod count after second
forget")
+}
+
+// this test verifies that forgetting a pod which is bound in the cluster, and
thus is not assumed
+// anymore, leaves the node assignment in place
+func TestForgetBoundPod(t *testing.T) {
+ cache := NewSchedulerCache(client.NewMockedAPIProvider(false).GetAPIs())
+ cache.UpdateNode(newTestNode())
+
+ pod := newTestPod()
+ pod.Spec.NodeName = host1
+ pod.Status.Phase = v1.PodRunning
+ cache.UpdatePod(pod)
+ assert.Check(t, !cache.IsAssumedPod(podUID1), "running pod is assumed")
+
+ cache.ForgetPod(cache.GetPod(podUID1))
+ assert.Equal(t, cache.GetPod(podUID1).Spec.NodeName, host1, "bound pod
lost its node assignment")
+ assert.Equal(t, cache.assignedPods[podUID1], host1, "bound pod is not
in the assigned pods")
+ // nolint:staticcheck
+ assert.Equal(t, len(cache.GetNode(host1).Pods), 1, "bound pod is not
added to the node")
+}
+
+// newTestNode returns a node with a name of host1 and enough capacity for the
test pods
+func newTestNode() *v1.Node {
+ resourceList := make(map[v1.ResourceName]resource.Quantity)
+ resourceList[v1.ResourceName("memory")] =
*resource.NewQuantity(1024*1000*1000, resource.DecimalSI)
+ resourceList[v1.ResourceName("cpu")] = *resource.NewQuantity(10,
resource.DecimalSI)
+ return &v1.Node{
+ ObjectMeta: apis.ObjectMeta{
+ Name: host1,
+ Namespace: "default",
+ UID: nodeUID1,
+ },
+ Status: v1.NodeStatus{
+ Allocatable: resourceList,
+ },
+ Spec: v1.NodeSpec{
+ Unschedulable: false,
+ },
+ }
+}
+
+// newTestPod returns an unassigned pod with a name of podName1
+func newTestPod() *v1.Pod {
+ return &v1.Pod{
+ TypeMeta: apis.TypeMeta{
+ Kind: "Pod",
+ APIVersion: "v1",
+ },
+ ObjectMeta: apis.ObjectMeta{
+ Name: podName1,
+ Namespace: "default",
+ UID: podUID1,
+ },
+ Spec: v1.PodSpec{},
+ }
+}
+
func TestRemovePod(t *testing.T) {
cache := NewSchedulerCache(client.NewMockedAPIProvider(false).GetAPIs())
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
