func (r *TFJobReconciler) SetupWithManager(mgr ctrl.Manager) error {
	c, err := controller.New(r.ControllerName(), mgr, controller.Options{
		Reconciler:              r,
		MaxConcurrentReconciles: options.CtrlConfig.MaxConcurrentReconciles,
	})
	if err != nil {
		return err
	}
	if err = c.Watch(&source.Kind{Type: &training.TFJob{}}, &handler.EnqueueRequestForObject{}, predicate.Funcs{
		CreateFunc: onOwnerCreateFunc(r),
	}); err != nil {
		return err
	}
	if err = c.Watch(&source.Kind{Type: &corev1.Pod{}}, &handler.EnqueueRequestForOwner{
		OwnerType:    &training.TFJob{},
		IsController: true,
	}, predicate.Funcs{
		CreateFunc: r.ctrl.OnPodCreateFunc,
		UpdateFunc: r.ctrl.OnPodUpdateFunc,
		DeleteFunc: r.ctrl.OnPodDeleteFunc,
	}); err != nil {
		return err
	}
	return c.Watch(&source.Kind{Type: &corev1.Service{}}, &handler.EnqueueRequestForOwner{
		OwnerType:    &training.TFJob{},
		IsController: true,
	}, predicate.Funcs{
		CreateFunc: r.ctrl.OnServiceCreateFunc,
		UpdateFunc: r.ctrl.OnServiceUpdateFunc,
		DeleteFunc: r.ctrl.OnServiceDeleteFunc,
	})
}
SetupWithManager setup reconciler to the Manager with default RBAC. The Manager will set fields on the
Controller and Start it when the Manager is Started.
package tensorflow
import (
	"context"
	"fmt"
	"strconv"
	"strings"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/client-go/tools/record"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
	"sigs.k8s.io/controller-runtime/pkg/source"
	training "github.com/alibaba/kubedl/apis/training/v1alpha1"
	"github.com/alibaba/kubedl/cmd/options"
	"github.com/alibaba/kubedl/pkg/gang_schedule/registry"
	"github.com/alibaba/kubedl/pkg/job_controller"
	v1 "github.com/alibaba/kubedl/pkg/job_controller/api/v1"
	"github.com/alibaba/kubedl/pkg/metrics"
	"github.com/alibaba/kubedl/pkg/tensorboard"
	"github.com/alibaba/kubedl/pkg/util"
)
const (
	controllerName = "TensorFlowController"
)
const (
	tfConfig = "TF_CONFIG"
)
var log = logf.Log.WithName("tf-controller")
func NewReconciler(mgr ctrl.Manager, config options.JobControllerConfiguration) *TFJobReconciler {
	r := &TFJobReconciler{
		Client: mgr.GetClient(),
		scheme: mgr.GetScheme(),
	}
	r.recorder = mgr.GetEventRecorderFor(r.ControllerName())
	r.ctrl = job_controller.NewJobController(r.Client, r, config, r.recorder, metrics.NewJobMetrics(training.TFJobKind, r.Client))
	if r.ctrl.Config.EnableGangScheduling {
		r.ctrl.GangScheduler = registry.Get(r.ctrl.Config.GangSchedulerName)
	}
	return r
}
var _ reconcile.Reconciler = &TFJobReconciler{}
var _ v1.ControllerInterface = &TFJobReconciler{}
type TFJobReconciler struct {
	client.Client
	scheme   *runtime.Scheme
	recorder record.EventRecorder
	ctrl     job_controller.JobController
}
func (r *TFJobReconciler) GetNodeForModelOutput(pods []*corev1.Pod) string {
	var chiefPod *corev1.Pod
	var masterPod *corev1.Pod
	var worker0Pod *corev1.Pod
	for _, pod := range pods {
		rtype := pod.Labels[v1.ReplicaTypeLabel]
		rIndex, _ := strconv.Atoi(pod.Labels[v1.ReplicaIndexLabel])
		if rtype == strings.ToLower(string(training.TFReplicaTypeMaster)) && rIndex == 0 {
			masterPod = pod
		}
		if rtype == strings.ToLower(string(training.TFReplicaTypeChief)) && rIndex == 0 {
			chiefPod = pod
		}
		if rtype == strings.ToLower(string(training.TFReplicaTypeWorker)) && rIndex == 0 {
			worker0Pod = pod
		}
	}
	nodeName := ""
	if chiefPod != nil {
		nodeName = chiefPod.Spec.NodeName
	}
	if masterPod != nil {
		nodeName = masterPod.Spec.NodeName
	}
	if worker0Pod != nil {
		nodeName = worker0Pod.Spec.NodeName
	}
	if nodeName != "" {
		log.Info(fmt.Sprintf("select %s for model output", nodeName))
		return nodeName
	} else {
		log.Info("no node selected for model output")
		return nodeName
	}
}
func (r *TFJobReconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
	sharedTfJob := &training.TFJob{}
	err := util.GetObjectByPassCache(r.Client, req.NamespacedName, sharedTfJob)
	if err != nil {
		if errors.IsNotFound(err) {
			log.Info("try to get job but it has been deleted", "key", req.String())
			r.ctrl.Metrics.DeletedInc()
			return reconcile.Result{}, nil
		}
		return reconcile.Result{}, err
	}
	tfJob := sharedTfJob.DeepCopy()
	needSync := r.ctrl.SatisfyExpectations(tfJob, tfJob.Spec.TFReplicaSpecs)
	if !needSync || tfJob.DeletionTimestamp != nil {
		log.Info("reconcile cancelled, job does not need to do reconcile or has been deleted",
			"sync", needSync, "deleted", tfJob.DeletionTimestamp != nil)
		return reconcile.Result{}, nil
	}
	r.scheme.Default(tfJob)
	result, err := r.ctrl.ReconcileJobs(tfJob, tfJob.Spec.TFReplicaSpecs, tfJob.Status, &tfJob.Spec.RunPolicy, tfJob.Spec.ModelVersion)
	if err != nil {
		log.Error(err, "tensorflow job reconcile failed")
		return result, err
	}
	masterType, masterSpec := r.getMasterSpec(tfJob.Spec.TFReplicaSpecs)
	if err := tensorboard.ReconcileTensorBoard(r.ctrl, r.Client, tfJob,
		masterType, masterSpec, tfJob.Status, &result); err != nil {
		log.Error(err, "ReconcileTensorBoard error %v")
		return result, err
	}
	return result, err
}