add job priority for kube-batch scheduling #141

job_controller/api/v1/types.go

@@ -188,4 +188,8 @@ type RunPolicy struct {
 // job, for example `minAvailable` for gang-scheduling.
 type SchedulingPolicy struct {
 	MinAvailable *int32 `json:"minAvailable,omitempty"`
+
+	//add PriorityClassName
+	//PriorityClassName is a type of k8s resource.(kubectl get priorityclass)
+	PriorityClassName *string `json:"priorityClassName,omitempty"`
 }

job_controller/job.go

@@ -40,7 +40,7 @@ func (jc *JobController) deletePodsAndServices(runPolicy *apiv1.RunPolicy, job i
 	}
 	return nil
 }
-
+// TTL means time to live
 func (jc *JobController) cleanupJobIfTTL(runPolicy *apiv1.RunPolicy, jobStatus apiv1.JobStatus, job interface{}) error {
 	currentTime := time.Now()
 	metaObject, _ := job.(metav1.Object)
@@ -191,6 +191,16 @@ func (jc *JobController) ReconcileJobs(
 		return jc.Controller.UpdateJobStatusInApiServer(job, &jobStatus)
 	}
 
+	if jc.Config.EnableGangScheduling {
+		minAvailableReplicas := getTotalReplicas(replicas)
+		priorityClassName := getPriorityClassName(runPolicy)
+		//_, err := pc.SyncPodGroup(job, minAvailableReplicas)
+		_, err := jc.SyncPodGroup(metaObject, minAvailableReplicas, priorityClassName)
+		if err != nil {
+			log.Warnf("Sync PodGroup %v: %v", jobName, err)
+		}
+	}
+
 	// Save the current state of the replicas
 	replicasStatus := make(map[string]v1.PodPhase)
 
@@ -299,3 +309,14 @@ func (jc *JobController) cleanupJob(runPolicy *apiv1.RunPolicy, jobStatus apiv1.
 	jc.WorkQueue.AddRateLimited(key)
 	return nil
 }
+func getPriorityClassName(runPolicy *apiv1.RunPolicy) string {
+	priorityClassName := *runPolicy.SchedulingPolicy.PriorityClassName
+	return priorityClassName
+}
+func getTotalReplicas(replicas map[apiv1.ReplicaType]*apiv1.ReplicaSpec) int32 {
+	jobReplicas := int32(0)
+	for _, r := range replicas {
+		jobReplicas += *r.Replicas
+	}
+	return jobReplicas
+}

job_controller/job_controller.go

@@ -48,7 +48,7 @@ func (jc *JobController) GenLabels(jobName string) map[string]string {
 	}
 }
 
-func (jc *JobController) SyncPodGroup(job metav1.Object, minAvailableReplicas int32) (*v1alpha1.PodGroup, error) {
+func (jc *JobController) SyncPodGroup(job metav1.Object, minAvailableReplicas int32,priorityClassName string) (*v1alpha1.PodGroup, error) {
 
 	kubeBatchClientInterface := jc.KubeBatchClientSet
 	// Check whether podGroup exists or not
@@ -68,6 +68,7 @@ func (jc *JobController) SyncPodGroup(job metav1.Object, minAvailableReplicas in
 		},
 		Spec: v1alpha1.PodGroupSpec{
 			MinMember: minAvailable.IntVal,
+			PriorityClassName: priorityClassName,
 		},
 	}
 	return kubeBatchClientInterface.SchedulingV1alpha1().PodGroups(job.GetNamespace()).Create(createPodGroup)