feat(job): add kubeflow PaddleJob (#1150)

* feat(job): add kubeflow PaddleJob

* update ut

* change the annotation of job KindGroupVersion

* minor change

Author: D0m021ng

config/server/default/job/job_template.yaml

@@ -95,6 +95,33 @@ spec:
 # paddle-collective-job
 ---
 apiVersion: "kubeflow.org/v1"
+kind: PaddleJob
+metadata:
+  name: paddle-simple-cpu
+  namespace: kubeflow
+spec:
+  paddleReplicaSpecs:
+    Worker:
+      replicas: 2
+      restartPolicy: OnFailure
+      template:
+        spec:
+          containers:
+            - name: paddle
+              image: registry.baidubce.com/paddlepaddle/paddle:2.4.0rc0-cpu
+              command:
+                - python
+              args:
+                - "-m"
+                - paddle.distributed.launch
+                - "run_check"
+              ports:
+                - containerPort: 37777
+                  name: master
+              imagePullPolicy: Always
+# PaddleJob-kubeflow.org/v1-collective
+---
+apiVersion: "kubeflow.org/v1"
 kind: "PyTorchJob"
 metadata:
   name: "pytorch-dist-sendrecv"

pkg/apis/training-operator/kubeflow.org/v1/paddlepaddle_types.go

@@ -0,0 +1,132 @@
+/*
+Copyright (c) 2022 PaddlePaddle Authors. All Rights Reserve.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1
+
+import (
+	commonv1 "github.com/kubeflow/common/pkg/apis/common/v1"
+	autoscalingv2 "k8s.io/api/autoscaling/v2beta2"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const (
+	// PaddleJobDefaultPortName is name of the port used to communicate between Master and
+	// workers.
+	PaddleJobDefaultPortName = "master"
+	// PaddleJobDefaultContainerName is the name of the PaddleJob container.
+	PaddleJobDefaultContainerName = "paddle"
+	// PaddleJobDefaultPort is default value of the port.
+	PaddleJobDefaultPort = 36543
+	// PaddleJobDefaultRestartPolicy is default RestartPolicy for PaddleReplicaSpec.
+	PaddleJobDefaultRestartPolicy = commonv1.RestartPolicyOnFailure
+	// PaddleJobKind is the kind name.
+	PaddleJobKind = "PaddleJob"
+	// PaddleJobPlural is the PaddlePlural for paddleJob.
+	PaddleJobPlural = "paddlejobs"
+	// PaddleJobSingular is the singular for paddleJob.
+	PaddleJobSingular = "paddlejob"
+	// PaddleJobFrameworkName is the name of the ML Framework
+	PaddleJobFrameworkName = "paddle"
+	// PaddleJobReplicaTypeMaster is the type of Master of distributed Paddle
+	PaddleJobReplicaTypeMaster commonv1.ReplicaType = "Master"
+	// PaddleJobReplicaTypeWorker is the type for workers of distributed Paddle.
+	PaddleJobReplicaTypeWorker commonv1.ReplicaType = "Worker"
+)
+
+// +genclient
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+// +resource:path=paddlejob
+//+kubebuilder:object:root=true
+//+kubebuilder:subresource:status
+//+kubebuilder:printcolumn:name="State",type=string,JSONPath=`.status.conditions[-1:].type`
+//+kubebuilder:printcolumn:name="Age",type=date,JSONPath=`.metadata.creationTimestamp`
+// +kubebuilder:subresource:scale:specpath=.spec.paddleReplicaSpecs.Worker.replicas,statuspath=.status.replicaStatuses.Worker.active,selectorpath=.status.replicaStatuses.Worker.selector
+
+// PaddleJob Represents a PaddleJob resource.
+type PaddleJob struct {
+	// Standard Kubernetes type metadata.
+	metav1.TypeMeta `json:",inline"`
+
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	// Specification of the desired state of the PaddleJob.
+	Spec PaddleJobSpec `json:"spec,omitempty"`
+
+	// Most recently observed status of the PaddleJob.
+	// Read-only (modified by the system).
+	Status commonv1.JobStatus `json:"status,omitempty"`
+}
+
+// PaddleJobSpec is a desired state description of the PaddleJob.
+type PaddleJobSpec struct {
+	// 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.
+	//+kubebuilder:validation:Optional
+	RunPolicy commonv1.RunPolicy `json:"runPolicy"`
+
+	// ElasticPolicy holds the elastic policy for paddle job.
+	ElasticPolicy *PaddleElasticPolicy `json:"elasticPolicy,omitempty"`
+
+	// A map of PaddleReplicaType (type) to ReplicaSpec (value). Specifies the Paddle cluster configuration.
+	// For example,
+	//   {
+	//     "Master": PaddleReplicaSpec,
+	//     "Worker": PaddleReplicaSpec,
+	//   }
+	PaddleReplicaSpecs map[commonv1.ReplicaType]*commonv1.ReplicaSpec `json:"paddleReplicaSpecs"`
+}
+
+type PaddleElasticPolicy struct {
+	// minReplicas is the lower limit for the number of replicas to which the training job
+	// can scale down.  It defaults to null.
+	// +optional
+	MinReplicas *int32 `json:"minReplicas,omitempty"`
+	// upper limit for the number of pods that can be set by the autoscaler; cannot be smaller than MinReplicas, defaults to null.
+	// +optional
+	MaxReplicas *int32 `json:"maxReplicas,omitempty"`
+
+	// MaxRestarts is the limit for restart times of pods in elastic mode.
+	// +optional
+	MaxRestarts *int32 `json:"maxRestarts,omitempty"`
+
+	// Metrics contains the specifications which are used to calculate the
+	// desired replica count (the maximum replica count across all metrics will
+	// be used).  The desired replica count is calculated with multiplying the
+	// ratio between the target value and the current value by the current
+	// number of pods. Ergo, metrics used must decrease as the pod count is
+	// increased, and vice-versa.  See the individual metric source types for
+	// more information about how each type of metric must respond.
+	// If not set, the HPA will not be created.
+	// +optional
+	Metrics []autoscalingv2.MetricSpec `json:"metrics,omitempty"`
+}
+
+// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
+// +resource:path=paddlejobs
+//+kubebuilder:object:root=true
+
+// PaddleJobList is a list of PaddleJobs.
+type PaddleJobList struct {
+	// Standard type metadata.
+	metav1.TypeMeta `json:",inline"`
+
+	// Standard list metadata.
+	metav1.ListMeta `json:"metadata,omitempty"`
+
+	// List of PaddleJobs.
+	Items []PaddleJob `json:"items"`
+}

pkg/apiserver/controller/job/create.go

@@ -620,11 +620,13 @@ func getFrameworkRoles(framework schema.Framework) map[schema.MemberRole]int {
 func buildJob(request *CreateJobInfo) (*model.Job, error) {
 	log.Debugf("begin build job with request: %#v", request)
 	// build main job config
-	conf := buildMainConf(request)
+	conf, err := buildMainConf(request)
+	if err != nil {
+		return nil, err
+	}
 	// convert job members if necessary
 	var members []schema.Member
 	var templateJson string
-	var err error
 	if len(request.Members) != 0 {
 		members = buildMembers(request)
 		log.Debugf("members is %v", members)
@@ -652,7 +654,7 @@ func buildJob(request *CreateJobInfo) (*model.Job, error) {
 	return jobInfo, nil
 }
 
-func buildMainConf(request *CreateJobInfo) *schema.Conf {
+func buildMainConf(request *CreateJobInfo) (*schema.Conf, error) {
 	log.Debugf("buildMainConf request %v", request)
 	var conf = &schema.Conf{
 		Name: request.Name,
@@ -681,12 +683,16 @@ func buildMainConf(request *CreateJobInfo) *schema.Conf {
 	if request.Type == schema.TypeWorkflow {
 		conf.KindGroupVersion = schema.WorkflowKindGroupVersion
 	} else {
-		conf.KindGroupVersion = schema.ToKindGroupVersion("", request.Framework, conf.Annotations)
+		var err error
+		conf.KindGroupVersion, err = schema.ToKindGroupVersion("", request.Framework, conf.Annotations)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	// TODO: remove job mode
 	conf.SetEnv(schema.EnvJobMode, request.Mode)
-	return conf
+	return conf, nil
 }
 
 func buildMembers(request *CreateJobInfo) []schema.Member {

pkg/common/k8s/discovery.go

@@ -48,6 +48,7 @@ var DiscoveryHandlerFunc = http.HandlerFunc(func(w http.ResponseWriter, req *htt
 				{Name: "pytorchjobs", Namespaced: true, Kind: "PyTorchJob"},
 				{Name: "tfjobs", Namespaced: true, Kind: "TFJob"},
 				{Name: "mpijobs", Namespaced: true, Kind: "MPIJob"},
+				{Name: "paddlejobs", Namespaced: true, Kind: "PaddleJob"},
 			},
 		}
 	case "/apis/argoproj.io/v1alpha1":

pkg/common/schema/job.go

@@ -152,9 +152,8 @@ const (
 	JobIDLabel        = "paddleflow-job-id"
 	JobTTLSeconds     = "paddleflow/job-ttl-seconds"
 	JobLabelFramework = "paddleflow-job-framework"
-	// JobKindAnnotation kind for job
-	JobKindAnnotation         = "paddleflow/job-kind"
-	JobGroupVersionAnnotation = "paddleflow/job-group-version"
+	// JobKindGroupVersionAnnotation KindGroupVersion for job, format: {kind}.{group}/{version}
+	JobKindGroupVersionAnnotation = "paddleflow/job-kind-group-version"
 
 	VolcanoJobNameLabel  = "volcano.sh/job-name"
 	QueueLabelKey        = "volcano.sh/queue-name"

pkg/common/schema/kind_version.go

@@ -36,43 +36,53 @@ func NewKindGroupVersion(kind, group, version string) KindGroupVersion {
 }
 
 func (kv KindGroupVersion) String() string {
-	return fmt.Sprintf("kind: %s, groupVersion: %s/%s", kv.Kind, kv.Group, kv.APIVersion)
+	return fmt.Sprintf("kind: %s, groupVersion: %s", kv.Kind, kv.GroupVersion())
 }
 
-func ToKindGroupVersion(clusterType string, framework Framework, annotations map[string]string) KindGroupVersion {
+func (kv KindGroupVersion) GroupVersion() string {
+	return fmt.Sprintf("%s/%s", kv.Group, kv.APIVersion)
+}
+
+func ToKindGroupVersion(clusterType string, framework Framework, annotations map[string]string) (KindGroupVersion, error) {
 	// TODO: get KindGroupVersion for different cluster
-	// get KindGroupVersion from job annotations
-	kind := annotations[JobKindAnnotation]
-	groupVersion := annotations[JobGroupVersionAnnotation]
-	if kind != "" && groupVersion != "" {
-		group, version, err := FromGroupVersion(groupVersion)
+	// 1. get KindGroupVersion from default
+	kindGV := frameworkKindGroupVersionMap[framework]
+	// 2. get KindGroupVersion from job annotations if set
+	kindGroupVersionStr := annotations[JobKindGroupVersionAnnotation]
+	if kindGroupVersionStr != "" {
+		kv, err := parseKindGroupVersion(kindGroupVersionStr)
 		if err == nil {
-			return NewKindGroupVersion(kind, group, version)
+			kindGV = kv
+		} else {
+			return KindGroupVersion{}, err
 		}
 	}
-	// get KindGroupVersion from default
-	kv, ok := frameworkKindGroupVersionMap[framework]
-	if !ok {
-		return KindGroupVersion{}
+	// 3. check KindGroupVersion
+	_, find := JobKindGroupVersionMap[kindGV]
+	if !find {
+		return KindGroupVersion{}, fmt.Errorf("the KindGroupVersion %s is not supported", kindGV)
 	}
-	return kv
+	return kindGV, nil
 }
 
-func FromGroupVersion(groupVersion string) (string, string, error) {
-	group := ""
-	version := ""
-	var err error
-	items := strings.Split(groupVersion, "/")
-	switch len(items) {
-	case 1:
-		version = items[0]
-	case 2:
-		version = items[1]
-		group = items[0]
-	default:
-		err = fmt.Errorf("unexpected GroupVersion string: %v", groupVersion)
+// parseKindGroupVersion convert KinGroupVersion str to struct, str format is {kind}.{group}/{version}
+func parseKindGroupVersion(kindGVStr string) (KindGroupVersion, error) {
+	kindGV := KindGroupVersion{}
+	kindVersion := strings.Split(kindGVStr, "/")
+	if len(kindVersion) != 2 {
+		return kindGV, fmt.Errorf("the KindGroupVersion %s is invalid, "+
+			"and it's format must be {kind}.{group}/{version}", kindGVStr)
+	}
+	// get kind and group
+	kindGroups := strings.Split(kindVersion[0], ".")
+	if len(kindGroups) < 2 {
+		return kindGV, fmt.Errorf("the KindGroupVersion %s is invalid, "+
+			"and it's format must be {kind}.{group}/{version}", kindGVStr)
 	}
-	return group, version, err
+	kindGV.Kind = kindGroups[0]
+	kindGV.Group = strings.TrimPrefix(kindVersion[0], kindGV.Kind+".")
+	kindGV.APIVersion = kindVersion[1]
+	return kindGV, nil
 }
 
 var (
@@ -103,6 +113,7 @@ var (
 		StandaloneKindGroupVersion: true,
 		SparkKindGroupVersion:      true,
 		PaddleKindGroupVersion:     true,
+		KFPaddleKindGroupVersion:   true,
 		PyTorchKindGroupVersion:    true,
 		TFKindGroupVersion:         true,
 		MXNetKindGroupVersion:      true,