Merge pull request #119396 from wackxu/NodeUnschedulableHintFunc

NodeUnschedulable: scheduler queueing hints

pkg/scheduler/framework/plugins/nodeunschedulable/node_unschedulable.go

@@ -22,8 +22,10 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	v1helper "k8s.io/component-helpers/scheduling/corev1"
+	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
 	"k8s.io/kubernetes/pkg/scheduler/framework/plugins/names"
+	"k8s.io/kubernetes/pkg/scheduler/util"
 )
 
 // NodeUnschedulable plugin filters nodes that set node.Spec.Unschedulable=true unless
@@ -48,10 +50,34 @@ const (
 // failed by this plugin schedulable.
 func (pl *NodeUnschedulable) EventsToRegister() []framework.ClusterEventWithHint {
 	return []framework.ClusterEventWithHint{
-		{Event: framework.ClusterEvent{Resource: framework.Node, ActionType: framework.Add | framework.UpdateNodeTaint}},
+		{Event: framework.ClusterEvent{Resource: framework.Node, ActionType: framework.Add | framework.UpdateNodeTaint}, QueueingHintFn: pl.isSchedulableAfterNodeChange},
 	}
 }
 
+// isSchedulableAfterNodeChange is invoked for all node events reported by
+// an informer. It checks whether that change made a previously unschedulable
+// pod schedulable.
+func (pl *NodeUnschedulable) isSchedulableAfterNodeChange(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) framework.QueueingHint {
+	originalNode, modifiedNode, err := util.As[*v1.Node](oldObj, newObj)
+	if err != nil {
+		logger.Error(err, "unexpected objects in isSchedulableAfterNodeChange", "oldObj", oldObj, "newObj", newObj)
+		return framework.QueueAfterBackoff
+	}
+
+	originalNodeSchedulable, modifiedNodeSchedulable := false, !modifiedNode.Spec.Unschedulable
+	if originalNode != nil {
+		originalNodeSchedulable = !originalNode.Spec.Unschedulable
+	}
+
+	if !originalNodeSchedulable && modifiedNodeSchedulable {
+		logger.V(4).Info("node was created or updated, pod may be schedulable now", "pod", klog.KObj(pod), "node", klog.KObj(modifiedNode))
+		return framework.QueueAfterBackoff
+	}
+
+	logger.V(4).Info("node was created or updated, but it doesn't make this pod schedulable", "pod", klog.KObj(pod), "node", klog.KObj(modifiedNode))
+	return framework.QueueSkip
+}
+
 // Name returns name of the plugin. It is used in logs, etc.
 func (pl *NodeUnschedulable) Name() string {
 	return Name

pkg/scheduler/framework/plugins/nodeunschedulable/node_unschedulable_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 
 	v1 "k8s.io/api/core/v1"
+	"k8s.io/klog/v2/ktesting"
 	"k8s.io/kubernetes/pkg/scheduler/framework"
 )
 
@@ -82,3 +83,96 @@ func TestNodeUnschedulable(t *testing.T) {
 		}
 	}
 }
+
+func TestIsSchedulableAfterNodeChange(t *testing.T) {
+	testCases := []struct {
+		name           string
+		pod            *v1.Pod
+		oldObj, newObj interface{}
+		expectedHint   framework.QueueingHint
+	}{
+		{
+			name:         "backoff-wrong-new-object",
+			pod:          &v1.Pod{},
+			newObj:       "not-a-node",
+			expectedHint: framework.QueueAfterBackoff,
+		},
+		{
+			name: "backoff-wrong-old-object",
+			pod:  &v1.Pod{},
+			newObj: &v1.Node{
+				Spec: v1.NodeSpec{
+					Unschedulable: true,
+				},
+			},
+			oldObj:       "not-a-node",
+			expectedHint: framework.QueueAfterBackoff,
+		},
+		{
+			name: "skip-queue-on-unschedulable-node-added",
+			pod:  &v1.Pod{},
+			newObj: &v1.Node{
+				Spec: v1.NodeSpec{
+					Unschedulable: true,
+				},
+			},
+			expectedHint: framework.QueueSkip,
+		},
+		{
+			name: "queue-on-schedulable-node-added",
+			pod:  &v1.Pod{},
+			newObj: &v1.Node{
+				Spec: v1.NodeSpec{
+					Unschedulable: false,
+				},
+			},
+			expectedHint: framework.QueueAfterBackoff,
+		},
+		{
+			name: "skip-unrelated-change",
+			pod:  &v1.Pod{},
+			newObj: &v1.Node{
+				Spec: v1.NodeSpec{
+					Unschedulable: true,
+					Taints: []v1.Taint{
+						{
+							Key:    v1.TaintNodeNotReady,
+							Effect: v1.TaintEffectNoExecute,
+						},
+					},
+				},
+			},
+			oldObj: &v1.Node{
+				Spec: v1.NodeSpec{
+					Unschedulable: true,
+				},
+			},
+			expectedHint: framework.QueueSkip,
+		},
+		{
+			name: "queue-on-unschedulable-field-change",
+			pod:  &v1.Pod{},
+			newObj: &v1.Node{
+				Spec: v1.NodeSpec{
+					Unschedulable: false,
+				},
+			},
+			oldObj: &v1.Node{
+				Spec: v1.NodeSpec{
+					Unschedulable: true,
+				},
+			},
+			expectedHint: framework.QueueAfterBackoff,
+		},
+	}
+
+	for _, testCase := range testCases {
+		t.Run(testCase.name, func(t *testing.T) {
+			logger, _ := ktesting.NewTestContext(t)
+			pl := &NodeUnschedulable{}
+			if got := pl.isSchedulableAfterNodeChange(logger, testCase.pod, testCase.oldObj, testCase.newObj); got != testCase.expectedHint {
+				t.Errorf("isSchedulableAfterNodeChange() = %v, want %v", got, testCase.expectedHint)
+			}
+		})
+	}
+}