implement e2e test for lock feature

Signed-off-by: Yuji Ito <llamerada.jp@gmail.com>

Author: llamerada-jp

internal/controller/mantlebackup_controller.go

@@ -1895,6 +1895,8 @@ func (r *MantleBackupReconciler) startImport(
 	backup *mantlev1.MantleBackup,
 	target *snapshotTarget,
 ) (ctrl.Result, error) {
+	logger := log.FromContext(ctx)
+
 	if !r.doesMantleBackupHaveSyncModeAnnot(backup) {
 		// SetSynchronizing is not called yet or the cache is stale.
 		return ctrl.Result{}, nil
@@ -1925,7 +1927,7 @@ func (r *MantleBackupReconciler) startImport(
 		return ctrl.Result{}, fmt.Errorf("failed to lock the volume: %w", err)
 	}
 	if !locked {
-		// another process is holding the lock
+		logger.Info("the volume is locked by another process", "uid", string(backup.GetUID()))
 		return requeueReconciliation(), nil
 	}
 

test/e2e/multik8s/lock_test.go

@@ -0,0 +1,123 @@
+package multik8s
+
+import (
+	"encoding/json"
+	"time"
+
+	mantlev1 "github.com/cybozu-go/mantle/api/v1"
+	"github.com/cybozu-go/mantle/internal/ceph"
+	"github.com/cybozu-go/mantle/internal/controller"
+	. "github.com/cybozu-go/mantle/test/e2e/multik8s/testutil"
+	"github.com/cybozu-go/mantle/test/util"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+)
+
+var _ = Describe("Locking", func() {
+	namespace := util.GetUniqueName("ns-")
+	pvcName := util.GetUniqueName("pvc-")
+	backupName0 := util.GetUniqueName("mb-")
+	backupName1 := util.GetUniqueName("mb-")
+	dummyLockID := "dummy-lock-id"
+	var controllerPod string
+	var poolName, imageName string
+
+	It("should setup environment", func(ctx SpecContext) {
+		SetupEnvironment(namespace)
+		// Create PVC and MantleBackup in the primary cluster
+		CreatePVC(ctx, PrimaryK8sCluster, namespace, pvcName)
+		CreateMantleBackup(PrimaryK8sCluster, namespace, pvcName, backupName0)
+		WaitMantleBackupSynced(namespace, backupName0)
+		var err error
+		controllerPod, err = GetControllerPodName(SecondaryK8sCluster)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("should lock the volume in the secondary cluster", func() {
+		mb0, err := GetMB(SecondaryK8sCluster, namespace, backupName0)
+		Expect(err).NotTo(HaveOccurred())
+
+		pvStored := corev1.PersistentVolume{}
+		err = json.Unmarshal([]byte(mb0.Status.PVManifest), &pvStored)
+		Expect(err).NotTo(HaveOccurred())
+		poolName = pvStored.Spec.CSI.VolumeAttributes["pool"]
+		imageName = pvStored.Spec.CSI.VolumeAttributes["imageName"]
+
+		// locked
+		_, _, err = Kubectl(SecondaryK8sCluster, nil, "exec", "-n", CephClusterNamespace, controllerPod, "--",
+			"rbd", "-p", poolName, "lock", "add", imageName, dummyLockID)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("should create additional backup and wait for the log", func(ctx SpecContext) {
+		CreateMantleBackup(PrimaryK8sCluster, namespace, pvcName, backupName1)
+		var mb1 *mantlev1.MantleBackup
+		Eventually(func() error {
+			var err error
+			mb1, err = GetMB(SecondaryK8sCluster, namespace, backupName1)
+			return err
+		}).Should(Succeed())
+
+		err := WaitControllerLog(ctx, SecondaryK8sCluster,
+			"the volume is locked by another process.*"+string(mb1.GetUID()),
+			3*time.Minute)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("checks that the jobs are not created", func(ctx SpecContext) {
+		mb1, err := GetMB(SecondaryK8sCluster, namespace, backupName1)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(
+			CheckJobExist(SecondaryK8sCluster, CephClusterNamespace, controller.MantleZeroOutJobPrefix+string(mb1.GetUID())),
+		).To(BeFalse())
+		Expect(
+			CheckJobExist(SecondaryK8sCluster, CephClusterNamespace, controller.MantleImportJobPrefix+string(mb1.GetUID())),
+		).To(BeFalse())
+	})
+
+	It("should unlock the volume in the secondary cluster", func() {
+		stdout, _, err := Kubectl(SecondaryK8sCluster, nil, "exec", "-n", CephClusterNamespace, controllerPod, "--",
+			"rbd", "-p", poolName, "--format", "json", "lock", "ls", imageName)
+		Expect(err).NotTo(HaveOccurred())
+		var locks []*ceph.RBDLock
+		err = json.Unmarshal([]byte(stdout), &locks)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(locks).To(HaveLen(1))
+
+		// unlock
+		_, _, err = Kubectl(SecondaryK8sCluster, nil, "exec", "-n", CephClusterNamespace, controllerPod, "--",
+			"rbd", "-p", poolName, "lock", "rm", imageName, dummyLockID, locks[0].Locker)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("should resume backup creation and complete it", func() {
+		// check that the lock created by Mantle exists
+		mb1, err := GetMB(SecondaryK8sCluster, namespace, backupName1)
+		Expect(err).NotTo(HaveOccurred())
+		Eventually(func() error {
+			stdout, _, err := Kubectl(SecondaryK8sCluster, nil, "exec", "-n", CephClusterNamespace, controllerPod, "--",
+				"rbd", "-p", poolName, "--format", "json", "lock", "ls", imageName)
+			Expect(err).NotTo(HaveOccurred())
+			var locks []*ceph.RBDLock
+			err = json.Unmarshal([]byte(stdout), &locks)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(locks).To(HaveLen(1))
+			Expect(locks[0].LockID).To(Equal(string(mb1.GetUID())))
+			return nil
+		}).Should(Succeed())
+
+		// wait for completion
+		WaitMantleBackupSynced(namespace, backupName1)
+	})
+
+	It("should not exist locks after backup completion", func() {
+		stdout, _, err := Kubectl(SecondaryK8sCluster, nil, "exec", "-n", CephClusterNamespace, controllerPod, "--",
+			"rbd", "-p", poolName, "--format", "json", "lock", "ls", imageName)
+		Expect(err).NotTo(HaveOccurred())
+		var locks []*ceph.RBDLock
+		err = json.Unmarshal([]byte(stdout), &locks)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(locks).To(HaveLen(0))
+	})
+})

test/e2e/multik8s/testutil/util.go

@@ -1,6 +1,7 @@
 package testutil
 
 import (
+	"bufio"
 	"bytes"
 	"context"
 	_ "embed"
@@ -9,6 +10,7 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"regexp"
 	"slices"
 	"strings"
 	"time"
@@ -764,17 +766,13 @@ func WaitUploadJobCreated(ctx SpecContext, cluster int, namespace, backupName st
 	}).Should(Succeed())
 }
 
-func WaitJobDeleted(ctx SpecContext, cluster int, namespace, jobName string) {
+func CheckJobExist(clusterNo int, namespace, jobName string) bool {
 	GinkgoHelper()
-	By("waiting for a job to be deleted")
-	Eventually(ctx, func(g Gomega) {
-		jobs, err := GetJobList(cluster, CephClusterNamespace)
-		g.Expect(err).NotTo(HaveOccurred())
-		exist := slices.ContainsFunc(jobs.Items, func(job batchv1.Job) bool {
-			return job.GetName() == jobName
-		})
-		g.Expect(exist).To(BeFalse())
-	}).Should(Succeed())
+	jobs, err := GetJobList(clusterNo, CephClusterNamespace)
+	Expect(err).NotTo(HaveOccurred())
+	return slices.ContainsFunc(jobs.Items, func(job batchv1.Job) bool {
+		return job.GetName() == jobName
+	})
 }
 
 func WaitComponentJobsDeleted(
@@ -1079,3 +1077,55 @@ func ChangeComponentJobScript(
 		}
 	}).Should(Succeed())
 }
+
+func WaitControllerLog(ctx SpecContext, clusterNo int, pattern string, duration time.Duration) error {
+	timeoutCtx, cancel := context.WithTimeout(ctx, duration)
+	defer cancel()
+
+	matcher := regexp.MustCompile(pattern)
+
+	fields, err := getKubectlInvocation(clusterNo)
+	if err != nil {
+		panic(err)
+	}
+	fields = append(fields, "logs", "-n", CephClusterNamespace, "deployment/"+MantleControllerDeployName, "-f")
+
+	command := exec.CommandContext(timeoutCtx, fields[0], fields[1:]...)
+	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
+			}
+		}
+	}()
+
+	select {
+	case <-timeoutCtx.Done():
+		return timeoutCtx.Err()
+	case <-found:
+		return nil
+	}
+}