upstream/kubernetes
1
0
Fork 0
mirror of https://github.com/kubernetes/kubernetes.git synced 2026-07-15 20:53:45 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #140054 from googs1025/flake_scheduler

Browse source 
flake(scheduler): reactivate preemptor after in-memory preemption
This commit is contained in:
kubernetes-prow[bot] 2026-07-06 13:32:28 +00:00 committed by GitHub
commit 5f2bdc6de2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 257 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
pkg/scheduler/framework/plugins/defaultpreemption/default_preemption_test.go
View file

@ -2594,12 +2594,12 @@ func TestPreEnqueue(t *testing.T) {
finishPreemption := make(chan struct{})
p.Executor.PreemptPod = func(ctx context.Context, c preemption.Candidate, preemptor preemption.ExecutorPreemptor, victim *v1.Pod, pluginName string) error {
p.Executor.PreemptPod = func(ctx context.Context, c preemption.Candidate, preemptor preemption.ExecutorPreemptor, victim *v1.Pod, pluginName string) (bool, error) {
if !tt.features.EnableAsyncPreemption {
return nil
return false, nil
}
<-finishPreemption
return nil
return false, nil
}
// Fill the cycle state

42
pkg/scheduler/framework/preemption/executor.go
View file

@ -85,9 +85,10 @@ type Executor struct {
// to prevent the pods/podgroups from entering the scheduling cycle while waiting for preemption to complete.
lastVictimsPendingPreemption map[types.UID]pendingVictim
// PreemptPod is a function that actually makes API calls to preempt a specific Pod.
// PreemptPod is a function that actually preempts a specific Pod. It returns true
// when the victim was preempted only in scheduler memory, without a delete call.
// This is exposed to be replaced during tests.
PreemptPod func(ctx context.Context, c Candidate, preemptor ExecutorPreemptor, victim *v1.Pod, pluginName string) error
PreemptPod func(ctx context.Context, c Candidate, preemptor ExecutorPreemptor, victim *v1.Pod, pluginName string) (bool, error)
}
// NewExecutor creates a new preemption executor.
@ -100,28 +101,28 @@ func NewExecutor(fh fwk.Handle, fts feature.Features) *Executor {
fts: fts,
}
e.PreemptPod = func(ctx context.Context, c Candidate, preemptor ExecutorPreemptor, victim *v1.Pod, pluginName string) error {
e.PreemptPod = func(ctx context.Context, c Candidate, preemptor ExecutorPreemptor, victim *v1.Pod, pluginName string) (bool, error) {
logger := klog.FromContext(ctx)
skipAPICall := false
preemptedInMemory := false
eventMessage := fmt.Sprintf("Preempted by %s %v on node %v", preemptor.Type(), preemptor.UID(), c.Name())
// If the victim is a WaitingPod, try to preempt it without a delete call (victim will go back to backoff queue).
// Otherwise we should delete the victim.
if waitingPod := e.fh.GetWaitingPod(victim.UID); waitingPod != nil {
if waitingPod.Preempt(pluginName, "preempted") {
logger.V(2).Info("Preemptor preempted a waiting pod", "preemptorType", preemptor.Type(), "preemptor", klog.KObj(preemptor), "waitingPod", klog.KObj(victim), "node", c.Name())
skipAPICall = true
preemptedInMemory = true
}
} else if podInPreBind := e.fh.GetPodInPreBind(victim.UID); podInPreBind != nil {
// If the victim is in the preBind cancel the binding process.
if podInPreBind.CancelPod(fmt.Sprintf("preempted by %s", pluginName)) {
logger.V(2).Info("Preemptor rejected a pod in preBind", "preemptorType", preemptor.Type(), "preemptor", klog.KObj(preemptor), "podInPreBind", klog.KObj(victim), "node", c.Name())
skipAPICall = true
preemptedInMemory = true
} else {
logger.V(5).Info("Failed to reject a pod in preBind, falling back to deletion via api call", "preemptor", klog.KObj(preemptor), "podInPreBind", klog.KObj(victim), "node", c.Name())
}
}
if !skipAPICall {
if !preemptedInMemory {
condition := &v1.PodCondition{
Type: v1.DisruptionTarget,
ObservedGeneration: apipod.CalculatePodConditionObservedGeneration(&victim.Status, victim.Generation, v1.DisruptionTarget),
@ -135,19 +136,19 @@ func NewExecutor(fh fwk.Handle, fts feature.Features) *Executor {
if err := util.PatchPodStatus(ctx, fh.ClientSet(), victim.Name, victim.Namespace, "", &victim.Status, newStatus); err != nil {
if !apierrors.IsNotFound(err) {
logger.Error(err, "Could not add DisruptionTarget condition due to preemption", "preemptor", klog.KObj(preemptor), "victim", klog.KObj(victim))
return err
return false, err
}
logger.V(2).Info("Victim Pod is already deleted", "preemptor", klog.KObj(preemptor), "victim", klog.KObj(victim), "node", c.Name())
return nil
return false, nil
}
}
if err := util.DeletePod(ctx, fh.ClientSet(), victim); err != nil {
if !apierrors.IsNotFound(err) {
logger.Error(err, "Tried to preempted pod", "pod", klog.KObj(victim), "preemptor", klog.KObj(preemptor))
return err
return false, err
}
logger.V(2).Info("Victim Pod is already deleted", "preemptor", klog.KObj(preemptor), "victim", klog.KObj(victim), "node", c.Name())
return nil
return false, nil
}
logger.V(2).Info("Preemptor Pod preempted victim Pod", "preemptor", klog.KObj(preemptor), "victim", klog.KObj(victim), "node", c.Name())
} else {
@ -156,7 +157,7 @@ func NewExecutor(fh fwk.Handle, fts feature.Features) *Executor {
fh.EventRecorder().WithLogger(logger).Eventf(victim, preemptor.Obj(), v1.EventTypeNormal, "Preempted", "Preempting", eventMessage)
return nil
return preemptedInMemory, nil
}
return e
@ -223,9 +224,11 @@ func (e *Executor) prepareCandidateAsync(c Candidate, preemptor ExecutorPreempto
metrics.PreemptionVictims.Observe(float64(len(c.Victims().Pods)))
errCh := parallelize.NewResultChannel[error]()
// PreEnqueue only watches the last victim for completion, so activate when that victim won't emit a deletion event.
preemptedLastVictimInMemory := false
preemptPod := func(index int) {
victim := victimPods[index]
if err := e.PreemptPod(ctx, c, preemptor, victim, pluginName); err != nil {
if _, err := e.PreemptPod(ctx, c, preemptor, victim, pluginName); err != nil {
errCh.SendWithCancel(err, cancel)
}
}
@ -241,9 +244,9 @@ func (e *Executor) prepareCandidateAsync(c Candidate, preemptor ExecutorPreempto
defer metrics.PreemptionGoroutinesDuration.WithLabelValues(result).Observe(metrics.SinceInSeconds(startTime))
defer metrics.PreemptionGoroutinesExecutionTotal.WithLabelValues(result).Inc()
defer func() {
if result == metrics.GoroutineResultError {
// When API call isn't successful, the preemptor's Pods may get stuck in the unschedulable pod pool in the worst case.
// So, we should move the preemptor's Pods to the activeQ.
if result == metrics.GoroutineResultError || preemptedLastVictimInMemory {
// When API call isn't successful or no victim deletion event will be produced, the preemptor's
// Pods may get stuck in the unschedulable pod pool in the worst case.
e.fh.Activate(logger, preemptor.Pods())
}
}()
@ -290,9 +293,12 @@ func (e *Executor) prepareCandidateAsync(c Candidate, preemptor ExecutorPreempto
e.lastVictimsPendingPreemption[preemptor.UID()] = pendingVictim{namespace: lastVictim.Namespace, name: lastVictim.Name}
e.mu.Unlock()
if err := e.PreemptPod(ctx, c, preemptor, lastVictim, pluginName); err != nil {
preemptedInMemory, err := e.PreemptPod(ctx, c, preemptor, lastVictim, pluginName)
if err != nil {
utilruntime.HandleErrorWithContext(ctx, err, "Error occurred during async preemption of the last victim")
result = metrics.GoroutineResultError
} else if preemptedInMemory {
preemptedLastVictimInMemory = true
}
}
e.mu.Lock()
@ -325,7 +331,7 @@ func (e *Executor) prepareCandidate(ctx context.Context, c Candidate, preemptor
logger.V(2).Info("Victim Pod is already being deleted, skipping the API call for it", "preemptor", klog.KObj(preemptor), "node", c.Name(), "victim", klog.KObj(victim))
return
}
if err := e.PreemptPod(ctx, c, preemptor, victim, pluginName); err != nil {
if _, err := e.PreemptPod(ctx, c, preemptor, victim, pluginName); err != nil {
errCh.SendWithCancel(err, cancel)
}
}, pluginName)

227
pkg/scheduler/framework/preemption/executor_test.go
View file

@ -198,8 +198,12 @@ func TestIsPodRunningPreemption(t *testing.T) {
}
type fakePodActivator struct {
activatedPods map[string]*v1.Pod
mu *sync.RWMutex
activatedPods map[string]*v1.Pod
mu *sync.RWMutex
activatedCh chan struct{}
activatedOnce sync.Once
isPreempting func() bool
activatedWhilePreempting bool
}
func (f *fakePodActivator) Activate(logger klog.Logger, pods map[string]*v1.Pod) {
@ -208,6 +212,14 @@ func (f *fakePodActivator) Activate(logger klog.Logger, pods map[string]*v1.Pod)
for name, pod := range pods {
f.activatedPods[name] = pod
}
if f.isPreempting != nil && f.isPreempting() {
f.activatedWhilePreempting = true
}
if f.activatedCh != nil {
f.activatedOnce.Do(func() {
close(f.activatedCh)
})
}
}
type fakePodNominator struct {
@ -903,7 +915,7 @@ func TestPrepareCandidateAsyncSetsPreemptingSets(t *testing.T) {
// preemptPodCallsCounter helps verify if the last victim pod gets preempted after other victims.
preemptPodCallsCounter := 0
preemptFunc := executor.PreemptPod
executor.PreemptPod = func(ctx context.Context, c Candidate, preemptor ExecutorPreemptor, victim *v1.Pod, pluginName string) error {
executor.PreemptPod = func(ctx context.Context, c Candidate, preemptor ExecutorPreemptor, victim *v1.Pod, pluginName string) (bool, error) {
// Verify contents of the sets: preempting and lastVictimsPendingPreemption before preemption of subsequent pods.
executor.mu.RLock()
preemptPodCallsCounter++
@ -1391,10 +1403,13 @@ func TestPreemptPod(t *testing.T) {
preemptor = &podGroupExecutorPreemptor{pg: preemptorPodGroup, pods: preemptorPods}
}
err = pe.PreemptPod(ctx, &candidate{name: "fake-node"}, preemptor, victimPod, "test-plugin")
preemptedInMemory, err := pe.PreemptPod(ctx, &candidate{name: "fake-node"}, preemptor, victimPod, "test-plugin")
if err != nil {
t.Fatal(err)
}
if preemptedInMemory != (tt.addVictimToPrebind || tt.addVictimToWaiting) {
t.Errorf("PreemptPod() preemptedInMemory = %v, want %v", preemptedInMemory, tt.addVictimToPrebind || tt.addVictimToWaiting)
}
if tt.expectCancel {
if victimCtx.Err() == nil {
t.Errorf("Context of a binding pod should be cancelled")
@ -1420,6 +1435,210 @@ func TestPreemptPod(t *testing.T) {
}
}
func TestPrepareCandidateAsyncActivatesPreemptorAfterLastVictimInMemoryPreemption(t *testing.T) {
preemptorPod := st.MakePod().Name("p").UID("p").Priority(highPriority).Obj()
secondPreemptorPod := st.MakePod().Name("p2").UID("p2").Priority(highPriority).Obj()
preemptorPodGroup := &schedulingapi.PodGroup{ObjectMeta: metav1.ObjectMeta{Name: "pg", UID: "pg"}}
waitingVictim := st.MakePod().Name("waiting-v").UID("waiting-v").Priority(midPriority).Node("node1").Obj()
preBindVictim := st.MakePod().Name("prebind-v").UID("prebind-v").Priority(midPriority).Node("node1").Obj()
apiVictim := st.MakePod().Name("api-v").UID("api-v").Priority(midPriority).Node("node1").Obj()
tests := []struct {
name string
victimPods []*v1.Pod
inMemoryVictim *v1.Pod
addVictimToPrebind bool
addVictimToPrebindOnPreempt bool
addVictimToWaiting bool
preemptorPodGroup *schedulingapi.PodGroup
preemptorPods []*v1.Pod
wantPreemptorActivate bool
}{
{
name: "last waiting pod",
victimPods: []*v1.Pod{waitingVictim},
inMemoryVictim: waitingVictim,
addVictimToWaiting: true,
wantPreemptorActivate: true,
},
{
name: "last preBind pod",
victimPods: []*v1.Pod{preBindVictim},
inMemoryVictim: preBindVictim,
addVictimToPrebind: true,
wantPreemptorActivate: true,
},
{
name: "last pod enters preBind during preemption",
victimPods: []*v1.Pod{preBindVictim.DeepCopy()},
inMemoryVictim: preBindVictim.DeepCopy(),
addVictimToPrebindOnPreempt: true,
wantPreemptorActivate: true,
},
{
name: "last waiting pod after API-deleted victim",
victimPods: []*v1.Pod{apiVictim.DeepCopy(), waitingVictim.DeepCopy()},
inMemoryVictim: waitingVictim.DeepCopy(),
addVictimToWaiting: true,
wantPreemptorActivate: true,
},
{
name: "last waiting pod for pod group",
victimPods: []*v1.Pod{waitingVictim.DeepCopy()},
inMemoryVictim: waitingVictim.DeepCopy(),
addVictimToWaiting: true,
preemptorPodGroup: preemptorPodGroup,
preemptorPods: []*v1.Pod{preemptorPod.DeepCopy(), secondPreemptorPod.DeepCopy()},
wantPreemptorActivate: true,
},
{
name: "non-last waiting pod",
victimPods: []*v1.Pod{waitingVictim.DeepCopy(), apiVictim.DeepCopy()},
inMemoryVictim: waitingVictim.DeepCopy(),
addVictimToWaiting: true,
},
{
name: "non-last preBind pod",
victimPods: []*v1.Pod{preBindVictim.DeepCopy(), apiVictim.DeepCopy()},
inMemoryVictim: preBindVictim.DeepCopy(),
addVictimToPrebind: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
metrics.Register()
logger, ctx := ktesting.NewTestContext(t)
ctx, cancel := context.WithCancel(ctx)
defer cancel()
mu := &sync.RWMutex{}
fakeActivator := &fakePodActivator{activatedPods: make(map[string]*v1.Pod), mu: mu, activatedCh: make(chan struct{})}
podsInPreBind := frameworkruntime.NewPodsInPreBindMap()
waitingPods := frameworkruntime.NewWaitingPodsMap()
registeredPlugins := append([]tf.RegisterPluginFunc{
tf.RegisterQueueSortPlugin(queuesort.Name, queuesort.New)},
tf.RegisterBindPlugin(defaultbinder.Name, defaultbinder.New),
tf.RegisterPermitPlugin(waitingPermitPluginName, newWaitingPermitPlugin),
)
preemptorPods := tt.preemptorPods
if len(preemptorPods) == 0 {
preemptorPods = []*v1.Pod{preemptorPod.DeepCopy()}
}
var preemptor ExecutorPreemptor = &podExecutorPreemptor{Pod: preemptorPods[0]}
if tt.preemptorPodGroup != nil {
preemptor = &podGroupExecutorPreemptor{pg: tt.preemptorPodGroup, pods: preemptorPods}
}
objects := make([]runtime.Object, 0, len(preemptorPods)+len(tt.victimPods))
for _, pod := range preemptorPods {
objects = append(objects, pod)
}
podsForSnapshot := make([]*v1.Pod, 0, len(tt.victimPods))
for _, pod := range tt.victimPods {
objects = append(objects, pod)
podsForSnapshot = append(podsForSnapshot, pod)
}
cs := clientsetfake.NewClientset(objects...)
informerFactory := informers.NewSharedInformerFactory(cs, 0)
eventBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{Interface: cs.EventsV1()})
fwk, err := tf.NewFramework(
ctx,
registeredPlugins, "",
frameworkruntime.WithClientSet(cs),
frameworkruntime.WithSnapshotSharedLister(internalcache.NewSnapshot(podsForSnapshot, []*v1.Node{st.MakeNode().Name("node1").Obj()})),
frameworkruntime.WithInformerFactory(informerFactory),
frameworkruntime.WithWaitingPods(waitingPods),
frameworkruntime.WithPodsInPreBind(podsInPreBind),
frameworkruntime.WithLogger(logger),
frameworkruntime.WithEventRecorder(eventBroadcaster.NewRecorder(scheme.Scheme, "test-scheduler")),
frameworkruntime.WithPodActivator(fakeActivator),
frameworkruntime.WithPodNominator(internalqueue.NewSchedulingQueue(nil, informerFactory)),
)
if err != nil {
t.Fatal(err)
}
var victimCtx context.Context
var cancelVictim context.CancelCauseFunc
if tt.addVictimToPrebind {
victimCtx, cancelVictim = context.WithCancelCause(context.Background())
fwk.AddPodInPreBind(tt.inMemoryVictim.UID, cancelVictim)
}
if tt.addVictimToWaiting {
pluginsWaitTime, status := fwk.RunPermitPlugins(ctx, framework.NewCycleState(), tt.inMemoryVictim, "node1")
if !status.IsWait() {
t.Fatalf("Failed to add a pod to waiting list")
}
fwk.AddWaitingPod(tt.inMemoryVictim, pluginsWaitTime)
}
executor := NewExecutor(fwk, feature.Features{EnableAsyncPreemption: true})
fakeActivator.isPreempting = func() bool {
executor.mu.RLock()
defer executor.mu.RUnlock()
return executor.preempting.Has(preemptor.UID())
}
if tt.addVictimToPrebindOnPreempt {
preemptFunc := executor.PreemptPod
executor.PreemptPod = func(ctx context.Context, c Candidate, preemptor ExecutorPreemptor, victim *v1.Pod, pluginName string) (bool, error) {
if victim.UID == tt.inMemoryVictim.UID {
victimCtx, cancelVictim = context.WithCancelCause(context.Background())
fwk.AddPodInPreBind(victim.UID, cancelVictim)
}
return preemptFunc(ctx, c, preemptor, victim, pluginName)
}
}
candidate := &candidate{
name: "node1",
victims: &extenderv1.Victims{
Pods: tt.victimPods,
},
}
executor.prepareCandidateAsync(candidate, preemptor, "test-plugin")
if tt.wantPreemptorActivate {
select {
case <-fakeActivator.activatedCh:
case <-time.After(wait.ForeverTestTimeout):
t.Fatal("Timed out waiting for preemptor activation")
}
} else {
if err := wait.PollUntilContextTimeout(ctx, 10*time.Millisecond, wait.ForeverTestTimeout, false, func(ctx context.Context) (bool, error) {
executor.mu.RLock()
defer executor.mu.RUnlock()
return len(executor.preempting) == 0, nil
}); err != nil {
t.Fatalf("Timed out waiting for async preemption to finish: %v", err)
}
}
mu.RLock()
defer mu.RUnlock()
if fakeActivator.activatedWhilePreempting {
t.Fatal("Preemptor was activated before its preempting state was cleared")
}
wantActivatedPods := 0
if tt.wantPreemptorActivate {
wantActivatedPods = len(preemptor.Pods())
}
if len(fakeActivator.activatedPods) != wantActivatedPods {
t.Fatalf("Activated pod count = %d, want %d; activated pods: %v", len(fakeActivator.activatedPods), wantActivatedPods, fakeActivator.activatedPods)
}
for name := range preemptor.Pods() {
_, gotActivated := fakeActivator.activatedPods[name]
if gotActivated != tt.wantPreemptorActivate {
t.Fatalf("Preemptor pod %q activation = %v, want %v; activated pods: %v", name, gotActivated, tt.wantPreemptorActivate, fakeActivator.activatedPods)
}
}
if (tt.addVictimToPrebind || tt.addVictimToPrebindOnPreempt) && (victimCtx == nil || victimCtx.Err() == nil) {
t.Fatalf("Expected preBind victim context to be cancelled")
}
})
}
}
// waitingPermitPlugin is a PermitPlugin that always returns Wait.
type waitingPermitPlugin struct{}

2
test/integration/scheduler/nominated_node_name/nominated_node_name_test.go
View file

@ -492,7 +492,7 @@ func TestPreemptionAndNominatedNodeNameScenarios(t *testing.T) {
}
preemptPodFn := preemptionPlugin.Executor.PreemptPod
preemptionPlugin.Executor.PreemptPod = func(ctx context.Context, c preemption.Candidate, preemptor preemption.ExecutorPreemptor, victim *v1.Pod, pluginName string) error {
preemptionPlugin.Executor.PreemptPod = func(ctx context.Context, c preemption.Candidate, preemptor preemption.ExecutorPreemptor, victim *v1.Pod, pluginName string) (bool, error) {
// block the preemption goroutine to complete until the test case allows it to proceed.
lock.Lock()
ch, ok := preemptionDoneChannels[preemptor.GetName()]

7
test/integration/scheduler/preemption/preemption_test.go
View file

@ -1311,7 +1311,7 @@ func TestAsyncPreemption(t *testing.T) {
}
preemptPodFn := preemptionPlugin.Executor.PreemptPod
preemptionPlugin.Executor.PreemptPod = func(ctx context.Context, c preemption.Candidate, preemptor preemption.ExecutorPreemptor, victim *v1.Pod, pluginName string) error {
preemptionPlugin.Executor.PreemptPod = func(ctx context.Context, c preemption.Candidate, preemptor preemption.ExecutorPreemptor, victim *v1.Pod, pluginName string) (bool, error) {
// block the preemption goroutine to complete until the test case allows it to proceed.
lock.Lock()
ch, ok := preemptionDoneChannels[preemptor.GetName()]
@ -1851,6 +1851,11 @@ func TestPreemptionRespectsWaitingPod(t *testing.T) {
case <-time.After(wait.ForeverTestTimeout):
t.Fatalf("Timed out waiting for victim to reach WaitOnPermit")
}
if err := wait.PollUntilContextTimeout(testCtx.Ctx, 100*time.Millisecond, wait.ForeverTestTimeout, false, func(ctx context.Context) (bool, error) {
return testCtx.Scheduler.Profiles[v1.DefaultSchedulerName].GetWaitingPod(victim.UID) != nil, nil
}); err != nil {
t.Fatalf("Timed out waiting for victim to be recorded as a waiting pod: %v", err)
}
smallNodeRes := map[v1.ResourceName]string{
v1.ResourceCPU: "1",