controllers/dsmaster_controller.go - dolphinscheduler-operator - Git at Google

 /*
 Copyright 2022.

 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 controllers

 import (
 	"context"
 	"k8s.io/api/autoscaling/v2beta2"
 	"time"

 	dsv1alpha1 "dolphinscheduler-operator/api/v1alpha1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/selection"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes"
 	"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/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/event"
 )

 var (
 	masterLogger = ctrl.Log.WithName("DSMaster-controller")
 )

 // DSMasterReconciler reconciles a DSMaster object
 type DSMasterReconciler struct {
 	client.Client
 	Scheme   *runtime.Scheme
 	recorder record.EventRecorder
 }

 //+kubebuilder:rbac:groups=ds.apache.dolphinscheduler.dev,resources=dsmasters,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=ds.apache.dolphinscheduler.dev,resources=dsmasters/status,verbs=get;update;patch
 //+kubebuilder:rbac:groups=ds.apache.dolphinscheduler.dev,resources=dsmasters/finalizers,verbs=update
 //+kubebuilder:rbac:groups=autoscaling,resources=horizontalpodautoscalers,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups="",resources=persistentvolumeclaims,verbs=get;create;delete;list;watch
 //+kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch;create;update;patch;delete
 //+kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete

 // Reconcile is part of the main kubernetes reconciliation loop which aims to
 // move the current state of the cluster closer to the desired state.
 // the DSMaster object against the actual cluster state, and then
 // perform operations to make the cluster state reflect the state specified by
 // the user.
 //
 // For more details, check Reconcile and its Result here:
 // - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.11.0/pkg/reconcile
 func (r *DSMasterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	masterLogger.Info("dmMaster start reconcile logic")
 	defer masterLogger.Info("dmMaster Reconcile end ---------------------------------------------")

 	cluster := &dsv1alpha1.DSMaster{}

 	if err := r.Client.Get(ctx, req.NamespacedName, cluster); err != nil {
 		return ctrl.Result{}, client.IgnoreNotFound(err)
 	}
 	desired := cluster.DeepCopy()

 	// Handler finalizer
 	// examine DeletionTimestamp to determine if object is under deletion
 	if cluster.ObjectMeta.DeletionTimestamp.IsZero() {
 		// The object is not being deleted, so if it does not have our finalizer,
 		// then lets add the finalizer and update the object. This is equivalent
 		// registering our finalizer.
 		if !controllerutil.ContainsFinalizer(desired, dsv1alpha1.FinalizerName) {
 			controllerutil.AddFinalizer(desired, dsv1alpha1.FinalizerName)
 			if err := r.Update(ctx, desired); err != nil {
 				return ctrl.Result{}, err
 			}
 		}
 	} else {
 		// The object is being deleted
 		if controllerutil.ContainsFinalizer(desired, dsv1alpha1.FinalizerName) {
 			// our finalizer is present, so lets handle any external dependency
 			if err := r.ensureDSMasterDeleted(ctx, cluster); err != nil {
 				return ctrl.Result{}, err
 			}

 			// remove our finalizer from the list and update it.
 			controllerutil.RemoveFinalizer(desired, dsv1alpha1.FinalizerName)
 			if err := r.Update(ctx, desired); err != nil {
 				return ctrl.Result{}, err
 			}
 		}
 		// Stop reconciliation as the item is being deleted
 		return ctrl.Result{}, nil
 	}

 	// If dsmaster-cluster is paused, we do nothing on things changed.
 	// Until dsmaster-cluster is un-paused, we will reconcile to the  state of that point.
 	if cluster.Spec.Paused {
 		masterLogger.Info("ds-master control has been paused: ", "ds-master-name", cluster.Name)
 		desired.Status.ControlPaused = true
 		if err := r.Status().Patch(ctx, desired, client.MergeFrom(cluster)); err != nil {
 			if apierrors.IsConflict(err) {
 				return ctrl.Result{Requeue: true}, nil
 			} else {
 				masterLogger.Error(err, "unexpected error when master update status in paused")
 				return ctrl.Result{}, err
 			}
 		}
 		r.recorder.Event(cluster, corev1.EventTypeNormal, "the master spec status is paused", "do nothing")
 		return ctrl.Result{}, nil
 	}

 	// 1. First time we see the ds-master-cluster, initialize it
 	if cluster.Status.Phase == dsv1alpha1.DsPhaseNone {
 		desired.Status.Phase = dsv1alpha1.DsPhaseCreating
 		masterLogger.Info("phase had been changed from  none ---> creating")
 		if err := r.Client.Status().Patch(ctx, desired, client.MergeFrom(cluster)); err != nil {

 			if apierrors.IsConflict(err) {
 				return ctrl.Result{RequeueAfter: 100 * time.Millisecond}, err
 			} else {
 				masterLogger.Error(err, "unexpected error when master update status in creating")
 				return ctrl.Result{}, err
 			}
 		}
 	}

 	//2 ensure the headless service
 	masterLogger.Info("Ensuring cluster service")

 	if err := r.ensureMasterService(ctx, cluster); err != nil {
 		return ctrl.Result{}, err
 	}

 	//2 ensure the headless service
 	masterLogger.Info("Ensuring worker hpa")

 	if err := r.ensureHPA(ctx, cluster); err != nil {
 		return ctrl.Result{}, err
 	}

 	// 4. Ensure bootstrapped, we will block here util cluster is up and healthy
 	masterLogger.Info("Ensuring cluster members")
 	if requeue, err := r.ensureMembers(ctx, cluster); requeue {
 		return ctrl.Result{RequeueAfter: 5 * time.Second}, err
 	}

 	// 5. Ensure cluster scaled
 	masterLogger.Info("Ensuring cluster scaled")
 	if requeue, err := r.ensureScaled(ctx, cluster); requeue {
 		return ctrl.Result{Requeue: true, RequeueAfter: 5 * time.Second}, err
 	}

 	// 6. Ensure cluster upgraded
 	masterLogger.Info("Ensuring cluster upgraded")
 	if requeue, err := r.ensureUpgraded(ctx, cluster); requeue {
 		return ctrl.Result{Requeue: true}, err
 	}

 	desired.Status.Phase = dsv1alpha1.DsPhaseFinished
 	if err := r.Status().Patch(ctx, desired, client.MergeFrom(cluster)); err != nil {
 		if apierrors.IsConflict(err) {
 			return ctrl.Result{Requeue: true}, nil
 		} else {
 			masterLogger.Error(err, "unexpected error when master update status in finished")
 			return ctrl.Result{}, err
 		}
 	}

 	masterLogger.Info("******************************************************")
 	desired.Status.Phase = dsv1alpha1.DsPhaseNone
 	if err := r.Update(ctx, desired); err != nil {
 		return ctrl.Result{}, err
 	}
 	return ctrl.Result{Requeue: false}, nil
 }

 // SetupWithManager sets up the controller with the Manager.
 func (r *DSMasterReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	r.recorder = mgr.GetEventRecorderFor("master-controller")

 	filter := &Predicate{}
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&dsv1alpha1.DSMaster{}).
 		Owns(&corev1.Pod{}).
 		Owns(&corev1.Service{}).
 		Owns(&v2beta2.HorizontalPodAutoscaler{}).
 		// or use WithEventFilter()
 		WithEventFilter(filter).
 		Complete(r)
 }

 func (r *DSMasterReconciler) ensureMembers(ctx context.Context, cluster *dsv1alpha1.DSMaster) (bool, error) {

 	pms, err := r.podMemberSet(ctx, cluster)
 	if err != nil {
 		return true, err
 	}
 	if len(pms) > 0 {
 		return !allMembersHealth(pms), nil
 	} else {
 		return false, nil
 	}

 }

 func (r *DSMasterReconciler) ensureScaled(ctx context.Context, cluster *dsv1alpha1.DSMaster) (bool, error) {
 	// Get current members in this cluster
 	ms, err := r.podMemberSet(ctx, cluster)
 	if err != nil {
 		return true, err
 	}

 	// Scale up
 	if len(ms) < cluster.Spec.Replicas {
 		err = r.createMember(ctx, cluster)
 		if err != nil {
 			r.recorder.Event(cluster, corev1.EventTypeWarning, "cannot create the new ds-master pod", "the ds-master pod had been created failed")
 			return true, err
 		}
 		// Cluster modified, next reconcile will enter r.ensureMembers()
 		return true, err
 	}

 	// Scale down
 	if len(ms) > cluster.Spec.Replicas {
 		pod := &corev1.Pod{}
 		member := ms.PickOne()
 		pod.SetName(member.Name)
 		pod.SetNamespace(member.Namespace)
 		err = r.deletePod(ctx, pod)
 		if err != nil {
 			return true, err
 		}
 		return true, err
 	}

 	return false, nil
 }

 func (r *DSMasterReconciler) createMember(ctx context.Context, cluster *dsv1alpha1.DSMaster) error {
 	masterLogger.Info("Starting add new member to cluster", "cluster", cluster.Name)
 	defer masterLogger.Info("End add new member to cluster", "cluster", cluster.Name)

 	// New Pod
 	pod, err := r.newDSMasterPod(cluster)
 	if err != nil {
 		return err
 	}

 	// Create pod
 	if err = r.Client.Create(ctx, pod); err != nil && !apierrors.IsAlreadyExists(err) {
 		return err
 	}
 	return nil
 }

 func (r *DSMasterReconciler) deletePod(ctx context.Context, pod *corev1.Pod) error {
 	masterLogger.Info("begin delete pod", "pod name", pod.Name)
 	if err := r.Client.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) {
 		return err
 	}
 	return nil
 }

 func (r *DSMasterReconciler) ensureUpgraded(ctx context.Context, cluster *dsv1alpha1.DSMaster) (bool, error) {
 	ms, err := r.podMemberSet(ctx, cluster)
 	if err != nil {
 		return false, err
 	}

 	masterLogger.Info("cluster.Spec.Version", "cluster.Spec.Version", cluster.Spec.Version)
 	for _, memset := range ms {
 		if r.predicateUpdate(memset, cluster) {
 			pod := &corev1.Pod{}
 			pod.SetName(memset.Name)
 			pod.SetNamespace(memset.Namespace)
 			if err := r.deletePod(ctx, pod); err != nil {
 				return false, err
 			}
 			return true, nil
 		}
 	}
 	return false, nil
 }

 func getNeedUpgradePods(ctx context.Context, cli *kubernetes.Clientset, cluster *dsv1alpha1.DSMaster) (*corev1.PodList, error) {
 	podSelector, err := labels.NewRequirement(dsv1alpha1.DsVersionLabel, selection.NotIn, []string{cluster.Spec.Version})
 	if err != nil {
 		return nil, err
 	}
 	podAppSelect, err := labels.NewRequirement(dsv1alpha1.DsAppName, selection.Equals, []string{dsv1alpha1.DsMasterLabel})
 	if err != nil {
 		return nil, err
 	}
 	selector := labels.NewSelector()
 	selector = selector.Add(*podSelector).Add(*podAppSelect)
 	podListOptions := metav1.ListOptions{
 		LabelSelector: selector.String(),
 	}
 	return cli.CoreV1().Pods(cluster.Namespace).List(ctx, podListOptions)
 }

 func (r *DSMasterReconciler) ensureMasterService(ctx context.Context, cluster *dsv1alpha1.DSMaster) error {

 	// 1. Client service
 	service := &corev1.Service{}
 	namespacedName := types.NamespacedName{Namespace: cluster.Namespace, Name: dsv1alpha1.DsHeadLessServiceLabel}
 	if err := r.Client.Get(ctx, namespacedName, service); err != nil {
 		// Local cache not found
 		if apierrors.IsNotFound(err) && !apierrors.IsAlreadyExists(err) {
 			service = createMasterService(cluster)
 			if err := controllerutil.SetControllerReference(cluster, service, r.Scheme); err != nil {
 				return err
 			}
 			// Remote may already exist, so we will return err, for the next time, this code will not execute
 			if err := r.Client.Create(ctx, service); err != nil {
 				return err
 			}

 			r.recorder.Event(cluster, corev1.EventTypeNormal, "ds-operator-service created", "master headless service had been created")
 		}
 	}
 	return nil
 }

 type Predicate struct{}

 // Create will be trigger when object created or controller restart
 // first time see the object
 func (r *Predicate) Create(evt event.CreateEvent) bool {
 	switch evt.Object.(type) {
 	case *dsv1alpha1.DSMaster:
 		return true
 	}
 	return false
 }

 func (r *Predicate) Update(evt event.UpdateEvent) bool {
 	switch evt.ObjectNew.(type) {
 	case *dsv1alpha1.DSMaster:
 		oldC := evt.ObjectOld.(*dsv1alpha1.DSMaster)
 		newC := evt.ObjectNew.(*dsv1alpha1.DSMaster)

 		// Only care about size, repo,version and paused fields
 		if oldC.Spec.Replicas != newC.Spec.Replicas {
 			return true
 		}

 		if oldC.Spec.Paused != newC.Spec.Paused {
 			return true
 		}

 		if oldC.Spec.Version != newC.Spec.Version {
 			return true
 		}

 		if oldC.Spec.Repository != newC.Spec.Repository {
 			return true
 		}

 		// If cluster has been marked as deleted, check if we have removed our finalizer
 		// If it has our finalizer, indicating our cleaning up works has not been done.
 		if oldC.DeletionTimestamp.IsZero() && !newC.DeletionTimestamp.IsZero() {
 			if controllerutil.ContainsFinalizer(newC, dsv1alpha1.FinalizerName) {
 				return true
 			}
 		}
 	}
 	return false
 }

 func (r *Predicate) Delete(evt event.DeleteEvent) bool {
 	switch evt.Object.(type) {
 	case *dsv1alpha1.DSMaster:
 		return true
 	case *corev1.Pod:
 		return true
 	case *corev1.Service:
 		return true
 	}
 	return false
 }

 func (r *Predicate) Generic(evt event.GenericEvent) bool {
 	switch evt.Object.(type) {
 	case *dsv1alpha1.DSMaster:
 		return true
 	}
 	return false
 }

 func (r *DSMasterReconciler) predicateUpdate(member *Member, cluster *dsv1alpha1.DSMaster) bool {
 	return member.Version != cluster.Spec.Version
 }

 func (r *DSMasterReconciler) ensureHPA(ctx context.Context, cluster *dsv1alpha1.DSMaster) error {
 	hpa := &v2beta2.HorizontalPodAutoscaler{}
 	namespacedName := types.NamespacedName{Namespace: cluster.Namespace, Name: dsv1alpha1.DsWorkerHpa}
 	if err := r.Client.Get(ctx, namespacedName, hpa); err != nil {
 		// Local cache not found
 		if apierrors.IsNotFound(err) && cluster.Spec.HpaPolicy != nil {
 			hpa := r.createHPA(cluster)
 			if err := controllerutil.SetControllerReference(cluster, hpa, r.Scheme); err != nil {
 				masterLogger.Info("set controller worker hpa error")
 				return err
 			}
 			// Remote may already exist, so we will return err, for the next time, this code will not execute
 			if err := r.Client.Create(ctx, hpa); err != nil {
 				masterLogger.Info("create worker hpa error")
 				return err
 			}
 		}
 	}

 	if hpa.Kind != "" && cluster.Spec.HpaPolicy == nil {
 		if err := r.deleteHPA(ctx, hpa); err != nil {
 			masterLogger.Info("delete hpa error")
 			return err
 		}
 	}
 	return nil
 }
	/*
	Copyright 2022.

	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 controllers

	import (
	"context"
	"k8s.io/api/autoscaling/v2beta2"
	"time"

	dsv1alpha1 "dolphinscheduler-operator/api/v1alpha1"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/selection"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes"
	"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/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/event"
	)

	var (
	masterLogger = ctrl.Log.WithName("DSMaster-controller")
	)

	// DSMasterReconciler reconciles a DSMaster object
	type DSMasterReconciler struct {
	client.Client
	Scheme *runtime.Scheme
	recorder record.EventRecorder
	}

	//+kubebuilder:rbac:groups=ds.apache.dolphinscheduler.dev,resources=dsmasters,verbs=get;list;watch;create;update;patch;delete
	//+kubebuilder:rbac:groups=ds.apache.dolphinscheduler.dev,resources=dsmasters/status,verbs=get;update;patch
	//+kubebuilder:rbac:groups=ds.apache.dolphinscheduler.dev,resources=dsmasters/finalizers,verbs=update
	//+kubebuilder:rbac:groups=autoscaling,resources=horizontalpodautoscalers,verbs=get;list;watch;create;update;patch;delete
	//+kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch;delete
	//+kubebuilder:rbac:groups="",resources=persistentvolumeclaims,verbs=get;create;delete;list;watch
	//+kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch;create;update;patch;delete
	//+kubebuilder:rbac:groups="",resources=services,verbs=get;list;watch;create;update;patch;delete

	// Reconcile is part of the main kubernetes reconciliation loop which aims to
	// move the current state of the cluster closer to the desired state.
	// the DSMaster object against the actual cluster state, and then
	// perform operations to make the cluster state reflect the state specified by
	// the user.
	//
	// For more details, check Reconcile and its Result here:
	// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.11.0/pkg/reconcile
	func (r *DSMasterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	masterLogger.Info("dmMaster start reconcile logic")
	defer masterLogger.Info("dmMaster Reconcile end ---------------------------------------------")

	cluster := &dsv1alpha1.DSMaster{}

	if err := r.Client.Get(ctx, req.NamespacedName, cluster); err != nil {
	return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	desired := cluster.DeepCopy()

	// Handler finalizer
	// examine DeletionTimestamp to determine if object is under deletion
	if cluster.ObjectMeta.DeletionTimestamp.IsZero() {
	// The object is not being deleted, so if it does not have our finalizer,
	// then lets add the finalizer and update the object. This is equivalent
	// registering our finalizer.
	if !controllerutil.ContainsFinalizer(desired, dsv1alpha1.FinalizerName) {
	controllerutil.AddFinalizer(desired, dsv1alpha1.FinalizerName)
	if err := r.Update(ctx, desired); err != nil {
	return ctrl.Result{}, err
	}
	}
	} else {
	// The object is being deleted
	if controllerutil.ContainsFinalizer(desired, dsv1alpha1.FinalizerName) {
	// our finalizer is present, so lets handle any external dependency
	if err := r.ensureDSMasterDeleted(ctx, cluster); err != nil {
	return ctrl.Result{}, err
	}

	// remove our finalizer from the list and update it.
	controllerutil.RemoveFinalizer(desired, dsv1alpha1.FinalizerName)
	if err := r.Update(ctx, desired); err != nil {
	return ctrl.Result{}, err
	}
	}
	// Stop reconciliation as the item is being deleted
	return ctrl.Result{}, nil
	}

	// If dsmaster-cluster is paused, we do nothing on things changed.
	// Until dsmaster-cluster is un-paused, we will reconcile to the state of that point.
	if cluster.Spec.Paused {
	masterLogger.Info("ds-master control has been paused: ", "ds-master-name", cluster.Name)
	desired.Status.ControlPaused = true
	if err := r.Status().Patch(ctx, desired, client.MergeFrom(cluster)); err != nil {
	if apierrors.IsConflict(err) {
	return ctrl.Result{Requeue: true}, nil
	} else {
	masterLogger.Error(err, "unexpected error when master update status in paused")
	return ctrl.Result{}, err
	}
	}
	r.recorder.Event(cluster, corev1.EventTypeNormal, "the master spec status is paused", "do nothing")
	return ctrl.Result{}, nil
	}

	// 1. First time we see the ds-master-cluster, initialize it
	if cluster.Status.Phase == dsv1alpha1.DsPhaseNone {
	desired.Status.Phase = dsv1alpha1.DsPhaseCreating
	masterLogger.Info("phase had been changed from none ---> creating")
	if err := r.Client.Status().Patch(ctx, desired, client.MergeFrom(cluster)); err != nil {

	if apierrors.IsConflict(err) {
	return ctrl.Result{RequeueAfter: 100 * time.Millisecond}, err
	} else {
	masterLogger.Error(err, "unexpected error when master update status in creating")
	return ctrl.Result{}, err
	}
	}
	}

	//2 ensure the headless service
	masterLogger.Info("Ensuring cluster service")

	if err := r.ensureMasterService(ctx, cluster); err != nil {
	return ctrl.Result{}, err
	}

	//2 ensure the headless service
	masterLogger.Info("Ensuring worker hpa")

	if err := r.ensureHPA(ctx, cluster); err != nil {
	return ctrl.Result{}, err
	}

	// 4. Ensure bootstrapped, we will block here util cluster is up and healthy
	masterLogger.Info("Ensuring cluster members")
	if requeue, err := r.ensureMembers(ctx, cluster); requeue {
	return ctrl.Result{RequeueAfter: 5 * time.Second}, err
	}

	// 5. Ensure cluster scaled
	masterLogger.Info("Ensuring cluster scaled")
	if requeue, err := r.ensureScaled(ctx, cluster); requeue {
	return ctrl.Result{Requeue: true, RequeueAfter: 5 * time.Second}, err
	}

	// 6. Ensure cluster upgraded
	masterLogger.Info("Ensuring cluster upgraded")
	if requeue, err := r.ensureUpgraded(ctx, cluster); requeue {
	return ctrl.Result{Requeue: true}, err
	}

	desired.Status.Phase = dsv1alpha1.DsPhaseFinished
	if err := r.Status().Patch(ctx, desired, client.MergeFrom(cluster)); err != nil {
	if apierrors.IsConflict(err) {
	return ctrl.Result{Requeue: true}, nil
	} else {
	masterLogger.Error(err, "unexpected error when master update status in finished")
	return ctrl.Result{}, err
	}
	}

	masterLogger.Info("******************************************************")
	desired.Status.Phase = dsv1alpha1.DsPhaseNone
	if err := r.Update(ctx, desired); err != nil {
	return ctrl.Result{}, err
	}
	return ctrl.Result{Requeue: false}, nil
	}

	// SetupWithManager sets up the controller with the Manager.
	func (r *DSMasterReconciler) SetupWithManager(mgr ctrl.Manager) error {
	r.recorder = mgr.GetEventRecorderFor("master-controller")

	filter := &Predicate{}
	return ctrl.NewControllerManagedBy(mgr).
	For(&dsv1alpha1.DSMaster{}).
	Owns(&corev1.Pod{}).
	Owns(&corev1.Service{}).
	Owns(&v2beta2.HorizontalPodAutoscaler{}).
	// or use WithEventFilter()
	WithEventFilter(filter).
	Complete(r)
	}

	func (r DSMasterReconciler) ensureMembers(ctx context.Context, cluster dsv1alpha1.DSMaster) (bool, error) {

	pms, err := r.podMemberSet(ctx, cluster)
	if err != nil {
	return true, err
	}
	if len(pms) > 0 {
	return !allMembersHealth(pms), nil
	} else {
	return false, nil
	}

	}

	func (r DSMasterReconciler) ensureScaled(ctx context.Context, cluster dsv1alpha1.DSMaster) (bool, error) {
	// Get current members in this cluster
	ms, err := r.podMemberSet(ctx, cluster)
	if err != nil {
	return true, err
	}

	// Scale up
	if len(ms) < cluster.Spec.Replicas {
	err = r.createMember(ctx, cluster)
	if err != nil {
	r.recorder.Event(cluster, corev1.EventTypeWarning, "cannot create the new ds-master pod", "the ds-master pod had been created failed")
	return true, err
	}
	// Cluster modified, next reconcile will enter r.ensureMembers()
	return true, err
	}

	// Scale down
	if len(ms) > cluster.Spec.Replicas {
	pod := &corev1.Pod{}
	member := ms.PickOne()
	pod.SetName(member.Name)
	pod.SetNamespace(member.Namespace)
	err = r.deletePod(ctx, pod)
	if err != nil {
	return true, err
	}
	return true, err
	}

	return false, nil
	}

	func (r DSMasterReconciler) createMember(ctx context.Context, cluster dsv1alpha1.DSMaster) error {
	masterLogger.Info("Starting add new member to cluster", "cluster", cluster.Name)
	defer masterLogger.Info("End add new member to cluster", "cluster", cluster.Name)

	// New Pod
	pod, err := r.newDSMasterPod(cluster)
	if err != nil {
	return err
	}

	// Create pod
	if err = r.Client.Create(ctx, pod); err != nil && !apierrors.IsAlreadyExists(err) {
	return err
	}
	return nil
	}

	func (r DSMasterReconciler) deletePod(ctx context.Context, pod corev1.Pod) error {
	masterLogger.Info("begin delete pod", "pod name", pod.Name)
	if err := r.Client.Delete(ctx, pod); err != nil && !apierrors.IsNotFound(err) {
	return err
	}
	return nil
	}

	func (r DSMasterReconciler) ensureUpgraded(ctx context.Context, cluster dsv1alpha1.DSMaster) (bool, error) {
	ms, err := r.podMemberSet(ctx, cluster)
	if err != nil {
	return false, err
	}

	masterLogger.Info("cluster.Spec.Version", "cluster.Spec.Version", cluster.Spec.Version)
	for _, memset := range ms {
	if r.predicateUpdate(memset, cluster) {
	pod := &corev1.Pod{}
	pod.SetName(memset.Name)
	pod.SetNamespace(memset.Namespace)
	if err := r.deletePod(ctx, pod); err != nil {
	return false, err
	}
	return true, nil
	}
	}
	return false, nil
	}

	func getNeedUpgradePods(ctx context.Context, cli kubernetes.Clientset, cluster dsv1alpha1.DSMaster) (*corev1.PodList, error) {
	podSelector, err := labels.NewRequirement(dsv1alpha1.DsVersionLabel, selection.NotIn, []string{cluster.Spec.Version})
	if err != nil {
	return nil, err
	}
	podAppSelect, err := labels.NewRequirement(dsv1alpha1.DsAppName, selection.Equals, []string{dsv1alpha1.DsMasterLabel})
	if err != nil {
	return nil, err
	}
	selector := labels.NewSelector()
	selector = selector.Add(podSelector).Add(podAppSelect)
	podListOptions := metav1.ListOptions{
	LabelSelector: selector.String(),
	}
	return cli.CoreV1().Pods(cluster.Namespace).List(ctx, podListOptions)
	}

	func (r DSMasterReconciler) ensureMasterService(ctx context.Context, cluster dsv1alpha1.DSMaster) error {

	// 1. Client service
	service := &corev1.Service{}
	namespacedName := types.NamespacedName{Namespace: cluster.Namespace, Name: dsv1alpha1.DsHeadLessServiceLabel}
	if err := r.Client.Get(ctx, namespacedName, service); err != nil {
	// Local cache not found
	if apierrors.IsNotFound(err) && !apierrors.IsAlreadyExists(err) {
	service = createMasterService(cluster)
	if err := controllerutil.SetControllerReference(cluster, service, r.Scheme); err != nil {
	return err
	}
	// Remote may already exist, so we will return err, for the next time, this code will not execute
	if err := r.Client.Create(ctx, service); err != nil {
	return err
	}

	r.recorder.Event(cluster, corev1.EventTypeNormal, "ds-operator-service created", "master headless service had been created")
	}
	}
	return nil
	}

	type Predicate struct{}

	// Create will be trigger when object created or controller restart
	// first time see the object
	func (r *Predicate) Create(evt event.CreateEvent) bool {
	switch evt.Object.(type) {
	case *dsv1alpha1.DSMaster:
	return true
	}
	return false
	}

	func (r *Predicate) Update(evt event.UpdateEvent) bool {
	switch evt.ObjectNew.(type) {
	case *dsv1alpha1.DSMaster:
	oldC := evt.ObjectOld.(*dsv1alpha1.DSMaster)
	newC := evt.ObjectNew.(*dsv1alpha1.DSMaster)

	// Only care about size, repo,version and paused fields
	if oldC.Spec.Replicas != newC.Spec.Replicas {
	return true
	}

	if oldC.Spec.Paused != newC.Spec.Paused {
	return true
	}

	if oldC.Spec.Version != newC.Spec.Version {
	return true
	}

	if oldC.Spec.Repository != newC.Spec.Repository {
	return true
	}

	// If cluster has been marked as deleted, check if we have removed our finalizer
	// If it has our finalizer, indicating our cleaning up works has not been done.
	if oldC.DeletionTimestamp.IsZero() && !newC.DeletionTimestamp.IsZero() {
	if controllerutil.ContainsFinalizer(newC, dsv1alpha1.FinalizerName) {
	return true
	}
	}
	}
	return false
	}

	func (r *Predicate) Delete(evt event.DeleteEvent) bool {
	switch evt.Object.(type) {
	case *dsv1alpha1.DSMaster:
	return true
	case *corev1.Pod:
	return true
	case *corev1.Service:
	return true
	}
	return false
	}

	func (r *Predicate) Generic(evt event.GenericEvent) bool {
	switch evt.Object.(type) {
	case *dsv1alpha1.DSMaster:
	return true
	}
	return false
	}

	func (r DSMasterReconciler) predicateUpdate(member Member, cluster *dsv1alpha1.DSMaster) bool {
	return member.Version != cluster.Spec.Version
	}

	func (r DSMasterReconciler) ensureHPA(ctx context.Context, cluster dsv1alpha1.DSMaster) error {
	hpa := &v2beta2.HorizontalPodAutoscaler{}
	namespacedName := types.NamespacedName{Namespace: cluster.Namespace, Name: dsv1alpha1.DsWorkerHpa}
	if err := r.Client.Get(ctx, namespacedName, hpa); err != nil {
	// Local cache not found
	if apierrors.IsNotFound(err) && cluster.Spec.HpaPolicy != nil {
	hpa := r.createHPA(cluster)
	if err := controllerutil.SetControllerReference(cluster, hpa, r.Scheme); err != nil {
	masterLogger.Info("set controller worker hpa error")
	return err
	}
	// Remote may already exist, so we will return err, for the next time, this code will not execute
	if err := r.Client.Create(ctx, hpa); err != nil {
	masterLogger.Info("create worker hpa error")
	return err
	}
	}
	}

	if hpa.Kind != "" && cluster.Spec.HpaPolicy == nil {
	if err := r.deleteHPA(ctx, hpa); err != nil {
	masterLogger.Info("delete hpa error")
	return err
	}
	}
	return nil
	}