feat: adds job tracker functions and job status checks

Author: nxtcoder17

toolkit/job-helper/job-helper.go

@@ -23,6 +23,15 @@ const (
 	JobStatusSuccess JobStatus = "job-status-success"
 )
 
+type JobPhase string
+
+const (
+	JobPhasePending   JobPhase = "PENDING"
+	JobPhaseRunning   JobPhase = "RUNNING"
+	JobPhaseFailed    JobPhase = "FAILED"
+	JobPhaseSucceeded JobPhase = "SUCCEEDED"
+)
+
 func GetLatestPod(ctx context.Context, cli client.Client, jobNamespace string, jobName string) (*corev1.Pod, error) {
 	var podList corev1.PodList
 

toolkit/job-helper/job-runner.go

@@ -0,0 +1,93 @@
+package job_helper
+
+import (
+	"context"
+	"errors"
+
+	batchv1 "k8s.io/api/batch/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type JobTracker struct {
+	ctx    context.Context
+	client client.Client
+
+	job *batchv1.Job
+
+	isTargetJob func(job *batchv1.Job) bool
+}
+
+type JobTrackerArgs struct {
+	client client.Client
+
+	JobNamespace string
+	JobName      string
+
+	IsTargetJob func(job *batchv1.Job) bool
+}
+
+func NewJobTracker(ctx context.Context, kcli client.Client, args JobTrackerArgs) (*JobTracker, error) {
+	var job batchv1.Job
+	if err := kcli.Get(ctx, types.NamespacedName{Name: args.JobName, Namespace: args.JobNamespace}, &job); err != nil {
+		return nil, err
+	}
+
+	return &JobTracker{
+		ctx:         ctx,
+		client:      kcli,
+		job:         &job,
+		isTargetJob: args.IsTargetJob,
+	}, nil
+}
+
+func (jr *JobTracker) HasJobFinished() bool {
+	for _, v := range jr.job.Status.Conditions {
+		if v.Type == batchv1.JobComplete && v.Status == "True" {
+			return true
+		}
+
+		if v.Type == batchv1.JobFailed && v.Status == "True" {
+			return true
+		}
+
+		if v.Type == batchv1.JobSuspended && v.Status == "True" {
+			return true
+		}
+	}
+
+	return false
+}
+
+func (jr *JobTracker) Process(ctx context.Context) (phase JobPhase, message string, err error) {
+	if !jr.isTargetJob(jr.job) {
+		if !jr.HasJobFinished() {
+			return JobPhasePending, "waiting for previous jobs to finish execution", nil
+		}
+
+		if err := DeleteJob(jr.ctx, jr.client, jr.job.Namespace, jr.job.Name); err != nil {
+			return JobPhasePending, "", errors.Join(errors.New("failed to delete job"), err)
+		}
+
+		return JobPhasePending, "waiting for job to start", nil
+	}
+
+	if !jr.HasJobFinished() {
+		return JobPhaseRunning, "waiting for running job to finish", nil
+	}
+
+	// check.Message = job_manager.GetTerminationLog(ctx, r.Client, job.Namespace, job.Name)
+	if jr.job.Status.Succeeded > 0 {
+		return JobPhaseSucceeded, "", nil
+	}
+
+	if jr.job.Status.Active > 0 {
+		return JobPhaseRunning, "waiting for job to complete", nil
+	}
+
+	if jr.job.Status.Failed > 0 {
+		return JobPhaseFailed, "", errors.New("install or upgrade job failed")
+	}
+
+	return JobPhasePending, "", nil
+}

toolkit/reconciler/checks.go

@@ -50,13 +50,13 @@ func AreChecksEqual(c1 Check, c2 Check) bool {
 		c1.StartedAt.Sub(c2.StartedAt.Time) == 0
 }
 
-type checkWrapper[T Resource] struct {
+type CheckWrapper[T Resource] struct {
 	checkName string
 	request   *Request[T]
 	Check     `json:",inline"`
 }
 
-func (cw *checkWrapper[T]) Failed(err error) step_result.Result {
+func (cw *CheckWrapper[T]) Failed(err error) step_result.Result {
 	defer cw.request.LogPostCheck(cw.checkName)
 
 	_, file, line, _ := runtime.Caller(1)
@@ -82,7 +82,7 @@ func (cw *checkWrapper[T]) Failed(err error) step_result.Result {
 	// return cw.request.updateStatus().Continue(false).Err(err)
 }
 
-func (cw *checkWrapper[T]) StillRunning(err error) step_result.Result {
+func (cw *CheckWrapper[T]) StillRunning(err error) step_result.Result {
 	defer cw.request.LogPostCheck(cw.checkName)
 
 	cw.Check.State = RunningState
@@ -104,7 +104,7 @@ func (cw *checkWrapper[T]) StillRunning(err error) step_result.Result {
 	// return cw.request.updateStatus().Continue(false).Err(err)
 }
 
-func (cw *checkWrapper[T]) Completed() step_result.Result {
+func (cw *CheckWrapper[T]) Completed() step_result.Result {
 	defer cw.request.LogPostCheck(cw.checkName)
 
 	cw.Check.State = CompletedState
@@ -119,8 +119,8 @@ func (cw *checkWrapper[T]) Completed() step_result.Result {
 	return step_result.New().Continue(true)
 }
 
-func NewRunningCheck[T Resource](name string, req *Request[T]) *checkWrapper[T] {
-	cw := &checkWrapper[T]{
+func NewRunningCheck[T Resource](name string, req *Request[T]) *CheckWrapper[T] {
+	cw := &CheckWrapper[T]{
 		checkName: name,
 		request:   req,
 		Check: Check{

toolkit/reconciler/request.go

@@ -448,7 +448,7 @@ func (r *Request[T]) AddToOwnedResources(refs ...ResourceRef) {
 	r.resourceRefs = append(r.resourceRefs, refs...)
 }
 
-func (r *Request[T]) CleanupOwnedResources(check *checkWrapper[T]) stepResult.Result {
+func (r *Request[T]) CleanupOwnedResources(check *CheckWrapper[T]) stepResult.Result {
 	resources := r.Object.GetStatus().Resources
 	objects := make([]client.Object, 0, len(resources))
 	for i := range resources {
@@ -474,7 +474,7 @@ INFO: this should only be used for very specific cases, where there is no other
 Like, when deleting ManagedService
   - all managed resources should be deleted, but since owner is already getting deleted, there is no point in their proper cleanup
 */
-func (r *Request[T]) ForceCleanupOwnedResources(check *checkWrapper[T]) stepResult.Result {
+func (r *Request[T]) ForceCleanupOwnedResources(check *CheckWrapper[T]) stepResult.Result {
 	ctx := r.Context()
 	resources := r.Object.GetStatus().Resources