Merge pull request #123117 from kerthcet/fix/wild-resource

Fix registered wildcard clusterEvents doesn't work in scheduler requeueing

pkg/scheduler/framework/types.go

@@ -83,6 +83,14 @@ const (
 	CSINode               GVK = "storage.k8s.io/CSINode"
 	CSIDriver             GVK = "storage.k8s.io/CSIDriver"
 	CSIStorageCapacity    GVK = "storage.k8s.io/CSIStorageCapacity"
+
+	// WildCard is a special GVK to match all resources.
+	// e.g., If you register `{Resource: "*", ActionType: All}` in EventsToRegister,
+	// all coming clusterEvents will be admitted. Be careful to register it, it will
+	// increase the computing pressure in requeueing unless you really need it.
+	//
+	// Meanwhile, if the coming clusterEvent is a wildcard one, all pods
+	// will be moved from unschedulablePod pool to activeQ/backoffQ forcibly.
 	WildCard GVK = "*"
 )
 
@@ -144,6 +152,17 @@ func (ce ClusterEvent) IsWildCard() bool {
 	return ce.Resource == WildCard && ce.ActionType == All
 }
 
+// Match returns true if ClusterEvent is matched with the coming event.
+// If the ce.Resource is "*", there's no requirement for the coming event' Resource.
+// Contrarily, if the coming event's Resource is "*", the ce.Resource should only be "*".
+//
+// Note: we have a special case here when the coming event is a wildcard event,
+// it will force all Pods to move to activeQ/backoffQ,
+// but we take it as an unmatched event unless the ce is also a wildcard one.
+func (ce ClusterEvent) Match(event ClusterEvent) bool {
+	return ce.IsWildCard() || (ce.Resource == WildCard || ce.Resource == event.Resource) && ce.ActionType&event.ActionType != 0
+}
+
 func UnrollWildCardResource() []ClusterEventWithHint {
 	return []ClusterEventWithHint{
 		{Event: ClusterEvent{Resource: Pod, ActionType: All}},

pkg/scheduler/framework/types_test.go

@@ -1608,3 +1608,52 @@ func TestCalculatePodResourcesWithResize(t *testing.T) {
 		})
 	}
 }
+
+func TestCloudEvent_Match(t *testing.T) {
+	testCases := []struct {
+		name        string
+		event       ClusterEvent
+		comingEvent ClusterEvent
+		wantResult  bool
+	}{
+		{
+			name:        "wildcard event matches with all kinds of coming events",
+			event:       ClusterEvent{Resource: WildCard, ActionType: All},
+			comingEvent: ClusterEvent{Resource: Pod, ActionType: UpdateNodeLabel},
+			wantResult:  true,
+		},
+		{
+			name:        "event with resource = 'Pod' matching with coming events carries same actionType",
+			event:       ClusterEvent{Resource: Pod, ActionType: UpdateNodeLabel | UpdateNodeTaint},
+			comingEvent: ClusterEvent{Resource: Pod, ActionType: UpdateNodeLabel},
+			wantResult:  true,
+		},
+		{
+			name:        "event with resource = '*' matching with coming events carries same actionType",
+			event:       ClusterEvent{Resource: WildCard, ActionType: UpdateNodeLabel},
+			comingEvent: ClusterEvent{Resource: Pod, ActionType: UpdateNodeLabel},
+			wantResult:  true,
+		},
+		{
+			name:        "event with resource = '*' matching with coming events carries different actionType",
+			event:       ClusterEvent{Resource: WildCard, ActionType: UpdateNodeLabel},
+			comingEvent: ClusterEvent{Resource: Pod, ActionType: UpdateNodeAllocatable},
+			wantResult:  false,
+		},
+		{
+			name:        "event matching with coming events carries '*' resources",
+			event:       ClusterEvent{Resource: Pod, ActionType: UpdateNodeLabel},
+			comingEvent: ClusterEvent{Resource: WildCard, ActionType: UpdateNodeLabel},
+			wantResult:  false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := tc.event.Match(tc.comingEvent)
+			if got != tc.wantResult {
+				t.Fatalf("unexpected result")
+			}
+		})
+	}
+}

pkg/scheduler/internal/queue/scheduling_queue.go

@@ -441,7 +441,7 @@ func (p *PriorityQueue) isPodWorthRequeuing(logger klog.Logger, pInfo *framework
 	pod := pInfo.Pod
 	queueStrategy := queueSkip
 	for eventToMatch, hintfns := range hintMap {
-		if eventToMatch.Resource != event.Resource || eventToMatch.ActionType&event.ActionType == 0 {
+		if !eventToMatch.Match(event) {
 			continue
 		}
 

pkg/scheduler/internal/queue/scheduling_queue_test.go

@@ -3550,6 +3550,50 @@ func Test_isPodWorthRequeuing(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "If event with '*' Resource, queueing hint function for specified Resource is also executed",
+			podInfo: &framework.QueuedPodInfo{
+				UnschedulablePlugins: sets.New("fooPlugin1"),
+				PodInfo:              mustNewPodInfo(st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj()),
+			},
+			event:                  framework.ClusterEvent{Resource: framework.Node, ActionType: framework.Add},
+			oldObj:                 nil,
+			newObj:                 st.MakeNode().Obj(),
+			expected:               queueAfterBackoff,
+			expectedExecutionCount: 1,
+			queueingHintMap: QueueingHintMapPerProfile{
+				"": {
+					framework.ClusterEvent{Resource: framework.WildCard, ActionType: framework.Add}: {
+						{
+							PluginName:     "fooPlugin1",
+							QueueingHintFn: queueHintReturnQueue,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "If event is a wildcard one, queueing hint function for all kinds of events is executed",
+			podInfo: &framework.QueuedPodInfo{
+				UnschedulablePlugins: sets.New("fooPlugin1"),
+				PodInfo:              mustNewPodInfo(st.MakePod().Name("pod1").Namespace("ns1").UID("1").Obj()),
+			},
+			event:                  framework.ClusterEvent{Resource: framework.Node, ActionType: framework.UpdateNodeLabel | framework.UpdateNodeTaint},
+			oldObj:                 nil,
+			newObj:                 st.MakeNode().Obj(),
+			expected:               queueAfterBackoff,
+			expectedExecutionCount: 1,
+			queueingHintMap: QueueingHintMapPerProfile{
+				"": {
+					framework.ClusterEvent{Resource: framework.WildCard, ActionType: framework.All}: {
+						{
+							PluginName:     "fooPlugin1",
+							QueueingHintFn: queueHintReturnQueue,
+						},
+					},
+				},
+			},
+		},
 	}
 
 	for _, test := range tests {