github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2026-07-17 20:00:07 +00:00
Code Issues Projects Releases Wiki Activity

Adjust pod group scheduling cycle code

Browse Source
This commit is contained in:
Maciej Skoczeń
2026-02-23 11:10:54 +00:00
parent 2a5b28436c
commit 9b7ad1794d
5 changed files with 22 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/scheduler/backend/cache/snapshot.go vendored
View File

@@ -270,5 +270,5 @@ func (s *Snapshot) forgetAllAssumedPods(logger klog.Logger) {
utilruntime.HandleErrorWithLogger(logger, err, "Failed to forget assumed pod")
}
}
utilruntime.HandleErrorWithLogger(logger, nil, "Found assumed pods in the snapshot that were not forgotten", "assumedPodsCount", len(s.assumedPods))
logger.Error(nil, "Found assumed pods in the snapshot that were not forgotten", "assumedPodsCount", len(s.assumedPods))
}

4
pkg/scheduler/extender_test.go
View File

@@ -358,6 +358,10 @@ func TestSchedulerWithExtenders(t *testing.T) {
}
sched.applyDefaultHandlers()
if err := sched.Cache.UpdateSnapshot(logger, sched.nodeInfoSnapshot); err != nil {
t.Fatalf("Unexpected error updating snapshot: %v", err)
}
podInfoIgnored := queuedPodInfoForPod(&v1.Pod{})
result, err := sched.SchedulePod(ctx, fwk, framework.NewCycleState(), podInfoIgnored)
if test.expectsErr {

10
pkg/scheduler/schedule_one.go
View File

@@ -179,6 +179,10 @@ func (sched *Scheduler) schedulingCycle(
start time.Time,
podsToActivate *framework.PodsToActivate,
) (ScheduleResult, *framework.QueuedPodInfo, *fwk.Status) {
if err := sched.Cache.UpdateSnapshot(klog.FromContext(ctx), sched.nodeInfoSnapshot); err != nil {
return ScheduleResult{nominatingInfo: clearNominatedNode}, podInfo, fwk.AsStatus(err)
}
scheduleResult, status := sched.schedulingAlgorithm(ctx, state, schedFramework, podInfo, start)
if !status.IsSuccess() {
return scheduleResult, podInfo, status
@@ -563,12 +567,6 @@ func (sched *Scheduler) schedulePod(ctx context.Context, fwk framework.Framework
pod := podInfo.Pod
trace := utiltrace.New("Scheduling", utiltrace.Field{Key: "namespace", Value: pod.Namespace}, utiltrace.Field{Key: "name", Value: pod.Name})
defer trace.LogIfLong(100 * time.Millisecond)
if !podInfo.NeedsPodGroupScheduling {
if err := sched.Cache.UpdateSnapshot(klog.FromContext(ctx), sched.nodeInfoSnapshot); err != nil {
return result, err
}
trace.Step("Snapshotting scheduler cache and node infos done")
}
if sched.nodeInfoSnapshot.NumNodes() == 0 {
return result, ErrNoNodesAvailable

21
pkg/scheduler/schedule_one_podgroup.go
View File

@@ -48,11 +48,11 @@ func (sched *Scheduler) scheduleOnePodGroup(ctx context.Context, podGroupInfo *f
schedFwk, err := sched.frameworkForPodGroup(podGroupInfo)
if err != nil {
for _, podInfo := range podGroupInfo.QueuedPodInfos {
podFwk, err := sched.frameworkForPod(podInfo.Pod)
if err != nil {
podFwk, podFwkErr := sched.frameworkForPod(podInfo.Pod)
if podFwkErr != nil {
// This shouldn't happen, because we only accept for scheduling the pods
// which specify a scheduler name that matches one of the profiles.
logger.Error(err, "Error occurred")
logger.Error(podFwkErr, "Error occurred")
sched.SchedulingQueue.Done(podInfo.Pod.UID)
return
}
@@ -70,10 +70,7 @@ func (sched *Scheduler) scheduleOnePodGroup(ctx context.Context, podGroupInfo *f
logger.V(3).Info("Attempting to schedule pod group", "podGroup", klog.KObj(podGroupInfo))
// Synchronously attempt to find a fit for the pod group.
start := time.Now()
sched.podGroupCycle(ctx, schedFwk, podGroupInfo, start)
sched.podGroupCycle(ctx, schedFwk, podGroupInfo)
}
// frameworkForPodGroup obtains the concrete scheduler framework for the entire pod group.
@@ -202,7 +199,10 @@ func (sched *Scheduler) initPodSchedulingContext(ctx context.Context, pod *v1.Po
}
// podGroupCycle runs a pod group scheduling cycle for the given pod group in a single cluster snapshot.
func (sched *Scheduler) podGroupCycle(ctx context.Context, schedFwk framework.Framework, podGroupInfo *framework.QueuedPodGroupInfo, start time.Time) {
func (sched *Scheduler) podGroupCycle(ctx context.Context, schedFwk framework.Framework, podGroupInfo *framework.QueuedPodGroupInfo) {
// Synchronously attempt to find a fit for the pod group.
start := time.Now()
podGroupCycleCtx, cancel := context.WithCancel(ctx)
defer cancel()
logger := klog.FromContext(podGroupCycleCtx)
@@ -434,7 +434,10 @@ func (sched *Scheduler) submitPodGroupAlgorithmResult(ctx context.Context, sched
sched.FailureHandler(ctx, schedFwk, pInfo, status, nominatingInfo, podSchedulingStart)
unschedulablePods++
default:
utilruntime.HandleErrorWithLogger(podCtx.logger, nil, "Unexpected pod group scheduling algorithm result status", "result", result.status)
err := fmt.Errorf("got unexpected pod group scheduling algorithm result status: %q", result.status)
utilruntime.HandleErrorWithLogger(podCtx.logger, err, "Unexpected error", "podGroup", klog.KObj(podGroupInfo), "pod", klog.KObj(pInfo.Pod))
sched.FailureHandler(ctx, schedFwk, pInfo, fwk.AsStatus(err), nominatingInfo, podSchedulingStart)
unschedulablePods++
}
} else {
// TBD: Add a message to status if the pod used features for which finding a placement cannot be guaranteed,

2
pkg/scheduler/schedule_one_podgroup_test.go
View File

@@ -378,7 +378,7 @@ func TestPodGroupCycle_UpdateSnapshotError(t *testing.T) {
},
}
sched.podGroupCycle(ctx, schedFwk, pgInfo, time.Now())
sched.podGroupCycle(ctx, schedFwk, pgInfo)
if !failureHandlerCalled {
t.Errorf("Expected FailureHandler to be called after UpdateSnapshot failed")