K8SPSMDB-1378: Prevent parallel backups (#1920)

* K8SPSMDB-1378: Prevent parallel backups

* add backup existence checks

* fix finalizer

* fix error backups

Author: egegunes

e2e-tests/demand-backup-incremental-sharded/run

@@ -127,8 +127,16 @@ if [ -z "$SKIP_BACKUPS_TO_AWS_GCP_AZURE" ]; then
 	run_backup azure-blob ${backup_name_azure}
 
 	wait_backup "${backup_name_aws}"
+	check_backup_in_storage ${backup_name_aws} s3 rs0
+	check_backup_in_storage ${backup_name_aws} s3 cfg
+
 	wait_backup "${backup_name_gcp}"
+	check_backup_in_storage ${backup_name_gcp} gcs rs0
+	check_backup_in_storage ${backup_name_gcp} gcs cfg
+
 	wait_backup "${backup_name_azure}"
+	check_backup_in_storage ${backup_name_azure} azure rs0
+	check_backup_in_storage ${backup_name_azure} azure cfg
 fi
 
 backup_name_minio="backup-minio-sharded"

e2e-tests/demand-backup-incremental/run

@@ -111,8 +111,13 @@ if [ -z "$SKIP_BACKUPS_TO_AWS_GCP_AZURE" ]; then
 	run_backup azure-blob ${backup_name_azure}
 
 	wait_backup "${backup_name_aws}"
+	check_backup_in_storage ${backup_name_aws} s3 rs0
+
 	wait_backup "${backup_name_gcp}"
+	check_backup_in_storage ${backup_name_gcp} gcs rs0
+
 	wait_backup "${backup_name_azure}"
+	check_backup_in_storage ${backup_name_azure} azure rs0
 fi
 
 backup_name_minio="backup-minio"

e2e-tests/demand-backup-physical-sharded/run

@@ -112,8 +112,16 @@ if [ -z "$SKIP_BACKUPS_TO_AWS_GCP_AZURE" ]; then
 	run_backup azure-blob ${backup_name_azure}
 
 	wait_backup "${backup_name_aws}"
+	check_backup_in_storage ${backup_name_aws} s3 rs0
+	check_backup_in_storage ${backup_name_aws} s3 cfg
+
 	wait_backup "${backup_name_gcp}"
+	check_backup_in_storage ${backup_name_gcp} gcs rs0
+	check_backup_in_storage ${backup_name_gcp} gcs cfg
+
 	wait_backup "${backup_name_azure}"
+	check_backup_in_storage ${backup_name_azure} azure rs0
+	check_backup_in_storage ${backup_name_azure} azure cfg
 fi
 wait_backup "${backup_name_minio}"
 

e2e-tests/demand-backup-physical/run

@@ -98,8 +98,13 @@ if [ -z "$SKIP_BACKUPS_TO_AWS_GCP_AZURE" ]; then
 	run_backup azure-blob ${backup_name_azure}
 
 	wait_backup "${backup_name_aws}"
+	check_backup_in_storage ${backup_name_aws} s3 rs0
+
 	wait_backup "${backup_name_gcp}"
+	check_backup_in_storage ${backup_name_gcp} gcs rs0
+
 	wait_backup "${backup_name_azure}"
+	check_backup_in_storage ${backup_name_azure} azure rs0
 fi
 wait_backup "${backup_name_minio}"
 

e2e-tests/functions

@@ -17,6 +17,7 @@ IMAGE_PMM_CLIENT=${IMAGE_PMM_CLIENT:-"perconalab/pmm-client:dev-latest"}
 IMAGE_PMM_SERVER=${IMAGE_PMM_SERVER:-"perconalab/pmm-server:dev-latest"}
 CERT_MANAGER_VER="1.16.3"
 UPDATE_COMPARE_FILES=${UPDATE_COMPARE_FILES:-0}
+DELETE_CRD_ON_START=${DELETE_CRD_ON_START:-1}
 tmp_dir=$(mktemp -d)
 sed=$(which gsed || which sed)
 date=$(which gdate || which date)
@@ -1331,8 +1332,10 @@ check_backup_deletion() {
 create_infra() {
 	local ns="$1"
 
-	delete_crd
-	check_crd_for_deletion "${GIT_BRANCH}"
+	if [[ ${DELETE_CRD_ON_START} == 1 ]]; then
+		delete_crd
+		check_crd_for_deletion "${GIT_BRANCH}"
+	fi
 	if [ -n "$OPERATOR_NS" ]; then
 		create_namespace $OPERATOR_NS
 		deploy_operator
@@ -1867,3 +1870,31 @@ wait_for_oplogs() {
 		sleep 10
 	done
 }
+
+check_backup_in_storage() {
+	local backup=$1
+	local storage_type=$2
+	local replset=$3
+	local file=${4:-"filelist.pbm"}
+
+	local endpoint
+	case ${storage_type} in
+		s3)
+			endpoint="s3.amazonaws.com"
+			;;
+		gcs)
+			endpoint="storage.googleapis.com"
+			;;
+		azure)
+			endpoint="engk8soperators.blob.core.windows.net"
+			;;
+		*)
+			echo "unsupported storage type: ${storage_type}"
+			exit 1
+	esac
+
+	backup_dest=$(get_backup_dest "$backup")
+	local url="https://${endpoint}/${backup_dest}/${replset}/${file}"
+	log "checking if ${url} exists"
+	curl --fail --head "${url}"
+}

pkg/apis/psmdb/v1/perconaservermongodbbackup_types.go

@@ -11,6 +11,7 @@ import (
 
 // PerconaServerMongoDBBackupSpec defines the desired state of PerconaServerMongoDBBackup
 type PerconaServerMongoDBBackupSpec struct {
+	// Deprecated: Use ClusterName instead
 	PSMDBCluster     string                   `json:"psmdbCluster,omitempty"` // TODO: Remove after v1.15
 	ClusterName      string                   `json:"clusterName,omitempty"`
 	StorageName      string                   `json:"storageName,omitempty"`

pkg/controller/perconaservermongodbbackup/perconaservermongodbbackup_controller.go

@@ -30,6 +30,7 @@ import (
 
 	"github.com/percona/percona-server-mongodb-operator/clientcmd"
 	psmdbv1 "github.com/percona/percona-server-mongodb-operator/pkg/apis/psmdb/v1"
+	"github.com/percona/percona-server-mongodb-operator/pkg/k8s"
 	"github.com/percona/percona-server-mongodb-operator/pkg/naming"
 	"github.com/percona/percona-server-mongodb-operator/pkg/psmdb"
 	"github.com/percona/percona-server-mongodb-operator/pkg/psmdb/backup"
@@ -118,7 +119,7 @@ func (r *ReconcilePerconaServerMongoDBBackup) Reconcile(ctx context.Context, req
 			// Request object not found, could have been deleted after reconcile request.
 			// Owned objects are automatically garbage collected. For additional cleanup logic use finalizers.
 			// Return and don't requeue
-			return rr, nil
+			return reconcile.Result{}, nil
 		}
 		// Error reading the object - requeue the request.
 		return rr, err
@@ -144,6 +145,16 @@ func (r *ReconcilePerconaServerMongoDBBackup) Reconcile(ctx context.Context, req
 			if uerr != nil {
 				log.Error(uerr, "failed to update backup status", "backup", cr.Name)
 			}
+
+			switch cr.Status.State {
+			case psmdbv1.BackupStateReady, psmdbv1.BackupStateError:
+				log.Info("Releasing backup lock", "lease", naming.BackupLeaseName(cr.Spec.ClusterName))
+
+				err := k8s.ReleaseLease(ctx, r.client, naming.BackupLeaseName(cr.Spec.ClusterName), cr.Namespace, naming.BackupHolderId(cr))
+				if err != nil {
+					log.Error(err, "failed to release the lock")
+				}
+			}
 		}
 	}()
 
@@ -228,14 +239,28 @@ func (r *ReconcilePerconaServerMongoDBBackup) reconcile(
 	}
 
 	if cjobs {
-		if cr.Status.State != psmdbv1.BackupStateWaiting {
-			log.Info("Waiting to finish another backup/restore.")
-		}
+		log.Info("Waiting to finish another backup/restore.")
 		status.State = psmdbv1.BackupStateWaiting
 		return status, nil
 	}
 
-	switch cr.Status.State {
+	log.Info("Acquiring the backup lock")
+	lease, err := k8s.AcquireLease(ctx, r.client, naming.BackupLeaseName(cluster.Name), cr.Namespace, naming.BackupHolderId(cr))
+	if err != nil {
+		return status, errors.Wrap(err, "acquire backup lock")
+	}
+
+	if lease.Spec.HolderIdentity != nil && *lease.Spec.HolderIdentity != naming.BackupHolderId(cr) {
+		log.Info("Another backup is holding the lock", "holder", *lease.Spec.HolderIdentity)
+		status.State = psmdbv1.BackupStateWaiting
+		return status, nil
+	}
+
+	if err := r.ensureReleaseLockFinalizer(ctx, cluster, cr); err != nil {
+		return status, errors.Wrapf(err, "ensure %s finalizer", naming.FinalizerReleaseLock)
+	}
+
+	switch status.State {
 	case psmdbv1.BackupStateNew, psmdbv1.BackupStateWaiting:
 		return bcp.Start(ctx, r.client, cluster, cr)
 	}
@@ -251,6 +276,28 @@ func (r *ReconcilePerconaServerMongoDBBackup) reconcile(
 	return status, err
 }
 
+func (r *ReconcilePerconaServerMongoDBBackup) ensureReleaseLockFinalizer(
+	ctx context.Context,
+	cluster *psmdbv1.PerconaServerMongoDB,
+	cr *psmdbv1.PerconaServerMongoDBBackup,
+) error {
+	for _, f := range cr.GetFinalizers() {
+		if f == naming.FinalizerReleaseLock {
+			return nil
+		}
+	}
+
+	orig := cr.DeepCopy()
+	cr.SetFinalizers(append(cr.GetFinalizers(), naming.FinalizerReleaseLock))
+	if err := r.client.Patch(ctx, cr.DeepCopy(), client.MergeFrom(orig)); err != nil {
+		return errors.Wrap(err, "patch finalizers")
+	}
+
+	logf.FromContext(ctx).V(1).Info("Added finalizer", "finalizer", naming.FinalizerReleaseLock)
+
+	return nil
+}
+
 func (r *ReconcilePerconaServerMongoDBBackup) getPBMStorage(ctx context.Context, cluster *psmdbv1.PerconaServerMongoDB, cr *psmdbv1.PerconaServerMongoDBBackup) (storage.Storage, error) {
 	switch {
 	case cr.Status.Azure != nil:
@@ -379,18 +426,24 @@ func (r *ReconcilePerconaServerMongoDBBackup) checkFinalizers(ctx context.Contex
 
 	finalizers := []string{}
 
-	if cr.Status.State == psmdbv1.BackupStateReady {
-		for _, f := range cr.GetFinalizers() {
-			switch f {
-			case "delete-backup":
-				log.Info("delete-backup finalizer is deprecated and will be deleted in 1.20.0. Use percona.com/delete-backup instead")
-				fallthrough
-			case naming.FinalizerDeleteBackup:
+	for _, f := range cr.GetFinalizers() {
+		switch f {
+		case "delete-backup":
+			log.Info("delete-backup finalizer is deprecated and will be deleted in 1.20.0. Use percona.com/delete-backup instead")
+			fallthrough
+		case naming.FinalizerDeleteBackup:
+			if cr.Status.State == psmdbv1.BackupStateReady {
 				if err := r.deleteBackupFinalizer(ctx, cr, cluster, b); err != nil {
 					log.Error(err, "failed to run finalizer", "finalizer", f)
 					finalizers = append(finalizers, f)
 				}
 			}
+		case naming.FinalizerReleaseLock:
+			err = r.runReleaseLockFinalizer(ctx, cr)
+			if err != nil {
+				log.Error(err, "failed to release backup lock")
+				finalizers = append(finalizers, f)
+			}
 		}
 	}
 
@@ -400,6 +453,20 @@ func (r *ReconcilePerconaServerMongoDBBackup) checkFinalizers(ctx context.Contex
 	return err
 }
 
+func (r *ReconcilePerconaServerMongoDBBackup) runReleaseLockFinalizer(ctx context.Context, cr *psmdbv1.PerconaServerMongoDBBackup) error {
+	leaseName := naming.BackupLeaseName(cr.Spec.ClusterName)
+	holderId := naming.BackupHolderId(cr)
+	log := logf.FromContext(ctx).WithValues("lease", leaseName, "holder", holderId)
+
+	log.Info("releasing backup lock")
+	err := k8s.ReleaseLease(ctx, r.client, leaseName, cr.Namespace, holderId)
+	if k8serrors.IsNotFound(err) || errors.Is(err, k8s.ErrNotTheHolder) {
+		log.V(1).Info("failed to release backup lock", "error", err)
+		return nil
+	}
+	return errors.Wrap(err, "release backup lock")
+}
+
 func (r *ReconcilePerconaServerMongoDBBackup) deleteBackupFinalizer(ctx context.Context, cr *psmdbv1.PerconaServerMongoDBBackup, cluster *psmdbv1.PerconaServerMongoDB, b *Backup) error {
 	if len(cr.Status.PBMname) == 0 {
 		return nil

pkg/k8s/lease.go

@@ -0,0 +1,65 @@
+package k8s
+
+import (
+	"context"
+	"time"
+
+	"github.com/pkg/errors"
+	coordv1 "k8s.io/api/coordination/v1"
+	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+var (
+	ErrNotTheHolder = errors.New("not the holder")
+)
+
+func AcquireLease(ctx context.Context, c client.Client, name, namespace, holder string) (*coordv1.Lease, error) {
+	lease := new(coordv1.Lease)
+
+	if err := c.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, lease); err != nil {
+		if !k8serrors.IsNotFound(err) {
+			return lease, errors.Wrap(err, "get lease")
+		}
+
+		lease := &coordv1.Lease{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      name,
+				Namespace: namespace,
+			},
+			Spec: coordv1.LeaseSpec{
+				AcquireTime:    &metav1.MicroTime{Time: time.Now()},
+				HolderIdentity: &holder,
+			},
+		}
+
+		if err := c.Create(ctx, lease); err != nil {
+			return lease, errors.Wrap(err, "create lease")
+		}
+
+		return lease, nil
+	}
+
+	return lease, nil
+}
+
+func ReleaseLease(ctx context.Context, c client.Client, name, namespace, holder string) error {
+	lease := new(coordv1.Lease)
+
+	if err := c.Get(ctx, types.NamespacedName{Namespace: namespace, Name: name}, lease); err != nil {
+		return errors.Wrap(err, "get lease")
+	}
+
+	// HolderIdentity can be nil, we allow deleting the lease if that's the case
+	if lease.Spec.HolderIdentity != nil && *lease.Spec.HolderIdentity != holder {
+		return ErrNotTheHolder
+	}
+
+	if err := c.Delete(ctx, lease); err != nil {
+		return errors.Wrap(err, "delete lease")
+	}
+
+	return nil
+}