volcano-sh · Jason-Liu-Dream · Jul 19, 2022 · Jul 19, 2022
diff --git a/docs/user-guide/how_to_use_cdp_plugin.md b/docs/user-guide/how_to_use_cdp_plugin.md
@@ -4,6 +4,8 @@
 When we need to enable elastic training or serving, preemptible job's pods can be preempted or back to running repeatedly, if no cooldown protection set, these pods can be preempted again after they just started for a short time, this may cause service stability dropped.
 So we add "cdp" plugin to ensure preemptible job's pods can run for at least some time set by user.
 
+In another case, we do not want some tasks always be evicted, which makes it be starved to death, we need a way to protect them.
+
 ## Environment setup
 
 ### Install volcano
@@ -114,10 +116,11 @@ spec:
 1. add annotations in volcano job in format below.
    1. `volcano.sh/preemptable` annotation indicates that job or task is preemptable
    2. `volcano.sh/cooldown-time` annotation indicates cooldown time for the entire job or dedicated task. Value for the annotation indicates cooldown time, valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h". 
-
+   3. `volcano.sh/max-cooldown-times` annotation indicates max number of cooldown(eviction times) for entire job or dedicated task. if the task cooldown times are greater than this value, the task cannot be evicted.
         ```yaml
             volcano.sh/preemptable: "true"
             volcano.sh/cooldown-time: "600s"
+            volcano.sh/max-cooldown-times: "2"
         ```
 
 **Example 1**
@@ -198,4 +201,104 @@ spec:
                   cpu: "1"
           restartPolicy: OnFailure
 
+```
+
+**Example 3**
+
+Add annotation to dedicated task, as shown below, the only task in a job named "test-low-job" can be preempted and have the maximum number of times of cooldown support.
+
+In that case, if the cluster only has 6 CPU cores, the first step users apply a job named "test-low-job", then apply a job named "test-middle-job", after the job named "test-middle-job" is finished, the job named "test-low-job" running again.  At this moment, apply the task job named "test-high-job" can not preempted the task job named "test-low-job". because the cooldown is protected, the job named "test-low-job" can just be preempted one time when set annotation `volcano.sh/max-cooldown-times: "1"`
+```yaml
+apiVersion: batch.volcano.sh/v1alpha1
+kind: Job
+metadata:
+   name: test-low-job
+spec:
+   minAvailable: 3
+   schedulerName: volcano
+   priorityClassName: low-priority
+   maxRetry: 5
+   queue: default
+   tasks:
+      - replicas: 6
+        name: tasks
+        template:
+           metadata:
+              name: worker
+              annotations:
+                 volcano.sh/preemptable: 'true'
+                 volcano.sh/max-cooldown-times: '1'
+           spec:
+              containers:
+                 - image: alpine
+                   imagePullPolicy: IfNotPresent
+                   name: sleep
+                   command:
+                      - /bin/ash
+                      - '-c'
+                      - sleep 3600
+                   resources:
+                      requests:
+                         cpu: '1'
+              restartPolicy: OnFailure
+---
+apiVersion: batch.volcano.sh/v1alpha1
+kind: Job
+metadata:
+   name: test-middle-job
+spec:
+   minAvailable: 3
+   schedulerName: volcano
+   priorityClassName: middle-priority
+   maxRetry: 5
+   queue: default
+   tasks:
+      - replicas: 6
+        name: tasks
+        template:
+           metadata:
+              name: worker
+           spec:
+              containers:
+                 - image: alpine
+                   imagePullPolicy: IfNotPresent
+                   name: sleep
+                   command:
+                      - /bin/ash
+                      - '-c'
+                      - sleep 60
+                   resources:
+                      requests:
+                         cpu: '1'
+              restartPolicy: OnFailure
+---
+apiVersion: batch.volcano.sh/v1alpha1
+kind: Job
+metadata:
+   name: test-high-job
+spec:
+   minAvailable: 3
+   schedulerName: volcano
+   priorityClassName: high-priority
+   maxRetry: 5
+   queue: default
+   tasks:
+      - replicas: 6
+        name: tasks
+        template:
+           metadata:
+              name: worker
+           spec:
+              containers:
+                 - image: alpine
+                   imagePullPolicy: IfNotPresent
+                   name: sleep
+                   command:
+                      - /bin/ash
+                      - '-c'
+                      - sleep 3600
+                   resources:
+                      requests:
+                         cpu: '1'
+              restartPolicy: OnFailure
 ```
diff --git a/pkg/controllers/job/job_controller_util.go b/pkg/controllers/job/job_controller_util.go
@@ -118,6 +118,9 @@ func createJobPod(job *batch.Job, template *v1.PodTemplateSpec, topologyPolicy b
 		if value, found := job.Annotations[schedulingv2.CooldownTime]; found {
 			pod.Annotations[schedulingv2.CooldownTime] = value
 		}
+		if value, found := job.Annotations[schedulingv2.MaxCooldownTimes]; found {
+			pod.Annotations[schedulingv2.MaxCooldownTimes] = value
+		}
 		if value, found := job.Annotations[schedulingv2.RevocableZone]; found {
 			pod.Annotations[schedulingv2.RevocableZone] = value
 		}
@@ -145,6 +148,9 @@ func createJobPod(job *batch.Job, template *v1.PodTemplateSpec, topologyPolicy b
 		if value, found := job.Labels[schedulingv2.CooldownTime]; found {
 			pod.Labels[schedulingv2.CooldownTime] = value
 		}
+		if value, found := job.Labels[schedulingv2.MaxCooldownTimes]; found {
+			pod.Labels[schedulingv2.MaxCooldownTimes] = value
+		}
 	}
 
 	if jobForwarding {

diff --git a/pkg/controllers/podgroup/pg_controller_handler.go b/pkg/controllers/podgroup/pg_controller_handler.go
@@ -160,6 +160,9 @@ func (pg *pgcontroller) createNormalPodPGIfNotExist(pod *v1.Pod) error {
 		if value, ok := pod.Annotations[scheduling.CooldownTime]; ok {
 			obj.Annotations[scheduling.CooldownTime] = value
 		}
+		if value, ok := pod.Annotations[scheduling.MaxCooldownTimes]; ok {
+			obj.Annotations[scheduling.MaxCooldownTimes] = value
+		}
 		if value, ok := pod.Annotations[scheduling.RevocableZone]; ok {
 			obj.Annotations[scheduling.RevocableZone] = value
 		}
@@ -169,6 +172,9 @@ func (pg *pgcontroller) createNormalPodPGIfNotExist(pod *v1.Pod) error {
 		if value, ok := pod.Labels[scheduling.CooldownTime]; ok {
 			obj.Labels[scheduling.CooldownTime] = value
 		}
+		if value, ok := pod.Labels[scheduling.MaxCooldownTimes]; ok {
+			obj.Labels[scheduling.MaxCooldownTimes] = value
+		}
 
 		if value, found := pod.Annotations[scheduling.JDBMinAvailable]; found {
 			obj.Annotations[scheduling.JDBMinAvailable] = value

diff --git a/pkg/scheduler/api/job_info.go b/pkg/scheduler/api/job_info.go
@@ -323,6 +323,11 @@ type JobInfo struct {
 
 	Preemptable bool
 
+	// Max eviction times for each task
+	MaxCooldownTimes int32
+	// Eviction times record for each task
+	TaskCooldownTimesRecord map[string]*int32
+
 	// RevocableZone support set volcano.sh/revocable-zone annotaion or label for pod/podgroup
 	// we only support empty value or * value for this version and we will support specify revocable zone name for futrue release
 	// empty value means workload can not use revocable node
@@ -334,14 +339,15 @@ type JobInfo struct {
 // NewJobInfo creates a new jobInfo for set of tasks
 func NewJobInfo(uid JobID, tasks ...*TaskInfo) *JobInfo {
 	job := &JobInfo{
-		UID:              uid,
-		MinAvailable:     0,
-		NodesFitErrors:   make(map[TaskID]*FitErrors),
-		Allocated:        EmptyResource(),
-		TotalRequest:     EmptyResource(),
-		TaskStatusIndex:  map[TaskStatus]tasksMap{},
-		Tasks:            tasksMap{},
-		TaskMinAvailable: map[TaskID]int32{},
+		UID:                     uid,
+		MinAvailable:            0,
+		NodesFitErrors:          make(map[TaskID]*FitErrors),
+		Allocated:               EmptyResource(),
+		TotalRequest:            EmptyResource(),
+		TaskStatusIndex:         map[TaskStatus]tasksMap{},
+		Tasks:                   tasksMap{},
+		TaskMinAvailable:        map[TaskID]int32{},
+		TaskCooldownTimesRecord: map[string]*int32{},
 	}
 
 	for _, task := range tasks {
@@ -356,6 +362,12 @@ func (ji *JobInfo) UnsetPodGroup() {
 	ji.PodGroup = nil
 }
 
+// InitTaskCooldownTimesRecord initialize TaskCooldownTimesRecord for each task
+func (ji *JobInfo) InitTaskCooldownTimesRecord(name string) {
+	var zero = int32(0)
+	ji.TaskCooldownTimesRecord[name] = &zero
+}
+
 // SetPodGroup sets podGroup details to a job
 func (ji *JobInfo) SetPodGroup(pg *PodGroup) {
 	ji.Name = pg.Name
@@ -382,6 +394,7 @@ func (ji *JobInfo) SetPodGroup(pg *PodGroup) {
 
 	ji.Preemptable = ji.extractPreemptable(pg)
 	ji.RevocableZone = ji.extractRevocableZone(pg)
+	ji.MaxCooldownTimes = ji.extractMaxCooldownTimes(pg)
 	ji.Budget = ji.extractBudget(pg)
 
 	taskMinAvailableTotal := int32(0)
@@ -392,6 +405,12 @@ func (ji *JobInfo) SetPodGroup(pg *PodGroup) {
 	ji.TaskMinAvailableTotal = taskMinAvailableTotal
 
 	ji.PodGroup = pg
+
+	for _, task := range ji.Tasks {
+		if _, found := ji.TaskCooldownTimesRecord[task.Name]; !found {
+			ji.InitTaskCooldownTimesRecord(task.Name)
+		}
+	}
 }
 
 // extractWaitingTime reads sla waiting time for job from podgroup annotations
@@ -442,6 +461,35 @@ func (ji *JobInfo) extractPreemptable(pg *PodGroup) bool {
 	return false
 }
 
+// extractMaxCooldownTimes return volcano.sh/max-cooldown-times value for job
+func (ji *JobInfo) extractMaxCooldownTimes(pg *PodGroup) int32 {
+	// check annotaion first
+	if len(pg.Annotations) > 0 {
+		if value, found := pg.Annotations[v1beta1.MaxCooldownTimes]; found {
+			i, err := strconv.ParseInt(value, 10, 32)
+			if err != nil {
+				klog.Warningf("invalid %s=%s", v1beta1.MaxCooldownTimes, value)
+				return 0
+			}
+			return int32(i)
+		}
+	}
+
+	// it annotation does not exit, check label
+	if len(pg.Labels) > 0 {
+		if value, found := pg.Labels[v1beta1.MaxCooldownTimes]; found {
+			i, err := strconv.ParseInt(value, 10, 32)
+			if err != nil {
+				klog.Warningf("invalid %s=%s", v1beta1.MaxCooldownTimes, value)
+				return 0
+			}
+			return int32(i)
+		}
+	}
+
+	return 0
+}
+
 // extractRevocableZone return volcano.sh/revocable-zone value for pod/podgroup
 func (ji *JobInfo) extractRevocableZone(pg *PodGroup) string {
 	// check annotation first
@@ -574,13 +622,15 @@ func (ji *JobInfo) Clone() *JobInfo {
 
 		PodGroup: ji.PodGroup.Clone(),
 
-		TaskStatusIndex:       map[TaskStatus]tasksMap{},
-		TaskMinAvailable:      ji.TaskMinAvailable,
-		TaskMinAvailableTotal: ji.TaskMinAvailableTotal,
-		Tasks:                 tasksMap{},
-		Preemptable:           ji.Preemptable,
-		RevocableZone:         ji.RevocableZone,
-		Budget:                ji.Budget.Clone(),
+		TaskStatusIndex:         map[TaskStatus]tasksMap{},
+		TaskMinAvailable:        ji.TaskMinAvailable,
+		TaskMinAvailableTotal:   ji.TaskMinAvailableTotal,
+		Tasks:                   tasksMap{},
+		Preemptable:             ji.Preemptable,
+		MaxCooldownTimes:        ji.MaxCooldownTimes,
+		TaskCooldownTimesRecord: ji.TaskCooldownTimesRecord,
+		RevocableZone:           ji.RevocableZone,
+		Budget:                  ji.Budget.Clone(),
 	}
 
 	ji.CreationTimestamp.DeepCopyInto(&info.CreationTimestamp)

diff --git a/pkg/scheduler/cache/cache.go b/pkg/scheduler/cache/cache.go
@@ -23,6 +23,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/client_golang/api"
@@ -697,6 +698,13 @@ func (sc *SchedulerCache) Evict(taskInfo *schedulingapi.TaskInfo, reason string)
 		return err
 	}
 
+	// Update task eviction times recode in JobInfo
+	if job.Preemptable {
+		if times, ok := job.TaskCooldownTimesRecord[task.Name]; ok {
+			atomic.AddInt32(times, 1)
+		}
+	}
+
 	p := task.Pod
 
 	go func() {

diff --git a/pkg/scheduler/framework/statement.go b/pkg/scheduler/framework/statement.go
@@ -18,6 +18,7 @@ package framework
 
 import (
 	"fmt"
+	"sync/atomic"
 
 	"k8s.io/klog"
 
@@ -59,7 +60,8 @@ func NewStatement(ssn *Session) *Statement {
 // Evict the pod
 func (s *Statement) Evict(reclaimee *api.TaskInfo, reason string) error {
 	// Update status in session
-	if job, found := s.ssn.Jobs[reclaimee.Job]; found {
+	job, found := s.ssn.Jobs[reclaimee.Job]
+	if found {
 		if err := job.UpdateTaskStatus(reclaimee, api.Releasing); err != nil {
 			klog.Errorf("Failed to update task <%v/%v> status to %v in Session <%v>: %v",
 				reclaimee.Namespace, reclaimee.Name, api.Releasing, s.ssn.UID, err)
@@ -79,6 +81,13 @@ func (s *Statement) Evict(reclaimee *api.TaskInfo, reason string) error {
 		}
 	}
 
+	// Update task eviction times recode in JobInfo
+	if job.Preemptable {
+		if times, ok := job.TaskCooldownTimesRecord[reclaimee.Name]; ok {
+			atomic.AddInt32(times, 1)
+		}
+	}
+
 	for _, eh := range s.ssn.eventHandlers {
 		if eh.DeallocateFunc != nil {
 			eh.DeallocateFunc(&Event{
@@ -130,6 +139,12 @@ func (s *Statement) unevict(reclaimee *api.TaskInfo) error {
 			return err
 		}
 	}
+	// Update task eviction times recode in JobInfo
+	if job.Preemptable {
+		if times, ok := job.TaskCooldownTimesRecord[reclaimee.Name]; ok {
+			atomic.AddInt32(times, -1)
+		}
+	}
 
 	for _, eh := range s.ssn.eventHandlers {
 		if eh.AllocateFunc != nil {