scheduler: add reservation preemption

.licenseignore

@@ -4,6 +4,8 @@ pkg/descheduler/controllers/migration/controllerfinder
 pkg/scheduler/plugins/coscheduling
 pkg/scheduler/plugins/nodenumaresource/least_allocated.go
 pkg/scheduler/plugins/nodenumaresource/most_allocated.go
+pkg/scheduler/plugins/reservation/preemption.go
+pkg/scheduler/plugins/reservation/preemption_test.go
 pkg/scheduler/frameworkext/topologymanager/policy.go
 pkg/scheduler/frameworkext/topologymanager/policy_test.go
 pkg/scheduler/frameworkext/topologymanager/policy_best_effort.go

apis/extension/preemption.go

@@ -23,6 +23,9 @@ const (
 	// PreemptionPolicyTransformer is enabled.
 	// When PreemptionPolicyTransformer is enabled but the pod does not set the label, DefaultPreemptionPolicy is used.
 	LabelPodPreemptionPolicy = SchedulingDomainPrefix + "/preemption-policy"
+
+	// LabelDisablePreemptible determines whether the pod disables being a victim during the reservation preemption.
+	LabelDisablePreemptible = SchedulingDomainPrefix + "/disable-preemptible"
 )
 
 func GetPodKoordPreemptionPolicy(pod *corev1.Pod) *corev1.PreemptionPolicy {
@@ -40,3 +43,14 @@ func GetPodKoordPreemptionPolicy(pod *corev1.Pod) *corev1.PreemptionPolicy {
 func GetPreemptionPolicyPtr(policy corev1.PreemptionPolicy) *corev1.PreemptionPolicy {
 	return &policy
 }
+
+func IsPodPreemptible(pod *corev1.Pod) bool {
+	if pod == nil || pod.Labels == nil {
+		return true
+	}
+	v, ok := pod.Labels[LabelDisablePreemptible]
+	if !ok {
+		return true
+	}
+	return v != "true"
+}

apis/extension/preemption_test.go

@@ -92,3 +92,56 @@ func TestGetPodKoordPreemptionPolicy(t *testing.T) {
 		})
 	}
 }
+
+func TestIsPodPreemptible(t *testing.T) {
+	tests := []struct {
+		name string
+		arg  *corev1.Pod
+		want bool
+	}{
+		{
+			name: "pod is nil",
+			arg:  nil,
+			want: true,
+		},
+		{
+			name: "pod has no label",
+			arg: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-pod",
+				},
+			},
+			want: true,
+		},
+		{
+			name: "pod is marked as non-preemptible",
+			arg: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-pod",
+					Labels: map[string]string{
+						LabelDisablePreemptible: "true",
+					},
+				},
+			},
+			want: false,
+		},
+		{
+			name: "pod is marked as preemptible",
+			arg: &corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test-pod",
+					Labels: map[string]string{
+						LabelDisablePreemptible: "false",
+					},
+				},
+			},
+			want: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := IsPodPreemptible(tt.arg)
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}

pkg/scheduler/apis/config/types.go

@@ -160,6 +160,19 @@ type ReservationArgs struct {
 
 	// EnablePreemption indicates whether to enable preemption for reservations.
 	EnablePreemption bool
+	// MinCandidateNodesPercentage is the minimum number of candidates to
+	// shortlist when dry running preemption as a percentage of number of nodes.
+	// Must be in the range [0, 100]. Defaults to 10% of the cluster size if
+	// unspecified.
+	MinCandidateNodesPercentage int32
+	// MinCandidateNodesAbsolute is the absolute minimum number of candidates to
+	// shortlist. The likely number of candidates enumerated for dry running
+	// preemption is given by the formula:
+	// numCandidates = max(numNodes * minCandidateNodesPercentage, minCandidateNodesAbsolute)
+	// We say "likely" because there are other factors such as PDB violations
+	// that play a role in the number of candidates shortlisted. Must be at least
+	// 0 nodes. Defaults to 100 nodes if unspecified.
+	MinCandidateNodesAbsolute int32
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/scheduler/apis/config/v1/types.go

@@ -155,6 +155,19 @@ type ReservationArgs struct {
 
 	// EnablePreemption indicates whether to enable preemption for reservations.
 	EnablePreemption *bool `json:"enablePreemption,omitempty"`
+	// MinCandidateNodesPercentage is the minimum number of candidates to
+	// shortlist when dry running preemption as a percentage of number of nodes.
+	// Must be in the range [0, 100]. Defaults to 10% of the cluster size if
+	// unspecified.
+	MinCandidateNodesPercentage *int32 `json:"minCandidateNodesPercentage,omitempty"`
+	// MinCandidateNodesAbsolute is the absolute minimum number of candidates to
+	// shortlist. The likely number of candidates enumerated for dry running
+	// preemption is given by the formula:
+	// numCandidates = max(numNodes * minCandidateNodesPercentage, minCandidateNodesAbsolute)
+	// We say "likely" because there are other factors such as PDB violations
+	// that play a role in the number of candidates shortlisted. Must be at least
+	// 0 nodes. Defaults to 100 nodes if unspecified.
+	MinCandidateNodesAbsolute *int32 `json:"minCandidateNodesAbsolute,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/scheduler/apis/config/v1beta3/defaults.go

@@ -49,7 +49,9 @@ var (
 
 	defaultPreferredCPUBindPolicy = CPUBindPolicyFullPCPUs
 
-	defaultEnablePreemption = pointer.Bool(false)
+	defaultEnablePreemption            = pointer.Bool(false)
+	defaultMinCandidateNodesPercentage = pointer.Int32(10)
+	defaultMinCandidateNodesAbsolute   = pointer.Int32(100)
 
 	defaultDelayEvictTime       = 120 * time.Second
 	defaultRevokePodInterval    = 1 * time.Second
@@ -143,6 +145,12 @@ func SetDefaults_ReservationArgs(obj *ReservationArgs) {
 	if obj.EnablePreemption == nil {
 		obj.EnablePreemption = defaultEnablePreemption
 	}
+	if obj.MinCandidateNodesPercentage == nil {
+		obj.MinCandidateNodesPercentage = defaultMinCandidateNodesPercentage
+	}
+	if obj.MinCandidateNodesAbsolute == nil {
+		obj.MinCandidateNodesAbsolute = defaultMinCandidateNodesAbsolute
+	}
 }
 
 func SetDefaults_ElasticQuotaArgs(obj *ElasticQuotaArgs) {

pkg/scheduler/apis/config/v1beta3/types.go

@@ -155,6 +155,19 @@ type ReservationArgs struct {
 
 	// EnablePreemption indicates whether to enable preemption for reservations.
 	EnablePreemption *bool `json:"enablePreemption,omitempty"`
+	// MinCandidateNodesPercentage is the minimum number of candidates to
+	// shortlist when dry running preemption as a percentage of number of nodes.
+	// Must be in the range [0, 100]. Defaults to 10% of the cluster size if
+	// unspecified.
+	MinCandidateNodesPercentage *int32 `json:"minCandidateNodesPercentage,omitempty"`
+	// MinCandidateNodesAbsolute is the absolute minimum number of candidates to
+	// shortlist. The likely number of candidates enumerated for dry running
+	// preemption is given by the formula:
+	// numCandidates = max(numNodes * minCandidateNodesPercentage, minCandidateNodesAbsolute)
+	// We say "likely" because there are other factors such as PDB violations
+	// that play a role in the number of candidates shortlisted. Must be at least
+	// 0 nodes. Defaults to 100 nodes if unspecified.
+	MinCandidateNodesAbsolute *int32 `json:"minCandidateNodesAbsolute,omitempty"`
 }
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

pkg/scheduler/frameworkext/eventhandlers/reservation_handler.go

@@ -91,6 +91,8 @@ func MakeReservationErrorHandler(
 			}
 			return false
 		} else if reservationutil.IsReservePod(pod) {
+			// NOTE: Since the failure handler is asynchronous and not locked with the reservation event handler,
+			// please make sure the status of the object is expected before adding or updating it.
 			// for reservation CR, which is treated as pod internal
 			reservationErrorFn(ctx, fwk, podInfo, status, nominatingInfo, start)
 
@@ -100,6 +102,10 @@ func MakeReservationErrorHandler(
 				return true
 			}
 
+			// nominate for the reserve pod if it is
+			// TODO: use the default nominator
+			addNominatedReservation(f, podInfo, nominatingInfo)
+
 			msg := truncateMessage(schedulingErr.Error())
 			fwk.EventRecorder().Eventf(r, nil, corev1.EventTypeWarning, "FailedScheduling", "Scheduling", msg)
 
@@ -111,6 +117,19 @@ func MakeReservationErrorHandler(
 	}
 }
 
+func addNominatedReservation(f framework.Framework, podInfo *framework.QueuedPodInfo, nominatingInfo *framework.NominatingInfo) {
+	frameworkExtender, ok := f.(frameworkext.FrameworkExtender)
+	if !ok {
+		return
+	}
+
+	reservationNominator := frameworkExtender.GetReservationNominator()
+	if reservationNominator == nil {
+		return
+	}
+	reservationNominator.AddNominatedReservePod(podInfo.Pod, nominatingInfo.NominatedNodeName)
+}
+
 // input:
 // "0/1 nodes are available: 3 Reservation(s) didn't match affinity rules, 1 Reservation(s) is unshedulable, 1 Reservation(s) is unavailable,
 // 2 Reservation(s) Insufficient cpu, 1 Reservation(s) Insufficient memory, 1 Insufficient cpu, 1 Insufficient memory.

pkg/scheduler/frameworkext/reservation_info.go

@@ -324,7 +324,9 @@ func (ri *ReservationInfo) UpdateReservation(r *schedulingv1alpha1.Reservation)
 	ri.Reservation = r
 	ri.Pod = reservationutil.NewReservePod(r)
 	ri.AllocatablePorts = util.RequestedHostPorts(ri.Pod)
-	ri.Allocated = quotav1.Mask(ri.Allocated, ri.ResourceNames)
+	if ri.Allocated != nil {
+		ri.Allocated = quotav1.Mask(ri.Allocated, ri.ResourceNames)
+	}
 	ownerMatchers, err := reservationutil.ParseReservationOwnerMatchers(r.Spec.Owners)
 	if err != nil {
 		klog.ErrorS(err, "Failed to parse reservation owner matchers", "reservation", klog.KObj(r))

pkg/scheduler/frameworkext/reservation_info_test.go

@@ -384,7 +384,6 @@ func TestReservationInfoUpdateReservation(t *testing.T) {
 			want: &ReservationInfo{
 				ResourceNames: []corev1.ResourceName{corev1.ResourceCPU, corev1.ResourceMemory},
 				Allocatable:   allocatable.DeepCopy(),
-				Allocated:     corev1.ResourceList{},
 				AssignedPods:  map[types.UID]*PodRequirement{},
 				OwnerMatchers: ownerMatchers,
 				ParseError:    parseError,
@@ -407,7 +406,6 @@ func TestReservationInfoUpdateReservation(t *testing.T) {
 			want: &ReservationInfo{
 				ResourceNames: []corev1.ResourceName{corev1.ResourceCPU, corev1.ResourceMemory},
 				Allocatable:   allocatable.DeepCopy(),
-				Allocated:     corev1.ResourceList{},
 				AssignedPods:  map[types.UID]*PodRequirement{},
 				OwnerMatchers: func() []reservationutil.ReservationOwnerMatcher {
 					m, _ := reservationutil.ParseReservationOwnerMatchers([]schedulingv1alpha1.ReservationOwner{
@@ -444,7 +442,6 @@ func TestReservationInfoUpdateReservation(t *testing.T) {
 			want: &ReservationInfo{
 				ResourceNames: []corev1.ResourceName{corev1.ResourceCPU, corev1.ResourceMemory},
 				Allocatable:   allocatable.DeepCopy(),
-				Allocated:     corev1.ResourceList{},
 				AssignedPods:  map[types.UID]*PodRequirement{},
 				OwnerMatchers: nil,
 				ParseError: func() error {
@@ -481,7 +478,6 @@ func TestReservationInfoUpdateReservation(t *testing.T) {
 					corev1.ResourceCPU:    resource.MustParse("8"),
 					corev1.ResourceMemory: resource.MustParse("16Gi"),
 				},
-				Allocated:     corev1.ResourceList{},
 				AssignedPods:  map[types.UID]*PodRequirement{},
 				OwnerMatchers: ownerMatchers,
 				ParseError:    nil,

pkg/scheduler/plugins/reservation/cache_test.go

@@ -96,7 +96,6 @@ func TestCacheUpdateReservation(t *testing.T) {
 	reservationInfos = cache.listAvailableReservationInfosOnNode(reservation.Status.NodeName)
 	assert.Len(t, reservationInfos, 1)
 	rInfo = reservationInfos[0]
-	expectReservationInfo.Allocated = corev1.ResourceList{}
 	sort.Slice(rInfo.ResourceNames, func(i, j int) bool {
 		return rInfo.ResourceNames[i] < rInfo.ResourceNames[j]
 	})

pkg/scheduler/plugins/reservation/eventhandler_test.go

@@ -196,7 +196,7 @@ func TestEventHandlerUpdate(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			cache := newReservationCache(nil)
-			eh := &reservationEventHandler{cache: cache, rrNominator: newNominator()}
+			eh := &reservationEventHandler{cache: cache, rrNominator: newNominator(nil, nil)}
 			eh.OnUpdate(tt.oldReservation, tt.newReservation)
 			if tt.wantReservation == nil {
 				rInfo := cache.getReservationInfoByUID(tt.newReservation.UID)
@@ -241,7 +241,7 @@ func TestEventHandlerDelete(t *testing.T) {
 		},
 	}
 	cache := newReservationCache(nil)
-	eh := &reservationEventHandler{cache: cache, rrNominator: newNominator()}
+	eh := &reservationEventHandler{cache: cache, rrNominator: newNominator(nil, nil)}
 	eh.OnAdd(activeReservation, true)
 	rInfo := cache.getReservationInfoByUID(activeReservation.UID)
 	assert.NotNil(t, rInfo)