func (r *XDLJobReconciler) updateGeneralJobStatus(xdlJob *xdlv1alpha1.XDLJob, replicaSpecs map[v1.ReplicaType]*v1.ReplicaSpec,
	jobStatus *v1.JobStatus, restart bool) error {
	log.Info("Updating status", "XDLJob name", xdlJob.Name)
	previousRestarting := commonutil.IsRestarting(*jobStatus)
	previousFailed := commonutil.IsFailed(*jobStatus)
	workerNum, workerSucceeded := int32(0), int32(0)
	for rtype, spec := range replicaSpecs {
		replicas := *spec.Replicas
		status, ok := jobStatus.ReplicaStatuses[rtype]
		if !ok {
			log.Info("skipping invalid replica type", "rtype", rtype)
			continue
		}
		failed := status.Failed
		if rtype == xdlv1alpha1.XDLReplicaTypeWorker || rtype == xdlv1alpha1.XDLReplicaTypeExtendRole {
			workerNum += replicas
			workerSucceeded += status.Succeeded
		}
		if status.Active == replicas && jobStatus.StartTime == nil {
			now := metav1.Now()
			jobStatus.StartTime = &now
		}
		if failed > 0 {
			if restart {
				msg := fmt.Sprintf("XDLJob %s is restarting because %d %s replica(s) failed.", xdlJob.Name, failed, rtype)
				r.recorder.Event(xdlJob, corev1.EventTypeWarning, xdlJobRestartingReason, msg)
				err := commonutil.UpdateJobConditions(jobStatus, v1.JobRestarting, commonutil.JobRestartingReason, msg)
				if err != nil {
					log.Error(err, "failed to update xdl job conditions")
					return err
				}
				if !previousRestarting {
					r.ctrl.Metrics.FailureInc()
					r.ctrl.Metrics.RestartInc()
				}
			} else {
				msg := fmt.Sprintf("XDLJob %s is failed because %d %s replica(s) failed.", xdlJob.Name, failed, rtype)
				r.recorder.Event(xdlJob, corev1.EventTypeNormal, xdlJobFailedReason, msg)
				if jobStatus.CompletionTime == nil {
					now := metav1.Now()
					jobStatus.CompletionTime = &now
				}
				err := commonutil.UpdateJobConditions(jobStatus, v1.JobFailed, commonutil.JobFailedReason, msg)
				if err != nil {
					log.Error(err, "failed to update xdl job conditions")
					return err
				}
				if !previousFailed {
					r.ctrl.Metrics.FailureInc()
				}
			}
			return nil
		}
	}
	workerMinFinish := calculateMinFinish(xdlJob, workerNum)
	if workerSucceeded >= workerMinFinish {
		msg := fmt.Sprintf("XDLJob %s is successfully completed.", xdlJob.Name)
		if jobStatus.CompletionTime == nil {
			now := metav1.Now()
			jobStatus.CompletionTime = &now
		}
		if err := commonutil.UpdateJobConditions(jobStatus, v1.JobSucceeded, commonutil.JobSucceededReason, msg); err != nil {
			log.Error(err, "failed to update xdl job conditions")
			return err
		}
		r.ctrl.Metrics.SuccessInc()
		return nil
	}
	msg := fmt.Sprintf("XDLJob %s is running.", xdlJob.Name)
	if err := commonutil.UpdateJobConditions(jobStatus, v1.JobRunning, commonutil.JobRunningReason, msg); err != nil {
		log.Error(err, "failed to update xdl job conditions")
		return err
	}
	return nil
}
updateGeneralJobStatus updates the status of job with given replica specs and job status.
package xdljob
import (
	"fmt"
	"math"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/event"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	xdlv1alpha1 "github.com/alibaba/kubedl/apis/training/v1alpha1"
	v1 "github.com/alibaba/kubedl/pkg/job_controller/api/v1"
	commonutil "github.com/alibaba/kubedl/pkg/util"
)
const (
	xdlJobFailedReason     = "XdlJobFailed"
	xdlJobRestartingReason = "XdlJobRestarting"
)
func onOwnerCreateFunc(r reconcile.Reconciler) func(event.CreateEvent) bool {
	return func(e event.CreateEvent) bool {
		xdlJob, ok := e.Meta.(*xdlv1alpha1.XDLJob)
		if !ok {
			return true
		}
		reconciler, ok := r.(*XDLJobReconciler)
		if !ok {
			return true
		}
		reconciler.scheme.Default(xdlJob)
		msg := fmt.Sprintf("XdlJob %s is created.", e.Meta.GetName())
		if err := commonutil.UpdateJobConditions(&xdlJob.Status, v1.JobCreated, commonutil.JobCreatedReason, msg); err != nil {
			log.Error(err, "append job condition error")
			return false
		}
		reconciler.ctrl.Metrics.CreatedInc()
		return true
	}
}