func (r *DIJobReconciler) addDIJob(obj client.Object) {
	log := r.Log.WithValues("dijob", diutil.NamespacedName(obj.GetNamespace(), obj.GetName()))
	job, ok := obj.(*div1alpha1.DIJob)
	if !ok {
		log.Error(fmt.Errorf("failed to convert object DIJob: %s/%s", obj.GetNamespace(), obj.GetName()), "")
		r.markIncorrectJobFailed(obj)
		return
	}
	oldStatus := job.Status.DeepCopy()
	msg := fmt.Sprintf("DIJob %s created", job.Name)
	if err := r.updateJobPhase(context.Background(), job, div1alpha1.JobCreated, DIJobCreatedReason, msg); err != nil {
		log.Error(err, "failed to update job status")
		return
	}
	log.Info(fmt.Sprintf("DIJob %s/%s created", job.Namespace, job.Name))
	if !apiequality.Semantic.DeepEqual(*oldStatus, job.Status) {
		if err := r.updateDIJobStatusInCluster(context.Background(), job); err != nil {
			log.Error(err, fmt.Sprintf("failed to update DIJob %s/%s status", job.Namespace, job.Name))
		}
	}
}
addDIJob is the event handler responsible for handling job add events
package controllers
import (
	"context"
	"fmt"
	corev1 "k8s.io/api/core/v1"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/dynamic"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	div1alpha1 "opendilab.org/di-orchestrator/api/v1alpha1"
	dicommon "opendilab.org/di-orchestrator/common"
	diutil "opendilab.org/di-orchestrator/utils"
)
func isSucceeded(job *div1alpha1.DIJob) bool {
	return job.Status.Phase == div1alpha1.JobSucceeded
}
func isFailed(job *div1alpha1.DIJob) bool {
	return job.Status.Phase == div1alpha1.JobFailed
}
func (r *DIJobReconciler) reconcileReplicas(ctx context.Context, job *div1alpha1.DIJob, pods []*corev1.Pod, services []*corev1.Service) error {
	log := r.Log.WithValues("dijob", diutil.NamespacedName(job.Namespace, job.Name))
	if err := r.reconcilePodsAndServices(ctx, job, pods, services); err != nil {
		log.Error(err, "failed to reconcile pods and services")
		return err
	}
	collectors, learners, coordinator, ags, _, err := diutil.ClassifyPods(pods)
	if err != nil {
		log.Error(err, "unable to classify pods")
		return err
	}
	if coordinator != nil {
		if err := r.updateDIJobStatus(ctx, job, collectors, learners, coordinator, ags); err != nil {
			return err
		}
	} else {
		volumes := job.Spec.Volumes
		template := job.Spec.Coordinator.Template.DeepCopy()
		coorpod, coorsvc, coorurl, err := buildPodAndServiceForReplica(template, job, dicommon.CoordinatorName, volumes)
		if err != nil {
			msg := fmt.Sprintf("build coordinator pod for job %s failed", job.Name)
			log.Error(err, msg)
			return err
		}
		envs := make(map[string]string)
		envs[dicommon.CoordinatorURLEnv] = coorurl
		diutil.AddEnvsToPod(coorpod, envs)
		if err := r.createPodAndService(ctx, job, coorpod, coorsvc); err != nil {
			return err
		}
	}
	return nil
}