Merge pull request kubernetes#127497 from pohly/dra-scheduler-queueing-hints-fix

DRA scheduler: fix queuing hint support

Author: k8s-ci-robot

pkg/scheduler/framework/events.go

@@ -20,6 +20,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/equality"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/dynamic-resource-allocation/resourceclaim"
 	"k8s.io/kubernetes/pkg/api/v1/resource"
 	"k8s.io/kubernetes/pkg/features"
 )
@@ -65,6 +66,8 @@ var (
 	PodTolerationChange = ClusterEvent{Resource: Pod, ActionType: UpdatePodTolerations, Label: "PodTolerationChange"}
 	// PodSchedulingGateEliminatedChange is the event when a pod's scheduling gate is changed.
 	PodSchedulingGateEliminatedChange = ClusterEvent{Resource: Pod, ActionType: UpdatePodSchedulingGatesEliminated, Label: "PodSchedulingGateChange"}
+	// PodGeneratedResourceClaimChange is the event when a pod's list of generated ResourceClaims changes.
+	PodGeneratedResourceClaimChange = ClusterEvent{Resource: Pod, ActionType: UpdatePodGeneratedResourceClaim, Label: "PodGeneratedResourceClaimChange"}
 	// NodeSpecUnschedulableChange is the event when unschedulable node spec is changed.
 	NodeSpecUnschedulableChange = ClusterEvent{Resource: Node, ActionType: UpdateNodeTaint, Label: "NodeSpecUnschedulableChange"}
 	// NodeAllocatableChange is the event when node allocatable is changed.
@@ -152,6 +155,9 @@ func PodSchedulingPropertiesChange(newPod *v1.Pod, oldPod *v1.Pod) (events []Clu
 		extractPodSchedulingGateEliminatedChange,
 		extractPodTolerationChange,
 	}
+	if utilfeature.DefaultFeatureGate.Enabled(features.DynamicResourceAllocation) {
+		podChangeExtracters = append(podChangeExtracters, extractPodGeneratedResourceClaimChange)
+	}
 
 	for _, fn := range podChangeExtracters {
 		if event := fn(newPod, oldPod); event != nil {
@@ -222,6 +228,14 @@ func extractPodSchedulingGateEliminatedChange(newPod *v1.Pod, oldPod *v1.Pod) *C
 	return nil
 }
 
+func extractPodGeneratedResourceClaimChange(newPod *v1.Pod, oldPod *v1.Pod) *ClusterEvent {
+	if !resourceclaim.PodStatusEqual(newPod.Status.ResourceClaimStatuses, oldPod.Status.ResourceClaimStatuses) {
+		return &PodGeneratedResourceClaimChange
+	}
+
+	return nil
+}
+
 // NodeSchedulingPropertiesChange interprets the update of a node and returns corresponding UpdateNodeXYZ event(s).
 func NodeSchedulingPropertiesChange(newNode *v1.Node, oldNode *v1.Node) (events []ClusterEvent) {
 	nodeChangeExtracters := []nodeChangeExtractor{

pkg/scheduler/framework/events_test.go

@@ -25,7 +25,11 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	featuregatetesting "k8s.io/component-base/featuregate/testing"
+	"k8s.io/kubernetes/pkg/features"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
+	"k8s.io/utils/ptr"
 )
 
 func TestNodeAllocatableChange(t *testing.T) {
@@ -335,11 +339,20 @@ func Test_podSchedulingPropertiesChange(t *testing.T) {
 			},
 		},
 	}
+	claimStatusA := v1.PodResourceClaimStatus{
+		Name:              "my-claim",
+		ResourceClaimName: ptr.To("claim"),
+	}
+	claimStatusB := v1.PodResourceClaimStatus{
+		Name:              "my-claim-2",
+		ResourceClaimName: ptr.To("claim-2"),
+	}
 	tests := []struct {
-		name   string
-		newPod *v1.Pod
-		oldPod *v1.Pod
-		want   []ClusterEvent
+		name        string
+		newPod      *v1.Pod
+		oldPod      *v1.Pod
+		draDisabled bool
+		want        []ClusterEvent
 	}{
 		{
 			name:   "only label is updated",
@@ -389,9 +402,35 @@ func Test_podSchedulingPropertiesChange(t *testing.T) {
 			oldPod: st.MakePod().Toleration("key").Obj(),
 			want:   []ClusterEvent{PodTolerationChange},
 		},
+		{
+			name:        "pod claim statuses change, feature disabled",
+			draDisabled: true,
+			newPod:      st.MakePod().ResourceClaimStatuses(claimStatusA).Obj(),
+			oldPod:      st.MakePod().Obj(),
+			want:        []ClusterEvent{assignedPodOtherUpdate},
+		},
+		{
+			name:   "pod claim statuses change, feature enabled",
+			newPod: st.MakePod().ResourceClaimStatuses(claimStatusA).Obj(),
+			oldPod: st.MakePod().Obj(),
+			want:   []ClusterEvent{PodGeneratedResourceClaimChange},
+		},
+		{
+			name:   "pod claim statuses swapped",
+			newPod: st.MakePod().ResourceClaimStatuses(claimStatusA, claimStatusB).Obj(),
+			oldPod: st.MakePod().ResourceClaimStatuses(claimStatusB, claimStatusA).Obj(),
+			want:   []ClusterEvent{PodGeneratedResourceClaimChange},
+		},
+		{
+			name:   "pod claim statuses extended",
+			newPod: st.MakePod().ResourceClaimStatuses(claimStatusA, claimStatusB).Obj(),
+			oldPod: st.MakePod().ResourceClaimStatuses(claimStatusA).Obj(),
+			want:   []ClusterEvent{PodGeneratedResourceClaimChange},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
+			featuregatetesting.SetFeatureGateDuringTest(t, utilfeature.DefaultFeatureGate, features.DynamicResourceAllocation, !tt.draDisabled)
 			got := PodSchedulingPropertiesChange(tt.newPod, tt.oldPod)
 			if diff := cmp.Diff(tt.want, got); diff != "" {
 				t.Errorf("unexpected event is returned from podSchedulingPropertiesChange (-want, +got):\n%s", diff)

pkg/scheduler/framework/plugins/dynamicresources/dynamicresources.go

@@ -399,6 +399,8 @@ func (pl *dynamicResources) EventsToRegister(_ context.Context) ([]framework.Clu
 		{Event: framework.ClusterEvent{Resource: framework.Node, ActionType: nodeActionType}},
 		// Allocation is tracked in ResourceClaims, so any changes may make the pods schedulable.
 		{Event: framework.ClusterEvent{Resource: framework.ResourceClaim, ActionType: framework.Add | framework.Update}, QueueingHintFn: pl.isSchedulableAfterClaimChange},
+		// Adding the ResourceClaim name to the pod status makes pods waiting for their ResourceClaim schedulable.
+		{Event: framework.ClusterEvent{Resource: framework.Pod, ActionType: framework.UpdatePodGeneratedResourceClaim}, QueueingHintFn: pl.isSchedulableAfterPodChange},
 		// A pod might be waiting for a class to get created or modified.
 		{Event: framework.ClusterEvent{Resource: framework.DeviceClass, ActionType: framework.Add | framework.Update}},
 		// Adding or updating a ResourceSlice might make a pod schedulable because new resources became available.
@@ -450,7 +452,10 @@ func (pl *dynamicResources) isSchedulableAfterClaimChange(logger klog.Logger, po
 		// This is not an unexpected error: we know that
 		// foreachPodResourceClaim only returns errors for "not
 		// schedulable".
-		logger.V(6).Info("pod is not schedulable after resource claim change", "pod", klog.KObj(pod), "claim", klog.KObj(modifiedClaim), "reason", err.Error())
+		if loggerV := logger.V(6); loggerV.Enabled() {
+			owner := metav1.GetControllerOf(modifiedClaim)
+			loggerV.Info("pod is not schedulable after resource claim change", "pod", klog.KObj(pod), "claim", klog.KObj(modifiedClaim), "claimOwner", owner, "reason", err.Error())
+		}
 		return framework.QueueSkip, nil
 	}
 
@@ -474,7 +479,7 @@ func (pl *dynamicResources) isSchedulableAfterClaimChange(logger klog.Logger, po
 	}
 
 	if originalClaim == nil {
-		logger.V(4).Info("claim for pod got created", "pod", klog.KObj(pod), "claim", klog.KObj(modifiedClaim))
+		logger.V(5).Info("claim for pod got created", "pod", klog.KObj(pod), "claim", klog.KObj(modifiedClaim))
 		return framework.Queue, nil
 	}
 
@@ -492,7 +497,34 @@ func (pl *dynamicResources) isSchedulableAfterClaimChange(logger klog.Logger, po
 		return framework.QueueSkip, nil
 	}
 
-	logger.V(4).Info("status of claim for pod got updated", "pod", klog.KObj(pod), "claim", klog.KObj(modifiedClaim))
+	logger.V(5).Info("status of claim for pod got updated", "pod", klog.KObj(pod), "claim", klog.KObj(modifiedClaim))
+	return framework.Queue, nil
+}
+
+// isSchedulableAfterPodChange is invoked for update pod events reported by
+// an informer. It checks whether that change adds the ResourceClaim(s) that the
+// pod has been waiting for.
+func (pl *dynamicResources) isSchedulableAfterPodChange(logger klog.Logger, pod *v1.Pod, oldObj, newObj interface{}) (framework.QueueingHint, error) {
+	_, modifiedPod, err := schedutil.As[*v1.Pod](nil, newObj)
+	if err != nil {
+		// Shouldn't happen.
+		return framework.Queue, fmt.Errorf("unexpected object in isSchedulableAfterClaimChange: %w", err)
+	}
+
+	if pod.UID != modifiedPod.UID {
+		logger.V(7).Info("pod is not schedulable after change in other pod", "pod", klog.KObj(pod), "modifiedPod", klog.KObj(modifiedPod))
+		return framework.QueueSkip, nil
+	}
+
+	if err := pl.foreachPodResourceClaim(modifiedPod, nil); err != nil {
+		// This is not an unexpected error: we know that
+		// foreachPodResourceClaim only returns errors for "not
+		// schedulable".
+		logger.V(6).Info("pod is not schedulable after being updated", "pod", klog.KObj(pod))
+		return framework.QueueSkip, nil
+	}
+
+	logger.V(5).Info("pod got updated and is schedulable", "pod", klog.KObj(pod))
 	return framework.Queue, nil
 }
 
@@ -537,7 +569,7 @@ func (pl *dynamicResources) isSchedulableAfterPodSchedulingContextChange(logger
 	// Deleted? That can happen because we ourselves delete the PodSchedulingContext while
 	// working on the pod. This can be ignored.
 	if oldObj != nil && newObj == nil {
-		logger.V(4).Info("PodSchedulingContext got deleted")
+		logger.V(5).Info("PodSchedulingContext got deleted")
 		return framework.QueueSkip, nil
 	}
 
@@ -568,7 +600,7 @@ func (pl *dynamicResources) isSchedulableAfterPodSchedulingContextChange(logger
 		// This is not an unexpected error: we know that
 		// foreachPodResourceClaim only returns errors for "not
 		// schedulable".
-		logger.V(4).Info("pod is not schedulable, keep waiting", "pod", klog.KObj(pod), "reason", err.Error())
+		logger.V(5).Info("pod is not schedulable, keep waiting", "pod", klog.KObj(pod), "reason", err.Error())
 		return framework.QueueSkip, nil
 	}
 
@@ -589,7 +621,7 @@ func (pl *dynamicResources) isSchedulableAfterPodSchedulingContextChange(logger
 	if oldPodScheduling == nil /* create */ ||
 		len(oldPodScheduling.Status.ResourceClaims) < len(podScheduling.Status.ResourceClaims) /* new information and not incomplete (checked above) */ {
 		// This definitely is new information for the scheduler. Try again immediately.
-		logger.V(4).Info("PodSchedulingContext for pod has all required information, schedule immediately", "pod", klog.KObj(pod))
+		logger.V(5).Info("PodSchedulingContext for pod has all required information, schedule immediately", "pod", klog.KObj(pod))
 		return framework.Queue, nil
 	}