mirror of
https://github.com/k3s-io/kubernetes.git
synced 2026-07-18 04:54:54 +00:00
Move unschedulablePods struct to a separate file
This commit is contained in:
Omar Nasser
Omar Nasser
@@ -64,9 +64,9 @@ const (
|
||||
// will be used.
|
||||
DefaultPodMaxInUnschedulablePodsDuration time.Duration = 5 * time.Minute
|
||||
// Scheduling queue names
|
||||
activeQ = "Active"
|
||||
backoffQ = "Backoff"
|
||||
unschedulablePods = "Unschedulable"
|
||||
activeQ = "Active"
|
||||
backoffQ = "Backoff"
|
||||
unschedulableQ = "Unschedulable"
|
||||
|
||||
preEnqueue = "PreEnqueue"
|
||||
)
|
||||
@@ -172,7 +172,7 @@ type PriorityQueue struct {
|
||||
activeQ activeQueuer
|
||||
backoffQ backoffQueuer
|
||||
// unschedulablePods holds pods that have been tried and determined unschedulable.
|
||||
unschedulablePods *UnschedulablePods
|
||||
unschedulablePods *unschedulablePods
|
||||
// moveRequestCycle caches the sequence number of scheduling cycle when we
|
||||
// received a move request. Unschedulable pods in and before this scheduling
|
||||
// cycle will be put back to activeQueue if we were trying to schedule them
|
||||
@@ -623,7 +623,7 @@ func (p *PriorityQueue) moveToActiveQ(logger klog.Logger, pInfo *framework.Queue
|
||||
return
|
||||
}
|
||||
p.unschedulablePods.addOrUpdate(pInfo, event)
|
||||
logger.V(5).Info("Pod moved to an internal scheduling queue, because the pod is gated", "pod", klog.KObj(pInfo.Pod), "event", event, "queue", unschedulablePods)
|
||||
logger.V(5).Info("Pod moved to an internal scheduling queue, because the pod is gated", "pod", klog.KObj(pInfo.Pod), "event", event, "queue", unschedulableQ)
|
||||
return
|
||||
}
|
||||
if pInfo.InitialAttemptTimestamp == nil {
|
||||
@@ -656,7 +656,7 @@ func (p *PriorityQueue) moveToBackoffQ(logger klog.Logger, pInfo *framework.Queu
|
||||
if pInfo.Gated() {
|
||||
if p.unschedulablePods.get(pInfo.Pod) == nil {
|
||||
p.unschedulablePods.addOrUpdate(pInfo, event)
|
||||
logger.V(5).Info("Pod moved to an internal scheduling queue", "pod", klog.KObj(pInfo.Pod), "event", event, "queue", unschedulablePods)
|
||||
logger.V(5).Info("Pod moved to an internal scheduling queue", "pod", klog.KObj(pInfo.Pod), "event", event, "queue", unschedulableQ)
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -813,7 +813,7 @@ func (p *PriorityQueue) addUnschedulableWithoutQueueingHint(logger klog.Logger,
|
||||
}
|
||||
} else {
|
||||
p.unschedulablePods.addOrUpdate(pInfo, framework.ScheduleAttemptFailure)
|
||||
logger.V(5).Info("Pod moved to an internal scheduling queue", "pod", klog.KObj(pod), "event", framework.ScheduleAttemptFailure, "queue", unschedulablePods)
|
||||
logger.V(5).Info("Pod moved to an internal scheduling queue", "pod", klog.KObj(pod), "event", framework.ScheduleAttemptFailure, "queue", unschedulableQ)
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -1017,7 +1017,7 @@ func (p *PriorityQueue) Update(logger klog.Logger, oldPod, newPod *v1.Pod) {
|
||||
for _, evt := range events {
|
||||
hint := p.isPodWorthRequeuing(logger, pInfo, evt, oldPod, newPod)
|
||||
queue := p.requeuePodWithQueueingStrategy(logger, pInfo, hint, evt.Label())
|
||||
if queue != unschedulablePods {
|
||||
if queue != unschedulableQ {
|
||||
logger.V(5).Info("Pod moved to an internal scheduling queue because the Pod is updated", "pod", klog.KObj(newPod), "event", evt.Label(), "queue", queue)
|
||||
p.unschedulablePods.delete(pInfo.Pod, gated)
|
||||
}
|
||||
@@ -1141,7 +1141,7 @@ func (p *PriorityQueue) MoveAllToActiveOrBackoffQueue(logger klog.Logger, event
|
||||
func (p *PriorityQueue) requeuePodWithQueueingStrategy(logger klog.Logger, pInfo *framework.QueuedPodInfo, strategy queueingStrategy, event string) string {
|
||||
if strategy == queueSkip {
|
||||
p.unschedulablePods.addOrUpdate(pInfo, event)
|
||||
return unschedulablePods
|
||||
return unschedulableQ
|
||||
}
|
||||
|
||||
// Pod might have completed its backoff time while being in unschedulablePods,
|
||||
@@ -1150,7 +1150,7 @@ func (p *PriorityQueue) requeuePodWithQueueingStrategy(logger klog.Logger, pInfo
|
||||
if added := p.moveToBackoffQ(logger, pInfo, event); added {
|
||||
return backoffQ
|
||||
}
|
||||
return unschedulablePods
|
||||
return unschedulableQ
|
||||
}
|
||||
|
||||
// Reach here if schedulingHint is QueueImmediately, or schedulingHint is Queue but the pod is not backing off.
|
||||
@@ -1158,7 +1158,7 @@ func (p *PriorityQueue) requeuePodWithQueueingStrategy(logger klog.Logger, pInfo
|
||||
return activeQ
|
||||
}
|
||||
// Pod is gated. We don't have to push it back to unschedulable queue, because moveToActiveQ should already have done that.
|
||||
return unschedulablePods
|
||||
return unschedulableQ
|
||||
}
|
||||
|
||||
// NOTE: this function assumes lock has been acquired in caller
|
||||
@@ -1362,77 +1362,6 @@ func (p *PriorityQueue) newQueuedPodInfo(pod *v1.Pod, plugins ...string) *framew
|
||||
}
|
||||
}
|
||||
|
||||
// UnschedulablePods holds pods that cannot be scheduled. This data structure
|
||||
// is used to implement unschedulablePods.
|
||||
type UnschedulablePods struct {
|
||||
// podInfoMap is a map key by a pod's full-name and the value is a pointer to the QueuedPodInfo.
|
||||
podInfoMap map[string]*framework.QueuedPodInfo
|
||||
keyFunc func(*v1.Pod) string
|
||||
// unschedulableRecorder/gatedRecorder updates the counter when elements of an unschedulablePodsMap
|
||||
// get added or removed, and it does nothing if it's nil.
|
||||
unschedulableRecorder, gatedRecorder metrics.MetricRecorder
|
||||
}
|
||||
|
||||
// addOrUpdate adds a pod to the unschedulable podInfoMap.
|
||||
// The event should show which event triggered the addition and is used for the metric recording.
|
||||
func (u *UnschedulablePods) addOrUpdate(pInfo *framework.QueuedPodInfo, event string) {
|
||||
podID := u.keyFunc(pInfo.Pod)
|
||||
if _, exists := u.podInfoMap[podID]; !exists {
|
||||
if pInfo.Gated() && u.gatedRecorder != nil {
|
||||
u.gatedRecorder.Inc()
|
||||
} else if !pInfo.Gated() && u.unschedulableRecorder != nil {
|
||||
u.unschedulableRecorder.Inc()
|
||||
}
|
||||
metrics.SchedulerQueueIncomingPods.WithLabelValues("unschedulable", event).Inc()
|
||||
}
|
||||
u.podInfoMap[podID] = pInfo
|
||||
}
|
||||
|
||||
// delete deletes a pod from the unschedulable podInfoMap.
|
||||
// The `gated` parameter is used to figure out which metric should be decreased.
|
||||
func (u *UnschedulablePods) delete(pod *v1.Pod, gated bool) {
|
||||
podID := u.keyFunc(pod)
|
||||
if _, exists := u.podInfoMap[podID]; exists {
|
||||
if gated && u.gatedRecorder != nil {
|
||||
u.gatedRecorder.Dec()
|
||||
} else if !gated && u.unschedulableRecorder != nil {
|
||||
u.unschedulableRecorder.Dec()
|
||||
}
|
||||
}
|
||||
delete(u.podInfoMap, podID)
|
||||
}
|
||||
|
||||
// get returns the QueuedPodInfo if a pod with the same key as the key of the given "pod"
|
||||
// is found in the map. It returns nil otherwise.
|
||||
func (u *UnschedulablePods) get(pod *v1.Pod) *framework.QueuedPodInfo {
|
||||
podKey := u.keyFunc(pod)
|
||||
if pInfo, exists := u.podInfoMap[podKey]; exists {
|
||||
return pInfo
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// clear removes all the entries from the unschedulable podInfoMap.
|
||||
func (u *UnschedulablePods) clear() {
|
||||
u.podInfoMap = make(map[string]*framework.QueuedPodInfo)
|
||||
if u.unschedulableRecorder != nil {
|
||||
u.unschedulableRecorder.Clear()
|
||||
}
|
||||
if u.gatedRecorder != nil {
|
||||
u.gatedRecorder.Clear()
|
||||
}
|
||||
}
|
||||
|
||||
// newUnschedulablePods initializes a new object of UnschedulablePods.
|
||||
func newUnschedulablePods(unschedulableRecorder, gatedRecorder metrics.MetricRecorder) *UnschedulablePods {
|
||||
return &UnschedulablePods{
|
||||
podInfoMap: make(map[string]*framework.QueuedPodInfo),
|
||||
keyFunc: util.GetPodFullName,
|
||||
unschedulableRecorder: unschedulableRecorder,
|
||||
gatedRecorder: gatedRecorder,
|
||||
}
|
||||
}
|
||||
|
||||
func podInfoKeyFunc(pInfo *framework.QueuedPodInfo) string {
|
||||
return cache.NewObjectName(pInfo.Pod.Namespace, pInfo.Pod.Name).String()
|
||||
}
|
||||
|
||||
@@ -49,7 +49,6 @@ import (
|
||||
"k8s.io/kubernetes/pkg/scheduler/framework/plugins/schedulinggates"
|
||||
"k8s.io/kubernetes/pkg/scheduler/metrics"
|
||||
st "k8s.io/kubernetes/pkg/scheduler/testing"
|
||||
"k8s.io/kubernetes/pkg/scheduler/util"
|
||||
testingclock "k8s.io/utils/clock/testing"
|
||||
)
|
||||
|
||||
@@ -1225,7 +1224,7 @@ func TestPriorityQueue_Update(t *testing.T) {
|
||||
},
|
||||
{
|
||||
name: "when updating a pod which is in unschedulable pods but the plugin returns skip, it will remain in unschedulablePods",
|
||||
wantQ: unschedulablePods,
|
||||
wantQ: unschedulableQ,
|
||||
prepareFunc: func(t *testing.T, logger klog.Logger, q *PriorityQueue) (oldPod, newPod *v1.Pod) {
|
||||
q.unschedulablePods.addOrUpdate(q.newQueuedPodInfo(medPriorityPodInfo.Pod, skipPlugin), framework.EventUnscheduledPodAdd.Label())
|
||||
updatedPod := medPriorityPodInfo.Pod.DeepCopy()
|
||||
@@ -1289,7 +1288,7 @@ func TestPriorityQueue_Update(t *testing.T) {
|
||||
})
|
||||
|
||||
if pInfoFromUnsched := q.unschedulablePods.get(newPod); pInfoFromUnsched != nil {
|
||||
if tt.wantQ != unschedulablePods {
|
||||
if tt.wantQ != unschedulableQ {
|
||||
t.Errorf("expected pod %s to not be queued to unschedulablePods, but it was", newPod.Name)
|
||||
}
|
||||
pInfo = pInfoFromUnsched
|
||||
@@ -1920,7 +1919,7 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithQueueingHint(t *testing.
|
||||
name: "Skip queues pod to unschedulablePods",
|
||||
podInfo: &framework.QueuedPodInfo{PodInfo: mustNewPodInfo(p), UnschedulablePlugins: sets.New("foo")},
|
||||
hint: queueHintReturnSkip,
|
||||
expectedQ: unschedulablePods,
|
||||
expectedQ: unschedulableQ,
|
||||
},
|
||||
{
|
||||
name: "QueueHintFunction is not called when Pod is gated by the plugin that isn't interested in the event",
|
||||
@@ -1931,7 +1930,7 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithQueueingHint(t *testing.
|
||||
hint: func(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (fwk.QueueingHint, error) {
|
||||
return fwk.Queue, fmt.Errorf("QueueingHintFn should not be called as pod is gated")
|
||||
},
|
||||
expectedQ: unschedulablePods,
|
||||
expectedQ: unschedulableQ,
|
||||
},
|
||||
{
|
||||
name: "QueueHintFunction is called when Pod is gated by the plugin that is interested in the event",
|
||||
@@ -1987,7 +1986,7 @@ func TestPriorityQueue_MoveAllToActiveOrBackoffQueueWithQueueingHint(t *testing.
|
||||
t.Fatalf("expected pod to be queued to activeQ, but it was not")
|
||||
}
|
||||
|
||||
if q.unschedulablePods.get(test.podInfo.Pod) == nil && test.expectedQ == unschedulablePods {
|
||||
if q.unschedulablePods.get(test.podInfo.Pod) == nil && test.expectedQ == unschedulableQ {
|
||||
t.Fatalf("expected pod to be queued to unschedulablePods, but it was not")
|
||||
}
|
||||
})
|
||||
@@ -2732,115 +2731,6 @@ func TestPriorityQueue_NewWithOptions(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestUnschedulablePodsMap(t *testing.T) {
|
||||
var pods = []*v1.Pod{
|
||||
st.MakePod().Name("p0").Namespace("ns1").Annotation("annot1", "val1").NominatedNodeName("node1").Obj(),
|
||||
st.MakePod().Name("p1").Namespace("ns1").Annotation("annot", "val").Obj(),
|
||||
st.MakePod().Name("p2").Namespace("ns2").Annotation("annot2", "val2").Annotation("annot3", "val3").NominatedNodeName("node3").Obj(),
|
||||
st.MakePod().Name("p3").Namespace("ns4").Annotation("annot2", "val2").Annotation("annot3", "val3").NominatedNodeName("node1").Obj(),
|
||||
}
|
||||
var updatedPods = make([]*v1.Pod, len(pods))
|
||||
updatedPods[0] = pods[0].DeepCopy()
|
||||
updatedPods[1] = pods[1].DeepCopy()
|
||||
updatedPods[3] = pods[3].DeepCopy()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
podsToAdd []*v1.Pod
|
||||
expectedMapAfterAdd map[string]*framework.QueuedPodInfo
|
||||
podsToUpdate []*v1.Pod
|
||||
expectedMapAfterUpdate map[string]*framework.QueuedPodInfo
|
||||
podsToDelete []*v1.Pod
|
||||
expectedMapAfterDelete map[string]*framework.QueuedPodInfo
|
||||
}{
|
||||
{
|
||||
name: "create, update, delete subset of pods",
|
||||
podsToAdd: []*v1.Pod{pods[0], pods[1], pods[2], pods[3]},
|
||||
expectedMapAfterAdd: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[0]): {PodInfo: mustNewTestPodInfo(t, pods[0]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, pods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, pods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToUpdate: []*v1.Pod{updatedPods[0]},
|
||||
expectedMapAfterUpdate: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[0]): {PodInfo: mustNewTestPodInfo(t, updatedPods[0]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, pods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, pods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToDelete: []*v1.Pod{pods[0], pods[1]},
|
||||
expectedMapAfterDelete: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, pods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "create, update, delete all",
|
||||
podsToAdd: []*v1.Pod{pods[0], pods[3]},
|
||||
expectedMapAfterAdd: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[0]): {PodInfo: mustNewTestPodInfo(t, pods[0]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, pods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToUpdate: []*v1.Pod{updatedPods[3]},
|
||||
expectedMapAfterUpdate: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[0]): {PodInfo: mustNewTestPodInfo(t, pods[0]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, updatedPods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToDelete: []*v1.Pod{pods[0], pods[3]},
|
||||
expectedMapAfterDelete: map[string]*framework.QueuedPodInfo{},
|
||||
},
|
||||
{
|
||||
name: "delete non-existing and existing pods",
|
||||
podsToAdd: []*v1.Pod{pods[1], pods[2]},
|
||||
expectedMapAfterAdd: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, pods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToUpdate: []*v1.Pod{updatedPods[1]},
|
||||
expectedMapAfterUpdate: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, updatedPods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToDelete: []*v1.Pod{pods[2], pods[3]},
|
||||
expectedMapAfterDelete: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, updatedPods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
upm := newUnschedulablePods(nil, nil)
|
||||
for _, p := range test.podsToAdd {
|
||||
upm.addOrUpdate(newQueuedPodInfoForLookup(p), framework.EventUnscheduledPodAdd.Label())
|
||||
}
|
||||
if diff := cmp.Diff(test.expectedMapAfterAdd, upm.podInfoMap, cmpopts.IgnoreUnexported(framework.PodInfo{})); diff != "" {
|
||||
t.Errorf("Unexpected map after adding pods(-want, +got):\n%s", diff)
|
||||
}
|
||||
|
||||
if len(test.podsToUpdate) > 0 {
|
||||
for _, p := range test.podsToUpdate {
|
||||
upm.addOrUpdate(newQueuedPodInfoForLookup(p), framework.EventUnscheduledPodUpdate.Label())
|
||||
}
|
||||
if diff := cmp.Diff(test.expectedMapAfterUpdate, upm.podInfoMap, cmpopts.IgnoreUnexported(framework.PodInfo{})); diff != "" {
|
||||
t.Errorf("Unexpected map after updating pods (-want, +got):\n%s", diff)
|
||||
}
|
||||
}
|
||||
for _, p := range test.podsToDelete {
|
||||
upm.delete(p, false)
|
||||
}
|
||||
if diff := cmp.Diff(test.expectedMapAfterDelete, upm.podInfoMap, cmpopts.IgnoreUnexported(framework.PodInfo{})); diff != "" {
|
||||
t.Errorf("Unexpected map after deleting pods (-want, +got):\n%s", diff)
|
||||
}
|
||||
upm.clear()
|
||||
if len(upm.podInfoMap) != 0 {
|
||||
t.Errorf("Expected the map to be empty, but has %v elements.", len(upm.podInfoMap))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSchedulingQueue_Close(t *testing.T) {
|
||||
logger, ctx := ktesting.NewTestContext(t)
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
|
||||
94
pkg/scheduler/backend/queue/unschedulable_pods.go
Normal file
94
pkg/scheduler/backend/queue/unschedulable_pods.go
Normal file
@@ -0,0 +1,94 @@
|
||||
/*
|
||||
Copyright 2025 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package queue
|
||||
|
||||
import (
|
||||
v1 "k8s.io/api/core/v1"
|
||||
"k8s.io/kubernetes/pkg/scheduler/framework"
|
||||
"k8s.io/kubernetes/pkg/scheduler/metrics"
|
||||
"k8s.io/kubernetes/pkg/scheduler/util"
|
||||
)
|
||||
|
||||
// unschedulablePods holds pods that cannot be scheduled.
|
||||
type unschedulablePods struct {
|
||||
// podInfoMap is a map key by a pod's full-name and the value is a pointer to the QueuedPodInfo.
|
||||
podInfoMap map[string]*framework.QueuedPodInfo
|
||||
keyFunc func(*v1.Pod) string
|
||||
// unschedulableRecorder/gatedRecorder updates the counter when elements of an unschedulablePods
|
||||
// get added or removed, and it does nothing if it's nil.
|
||||
unschedulableRecorder, gatedRecorder metrics.MetricRecorder
|
||||
}
|
||||
|
||||
// newUnschedulablePods initializes a new object of unschedulablePods.
|
||||
func newUnschedulablePods(unschedulableRecorder, gatedRecorder metrics.MetricRecorder) *unschedulablePods {
|
||||
return &unschedulablePods{
|
||||
podInfoMap: make(map[string]*framework.QueuedPodInfo),
|
||||
keyFunc: util.GetPodFullName,
|
||||
unschedulableRecorder: unschedulableRecorder,
|
||||
gatedRecorder: gatedRecorder,
|
||||
}
|
||||
}
|
||||
|
||||
// addOrUpdate adds a pod to the unschedulable podInfoMap.
|
||||
// The event should show which event triggered the addition and is used for the metric recording.
|
||||
func (u *unschedulablePods) addOrUpdate(pInfo *framework.QueuedPodInfo, event string) {
|
||||
podID := u.keyFunc(pInfo.Pod)
|
||||
if _, exists := u.podInfoMap[podID]; !exists {
|
||||
if pInfo.Gated() && u.gatedRecorder != nil {
|
||||
u.gatedRecorder.Inc()
|
||||
} else if !pInfo.Gated() && u.unschedulableRecorder != nil {
|
||||
u.unschedulableRecorder.Inc()
|
||||
}
|
||||
metrics.SchedulerQueueIncomingPods.WithLabelValues("unschedulable", event).Inc()
|
||||
}
|
||||
u.podInfoMap[podID] = pInfo
|
||||
}
|
||||
|
||||
// delete deletes a pod from the unschedulable podInfoMap.
|
||||
// The `gated` parameter is used to figure out which metric should be decreased.
|
||||
func (u *unschedulablePods) delete(pod *v1.Pod, gated bool) {
|
||||
podID := u.keyFunc(pod)
|
||||
if _, exists := u.podInfoMap[podID]; exists {
|
||||
if gated && u.gatedRecorder != nil {
|
||||
u.gatedRecorder.Dec()
|
||||
} else if !gated && u.unschedulableRecorder != nil {
|
||||
u.unschedulableRecorder.Dec()
|
||||
}
|
||||
}
|
||||
delete(u.podInfoMap, podID)
|
||||
}
|
||||
|
||||
// get returns the QueuedPodInfo if a pod with the same key as the key of the given "pod"
|
||||
// is found in the map. It returns nil otherwise.
|
||||
func (u *unschedulablePods) get(pod *v1.Pod) *framework.QueuedPodInfo {
|
||||
podKey := u.keyFunc(pod)
|
||||
if pInfo, exists := u.podInfoMap[podKey]; exists {
|
||||
return pInfo
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// clear removes all the entries from the unschedulable podInfoMap.
|
||||
func (u *unschedulablePods) clear() {
|
||||
u.podInfoMap = make(map[string]*framework.QueuedPodInfo)
|
||||
if u.unschedulableRecorder != nil {
|
||||
u.unschedulableRecorder.Clear()
|
||||
}
|
||||
if u.gatedRecorder != nil {
|
||||
u.gatedRecorder.Clear()
|
||||
}
|
||||
}
|
||||
138
pkg/scheduler/backend/queue/unschedulable_pods_test.go
Normal file
138
pkg/scheduler/backend/queue/unschedulable_pods_test.go
Normal file
@@ -0,0 +1,138 @@
|
||||
/*
|
||||
Copyright 2025 The Kubernetes Authors.
|
||||
|
||||
Licensed under the Apache License, Version 2.0 (the "License");
|
||||
you may not use this file except in compliance with the License.
|
||||
You may obtain a copy of the License at
|
||||
|
||||
http://www.apache.org/licenses/LICENSE-2.0
|
||||
|
||||
Unless required by applicable law or agreed to in writing, software
|
||||
distributed under the License is distributed on an "AS IS" BASIS,
|
||||
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
package queue
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/google/go-cmp/cmp"
|
||||
"github.com/google/go-cmp/cmp/cmpopts"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
"k8s.io/kubernetes/pkg/scheduler/framework"
|
||||
st "k8s.io/kubernetes/pkg/scheduler/testing"
|
||||
"k8s.io/kubernetes/pkg/scheduler/util"
|
||||
)
|
||||
|
||||
func TestUnschedulablePods(t *testing.T) {
|
||||
var pods = []*v1.Pod{
|
||||
st.MakePod().Name("p0").Namespace("ns1").Annotation("annot1", "val1").NominatedNodeName("node1").Obj(),
|
||||
st.MakePod().Name("p1").Namespace("ns1").Annotation("annot", "val").Obj(),
|
||||
st.MakePod().Name("p2").Namespace("ns2").Annotation("annot2", "val2").Annotation("annot3", "val3").NominatedNodeName("node3").Obj(),
|
||||
st.MakePod().Name("p3").Namespace("ns4").Annotation("annot2", "val2").Annotation("annot3", "val3").NominatedNodeName("node1").Obj(),
|
||||
}
|
||||
var updatedPods = make([]*v1.Pod, len(pods))
|
||||
updatedPods[0] = pods[0].DeepCopy()
|
||||
updatedPods[1] = pods[1].DeepCopy()
|
||||
updatedPods[3] = pods[3].DeepCopy()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
podsToAdd []*v1.Pod
|
||||
expectedMapAfterAdd map[string]*framework.QueuedPodInfo
|
||||
podsToUpdate []*v1.Pod
|
||||
expectedMapAfterUpdate map[string]*framework.QueuedPodInfo
|
||||
podsToDelete []*v1.Pod
|
||||
expectedMapAfterDelete map[string]*framework.QueuedPodInfo
|
||||
}{
|
||||
{
|
||||
name: "create, update, delete subset of pods",
|
||||
podsToAdd: []*v1.Pod{pods[0], pods[1], pods[2], pods[3]},
|
||||
expectedMapAfterAdd: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[0]): {PodInfo: mustNewTestPodInfo(t, pods[0]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, pods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, pods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToUpdate: []*v1.Pod{updatedPods[0]},
|
||||
expectedMapAfterUpdate: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[0]): {PodInfo: mustNewTestPodInfo(t, updatedPods[0]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, pods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, pods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToDelete: []*v1.Pod{pods[0], pods[1]},
|
||||
expectedMapAfterDelete: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, pods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "create, update, delete all",
|
||||
podsToAdd: []*v1.Pod{pods[0], pods[3]},
|
||||
expectedMapAfterAdd: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[0]): {PodInfo: mustNewTestPodInfo(t, pods[0]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, pods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToUpdate: []*v1.Pod{updatedPods[3]},
|
||||
expectedMapAfterUpdate: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[0]): {PodInfo: mustNewTestPodInfo(t, pods[0]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[3]): {PodInfo: mustNewTestPodInfo(t, updatedPods[3]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToDelete: []*v1.Pod{pods[0], pods[3]},
|
||||
expectedMapAfterDelete: map[string]*framework.QueuedPodInfo{},
|
||||
},
|
||||
{
|
||||
name: "delete non-existing and existing pods",
|
||||
podsToAdd: []*v1.Pod{pods[1], pods[2]},
|
||||
expectedMapAfterAdd: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, pods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToUpdate: []*v1.Pod{updatedPods[1]},
|
||||
expectedMapAfterUpdate: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, updatedPods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
util.GetPodFullName(pods[2]): {PodInfo: mustNewTestPodInfo(t, pods[2]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
podsToDelete: []*v1.Pod{pods[2], pods[3]},
|
||||
expectedMapAfterDelete: map[string]*framework.QueuedPodInfo{
|
||||
util.GetPodFullName(pods[1]): {PodInfo: mustNewTestPodInfo(t, updatedPods[1]), UnschedulablePlugins: sets.New[string]()},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
upm := newUnschedulablePods(nil, nil)
|
||||
for _, p := range test.podsToAdd {
|
||||
upm.addOrUpdate(newQueuedPodInfoForLookup(p), framework.EventUnscheduledPodAdd.Label())
|
||||
}
|
||||
if diff := cmp.Diff(test.expectedMapAfterAdd, upm.podInfoMap, cmpopts.IgnoreUnexported(framework.PodInfo{})); diff != "" {
|
||||
t.Errorf("Unexpected map after adding pods(-want, +got):\n%s", diff)
|
||||
}
|
||||
|
||||
if len(test.podsToUpdate) > 0 {
|
||||
for _, p := range test.podsToUpdate {
|
||||
upm.addOrUpdate(newQueuedPodInfoForLookup(p), framework.EventUnscheduledPodUpdate.Label())
|
||||
}
|
||||
if diff := cmp.Diff(test.expectedMapAfterUpdate, upm.podInfoMap, cmpopts.IgnoreUnexported(framework.PodInfo{})); diff != "" {
|
||||
t.Errorf("Unexpected map after updating pods (-want, +got):\n%s", diff)
|
||||
}
|
||||
}
|
||||
for _, p := range test.podsToDelete {
|
||||
upm.delete(p, false)
|
||||
}
|
||||
if diff := cmp.Diff(test.expectedMapAfterDelete, upm.podInfoMap, cmpopts.IgnoreUnexported(framework.PodInfo{})); diff != "" {
|
||||
t.Errorf("Unexpected map after deleting pods (-want, +got):\n%s", diff)
|
||||
}
|
||||
upm.clear()
|
||||
if len(upm.podInfoMap) != 0 {
|
||||
t.Errorf("Expected the map to be empty, but has %v elements.", len(upm.podInfoMap))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user