Repository: kubeflow/training-operator (language: python)

Issue:
Job TTLs not working
I'm running into an issue where the "Ttl Seconds After Finished" on my TFJobs isn't being respected and I suspect it's because the reconcile loop where `CleanupJob` runs isn't run frequently enough for all jobs. 

As an example:  I start a TFJob with a TTL of 1 minute. The reconcile loop runs anytime there's a state change (including when the job successfully finishes). It doesn't get deleted upon finishing because the TTL hasn't passed. Then, the reconcile loop doesn't run, in this case, until 7 hours after the job was finished. At that point the job does get cleaned up. (this and other jobs past TTL)

My question is: Is the fact that the reconcile loop doesn't run at shorter intervals for finished jobs expected behaviour? Is there a way to find out where that default 7 hours period if set? If so, would changing it to a smaller default to accommodate the TTL feature make sense?

Here are the training-operator logs for the job from start to finish:

```2022-02-08 17:42:43.836 PST
training-operator
time="2022-02-09T01:42:43Z" level=info msg="TFJob test-ttl-6 is created."
Error
2022-02-08 17:42:43.836 PST
training-operator
time="2022-02-09T01:42:43Z" level=info msg="Reconciling for job test-ttl-6"
Error
2022-02-08 17:42:43.836 PST
training-operator
time="2022-02-09T01:42:43Z" level=warning msg="The restart policy of replica Chief of the job test-ttl-6 is not OnFailure or Always. Not counted in backoff limit."
Error
2022-02-08 17:42:43.836 PST
training-operator
time="2022-02-09T01:42:43Z" level=info msg="Need to create new pod: chief-0" job=adelijani.test-ttl-6 uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
Error
2022-02-08 17:42:44.424 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="Controller test-ttl-6 created pod test-ttl-6-chief-0" job=.test-ttl-6 pod=.test-ttl-6-chief-0 uid=
Error
2022-02-08 17:42:44.424 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="need to create new service: chief-0" job=adelijani.test-ttl-6 replica-type=chief uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
Error
2022-02-08 17:42:44.424 PST
training-operator
2022-02-09T01:42:44.424Z DEBUG controller-runtime.manager.events Normal {"object": {"kind":"TFJob","namespace":"adelijani","name":"test-ttl-6","uid":"5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39","apiVersion":"kubeflow.org/v1","resourceVersion":"806803740"}, "reason": "SuccessfulCreatePod", "message": "Created pod: test-ttl-6-chief-0"}
Error
2022-02-08 17:42:44.436 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="Controller test-ttl-6 created service test-ttl-6-chief-0"
Error
2022-02-08 17:42:44.436 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="TFJob=adelijani/test-ttl-6, ReplicaType=Chief expected=1, running=0, failed=0" job=adelijani.test-ttl-6 uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
Error
2022-02-08 17:42:44.436 PST
training-operator
2022-02-09T01:42:44.436Z DEBUG controller-runtime.manager.events Normal {"object": {"kind":"TFJob","namespace":"adelijani","name":"test-ttl-6","uid":"5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39","apiVersion":"kubeflow.org/v1","resourceVersion":"806803740"}, "reason": "SuccessfulCreateService", "message": "Created service: test-ttl-6-chief-0"}
Error
2022-02-08 17:42:44.444 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="Finished updating TFJobs Status \"test-ttl-6\" (8.171651ms)" job=adelijani.test-ttl-6 uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
Error
2022-02-08 17:42:44.444 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="Reconciling for job test-ttl-6"
Error
2022-02-08 17:42:44.444 PST
training-operator
time="2022-02-09T01:42:44Z" level=warning msg="The restart policy of replica Chief of the job test-ttl-6 is not OnFailure or Always. Not counted in backoff limit."
Error
2022-02-08 17:42:44.444 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="TFJob=adelijani/test-ttl-6, ReplicaType=Chief expected=1, running=0, failed=0" job=adelijani.test-ttl-6 uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
Error
2022-02-08 17:42:44.450 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="Finished updating TFJobs Status \"test-ttl-6\" (5.933567ms)" job=adelijani.test-ttl-6 uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
Error
2022-02-08 17:42:44.451 PST
training-operator
time="2022-02-09T01:42:44Z" level=warning msg="Reconcile Tensorflow Job error Operation cannot be fulfilled on tfjobs.kubeflow.org \"test-ttl-6\": the object has been modified; please apply your changes to the latest version and try again"
Error
2022-02-08 17:42:44.451 PST
training-operator
2022-02-09T01:42:44.450Z ERROR controller-runtime.manager.controller.tfjob-controller Reconciler error {"name": "test-ttl-6", "namespace": "adelijani", "error": "Operation cannot be fulfilled on tfjobs.kubeflow.org \"test-ttl-6\": the object has been modified; please apply your changes to the latest version and try again"}
Error
2022-02-08 17:42:44.451 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="Reconciling for job test-ttl-6"
Error
2022-02-08 17:42:44.451 PST
training-operator
time="2022-02-09T01:42:44Z" level=warning msg="The restart policy of replica Chief of the job test-ttl-6 is not OnFailure or Always. Not counted in backoff limit."
Error
2022-02-08 17:42:44.451 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="TFJob=adelijani/test-ttl-6, ReplicaType=Chief expected=1, running=0, failed=0" job=adelijani.test-ttl-6 uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
Error
2022-02-08 17:42:44.456 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="Reconciling for job test-ttl-6"
Error
2022-02-08 17:42:44.456 PST
training-operator
time="2022-02-09T01:42:44Z" level=warning msg="The restart policy of replica Chief of the job test-ttl-6 is not OnFailure or Always. Not counted in backoff limit."
Error
2022-02-08 17:42:44.456 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="TFJob=adelijani/test-ttl-6, ReplicaType=Chief expected=1, running=0, failed=0" job=adelijani.test-ttl-6 uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
Error
2022-02-08 17:42:44.457 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="Reconciling for job test-ttl-6"
Error
2022-02-08 17:42:44.457 PST
training-operator
time="2022-02-09T01:42:44Z" level=warning msg="The restart policy of replica Chief of the job test-ttl-6 is not OnFailure or Always. Not counted in backoff limit."
Error
2022-02-08 17:42:44.457 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="TFJob=adelijani/test-ttl-6, ReplicaType=Chief expected=1, running=0, failed=0" job=adelijani.test-ttl-6 uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
Error
2022-02-08 17:42:44.835 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="Reconciling for job test-ttl-6"
Error
2022-02-08 17:42:44.835 PST
training-operator
time="2022-02-09T01:42:44Z" level=warning msg="The restart policy of replica Chief of the job test-ttl-6 is not OnFailure or Always. Not counted in backoff limit."
Error
2022-02-08 17:42:44.835 PST
training-operator
time="2022-02-09T01:42:44Z" level=info msg="TFJob=adelijani/test-ttl-6, ReplicaType=Chief expected=1, running=0, failed=0" job=adelijani.test-ttl-6 uid=5eeb5c65-b33b-4a56-ba8d-c02ce3bccb39
```


Discussion / hints:
>  Is there a way to find out where that default 7 hours period if set?

I believe so. When [constructing the manager](https://github.com/kubeflow/training-operator/blob/master/cmd/training-operator.v1/main.go#L101), an option named [`SyncPeriod`](https://github.com/kubernetes-sigs/controller-runtime/blob/6c7c1d73fc5ed965c2962d6f4552b15f9841ce86/pkg/manager/manager.go#L109) should do the work. *You will need to modify the `main.go` file to enable such option.*

However, I believe we might find another way to re-support such feature after the workqueue disabled in the reconcile mode. @Jeffwan 
Thanks for the helpful guidance.  Looking forward to hearing if there's a way to change this without touching code.

Fix patch:
diff --git a/pkg/common/util/util.go b/pkg/common/util/util.go
index 0ddef5ed74..8f4a1d7dea 100644
--- a/pkg/common/util/util.go
+++ b/pkg/common/util/util.go
@@ -15,7 +15,11 @@
 package util
 
 import (
+	"fmt"
+	"time"
+
 	commonv1 "github.com/kubeflow/common/pkg/apis/common/v1"
+	commonutil "github.com/kubeflow/common/pkg/util"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -72,3 +76,26 @@ func GetReplicaTypes(specs map[commonv1.ReplicaType]*commonv1.ReplicaSpec) []com
 	}
 	return keys
 }
+
+// DurationUntilExpireTime returns the duration until job needs to be cleaned up, or -1 if it's infinite.
+func DurationUntilExpireTime(runPolicy *commonv1.RunPolicy, jobStatus commonv1.JobStatus) (time.Duration, error) {
+	if !commonutil.IsSucceeded(jobStatus) && !commonutil.IsFailed(jobStatus) {
+		return -1, nil
+	}
+	currentTime := time.Now()
+	ttl := runPolicy.TTLSecondsAfterFinished
+	if ttl == nil {
+		return -1, nil
+	}
+	duration := time.Second * time.Duration(*ttl)
+	if jobStatus.CompletionTime == nil {
+		return -1, fmt.Errorf("job completion time is nil, cannot cleanup")
+	}
+	finishTime := jobStatus.CompletionTime
+	expireTime := finishTime.Add(duration)
+	if currentTime.After(expireTime) {
+		return 0, nil
+	} else {
+		return expireTime.Sub(currentTime), nil
+	}
+}
diff --git a/pkg/controller.v1/mpi/mpijob_controller.go b/pkg/controller.v1/mpi/mpijob_controller.go
index f0a93f1bbc..13991f91b7 100644
--- a/pkg/controller.v1/mpi/mpijob_controller.go
+++ b/pkg/controller.v1/mpi/mpijob_controller.go
@@ -162,6 +162,15 @@ func (jc *MPIJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		return ctrl.Result{}, err
 	}
 
+	t, err := util.DurationUntilExpireTime(&mpijob.Spec.RunPolicy, mpijob.Status)
+	if err != nil {
+		logrus.Warnf("Reconcile MPIJob Job error %v", err)
+		return ctrl.Result{}, err
+	}
+	if t >= 0 {
+		return ctrl.Result{Requeue: true, RequeueAfter: t}, nil
+	}
+
 	return ctrl.Result{}, nil
 }
 
diff --git a/pkg/controller.v1/mxnet/mxjob_controller.go b/pkg/controller.v1/mxnet/mxjob_controller.go
index 45169703b9..6183a983bc 100644
--- a/pkg/controller.v1/mxnet/mxjob_controller.go
+++ b/pkg/controller.v1/mxnet/mxjob_controller.go
@@ -163,6 +163,15 @@ func (r *MXJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl
 		return ctrl.Result{}, err
 	}
 
+	t, err := util.DurationUntilExpireTime(&mxjob.Spec.RunPolicy, mxjob.Status)
+	if err != nil {
+		logrus.Warnf("Reconcile MX Job error %v", err)
+		return ctrl.Result{}, err
+	}
+	if t >= 0 {
+		return ctrl.Result{Requeue: true, RequeueAfter: t}, nil
+	}
+
 	return ctrl.Result{}, nil
 }
 
diff --git a/pkg/controller.v1/pytorch/pytorchjob_controller.go b/pkg/controller.v1/pytorch/pytorchjob_controller.go
index e376aa899e..1c219ffa77 100644
--- a/pkg/controller.v1/pytorch/pytorchjob_controller.go
+++ b/pkg/controller.v1/pytorch/pytorchjob_controller.go
@@ -162,6 +162,15 @@ func (r *PyTorchJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, err
 	}
 
+	t, err := util.DurationUntilExpireTime(&pytorchjob.Spec.RunPolicy, pytorchjob.Status)
+	if err != nil {
+		logrus.Warnf("Reconcile PyTorchJob error %v", err)
+		return ctrl.Result{}, err
+	}
+	if t >= 0 {
+		return ctrl.Result{Requeue: true, RequeueAfter: t}, nil
+	}
+
 	return ctrl.Result{}, nil
 }
 
diff --git a/pkg/controller.v1/tensorflow/tfjob_controller.go b/pkg/controller.v1/tensorflow/tfjob_controller.go
index 23e551a915..c936086c12 100644
--- a/pkg/controller.v1/tensorflow/tfjob_controller.go
+++ b/pkg/controller.v1/tensorflow/tfjob_controller.go
@@ -183,6 +183,15 @@ func (r *TFJobReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl
 		return ctrl.Result{}, err
 	}
 
+	t, err := util.DurationUntilExpireTime(&tfjob.Spec.RunPolicy, tfjob.Status)
+	if err != nil {
+		logrus.Warnf("Reconcile Tensorflow Job error %v", err)
+		return ctrl.Result{}, err
+	}
+	if t >= 0 {
+		return ctrl.Result{Requeue: true, RequeueAfter: t}, nil
+	}
+
 	return ctrl.Result{}, nil
 }
 
diff --git a/pkg/controller.v1/xgboost/xgboostjob_controller.go b/pkg/controller.v1/xgboost/xgboostjob_controller.go
index 339feabd85..2d6da31143 100644
--- a/pkg/controller.v1/xgboost/xgboostjob_controller.go
+++ b/pkg/controller.v1/xgboost/xgboostjob_controller.go
@@ -169,6 +169,15 @@ func (r *XGBoostJobReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		return ctrl.Result{}, err
 	}
 
+	t, err := util.DurationUntilExpireTime(&xgboostjob.Spec.RunPolicy, xgboostjob.Status)
+	if err != nil {
+		logrus.Warnf("Reconcile XGBoost Job error %v", err)
+		return ctrl.Result{}, err
+	}
+	if t >= 0 {
+		return ctrl.Result{Requeue: true, RequeueAfter: t}, nil
+	}
+
 	return reconcile.Result{}, nil
 }
 


Test patch:
diff --git a/pkg/common/util/util_test.go b/pkg/common/util/util_test.go
new file mode 100644
index 0000000000..b5c4d2d7c4
--- /dev/null
+++ b/pkg/common/util/util_test.go
@@ -0,0 +1,124 @@
+// Copyright 2022 The Kubeflow Authors
+//
+// 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 util
+
+import (
+	"testing"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+
+	commonv1 "github.com/kubeflow/common/pkg/apis/common/v1"
+)
+
+func TestDurationUntilExpireTime(t *testing.T) {
+	tests := []struct {
+		name      string
+		runPolicy *commonv1.RunPolicy
+		jobStatus commonv1.JobStatus
+		want      time.Duration
+		wantErr   bool
+	}{
+		{
+			name:      "running job",
+			runPolicy: &commonv1.RunPolicy{},
+			jobStatus: commonv1.JobStatus{
+				Conditions: []commonv1.JobCondition{newJobCondition(commonv1.JobRunning)},
+			},
+			want:    -1,
+			wantErr: false,
+		},
+		{
+			name: "succeeded job with remaining time 1s",
+			runPolicy: &commonv1.RunPolicy{
+				TTLSecondsAfterFinished: pointer.Int32(5),
+			},
+			jobStatus: commonv1.JobStatus{
+				Conditions:     []commonv1.JobCondition{newJobCondition(commonv1.JobSucceeded)},
+				CompletionTime: &metav1.Time{Time: time.Now().Add(4 * time.Second)},
+			},
+			want:    1,
+			wantErr: false,
+		},
+		{
+			name: "failed job with remaining time 1s",
+			runPolicy: &commonv1.RunPolicy{
+				TTLSecondsAfterFinished: pointer.Int32(5),
+			},
+			jobStatus: commonv1.JobStatus{
+				Conditions:     []commonv1.JobCondition{newJobCondition(commonv1.JobFailed)},
+				CompletionTime: &metav1.Time{Time: time.Now().Add(4 * time.Second)},
+			},
+			want:    1,
+			wantErr: false,
+		},
+		{
+			name:      "succeeded job with infinite TTL",
+			runPolicy: &commonv1.RunPolicy{},
+			jobStatus: commonv1.JobStatus{
+				Conditions:     []commonv1.JobCondition{newJobCondition(commonv1.JobSucceeded)},
+				CompletionTime: &metav1.Time{Time: time.Now().Add(4 * time.Second)},
+			},
+			want:    -1,
+			wantErr: false,
+		},
+		{
+			name: "succeeded job without remaining time",
+			runPolicy: &commonv1.RunPolicy{
+				TTLSecondsAfterFinished: pointer.Int32(5),
+			},
+			jobStatus: commonv1.JobStatus{
+				Conditions:     []commonv1.JobCondition{newJobCondition(commonv1.JobSucceeded)},
+				CompletionTime: &metav1.Time{Time: time.Now().Add(6 * time.Second)},
+			},
+			want:    0,
+			wantErr: false,
+		},
+		{
+			name: "succeeded job with nil completion time error",
+			runPolicy: &commonv1.RunPolicy{
+				TTLSecondsAfterFinished: pointer.Int32(5),
+			},
+			jobStatus: commonv1.JobStatus{
+				Conditions: []commonv1.JobCondition{newJobCondition(commonv1.JobSucceeded)},
+			},
+			want:    -1,
+			wantErr: true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := DurationUntilExpireTime(tt.runPolicy, tt.jobStatus)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("DurationUntilExpireTime() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if got != tt.want {
+				if tt.want < 0 || tt.want >= 0 && tt.want > got {
+					t.Errorf("DurationUntilExpireTime() got = %v, want %v", got, tt.want)
+				}
+			}
+		})
+	}
+}
+
+func newJobCondition(t commonv1.JobConditionType) commonv1.JobCondition {
+	return commonv1.JobCondition{
+		Type:   t,
+		Status: corev1.ConditionTrue,
+	}
+}
diff --git a/pkg/controller.v1/tensorflow/job_test.go b/pkg/controller.v1/tensorflow/job_test.go
index 93d5eb953b..bd9930ba02 100644
--- a/pkg/controller.v1/tensorflow/job_test.go
+++ b/pkg/controller.v1/tensorflow/job_test.go
@@ -25,9 +25,11 @@ import (
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/utils/pointer"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	kubeflowv1 "github.com/kubeflow/training-operator/pkg/apis/kubeflow.org/v1"
@@ -522,4 +524,73 @@ var _ = Describe("TFJob controller", func() {
 		})
 	})
 
+	Context("Test TTL Seconds After Finished", func() {
+		It("should delete job when expired time is up", func() {
+			type testCase struct {
+				description string
+				tfJob       *kubeflowv1.TFJob
+				phase       corev1.PodPhase
+			}
+			testCases := []testCase{
+				{
+					description: "succeeded job with TTL 3s",
+					tfJob:       testutil.NewTFJobWithCleanupJobDelay(0, 1, 0, pointer.Int32(3)),
+					phase:       corev1.PodSucceeded,
+				},
+				{
+					description: "failed job with TTL 3s",
+					tfJob:       testutil.NewTFJobWithCleanupJobDelay(0, 1, 0, pointer.Int32(3)),
+					phase:       corev1.PodFailed,
+				},
+			}
+			jobNameTemplate := "test-bof-%d"
+			for idx, tc := range testCases {
+				By(fmt.Sprintf("preparing cases %s", tc.description))
+				ctx := context.Background()
+				name := fmt.Sprintf(jobNameTemplate, idx)
+				tc.tfJob.SetName(name)
+
+				By("creating a TFJob")
+				Expect(reconciler.Create(ctx, tc.tfJob)).Should(Succeed())
+
+				initializeReplicaStatuses(&tc.tfJob.Status, kubeflowv1.TFJobReplicaTypeWorker)
+
+				By("prepare pod")
+				refs := []metav1.OwnerReference{
+					*reconciler.GenOwnerReference(tc.tfJob),
+				}
+				pod := testutil.NewBasePod("pod", tc.tfJob, refs)
+				pod.Status.Phase = tc.phase
+
+				By("update job replica statuses")
+				updateJobReplicaStatuses(&tc.tfJob.Status, kubeflowv1.TFJobReplicaTypeWorker, pod)
+
+				By("update job status")
+				Expect(reconciler.UpdateJobStatus(tc.tfJob, tc.tfJob.Spec.TFReplicaSpecs, &tc.tfJob.Status)).To(Succeed())
+				Expect(reconciler.Status().Update(ctx, tc.tfJob)).Should(Succeed())
+
+				ttl := tc.tfJob.Spec.RunPolicy.TTLSecondsAfterFinished
+				if ttl != nil {
+					dur := time.Second * time.Duration(*ttl)
+					time.Sleep(dur)
+				}
+
+				Eventually(func() error {
+					tfJob := &kubeflowv1.TFJob{}
+					key := types.NamespacedName{
+						Namespace: metav1.NamespaceDefault,
+						Name:      name,
+					}
+					if err := reconciler.Get(ctx, key, tfJob); err != nil {
+						if errors.IsNotFound(err) {
+							return nil
+						}
+						return err
+					}
+					return fmt.Errorf("job %s still remains", name)
+				}, timeout, interval).Should(BeNil())
+			}
+		})
+	})
+
 })