controllers/dsmaster_controller.go

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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"
	v1 "k8s.io/api/rbac/v1"
	"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
//+kubebuilder:rbac:groups="",resources=serviceaccounts,verbs=get;list;create;delete
//+kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=role,verbs=get;list;create;delete
//+kubebuilder:rbac:groups="rbac.authorization.k8s.io",resources=rolebinding,verbs=get;list;create;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()

	sa := &corev1.ServiceAccount{}
	saReq := ctrl.Request{
		NamespacedName: types.NamespacedName{
			Namespace: req.Namespace,
			Name:      dsv1alpha1.DsServiceAccount,
		},
	}
	err := r.Get(ctx, saReq.NamespacedName, sa)
	if apierrors.IsNotFound(err) {
		err := r.createServiceAccountIfNotExists(ctx, cluster)
		if err != nil {
			return ctrl.Result{}, err
		}
	}

	// 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{}).
		Owns(&corev1.ServiceAccount{}).
		Owns(&v1.Role{}).
		Owns(&v1.RoleBinding{}).
		// 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
}

// 创建 ServiceAccount
func (r *DSMasterReconciler) createServiceAccountIfNotExists(ctx context.Context, cluster *dsv1alpha1.DSMaster) (err error) {

	masterLogger.Info("start create service account.")

	sa := &corev1.ServiceAccount{
		ObjectMeta: metav1.ObjectMeta{
			Name:      dsv1alpha1.DsServiceAccount,
			Namespace: cluster.Namespace,
		},
	}

	err = r.Create(ctx, sa)

	if err != nil {
		masterLogger.Error(err, "create service account error")
		return err
	}
	// binding the sa
	err = controllerutil.SetControllerReference(cluster, sa, r.Scheme)
	if err != nil {
		masterLogger.Error(err, "sa SetControllerReference error")
		return err
	}

	ro := &v1.Role{}
	namespacedName := types.NamespacedName{Namespace: cluster.Namespace, Name: dsv1alpha1.DsRole}
	if err := r.Client.Get(ctx, namespacedName, ro); err != nil {
		if apierrors.IsNotFound(err) && !apierrors.IsAlreadyExists(err) {
			// Remote may already exist, so we will return err, for the next time, this code will not execute
			ro := r.createRole(cluster)
			if err := controllerutil.SetControllerReference(cluster, ro, r.Scheme); err != nil {
				masterLogger.Info("set controller role  error")
				return err
			}
			masterLogger.Info("set  role  begin")
			if err := r.Client.Create(ctx, ro); err != nil {
				return err
			}
		}
	}

	rb := &v1.RoleBinding{}
	rbNamespacedName := types.NamespacedName{Namespace: cluster.Namespace, Name: dsv1alpha1.DsRoleBinding}
	if err := r.Client.Get(ctx, rbNamespacedName, rb); err != nil {
		if apierrors.IsNotFound(err) && !apierrors.IsAlreadyExists(err) {
			rb := r.createRoleBinding(cluster)
			if err := controllerutil.SetControllerReference(cluster, rb, r.Scheme); err != nil {
				masterLogger.Info("set controller  rolebinding error")
				return err
			}

			masterLogger.Info("set  rolebinding  begin")
			if err := r.Client.Create(ctx, rb); err != nil {
				return err
			}
		}
	}
	return nil
}