implement e2e test for lock feature

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

Author: llamerada-jp

internal/controller/finbackup_controller.go

@@ -45,7 +45,7 @@ const (
 	labelComponentDeletionJob = "deletion-job"
 
 	// Annotations
-	annotationBackupTargetRBDImage = "fin.cybozu.io/backup-target-rbd-image"
+	AnnotationBackupTargetRBDImage = "fin.cybozu.io/backup-target-rbd-image"
 	annotationDiffFrom             = "fin.cybozu.io/diff-from"
 	annotationFinBackupName        = "fin.cybozu.io/finbackup-name"
 	annotationFinBackupNamespace   = "fin.cybozu.io/finbackup-namespace"
@@ -390,7 +390,7 @@ func (r *FinBackupReconciler) createSnapshot(ctx context.Context, backup *finv1.
 	if annotations == nil {
 		annotations = map[string]string{}
 	}
-	annotations[annotationBackupTargetRBDImage] = rbdImage
+	annotations[AnnotationBackupTargetRBDImage] = rbdImage
 	annotations[annotationRBDPool] = rbdPool
 	backup.SetAnnotations(annotations)
 
@@ -723,7 +723,7 @@ func (r *FinBackupReconciler) getRBDPoolAndImageFromPVC(
 
 func (r *FinBackupReconciler) getRBDPoolAndImage(ctx context.Context, backup *finv1.FinBackup) (string, string, error) {
 	rbdPool := backup.GetAnnotations()[annotationRBDPool]
-	rbdImage := backup.GetAnnotations()[annotationBackupTargetRBDImage]
+	rbdImage := backup.GetAnnotations()[AnnotationBackupTargetRBDImage]
 	if rbdPool != "" && rbdImage != "" {
 		return rbdPool, rbdImage, nil
 	}
@@ -884,7 +884,7 @@ func (r *FinBackupReconciler) createOrUpdateBackupJob(
 					},
 					{
 						Name:  "RBD_IMAGE_NAME",
-						Value: backup.GetAnnotations()[annotationBackupTargetRBDImage],
+						Value: backup.GetAnnotations()[AnnotationBackupTargetRBDImage],
 					},
 					{
 						Name:  "BACKUP_SNAPSHOT_ID",

internal/controller/finbackup_controller_test.go

@@ -326,7 +326,7 @@ var _ = Describe("FinBackup Controller integration test", Ordered, func() {
 			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(finbackup2), finbackup2)).Should(Succeed())
 			Expect(finbackup2.GetLabels()).To(HaveKeyWithValue(labelBackupTargetPVCUID, string(pvc1.GetUID())))
 			annotations := finbackup2.GetAnnotations()
-			Expect(annotations).To(HaveKeyWithValue(annotationBackupTargetRBDImage, rbdImageName))
+			Expect(annotations).To(HaveKeyWithValue(AnnotationBackupTargetRBDImage, rbdImageName))
 			Expect(annotations).To(HaveKeyWithValue(annotationRBDPool, rbdPoolName))
 
 			// Incremental backup specific: the diff-from annotation should exist and point to the SnapID of the full backup.
@@ -813,7 +813,7 @@ var _ = Describe("FinBackup Controller Reconcile Test", Ordered, func() {
 			var updated finv1.FinBackup
 			Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(finbackup), &updated)).Should(Succeed())
 			Expect(updated.GetLabels()).To(HaveKeyWithValue(labelBackupTargetPVCUID, string(pvc.GetUID())))
-			Expect(updated.GetAnnotations()).To(HaveKeyWithValue(annotationBackupTargetRBDImage, rbdImageName))
+			Expect(updated.GetAnnotations()).To(HaveKeyWithValue(AnnotationBackupTargetRBDImage, rbdImageName))
 			Expect(updated.GetAnnotations()).To(HaveKeyWithValue(annotationRBDPool, rbdPoolName))
 		})
 	})

test/e2e/e2e_suite_test.go

@@ -74,6 +74,7 @@ var _ = Describe("Fin", func() {
 	Context("wait environment", waitEnvironment)
 	Context("full backup", Label("full-backup"), Ordered, fullBackupTestSuite)
 	Context("incremental backup", Label("incremental-backup"), Ordered, incrementalBackupTestSuite)
+	Context("lock", Label("lock"), Label("misc"), Ordered, lockTestSuite)
 	Context("verification", Label("verification"), Label("misc"), Ordered, verificationTestSuite)
 	Context("delete incremental backup", Label("delete-incremental-backup"), Label("misc"), Ordered,
 		deleteIncrementalBackupTestSuite)

test/e2e/full_backup_test.go

@@ -51,6 +51,10 @@ func fullBackupTestSuite() {
 	It("should create full backup", func(ctx SpecContext) {
 		finbackup = CreateBackup(ctx, ctrlClient, rookNamespace, pvc, nodes[0])
 
+		var err error
+		finbackup, err = WaitForFinBackupStoredToNodeAndVerified(ctx, ctrlClient, finbackup, 5*time.Minute)
+		Expect(err).NotTo(HaveOccurred())
+
 		VerifyRawImage(pvc, nodes[0], writtenData)
 	})
 
@@ -195,7 +199,8 @@ func fullBackupTestSuite() {
 
 		VerifyNonExistenceOfRawImage(pvc, nodes[0])
 		VerifyDeletionOfJobsForBackup(ctx, k8sClient, finbackup)
-		VerifyDeletionOfSnapshotInFinBackup(ctx, ctrlClient, finbackup)
+		err = VerifyDeletionOfSnapshotInFinBackup(ctx, finbackup)
+		Expect(err).NotTo(HaveOccurred())
 	})
 
 	AfterAll(func(ctx SpecContext) {

test/e2e/incremental_test.go

@@ -38,6 +38,10 @@ func incrementalBackupTestSuite() {
 
 		finbackup1 = CreateBackup(ctx, ctrlClient, rookNamespace, pvc, nodes[0])
 
+		var err error
+		finbackup1, err = WaitForFinBackupStoredToNodeAndVerified(ctx, ctrlClient, finbackup1, 5*time.Minute)
+		Expect(err).NotTo(HaveOccurred())
+
 		By("verifying the data in raw.img from the full backup")
 		volumePath = filepath.Join("/fin", ns.Name, pvc.Name)
 		// `--native-ssh=false` is used to avoid issues of conversion from LF to CRLF.
@@ -240,7 +244,8 @@ func incrementalBackupTestSuite() {
 		Expect(rawImageData).To(Equal(dataOnIncrementalBackup), "Data in raw.img does not match the expected data")
 
 		VerifyDeletionOfJobsForBackup(ctx, k8sClient, finbackup1)
-		VerifyDeletionOfSnapshotInFinBackup(ctx, ctrlClient, finbackup1)
+		err = VerifyDeletionOfSnapshotInFinBackup(ctx, finbackup1)
+		Expect(err).NotTo(HaveOccurred())
 	})
 
 	// Description:

test/e2e/lock_test.go

@@ -0,0 +1,92 @@
+package e2e
+
+import (
+	"encoding/json"
+	"time"
+
+	finv1 "github.com/cybozu-go/fin/api/v1"
+	"github.com/cybozu-go/fin/internal/model"
+	"github.com/cybozu-go/fin/test/utils"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+)
+
+func lockTestSuite() {
+	var ns *corev1.Namespace
+	var pvc *corev1.PersistentVolumeClaim
+	var fb *finv1.FinBackup
+	dummyLockID := "dummy-lock-id"
+	var poolName, imageName string
+
+	It("should setup environment", func(ctx SpecContext) {
+		ns = NewNamespace(utils.GetUniqueName("ns-"))
+		err := CreateNamespace(ctx, k8sClient, ns)
+		Expect(err).NotTo(HaveOccurred())
+
+		pvc = CreateBackupTargetPVC(ctx, k8sClient, ns, "Filesystem", rookStorageClass, "ReadWriteOnce", "100Mi")
+		pvc, err = WaitForPVCBound(ctx, ctrlClient, pvc, 1*time.Minute)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("should lock the volume", func(ctx SpecContext) {
+		pv, err := GetPvByPvc(ctx, k8sClient, pvc)
+		Expect(err).NotTo(HaveOccurred())
+		poolName = pv.Spec.CSI.VolumeAttributes["pool"]
+		imageName = pv.Spec.CSI.VolumeAttributes["imageName"]
+
+		// locked
+		_, _, err = kubectl("exec", "-n", rookNamespace, "deployment/"+finDeploymentName, "--",
+			"rbd", "-p", poolName, "lock", "add", imageName, dummyLockID)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("should create backup and wait for the log", func(ctx SpecContext) {
+		var err error
+		fb, err = NewFinBackup(rookNamespace, utils.GetUniqueName("fb-"), pvc, nodes[0])
+		Expect(err).NotTo(HaveOccurred())
+		err = CreateFinBackup(ctx, ctrlClient, fb)
+		Expect(err).NotTo(HaveOccurred())
+
+		err = WaitControllerLog(ctx,
+			"the volume is locked by another process.*"+string(fb.GetUID()),
+			3*time.Minute)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("checks that the snapshot is not created", func(ctx SpecContext) {
+		snapshots, err := ListRBDSnapshots(ctx, poolName, imageName)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(snapshots).To(BeEmpty())
+	})
+
+	It("should unlock the volume", func() {
+		stdout, _, err := kubectl("exec", "-n", rookNamespace, "deployment/"+finDeploymentName, "--",
+			"rbd", "-p", poolName, "--format", "json", "lock", "ls", imageName)
+		Expect(err).NotTo(HaveOccurred())
+		var locks []*model.RBDLock
+		err = json.Unmarshal(stdout, &locks)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(locks).To(HaveLen(1))
+
+		// unlock
+		_, _, err = kubectl("exec", "-n", rookNamespace, "deployment/"+finDeploymentName, "--",
+			"rbd", "-p", poolName, "lock", "rm", imageName, dummyLockID, locks[0].Locker)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("should resume backup creation and complete it", func(ctx SpecContext) {
+		_, err := WaitForFinBackupStoredToNodeAndVerified(ctx, ctrlClient, fb, 1*time.Minute)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("should not exist locks after backup completion", func() {
+		stdout, _, err := kubectl("exec", "-n", rookNamespace, "deployment/"+finDeploymentName, "--",
+			"rbd", "-p", poolName, "--format", "json", "lock", "ls", imageName)
+		Expect(err).NotTo(HaveOccurred())
+		var locks []*model.RBDLock
+		err = json.Unmarshal(stdout, &locks)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(locks).To(HaveLen(0))
+	})
+}

test/e2e/util_test.go

@@ -1,16 +1,21 @@
 package e2e
 
 import (
+	"bufio"
 	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
 	"html/template"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"regexp"
 	"time"
 
 	finv1 "github.com/cybozu-go/fin/api/v1"
+	"github.com/cybozu-go/fin/internal/controller"
+	"github.com/cybozu-go/fin/internal/model"
 	"github.com/cybozu-go/fin/test/utils"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -26,6 +31,7 @@ import (
 )
 
 const (
+	finDeploymentName      = "fin-controller-manager"
 	rookNamespace          = "rook-ceph"
 	rookStorageClass       = "rook-ceph-block"
 	poolName               = "rook-ceph-block-pool"
@@ -80,7 +86,7 @@ func checkDeploymentReady(namespace, name string) error {
 func waitEnvironment() {
 	It("wait for fin-controller to be ready", func() {
 		Eventually(func() error {
-			return checkDeploymentReady(rookNamespace, "fin-controller-manager")
+			return checkDeploymentReady(rookNamespace, finDeploymentName)
 		}).Should(Succeed())
 	})
 }
@@ -372,16 +378,31 @@ func DeleteFinRestore(ctx context.Context, client client.Client, finrestore *fin
 	return client.Delete(ctx, target)
 }
 
-func WaitForFinBackupStoredToNodeAndVerified(ctx context.Context, c client.Client, finbackup *finv1.FinBackup, timeout time.Duration) error {
-	return wait.PollUntilContextTimeout(ctx, time.Second, timeout, true, func(ctx context.Context) (bool, error) {
-		fb := &finv1.FinBackup{}
-		err := c.Get(ctx, client.ObjectKeyFromObject(finbackup), fb)
+func WaitForPVCBound(ctx context.Context, c client.Client, pvc *corev1.PersistentVolumeClaim, timeout time.Duration) (*corev1.PersistentVolumeClaim, error) {
+	GinkgoHelper()
+	currentPVC := &corev1.PersistentVolumeClaim{}
+	err := wait.PollUntilContextTimeout(ctx, time.Second, timeout, true, func(ctx context.Context) (bool, error) {
+		err := c.Get(ctx, client.ObjectKeyFromObject(pvc), currentPVC)
+		if err != nil {
+			return false, err
+		}
+		return currentPVC.Status.Phase == corev1.ClaimBound, nil
+	})
+	return currentPVC, err
+}
+
+func WaitForFinBackupStoredToNodeAndVerified(ctx context.Context, c client.Client, finbackup *finv1.FinBackup, timeout time.Duration) (*finv1.FinBackup, error) {
+	GinkgoHelper()
+	currentFB := &finv1.FinBackup{}
+	err := wait.PollUntilContextTimeout(ctx, time.Second, timeout, true, func(ctx context.Context) (bool, error) {
+		err := c.Get(ctx, client.ObjectKeyFromObject(finbackup), currentFB)
 		if err != nil {
 			return false, err
 		}
 
-		return fb.IsStoredToNode() && fb.IsVerifiedTrue(), nil
+		return currentFB.IsStoredToNode() && currentFB.IsVerifiedTrue(), nil
 	})
+	return currentFB, err
 }
 
 func WaitForFinRestoreReady(ctx context.Context, c client.Client, finrestore *finv1.FinRestore, timeout time.Duration) error {
@@ -455,6 +476,58 @@ func WaitForPVCDeletion(ctx context.Context, k8sClient kubernetes.Interface, pvc
 	})
 }
 
+// WaitControllerLog waits until the controller log matches the given pattern or the duration is exceeded.
+func WaitControllerLog(ctx SpecContext, pattern string, duration time.Duration) error {
+	GinkgoHelper()
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, duration)
+	defer cancel()
+
+	matcher := regexp.MustCompile(pattern)
+
+	command := exec.CommandContext(timeoutCtx, "kubectl", "logs", "-n", rookNamespace, "deployment/"+finDeploymentName, "-f")
+	stdoutPipe, err := command.StdoutPipe()
+	if err != nil {
+		panic(err)
+	}
+	err = command.Start()
+	if err != nil {
+		panic(err)
+	}
+	defer func() {
+		_ = command.Process.Kill()
+		_ = command.Wait()
+	}()
+
+	// read stdout line by line until the pattern is found
+	scanner := bufio.NewScanner(stdoutPipe)
+	found := make(chan struct{})
+	go func() {
+		for scanner.Scan() {
+			select {
+			case <-timeoutCtx.Done():
+				return
+			default:
+			}
+			line := scanner.Text()
+			if matcher.MatchString(line) {
+				close(found)
+				return
+			}
+		}
+		if scanner.Err() != nil {
+			panic(scanner.Err())
+		}
+	}()
+
+	select {
+	case <-timeoutCtx.Done():
+		return timeoutCtx.Err()
+	case <-found:
+		return nil
+	}
+}
+
 func VerifySizeOfRestorePVC(ctx context.Context, c client.Client, restore *finv1.FinRestore) {
 	GinkgoHelper()
 
@@ -679,6 +752,30 @@ func GetNodeNames(ctx context.Context, k8sClient kubernetes.Interface) ([]string
 	return nodeNames, nil
 }
 
+func GetPvByPvc(ctx context.Context, k8sClient kubernetes.Interface, pvc *corev1.PersistentVolumeClaim) (*corev1.PersistentVolume, error) {
+	GinkgoHelper()
+
+	return k8sClient.CoreV1().PersistentVolumes().Get(ctx, pvc.Spec.VolumeName, metav1.GetOptions{})
+}
+
+func ListRBDSnapshots(ctx context.Context, poolName, imageName string) ([]*model.RBDSnapshot, error) {
+	GinkgoHelper()
+
+	stdout, stderr, err := kubectl("exec", "-n", rookNamespace, "deploy/rook-ceph-tools", "--",
+		"rbd", "-p", poolName, "snap", "ls", imageName, "--format", "json")
+	if err != nil {
+		return nil, fmt.Errorf("failed to list RBD snapshots. stdout: %s, stderr: %s, err: %w",
+			string(stdout), string(stderr), err)
+	}
+
+	var snapshots []*model.RBDSnapshot
+	if err := json.Unmarshal(stdout, &snapshots); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal RBD snapshot list. err: %w", err)
+	}
+
+	return snapshots, nil
+}
+
 func VerifyRawImage(pvc *corev1.PersistentVolumeClaim, node string, expected []byte) {
 	GinkgoHelper()
 
@@ -706,13 +803,26 @@ func VerifyDeletionOfJobsForBackup(ctx context.Context, client kubernetes.Interf
 	Expect(err).NotTo(HaveOccurred(), "Deletion job should be deleted.")
 }
 
-func VerifyDeletionOfSnapshotInFinBackup(ctx context.Context, ctrlClient client.Client, finbackup *finv1.FinBackup) {
+func VerifyDeletionOfSnapshotInFinBackup(ctx context.Context, finbackup *finv1.FinBackup) error {
 	GinkgoHelper()
 
-	rbdImage := finbackup.Annotations["fin.cybozu.io/backup-target-rbd-image"]
-	stdout, stderr, err := kubectl("exec", "-n", rookNamespace, "deploy/rook-ceph-tools", "--",
-		"rbd", "info", fmt.Sprintf("%s/%s@fin-backup-%s", poolName, rbdImage, finbackup.UID))
-	Expect(err).To(HaveOccurred(), "Snapshot should be deleted. stdout: %s, stderr: %s", stdout, stderr)
+	imageName := finbackup.Annotations[controller.AnnotationBackupTargetRBDImage]
+	if len(imageName) == 0 {
+		return fmt.Errorf("finbackup %s/%s does not have %s annotation",
+			finbackup.Namespace, finbackup.Name, controller.AnnotationBackupTargetRBDImage)
+	}
+	snapshots, err := ListRBDSnapshots(ctx, poolName, imageName)
+	if err != nil {
+		return err
+	}
+
+	expectedSnapName := fmt.Sprintf("fin-backup-%s", finbackup.UID)
+	for _, snapshot := range snapshots {
+		if snapshot.Name == expectedSnapName {
+			return fmt.Errorf("snapshot %s still exists", expectedSnapName)
+		}
+	}
+	return nil
 }
 
 func VerifyDeletionOfResourcesForRestore(