K8SPSMDB-876: delete-psmdb-pods-in-order for cfg (#1267)

* K8SPSMDB-876: `delete-psmdb-pods-in-order` for cfg

https://jira.percona.com/browse/K8SPSMDB-876

* shut down `cfg` replset at the end

* delete replsets fully in order

* stop balancer and delete mongos before rs deletion

* fix pause

* don't disable backup before pausing cluster

---------

Co-authored-by: Inel Pandzic <[email protected]>
Co-authored-by: Viacheslav Sarzhan <[email protected]>

Author: 3 people

pkg/apis/psmdb/v1/psmdb_defaults.go

@@ -134,9 +134,7 @@ func (cr *PerconaServerMongoDB) CheckNSetDefaults(platform version.Platform, log
 			return errors.New("mongos should be specified")
 		}
 
-		if cr.Spec.Pause {
-			cr.Spec.Sharding.Mongos.Size = 0
-		} else {
+		if !cr.Spec.Pause && cr.DeletionTimestamp == nil {
 			if !cr.Spec.UnsafeConf && cr.Spec.Sharding.Mongos.Size < minSafeMongosSize {
 				log.Info("Safe config set, updating mongos size",
 					"oldSize", cr.Spec.Sharding.Mongos.Size, "newSize", minSafeMongosSize)
@@ -443,20 +441,6 @@ func (cr *PerconaServerMongoDB) CheckNSetDefaults(platform version.Platform, log
 		if err := replset.NonVoting.SetDefaults(cr, replset); err != nil {
 			return errors.Wrap(err, "set nonvoting defaults")
 		}
-
-		if cr.Spec.Pause {
-			if cr.Status.State == AppStateStopping {
-				log.Info("Pausing cluster", "replset", replset.Name, "oldSize", replset.Size, "newSize", 0)
-			}
-			replset.Size = 0
-			replset.Arbiter.Enabled = false
-			replset.NonVoting.Enabled = false
-		}
-	}
-
-	// there is shouldn't be any backups while pause
-	if cr.Spec.Pause {
-		cr.Spec.Backup.Enabled = false
 	}
 
 	if cr.Spec.Backup.Enabled {
@@ -564,7 +548,7 @@ func (rs *ReplsetSpec) SetDefaults(platform version.Platform, cr *PerconaServerM
 		rs.Arbiter.MultiAZ.reconcileOpts(cr)
 	}
 
-	if !cr.Spec.UnsafeConf && cr.DeletionTimestamp == nil {
+	if !cr.Spec.UnsafeConf && (cr.DeletionTimestamp == nil && !cr.Spec.Pause) {
 		rs.setSafeDefaults(log)
 	}
 

pkg/controller/perconaservermongodb/balancer.go

@@ -17,7 +17,7 @@ import (
 func (r *ReconcilePerconaServerMongoDB) enableBalancerIfNeeded(ctx context.Context, cr *api.PerconaServerMongoDB) error {
 	log := logf.FromContext(ctx)
 
-	if !cr.Spec.Sharding.Enabled || cr.Spec.Sharding.Mongos.Size == 0 || cr.Spec.Unmanaged {
+	if !cr.Spec.Sharding.Enabled || cr.Spec.Sharding.Mongos.Size == 0 || cr.Spec.Unmanaged || cr.DeletionTimestamp != nil || cr.Spec.Pause {
 		return nil
 	}
 

pkg/controller/perconaservermongodb/connections.go

@@ -63,3 +63,7 @@ func (r *ReconcilePerconaServerMongoDB) mongoClientWithRole(ctx context.Context,
 func (r *ReconcilePerconaServerMongoDB) mongosClientWithRole(ctx context.Context, cr *api.PerconaServerMongoDB, role UserRole) (mongo.Client, error) {
 	return r.MongoClientProvider().Mongos(ctx, cr, role)
 }
+
+func (r *ReconcilePerconaServerMongoDB) standaloneClientWithRole(ctx context.Context, cr *api.PerconaServerMongoDB, role UserRole, host string) (mongo.Client, error) {
+	return r.MongoClientProvider().Standalone(ctx, cr, role, host)
+}

pkg/controller/perconaservermongodb/finalizers.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/pkg/errors"
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/labels"
@@ -55,44 +56,106 @@ func (r *ReconcilePerconaServerMongoDB) checkFinalizers(ctx context.Context, cr
 }
 
 func (r *ReconcilePerconaServerMongoDB) deletePSMDBPods(ctx context.Context, cr *api.PerconaServerMongoDB) (err error) {
-	done := true
-	for _, rs := range cr.Spec.Replsets {
-		sts, err := r.getRsStatefulset(ctx, cr, rs.Name)
-		if err != nil {
-			if k8serrors.IsNotFound(err) {
-				continue
+	if cr.Spec.Sharding.Enabled {
+		cr.Spec.Sharding.Mongos.Size = 0
+
+		sts := new(appsv1.StatefulSet)
+		err := r.client.Get(ctx, cr.MongosNamespacedName(), sts)
+		if client.IgnoreNotFound(err) != nil {
+			return errors.Wrap(err, "failed to get mongos statefulset")
+		}
+		if sts.Spec.Replicas != nil && *sts.Spec.Replicas > 0 {
+			err = r.disableBalancer(ctx, cr)
+			if err != nil {
+				return errors.Wrap(err, "failed to disable balancer")
 			}
-			return errors.Wrap(err, "get rs statefulset")
 		}
-
-		pods := &corev1.PodList{}
-		err = r.client.List(ctx,
-			pods,
-			&client.ListOptions{
-				Namespace:     cr.Namespace,
-				LabelSelector: labels.SelectorFromSet(sts.Spec.Selector.MatchLabels),
-			},
-		)
+		list, err := r.getMongosPods(ctx, cr)
 		if err != nil {
-			if k8serrors.IsNotFound(err) {
-				continue
-			}
-			return errors.Wrap(err, "get rs statefulset")
+			return errors.Wrap(err, "get mongos pods")
 		}
-		if len(pods.Items) > int(*sts.Spec.Replicas) {
+		if len(list.Items) != 0 {
 			return errWaitingTermination
 		}
-		if *sts.Spec.Replicas != 1 {
-			rs.Size = 1
-			done = false
+	}
+
+	replsetsDeleted := true
+	for _, rs := range cr.Spec.Replsets {
+		if err := r.deleteRSPods(ctx, cr, rs); err != nil {
+			if err == errWaitingTermination {
+				replsetsDeleted = false
+				continue
+			}
+			return err
 		}
 	}
-	if !done {
+	if !replsetsDeleted {
 		return errWaitingTermination
 	}
+
+	if cr.Spec.Sharding.Enabled && cr.Spec.Sharding.ConfigsvrReplSet != nil {
+		if err := r.deleteRSPods(ctx, cr, cr.Spec.Sharding.ConfigsvrReplSet); err != nil {
+			return err
+		}
+	}
 	return nil
 }
 
+func (r *ReconcilePerconaServerMongoDB) deleteRSPods(ctx context.Context, cr *api.PerconaServerMongoDB, rs *api.ReplsetSpec) error {
+	sts, err := r.getRsStatefulset(ctx, cr, rs.Name)
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			return nil
+		}
+		return errors.Wrap(err, "get rs statefulset")
+	}
+
+	pods := &corev1.PodList{}
+	err = r.client.List(ctx,
+		pods,
+		&client.ListOptions{
+			Namespace:     cr.Namespace,
+			LabelSelector: labels.SelectorFromSet(sts.Spec.Selector.MatchLabels),
+		},
+	)
+	if err != nil {
+		if k8serrors.IsNotFound(err) {
+			return nil
+		}
+		return errors.Wrap(err, "get rs statefulset")
+	}
+
+	rs.Size = 1
+
+	switch *sts.Spec.Replicas {
+	case 0:
+		rs.Size = 0
+		if len(pods.Items) == 0 {
+			return nil
+		}
+		return errWaitingTermination
+	case 1:
+		// If there is one pod left, we should be sure that it's the primary
+		if len(pods.Items) != 1 {
+			return errWaitingTermination
+		}
+
+		isPrimary, err := r.isPodPrimary(ctx, cr, pods.Items[0], rs)
+		if err != nil {
+			return errors.Wrap(err, "is pod primary")
+		}
+		if !isPrimary {
+			return errWaitingTermination
+		}
+
+		// If true, we should resize the replset to 0
+		rs.Size = 0
+		return errWaitingTermination
+	default:
+		return errWaitingTermination
+	}
+}
+
 func (r *ReconcilePerconaServerMongoDB) deletePvcFinalizer(ctx context.Context, cr *api.PerconaServerMongoDB) error {
 	err := r.deleteAllStatefulsets(ctx, cr)
 	if err != nil {

pkg/controller/perconaservermongodb/mgo.go

@@ -165,7 +165,7 @@ func (r *ReconcilePerconaServerMongoDB) reconcileCluster(ctx context.Context, cr
 		cr.Status.Mongos != nil &&
 		cr.Status.Mongos.Status == api.AppStateReady &&
 		replset.ClusterRole == api.ClusterRoleShardSvr &&
-		len(mongosPods) > 0 {
+		len(mongosPods) > 0 && cr.Spec.Sharding.Mongos.Size > 0 {
 
 		mongosSession, err := r.mongosClientWithRole(ctx, cr, roleClusterAdmin)
 		if err != nil {

pkg/controller/perconaservermongodb/psmdb_controller.go

@@ -274,6 +274,11 @@ func (r *ReconcilePerconaServerMongoDB) Reconcile(ctx context.Context, request r
 		}
 	}
 
+	err = r.reconcilePause(ctx, cr)
+	if err != nil {
+		return reconcile.Result{}, err
+	}
+
 	err = r.checkConfiguration(ctx, cr)
 	if err != nil {
 		return reconcile.Result{}, err
@@ -553,6 +558,44 @@ func (r *ReconcilePerconaServerMongoDB) Reconcile(ctx context.Context, request r
 	return rr, nil
 }
 
+func (r *ReconcilePerconaServerMongoDB) reconcilePause(ctx context.Context, cr *api.PerconaServerMongoDB) error {
+	if !cr.Spec.Pause || cr.DeletionTimestamp != nil {
+		return nil
+	}
+
+	log := logf.FromContext(ctx)
+
+	backupRunning, err := r.isBackupRunning(ctx, cr)
+	if err != nil {
+		return errors.Wrap(err, "check if backup is running")
+	}
+	if backupRunning {
+		cr.Spec.Pause = false
+		if err := cr.CheckNSetDefaults(r.serverVersion.Platform, log); err != nil {
+			return errors.Wrap(err, "failed to set defaults")
+		}
+		log.Info("cluster will pause after all backups finished")
+		return nil
+	}
+
+	for _, rs := range cr.Spec.Replsets {
+		if cr.Status.State == api.AppStateStopping {
+			log.Info("Pausing cluster", "replset", rs.Name)
+		}
+		rs.Arbiter.Enabled = false
+		rs.NonVoting.Enabled = false
+	}
+
+	if err := r.deletePSMDBPods(ctx, cr); err != nil {
+		if err == errWaitingTermination {
+			log.Info("pausing cluster", "error", err.Error())
+			return nil
+		}
+		return errors.Wrap(err, "delete psmdb pods")
+	}
+	return nil
+}
+
 func (r *ReconcilePerconaServerMongoDB) setCRVersion(ctx context.Context, cr *api.PerconaServerMongoDB) error {
 	if len(cr.Spec.CRVersion) > 0 {
 		return nil

pkg/controller/perconaservermongodb/smart.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"sort"
-	"strings"
 	"time"
 
 	"github.com/pkg/errors"
@@ -115,24 +114,6 @@ func (r *ReconcilePerconaServerMongoDB) smartUpdate(ctx context.Context, cr *api
 		}
 	}
 
-	client, err := r.mongoClientWithRole(ctx, cr, *replset, roleClusterAdmin)
-	if err != nil {
-		return fmt.Errorf("failed to get mongo client: %v", err)
-	}
-
-	defer func() {
-		err := client.Disconnect(ctx)
-		if err != nil {
-			log.Error(err, "failed to close connection")
-		}
-	}()
-
-	primary, err := psmdb.GetPrimaryPod(ctx, client)
-	if err != nil {
-		return fmt.Errorf("get primary pod: %v", err)
-	}
-	log.Info("Got primary pod", "name", primary)
-
 	waitLimit := int(replset.LivenessProbe.InitialDelaySeconds)
 
 	sort.Slice(list.Items, func(i, j int) bool {
@@ -141,19 +122,11 @@ func (r *ReconcilePerconaServerMongoDB) smartUpdate(ctx context.Context, cr *api
 
 	var primaryPod corev1.Pod
 	for _, pod := range list.Items {
-		if replset.Expose.Enabled {
-			host, err := psmdb.MongoHost(ctx, r.client, cr, replset.Name, replset.Expose.Enabled, pod)
-			if err != nil {
-				return errors.Wrapf(err, "get mongo host for pod %s", pod.Name)
-			}
-
-			if host == primary {
-				primaryPod = pod
-				continue
-			}
+		isPrimary, err := r.isPodPrimary(ctx, cr, pod, replset)
+		if err != nil {
+			return errors.Wrap(err, "is pod primary")
 		}
-
-		if strings.HasPrefix(primary, fmt.Sprintf("%s.%s.%s", pod.Name, sfs.Name, sfs.Namespace)) {
+		if isPrimary {
 			primaryPod = pod
 			continue
 		}
@@ -180,6 +153,18 @@ func (r *ReconcilePerconaServerMongoDB) smartUpdate(ctx context.Context, cr *api
 	if sfs.Labels["app.kubernetes.io/component"] != "nonVoting" && len(primaryPod.Name) > 0 {
 		forceStepDown := replset.Size == 1
 		log.Info("doing step down...", "force", forceStepDown)
+		client, err := r.mongoClientWithRole(ctx, cr, *replset, roleClusterAdmin)
+		if err != nil {
+			return fmt.Errorf("failed to get mongo client: %v", err)
+		}
+
+		defer func() {
+			err := client.Disconnect(ctx)
+			if err != nil {
+				log.Error(err, "failed to close connection")
+			}
+		}()
+
 		err = client.StepDown(ctx, forceStepDown)
 		if err != nil {
 			return errors.Wrap(err, "failed to do step down")
@@ -196,6 +181,32 @@ func (r *ReconcilePerconaServerMongoDB) smartUpdate(ctx context.Context, cr *api
 	return nil
 }
 
+func (r *ReconcilePerconaServerMongoDB) isPodPrimary(ctx context.Context, cr *api.PerconaServerMongoDB, pod corev1.Pod, rs *api.ReplsetSpec) (bool, error) {
+	log := logf.FromContext(ctx)
+
+	host, err := psmdb.MongoHost(ctx, r.client, cr, rs.Name, rs.Expose.Enabled, pod)
+	if err != nil {
+		return false, errors.Wrap(err, "failed to get mongo host")
+	}
+	mgoClient, err := r.standaloneClientWithRole(ctx, cr, roleClusterAdmin, host)
+	if err != nil {
+		return false, errors.Wrap(err, "failed to create standalone client")
+	}
+	defer func() {
+		err := mgoClient.Disconnect(ctx)
+		if err != nil {
+			log.Error(err, "failed to close connection")
+		}
+	}()
+
+	isMaster, err := mgoClient.IsMaster(ctx)
+	if err != nil {
+		return false, errors.Wrap(err, "is master")
+	}
+
+	return isMaster.IsMaster, nil
+}
+
 func (r *ReconcilePerconaServerMongoDB) smartMongosUpdate(ctx context.Context, cr *api.PerconaServerMongoDB, sts *appsv1.StatefulSet) error {
 	log := logf.FromContext(ctx)