lifecycle management of a trainingJob #634

qizheng09 · 2018-03-05T06:43:20Z

@typhoonzero Hi~, I would like you to review this pr.
This pr contains the main process of lifecycle of a TrainingJob. Including:

creating resources;
maintaining a state machine;
releasing resources；

qizheng09 · 2018-03-06T04:55:49Z

@typhoonzero
This pr will pass the travis after #635 be merged.

typhoonzero · 2018-03-07T02:20:08Z

@qizheng09 you can simply add "Reviewers" on the right side of this page, so that they can receive an email notification.

typhoonzero · 2018-03-07T03:51:07Z

go/updater/trainingJobUpdater.go

 		job:               job,
 		kubeClient:        kubeClient,
 		trainingJobClient: trainingJobClient,
 		status:            job.Status,
 		eventCh:           make(chan *trainingJobEvent, 1000),
 	}
-	return jobber, nil
+	return updater, nil
 }

 // NewUpdater return a trainingJobUpdater for controller.


NewUpdater return a trainingJobUpdater for controller. => NewUpdater creates a new TrainingJobUpdater and start a goroutine to control current job.

typhoonzero · 2018-03-07T03:51:41Z

go/updater/trainingJobUpdater.go

 }

 // NewUpdater return a trainingJobUpdater for controller.
 func NewUpdater(job *padv1.TrainingJob, kubeClient kubernetes.Interface, trainingJobClient trainingJobClient.Interface) (*TrainingJobUpdater,
 	error) {
 	log.Infof("NewJobber namespace=%v name=%v", job.Namespace, job.Name)
-	jobber, err := initUpdater(job, kubeClient, trainingJobClient)
+	updater, err := initUpdater(job, kubeClient, trainingJobClient)


May not need another function call here, initUpdater is quite short.

typhoonzero · 2018-03-07T03:55:15Z

go/updater/trainingJobUpdater.go

+		return fmt.Errorf("unknow resource")
+	}
+	var replica int32
+	replica = 0


var replica int32 replica = 0

is same to replica := 0

typhoonzero · 2018-03-07T04:47:41Z

go/updater/trainingJobUpdater.go

+
+	err = updater.kubeClient.CoreV1().Pods(updater.job.Namespace).DeleteCollection(&v1.DeleteOptions{}, options)
+	if err != nil {
+		return fmt.Errorf("release resource failed , namespace=%v name=%v resourceName=%v", updater.job.Namespace, updater.job.Name, resource.Name)


Can return this err directly.

typhoonzero · 2018-03-07T04:51:00Z

go/updater/trainingJobUpdater.go

+	for j := 0; j <= retryTimes; j++ {
+		time.Sleep(confirmResourceTicker)
+		pl, err := updater.kubeClient.CoreV1().Pods(updater.job.Namespace).List(options)
+		if err == nil && len(pl.Items) == 0 {


Need to do something if the error still exists after retry.

Retry is used to make sure that the replica of resource have been scaled to 0. If error still exists or len(pl.Items) haven't changed to 0 after retry. I will delete pods forcely as shown below.

typhoonzero · 2018-03-07T04:51:32Z

go/updater/trainingJobUpdater.go

+	return nil
+}
+
+func (updater *TrainingJobUpdater) releaseMaPs() error {


Function name releaseMaPs does not make sense. Need a more clear name.

typhoonzero · 2018-03-07T05:04:31Z

go/updater/trainingJobUpdater.go

+}
+
+// SetUp validates the fields and parses the TrainingJob
+func (updater *TrainingJobUpdater) setUp() {


Just call this parseTrainingJob

Yancey1989 · 2018-03-09T01:37:54Z

go/updater/trainingJobUpdater.go

+	return err
+}
+
+func (updater *TrainingJobUpdater) releaseMasterPserver() error {


How about split releaseMasterPserver into two function releaseMaster and releasePserver ?

Yancey1989 · 2018-03-09T01:56:17Z

go/updater/trainingJobUpdater.go

+	}
+
+	err := updater.kubeClient.CoreV1().Pods(updater.job.Namespace).DeleteCollection(&v1.DeleteOptions{}, options)
+	if err != nil {


Looks no need to check err != nil, just

return updater.kubeClient.CoreV1().Pods(updater.job.Namespace).DeleteCollection(&v1.DeleteOptions{}, options)

Yancey1989 · 2018-03-09T02:00:31Z

go/updater/trainingJobUpdater.go

+		return err
+	}
+
+	err := updater.createTrainer()


return updater.createTrainer()

Yancey1989 · 2018-03-09T02:01:06Z

go/updater/trainingJobUpdater.go

+}
+
+func (updater *TrainingJobUpdater) createTrainingJob() error {
+


Delete redundant blank line.

@Yancey1989 Done and thanks for your comments~

helinwang · 2018-03-12T22:33:08Z

go/updater/trainingJobUpdater.go

+// it will notify updater to process the event. It send event to updater's eventCh.
+func (updater *TrainingJobUpdater) notify(te *trainingJobEvent) {
+	select {
+	case updater.eventCh <- te:


Is this code the same with:

// no more select updater.eventCh <- te // code in the case block.

Maybe the above is less verbose?

helinwang · 2018-03-12T22:33:34Z

go/updater/trainingJobUpdater.go

+	select {
+	case updater.eventCh <- te:
+		lene, cape := len(updater.eventCh), cap(updater.eventCh)
+		if lene > int(float64(cape)*0.8) {


Maybe put 0.8 as a constant in this file.

helinwang · 2018-03-12T22:36:39Z

go/updater/trainingJobUpdater.go

+		for v := range ticker.C {
+			rs, err := updater.kubeClient.ExtensionsV1beta1().ReplicaSets(updater.job.Namespace).Get(resource.Name, v1.GetOptions{})
+			log.Infof("Current time %v runing pod is %v, resourceName=%v", v.String(), rs.Status.ReadyReplicas, resource.Name)
+			if err != nil && !errors.IsServerTimeout(err) && !errors.IsTooManyRequests(err) {


What will happen if err is timeout or too many requests? I see the code below uses the returned rs even when the err happened.

helinwang · 2018-03-12T22:39:08Z

go/updater/trainingJobUpdater.go

+func (updater *TrainingJobUpdater) parseTrainingJob() {
+	if updater.job == nil {
+		updater.status.Phase = padv1.TrainingJobPhaseFailed
+		updater.status.Reason = "Internal error; Setup error; job is missing TainingJob"


Maybe need to return at here?

helinwang · 2018-03-12T22:43:37Z

go/updater/trainingJobUpdater.go

+			switch ev.pet {
+			case trainingJobEventDelete:
+				log.Infof("Delete updater, namespace=%v name=%v: ", updater.job.Namespace, updater.job.Name)
+				ticker.Stop()


There is already defer ticker.Stop() above, maybe it's not necessary?

helinwang · 2018-03-12T22:46:34Z

go/updater/trainingJobUpdater.go

+	}
+
+	if updater.status.Phase == padv1.TrainingJobPhaseSucceeded || updater.status.Phase == padv1.TrainingJobPhaseFailed {
+		if ticker != nil {


Maybe move this stop ticker logic to line 466, Convert function does not have much to do with ticker.

Done. Thank you very much for your comments, and I have fixed all of the comments you mentioned above.

Yancey1989 · 2018-03-13T09:08:18Z

go/updater/trainingJobUpdater.go

+	resource.Spec.Replicas = &replica
+	_, err := updater.kubeClient.ExtensionsV1beta1().ReplicaSets(updater.job.Namespace).Update(resource)
+	if errors.IsNotFound(err) {
+		return nil


May return err here?

Yancey1989 · 2018-03-13T09:14:02Z

go/updater/trainingJobUpdater.go

+			return err
+		}
+	}
+	if newResource != nil {


I think we need to deal with other error except IsNotFound, and the example code may be helpful:
https://github.com/kubernetes/kubernetes/blob/a7d6340ad28970a99a67feba3913d0bd653855b1/staging/src/k8s.io/client-go/examples/in-cluster-client-configuration/main.go#L51

Our goal is to try our best to ensure the normal operation of the job. In case of unexpected errors, should we deal with other errors by retrying？

Yancey1989 · 2018-03-13T09:15:51Z

go/updater/trainingJobUpdater.go

+
+func (updater *TrainingJobUpdater) createTrainer() error {
+	newTrainer, err := updater.kubeClient.BatchV1().Jobs(updater.job.Namespace).Get(updater.job.Spec.Trainer.ReplicaSpec.Name, v1.GetOptions{})
+	if errors.IsNotFound(err) {


The same as: https://github.com/PaddlePaddle/cloud/pull/634/files#r174057818

Done. Thank you very much for your comments, and I have fixed all of the comments you mentioned above.

typhoonzero

LGTM. Need some naming polishment later on.

typhoonzero · 2018-03-15T07:22:13Z

go/updater/trainingJobUpdater.go

+	log.Infof("End to delete TrainingJob namespace=%v name=%v", updater.job.Namespace, updater.job.Name)
+
+	if fault {
+		return fmt.Errorf("delete resource error namespace=%v name=%v", updater.job.Namespace, updater.job.Name)


This may lose the original k8s api call error message, return the error object from releaseXXX call.

qizheng09 force-pushed the lifecycle_managerment branch 2 times, most recently from d8cf99e to fc110b2 Compare March 6, 2018 13:44

typhoonzero requested review from Yancey1989, helinwang, gongweibao and typhoonzero March 7, 2018 02:26

typhoonzero reviewed Mar 7, 2018

View reviewed changes

qizheng09 force-pushed the lifecycle_managerment branch from fc110b2 to 8473a43 Compare March 7, 2018 06:27

Yancey1989 reviewed Mar 9, 2018

View reviewed changes

qizheng09 force-pushed the lifecycle_managerment branch from 8473a43 to 57eeb5e Compare March 9, 2018 03:23

lifecycle management of a trainingJob

d6447f6

qizheng09 force-pushed the lifecycle_managerment branch from 57eeb5e to d6447f6 Compare March 9, 2018 05:17

helinwang reviewed Mar 12, 2018

View reviewed changes

fix some comments

e46e836

Yancey1989 reviewed Mar 13, 2018

View reviewed changes

helinwang mentioned this pull request Mar 13, 2018

Move go/updater to PaddlePaddle/edl/pkg/updater #643

Open

error handle for creating resources

ac86ef9

qizheng09 force-pushed the lifecycle_managerment branch from d46a482 to ac86ef9 Compare March 14, 2018 11:40

typhoonzero approved these changes Mar 15, 2018

View reviewed changes

Merge branch 'develop' into lifecycle_managerment

c383a32

qizheng09 merged commit fdd97ba into PaddlePaddle:develop Mar 15, 2018

This was referenced Mar 19, 2018

migrate lifecycle management of a trainingJob elasticdeeplearning/edl#15

Closed

migrate lifecycle management of a trainingJob elasticdeeplearning/edl#16

Merged

		}

		func (updater *TrainingJobUpdater) createTrainingJob() error {

lifecycle management of a trainingJob #634

lifecycle management of a trainingJob #634

Conversation

qizheng09 commented Mar 5, 2018 • edited Loading

qizheng09 commented Mar 6, 2018

typhoonzero commented Mar 7, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

qizheng09 Mar 7, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

helinwang Mar 12, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

typhoonzero left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

qizheng09 commented Mar 5, 2018 •

edited

Loading

qizheng09 Mar 7, 2018 •

edited

Loading

helinwang Mar 12, 2018 •

edited

Loading