Copy SchedulingPolicy and CleanPodPolicy for RunPolicy

kubeflow · Jan 30, 2023 · 44db325 · 44db325
1 parent 502685c
commit 44db325
Show file tree

Hide file tree

Showing 19 changed files with 527 additions and 79 deletions.
diff --git a/pkg/apis/kubeflow/v2beta1/default.go b/pkg/apis/kubeflow/v2beta1/default.go
@@ -51,7 +51,7 @@ func setDefaultsTypeWorker(spec *common.ReplicaSpec) {
 
 func setDefaultsRunPolicy(policy *RunPolicy) {
 	if policy.CleanPodPolicy == nil {
-		policy.CleanPodPolicy = newCleanPodPolicy(common.CleanPodPolicyNone)
+		policy.CleanPodPolicy = newCleanPodPolicy(CleanPodPolicyNone)
 	}
 	// The remaining fields are passed as-is to the k8s Job API, which does its
 	// own defaulting.
@@ -80,6 +80,6 @@ func newInt32(v int32) *int32 {
 	return &v
 }
 
-func newCleanPodPolicy(policy common.CleanPodPolicy) *common.CleanPodPolicy {
+func newCleanPodPolicy(policy CleanPodPolicy) *CleanPodPolicy {
 	return &policy
 }
diff --git a/pkg/apis/kubeflow/v2beta1/default_test.go b/pkg/apis/kubeflow/v2beta1/default_test.go
@@ -31,7 +31,7 @@ func TestSetDefaults_MPIJob(t *testing.T) {
 				Spec: MPIJobSpec{
 					SlotsPerWorker: newInt32(1),
 					RunPolicy: RunPolicy{
-						CleanPodPolicy: newCleanPodPolicy(common.CleanPodPolicyNone),
+						CleanPodPolicy: newCleanPodPolicy(CleanPodPolicyNone),
 					},
 					SSHAuthMountPath:  "/root/.ssh",
 					MPIImplementation: MPIImplementationOpenMPI,
@@ -43,7 +43,7 @@ func TestSetDefaults_MPIJob(t *testing.T) {
 				Spec: MPIJobSpec{
 					SlotsPerWorker: newInt32(10),
 					RunPolicy: RunPolicy{
-						CleanPodPolicy:          newCleanPodPolicy(common.CleanPodPolicyRunning),
+						CleanPodPolicy:          newCleanPodPolicy(CleanPodPolicyRunning),
 						TTLSecondsAfterFinished: newInt32(2),
 						ActiveDeadlineSeconds:   newInt64(3),
 						BackoffLimit:            newInt32(4),
@@ -56,7 +56,7 @@ func TestSetDefaults_MPIJob(t *testing.T) {
 				Spec: MPIJobSpec{
 					SlotsPerWorker: newInt32(10),
 					RunPolicy: RunPolicy{
-						CleanPodPolicy:          newCleanPodPolicy(common.CleanPodPolicyRunning),
+						CleanPodPolicy:          newCleanPodPolicy(CleanPodPolicyRunning),
 						TTLSecondsAfterFinished: newInt32(2),
 						ActiveDeadlineSeconds:   newInt64(3),
 						BackoffLimit:            newInt32(4),
@@ -78,7 +78,7 @@ func TestSetDefaults_MPIJob(t *testing.T) {
 				Spec: MPIJobSpec{
 					SlotsPerWorker: newInt32(1),
 					RunPolicy: RunPolicy{
-						CleanPodPolicy: newCleanPodPolicy(common.CleanPodPolicyNone),
+						CleanPodPolicy: newCleanPodPolicy(CleanPodPolicyNone),
 					},
 					SSHAuthMountPath:  "/root/.ssh",
 					MPIImplementation: MPIImplementationOpenMPI,
@@ -103,7 +103,7 @@ func TestSetDefaults_MPIJob(t *testing.T) {
 				Spec: MPIJobSpec{
 					SlotsPerWorker: newInt32(1),
 					RunPolicy: RunPolicy{
-						CleanPodPolicy: newCleanPodPolicy(common.CleanPodPolicyNone),
+						CleanPodPolicy: newCleanPodPolicy(CleanPodPolicyNone),
 					},
 					SSHAuthMountPath:  "/root/.ssh",
 					MPIImplementation: MPIImplementationOpenMPI,

diff --git a/pkg/apis/kubeflow/v2beta1/openapi_generated.go b/pkg/apis/kubeflow/v2beta1/openapi_generated.go
diff --git a/pkg/apis/kubeflow/v2beta1/swagger.json b/pkg/apis/kubeflow/v2beta1/swagger.json
@@ -293,14 +293,41 @@
         },
         "schedulingPolicy": {
           "description": "SchedulingPolicy defines the policy related to scheduling, e.g. gang-scheduling",
-          "$ref": "#/definitions/v1.SchedulingPolicy"
+          "$ref": "#/definitions/v2beta1.SchedulingPolicy"
         },
         "ttlSecondsAfterFinished": {
           "description": "TTLSecondsAfterFinished is the TTL to clean up jobs. It may take extra ReconcilePeriod seconds for the cleanup, since reconcile gets called periodically. Default to infinite.",
           "type": "integer",
           "format": "int32"
         }
       }
+    },
+    "v2beta1.SchedulingPolicy": {
+      "description": "SchedulingPolicy encapsulates various scheduling policies of the distributed training job, for example `minAvailable` for gang-scheduling.",
+      "type": "object",
+      "properties": {
+        "minAvailable": {
+          "type": "integer",
+          "format": "int32"
+        },
+        "minResources": {
+          "type": "object",
+          "additionalProperties": {
+            "default": {},
+            "$ref": "#/definitions/resource.Quantity"
+          }
+        },
+        "priorityClass": {
+          "type": "string"
+        },
+        "queue": {
+          "type": "string"
+        },
+        "scheduleTimeoutSeconds": {
+          "type": "integer",
+          "format": "int32"
+        }
+      }
     }
   }
 }
diff --git a/pkg/apis/kubeflow/v2beta1/types.go b/pkg/apis/kubeflow/v2beta1/types.go
@@ -16,6 +16,7 @@ package v2beta1
 
 import (
 	common "github.com/kubeflow/common/pkg/apis/common/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
@@ -37,13 +38,33 @@ type MPIJobList struct {
 	Items           []MPIJob `json:"items"`
 }
 
+// CleanPodPolicy describes how to deal with pods when the job is finished.
+type CleanPodPolicy string
+
+const (
+	CleanPodPolicyUndefined CleanPodPolicy = ""
+	CleanPodPolicyAll       CleanPodPolicy = "All"
+	CleanPodPolicyRunning   CleanPodPolicy = "Running"
+	CleanPodPolicyNone      CleanPodPolicy = "None"
+)
+
+// SchedulingPolicy encapsulates various scheduling policies of the distributed training
+// job, for example `minAvailable` for gang-scheduling.
+type SchedulingPolicy struct {
+	MinAvailable           *int32           `json:"minAvailable,omitempty"`
+	Queue                  string           `json:"queue,omitempty"`
+	MinResources           *v1.ResourceList `json:"minResources,omitempty"`
+	PriorityClass          string           `json:"priorityClass,omitempty"`
+	ScheduleTimeoutSeconds *int32           `json:"scheduleTimeoutSeconds,omitempty"`
+}
+
 // RunPolicy encapsulates various runtime policies of the distributed training
 // job, for example how to clean up resources and how long the job can stay
 // active.
 type RunPolicy struct {
 	// CleanPodPolicy defines the policy to kill pods after the job completes.
 	// Default to Running.
-	CleanPodPolicy *common.CleanPodPolicy `json:"cleanPodPolicy,omitempty"`
+	CleanPodPolicy *CleanPodPolicy `json:"cleanPodPolicy,omitempty"`
 
 	// TTLSecondsAfterFinished is the TTL to clean up jobs.
 	// It may take extra ReconcilePeriod seconds for the cleanup, since
@@ -62,7 +83,7 @@ type RunPolicy struct {
 
 	// SchedulingPolicy defines the policy related to scheduling, e.g. gang-scheduling
 	// +optional
-	SchedulingPolicy *common.SchedulingPolicy `json:"schedulingPolicy,omitempty"`
+	SchedulingPolicy *SchedulingPolicy `json:"schedulingPolicy,omitempty"`
 }
 
 type MPIJobSpec struct {

diff --git a/pkg/apis/kubeflow/v2beta1/zz_generated.deepcopy.go b/pkg/apis/kubeflow/v2beta1/zz_generated.deepcopy.go
diff --git a/pkg/apis/kubeflow/validation/validation.go b/pkg/apis/kubeflow/validation/validation.go
@@ -29,9 +29,9 @@ import (
 
 var (
 	validCleanPolicies = sets.NewString(
-		string(common.CleanPodPolicyNone),
-		string(common.CleanPodPolicyRunning),
-		string(common.CleanPodPolicyAll))
+		string(kubeflow.CleanPodPolicyNone),
+		string(kubeflow.CleanPodPolicyRunning),
+		string(kubeflow.CleanPodPolicyAll))
 
 	validMPIImplementations = sets.NewString(
 		string(kubeflow.MPIImplementationOpenMPI),