implement lock volume feature

Signed-off-by: Yuji Ito <[email protected]>

Author: llamerada-jp

cmd/controller.go

@@ -162,7 +162,7 @@ func controllerMain(args []string) error {
 	// Register custom metrics
 	finmetrics.Register()
 
-	snapRepo := ceph.NewRBDRepository()
+	rbdRepo := ceph.NewRBDRepository()
 	maxPartSize, err := resource.ParseQuantity(os.Getenv("MAX_PART_SIZE"))
 	if err != nil {
 		return fmt.Errorf("failed to parse MAX_PART_SIZE environment variable: %w", err)
@@ -173,7 +173,8 @@ func controllerMain(args []string) error {
 		os.Getenv("POD_NAMESPACE"),
 		os.Getenv("POD_IMAGE"),
 		&maxPartSize,
-		snapRepo,
+		rbdRepo,
+		rbdRepo,
 		rawImgExpansionUnitSize,
 	)
 	if err = finBackupReconciler.SetupWithManager(mgr); err != nil {

internal/controller/finbackup_controller.go

@@ -68,6 +68,7 @@ const (
 var (
 	errNonRetryableReconcile = errors.New("non retryable reconciliation error; " +
 		"reconciliation must not keep going nor be retried")
+	errVolumeLockedByAnother = errors.New("the volume is locked by another process")
 )
 
 // FinBackupReconciler reconciles a FinBackup object
@@ -78,6 +79,7 @@ type FinBackupReconciler struct {
 	podImage                string
 	maxPartSize             *resource.Quantity
 	snapRepo                model.RBDSnapshotRepository
+	imageLocker             model.RBDImageLocker
 	rawImgExpansionUnitSize uint64
 }
 
@@ -88,6 +90,7 @@ func NewFinBackupReconciler(
 	podImage string,
 	maxPartSize *resource.Quantity,
 	snapRepo model.RBDSnapshotRepository,
+	imageLocker model.RBDImageLocker,
 	rawImgExpansionUnitSize uint64,
 ) *FinBackupReconciler {
 	return &FinBackupReconciler{
@@ -97,6 +100,7 @@ func NewFinBackupReconciler(
 		podImage:                podImage,
 		maxPartSize:             maxPartSize,
 		snapRepo:                snapRepo,
+		imageLocker:             imageLocker,
 		rawImgExpansionUnitSize: rawImgExpansionUnitSize,
 	}
 }
@@ -396,8 +400,13 @@ func (r *FinBackupReconciler) createSnapshot(ctx context.Context, backup *finv1.
 		return ctrl.Result{}, err
 	}
 
-	snap, err := r.createSnapshotIfNeeded(rbdPool, rbdImage, snapshotName(backup))
+	snap, err := r.createSnapshotIfNeeded(rbdPool, rbdImage, snapshotName(backup), lockID(backup))
 	if err != nil {
+		if errors.Is(err, errVolumeLockedByAnother) {
+			logger.Info("the volume is locked by another process", "uid", string(backup.GetUID()))
+			// FIXME: The following "requeue after" is temporary code.
+			return ctrl.Result{RequeueAfter: 5 * time.Second}, nil
+		}
 		logger.Error(err, "failed to create or get snapshot")
 		return ctrl.Result{}, err
 	}
@@ -571,12 +580,20 @@ func (r *FinBackupReconciler) reconcileDelete(
 	return ctrl.Result{}, nil
 }
 
-func (r *FinBackupReconciler) createSnapshotIfNeeded(rbdPool, rbdImage, snapName string) (*model.RBDSnapshot, error) {
+func (r *FinBackupReconciler) createSnapshotIfNeeded(rbdPool, rbdImage, snapName, lockID string) (*model.RBDSnapshot, error) {
 	snap, err := findSnapshot(r.snapRepo, rbdPool, rbdImage, snapName)
 	if err != nil {
 		if !errors.Is(err, model.ErrNotFound) {
 			return nil, fmt.Errorf("failed to get snapshot: %w", err)
 		}
+
+		lockSuccess, err := r.lockVolume(rbdPool, rbdImage, lockID)
+		if err != nil {
+			return nil, fmt.Errorf("failed to lock image: %w", err)
+		}
+		if !lockSuccess {
+			return nil, errVolumeLockedByAnother
+		}
 		err = r.snapRepo.CreateSnapshot(rbdPool, rbdImage, snapName)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create snapshot: %w", err)
@@ -586,6 +603,10 @@ func (r *FinBackupReconciler) createSnapshotIfNeeded(rbdPool, rbdImage, snapName
 			return nil, fmt.Errorf("failed to get snapshot after creation: %w", err)
 		}
 	}
+	if err := r.unlockVolume(rbdPool, rbdImage, lockID); err != nil {
+		return nil, fmt.Errorf("failed to unlock image: %w", err)
+	}
+
 	return snap, nil
 }
 
@@ -613,6 +634,70 @@ func (r *FinBackupReconciler) removeSnapshot(ctx context.Context, backup *finv1.
 	return nil
 }
 
+// lockVolume adds a lock to the specified RBD volume if the lock is not already held.
+// It returns true if the lock is held by this caller, false if another lock is held or an error occurs.
+func (r *FinBackupReconciler) lockVolume(
+	poolName, imageName, lockID string,
+) (bool, error) {
+	// Add a lock.
+	if errAdd := r.imageLocker.LockAdd(poolName, imageName, lockID); errAdd != nil {
+		locks, errLs := r.imageLocker.LockLs(poolName, imageName)
+		if errLs != nil {
+			return false, fmt.Errorf("failed to add a lock and list locks on volume %s/%s: %w", poolName, imageName, errors.Join(errAdd, errLs))
+		}
+
+		switch len(locks) {
+		case 0:
+			// It may have been unlocked after the lock failed, but since other causes are also possible, an error is returned.
+			return false, fmt.Errorf("failed to add a lock to the volume %s/%s: %w", poolName, imageName, errAdd)
+
+		case 1:
+			if locks[0].LockID == lockID {
+				// Already locked by this FB.
+				return true, nil
+			}
+			// Locked by another process.
+			return false, nil
+
+		default:
+			// Multiple locks found; unexpected state.
+			return false, fmt.Errorf("multiple locks found on volume %s/%s after failed lock attempt(%v)", poolName, imageName, locks)
+		}
+	}
+
+	// Locked
+	return true, nil
+}
+
+// unlockVolume removes the specified lock from the RBD volume if the lock is held.
+// No action is taken if the lock is not found.
+func (r *FinBackupReconciler) unlockVolume(
+	poolName, imageName, lockID string,
+) error {
+	// List up locks to check if the lock is held.
+	locks, err := r.imageLocker.LockLs(poolName, imageName)
+	if err != nil {
+		return fmt.Errorf("failed to list locks of the volume %s/%s: %w", poolName, imageName, err)
+	}
+
+	if len(locks) >= 2 {
+		return fmt.Errorf("multiple locks found on volume %s/%s when unlocking (%v)", poolName, imageName, locks)
+	}
+
+	for _, lock := range locks {
+		if lock.LockID == lockID {
+			// Unlock
+			if err := r.imageLocker.LockRm(poolName, imageName, lock); err != nil {
+				return fmt.Errorf("failed to remove the lock from the volume %s/%s: %w", poolName, imageName, err)
+			}
+			return nil
+		}
+	}
+
+	// Already unlocked.
+	return nil
+}
+
 func (r *FinBackupReconciler) getRBDPoolAndImageFromPVC(
 	ctx context.Context,
 	pvc *corev1.PersistentVolumeClaim,
@@ -733,6 +818,10 @@ func cleanupJobName(backup *finv1.FinBackup) string {
 	return "fin-cleanup-" + string(backup.GetUID())
 }
 
+func lockID(backup *finv1.FinBackup) string {
+	return string(backup.GetUID())
+}
+
 func (r *FinBackupReconciler) createOrUpdateBackupJob(
 	ctx context.Context, backup *finv1.FinBackup, diffFrom string,
 	backupTargetPVCUID string, maxPartSize *resource.Quantity,

internal/controller/finbackup_controller_test.go

@@ -2,6 +2,7 @@ package controller
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"slices"
 	"strconv"
@@ -119,6 +120,7 @@ var _ = Describe("FinBackup Controller integration test", Ordered, func() {
 			podImage:                podImage,
 			maxPartSize:             &defaultMaxPartSize,
 			snapRepo:                rbdRepo,
+			imageLocker:             rbdRepo,
 			rawImgExpansionUnitSize: 100 * 1 << 20,
 		}
 		err = reconciler.SetupWithManager(mgr)
@@ -766,6 +768,7 @@ var _ = Describe("FinBackup Controller Reconcile Test", Ordered, func() {
 			podImage:                podImage,
 			maxPartSize:             &defaultMaxPartSize,
 			snapRepo:                rbdRepo,
+			imageLocker:             rbdRepo,
 			rawImgExpansionUnitSize: 100 * 1 << 20,
 		}
 	})
@@ -1249,6 +1252,158 @@ var _ = Describe("FinBackup Controller Reconcile Test", Ordered, func() {
 	})
 })
 
+var _ = Describe("FinBackup Controller Unit Tests", Ordered, func() {
+	Context("lockVolume", func() {
+		var reconciler *FinBackupReconciler
+		var rbdRepo *fake.RBDRepository2
+
+		lock := func(poolName, imageName, lockID string, rbdErr error) (bool, error) {
+			rbdRepo.SetError(rbdErr)
+			defer rbdRepo.SetError(nil)
+
+			locked, err := reconciler.lockVolume(poolName, imageName, lockID)
+			if err != nil {
+				return locked, err
+			}
+			// Check if the lock with lockID exists
+			locks, err := rbdRepo.LockLs(poolName, imageName)
+			if err != nil {
+				return locked, err
+			}
+			lockExists := false
+			for _, lock := range locks {
+				if lock.LockID == lockID {
+					lockExists = true
+					break
+				}
+			}
+			Expect(locked).To(Equal(lockExists))
+			return locked, nil
+		}
+
+		It("setup", func(ctx SpecContext) {
+			volumeInfo := &fake.VolumeInfo{
+				Namespace: utils.GetUniqueName("ns-"),
+				PVCName:   utils.GetUniqueName("pvc-"),
+				PVName:    utils.GetUniqueName("pv-"),
+				PoolName:  rbdPoolName,
+				ImageName: rbdImageName,
+			}
+			rbdRepo = fake.NewRBDRepository2(volumeInfo.PoolName, volumeInfo.ImageName)
+
+			mgr, err := ctrl.NewManager(cfg, ctrl.Options{Scheme: scheme.Scheme})
+			Expect(err).ToNot(HaveOccurred())
+			reconciler = &FinBackupReconciler{
+				Client:                  mgr.GetClient(),
+				Scheme:                  mgr.GetScheme(),
+				cephClusterNamespace:    namespace,
+				podImage:                podImage,
+				maxPartSize:             &defaultMaxPartSize,
+				snapRepo:                rbdRepo,
+				imageLocker:             rbdRepo,
+				rawImgExpansionUnitSize: 100 * 1 << 20,
+			}
+		})
+
+		It("lock a volume successfully", func() {
+			locked, err := lock("pool", "image1", "lock1", nil)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locked).To(BeTrue())
+		})
+		It("lock a volume that is already locked by the same lockID", func() {
+			locked, err := lock("pool", "image1", "lock1", nil)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locked).To(BeTrue())
+		})
+		It("fail to lock a volume that is already locked by a different lockID", func() {
+			locked, err := lock("pool", "image1", "lock2", nil)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locked).To(BeFalse())
+		})
+		It("lock a different volume successfully", func() {
+			locked, err := lock("pool", "image2", "lock1", nil)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locked).To(BeTrue())
+		})
+		It("error when rbd lock ls fails", func() {
+			locked, err := lock("pool", "image1", "lock1", errors.New("rbd lock ls error"))
+			Expect(err).To(HaveOccurred())
+			Expect(locked).To(BeFalse())
+		})
+	})
+
+	Context("unlockVolume", func() {
+		var reconciler *FinBackupReconciler
+		var rbdRepo *fake.RBDRepository2
+
+		It("setup", func(ctx SpecContext) {
+			volumeInfo := &fake.VolumeInfo{
+				Namespace: utils.GetUniqueName("ns-"),
+				PVCName:   utils.GetUniqueName("pvc-"),
+				PVName:    utils.GetUniqueName("pv-"),
+				PoolName:  rbdPoolName,
+				ImageName: rbdImageName,
+			}
+			rbdRepo = fake.NewRBDRepository2(volumeInfo.PoolName, volumeInfo.ImageName)
+
+			mgr, err := ctrl.NewManager(cfg, ctrl.Options{Scheme: scheme.Scheme})
+			Expect(err).ToNot(HaveOccurred())
+			reconciler = &FinBackupReconciler{
+				Client:                  mgr.GetClient(),
+				Scheme:                  mgr.GetScheme(),
+				cephClusterNamespace:    namespace,
+				podImage:                podImage,
+				maxPartSize:             &defaultMaxPartSize,
+				snapRepo:                rbdRepo,
+				imageLocker:             rbdRepo,
+				rawImgExpansionUnitSize: 100 * 1 << 20,
+			}
+		})
+
+		It("locks a volume", func(ctx SpecContext) {
+			locked, err := reconciler.lockVolume("pool", "image", "lock1")
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locked).To(BeTrue())
+		})
+
+		DescribeTable("", func(
+			ctx SpecContext,
+			poolName, imageName, lockID string, rbdErr error,
+			expectErr bool,
+		) {
+			rbdRepo.SetError(rbdErr)
+			defer rbdRepo.SetError(nil)
+
+			err := reconciler.unlockVolume(poolName, imageName, lockID)
+			if expectErr {
+				Expect(err).To(HaveOccurred())
+			} else {
+				Expect(err).NotTo(HaveOccurred())
+			}
+			// Check if the lock is removed
+			locks, err := rbdRepo.LockLs(poolName, imageName)
+			if expectErr {
+				Expect(err).To(HaveOccurred())
+			} else {
+				Expect(err).NotTo(HaveOccurred())
+				for _, l := range locks {
+					Expect(l.LockID).NotTo(Equal(lockID))
+				}
+			}
+		},
+			Entry("unlock a volume successfully",
+				"pool", "image", "lock1", nil,
+				false),
+			Entry("unlock the same volume again (no-op)",
+				"pool", "image", "lock1", nil,
+				false),
+			Entry("error when rbd unlock fails",
+				"pool", "image", "lock1", errors.New("rbd unlock error"),
+				true),
+		)
+	})
+})
+
 // CSATEST-1627
 // Description:
 //