Merge pull request #194 from cybozu-go/5672-lock-volume-2

change api CreateOrUpdateMantleBackup to CreateMantleBackup

Author: satoru-takeuchi

docs/controller-protocol.md

@@ -4,8 +4,8 @@
 ## Table of Contents
 
 - [pkg/controller/proto/controller.proto](#pkg_controller_proto_controller-proto)
-    - [CreateOrUpdateMantleBackupRequest](#proto-CreateOrUpdateMantleBackupRequest)
-    - [CreateOrUpdateMantleBackupResponse](#proto-CreateOrUpdateMantleBackupResponse)
+    - [CreateMantleBackupRequest](#proto-CreateMantleBackupRequest)
+    - [CreateMantleBackupResponse](#proto-CreateMantleBackupResponse)
     - [CreateOrUpdatePVCRequest](#proto-CreateOrUpdatePVCRequest)
     - [CreateOrUpdatePVCResponse](#proto-CreateOrUpdatePVCResponse)
     - [ListMantleBackupRequest](#proto-ListMantleBackupRequest)
@@ -26,10 +26,10 @@
 
 
 
-<a name="proto-CreateOrUpdateMantleBackupRequest"></a>
+<a name="proto-CreateMantleBackupRequest"></a>
 
-### CreateOrUpdateMantleBackupRequest
-CreateOrUpdateMantleBackupRequest is a request message for CreateOrUpdateMantleBackup RPC.
+### CreateMantleBackupRequest
+CreateMantleBackupRequest is a request message for CreateMantleBackup RPC.
 
 
 | Field | Type | Label | Description |
@@ -41,10 +41,10 @@ CreateOrUpdateMantleBackupRequest is a request message for CreateOrUpdateMantleB
 
 
 
-<a name="proto-CreateOrUpdateMantleBackupResponse"></a>
+<a name="proto-CreateMantleBackupResponse"></a>
 
-### CreateOrUpdateMantleBackupResponse
-CreateOrUpdateMantleBackupResponse is a response message for CreateOrUpdateMantleBackup RPC.
+### CreateMantleBackupResponse
+CreateMantleBackupResponse is a response message for CreateMantleBackup RPC.
 
 nothing.
 
@@ -155,7 +155,7 @@ SetSynchronizingResponse is a response message for SetSynchronize RPC.
 | Method Name | Request Type | Response Type | Description |
 | ----------- | ------------ | ------------- | ------------|
 | CreateOrUpdatePVC | [CreateOrUpdatePVCRequest](#proto-CreateOrUpdatePVCRequest) | [CreateOrUpdatePVCResponse](#proto-CreateOrUpdatePVCResponse) |  |
-| CreateOrUpdateMantleBackup | [CreateOrUpdateMantleBackupRequest](#proto-CreateOrUpdateMantleBackupRequest) | [CreateOrUpdateMantleBackupResponse](#proto-CreateOrUpdateMantleBackupResponse) |  |
+| CreateMantleBackup | [CreateMantleBackupRequest](#proto-CreateMantleBackupRequest) | [CreateMantleBackupResponse](#proto-CreateMantleBackupResponse) |  |
 | ListMantleBackup | [ListMantleBackupRequest](#proto-ListMantleBackupRequest) | [ListMantleBackupResponse](#proto-ListMantleBackupResponse) |  |
 | SetSynchronizing | [SetSynchronizingRequest](#proto-SetSynchronizingRequest) | [SetSynchronizingResponse](#proto-SetSynchronizingResponse) |  |
 

internal/controller/mantlebackup_controller.go

@@ -617,10 +617,10 @@ func (r *MantleBackupReconciler) replicateManifests(
 		return ctrl.Result{}, err
 	}
 
-	// Call CreateOrUpdateMantleBackup.
-	if _, err := client.CreateOrUpdateMantleBackup(
+	// Call CreateMantleBackup.
+	if _, err := client.CreateMantleBackup(
 		ctx,
-		&proto.CreateOrUpdateMantleBackupRequest{
+		&proto.CreateMantleBackupRequest{
 			MantleBackup: backupSentJson,
 		},
 	); err != nil {

internal/controller/mantlebackup_controller_test.go

@@ -494,21 +494,21 @@ var _ = Describe("MantleBackup controller", func() {
 					Uid: "a7c9d5e2-4b8f-4e2a-9d3f-1b6a7c8e9f2b",
 				}, nil)
 			secondaryBackups := []*mantlev1.MantleBackup{}
-			grpcClient.EXPECT().CreateOrUpdateMantleBackup(gomock.Any(), gomock.Any()).
+			grpcClient.EXPECT().CreateMantleBackup(gomock.Any(), gomock.Any()).
 				MinTimes(1).
 				DoAndReturn(
 					func(
 						ctx context.Context,
-						req *proto.CreateOrUpdateMantleBackupRequest,
+						req *proto.CreateMantleBackupRequest,
 						opts ...grpc.CallOption,
-					) (*proto.CreateOrUpdateMantleBackupResponse, error) {
+					) (*proto.CreateMantleBackupResponse, error) {
 						var secondaryBackup mantlev1.MantleBackup
 						err := json.Unmarshal(req.GetMantleBackup(), &secondaryBackup)
 						if err != nil {
 							panic(err)
 						}
 						secondaryBackups = append(secondaryBackups, &secondaryBackup)
-						return &proto.CreateOrUpdateMantleBackupResponse{}, nil
+						return &proto.CreateMantleBackupResponse{}, nil
 					})
 			grpcClient.EXPECT().ListMantleBackup(gomock.Any(), gomock.Any()).
 				MinTimes(1).
@@ -1380,7 +1380,7 @@ var _ = Describe("export and upload", func() {
 			diffFrom:                          diffFrom,
 		})
 		Expect(err).NotTo(HaveOccurred())
-		Expect(ret.Requeue).To(BeTrue())
+		Expect(ret.RequeueAfter).NotTo(BeZero())
 
 		return target
 	}
@@ -1394,7 +1394,7 @@ var _ = Describe("export and upload", func() {
 			diffFrom:                          nil,
 		})
 		Expect(err).NotTo(HaveOccurred())
-		Expect(ret.Requeue).To(BeTrue())
+		Expect(ret.RequeueAfter).NotTo(BeZero())
 	}
 
 	completeJob := func(ctx SpecContext, jobName string) {
@@ -1739,7 +1739,7 @@ var _ = Describe("import", func() {
 			// The first call to reconcileImportJob should create an import Job
 			res, err := mbr.reconcileImportJob(ctx, backup, snapshotTarget, -1)
 			Expect(err).NotTo(HaveOccurred())
-			Expect(res.Requeue).To(BeTrue())
+			Expect(res.RequeueAfter).NotTo(BeZero())
 
 			var importJob batchv1.Job
 			err = k8sClient.Get(ctx, types.NamespacedName{
@@ -1751,7 +1751,7 @@ var _ = Describe("import", func() {
 			// The successive calls should return ctrl.Result{Requeue: true} until the import Job is completed.
 			res, err = mbr.reconcileImportJob(ctx, backup, snapshotTarget, -1)
 			Expect(err).NotTo(HaveOccurred())
-			Expect(res.Requeue).To(BeTrue())
+			Expect(res.RequeueAfter).NotTo(BeZero())
 
 			// Make the import Job completed.
 			completeJob(ctx, importJob.GetNamespace(), importJob.GetName())
@@ -2177,7 +2177,7 @@ var _ = Describe("import", func() {
 
 			result, err := mbr.reconcileZeroOutJob(ctx, backup, &snapshotTarget{})
 			Expect(err).NotTo(HaveOccurred())
-			Expect(result.Requeue).To(BeFalse())
+			Expect(result.RequeueAfter).To(BeZero())
 
 			var pv corev1.PersistentVolume
 			err = k8sClient.Get(ctx, types.NamespacedName{Name: MakeZeroOutPVName(backup)}, &pv)
@@ -2258,14 +2258,14 @@ var _ = Describe("import", func() {
 			// The first call to reconcileZeroOutJob should create a PV, PVC, and Job, and requeue.
 			result, err := mbr.reconcileZeroOutJob(ctx, backup, snapshotTarget)
 			Expect(err).NotTo(HaveOccurred())
-			Expect(result.Requeue).To(BeTrue())
+			Expect(result.RequeueAfter).NotTo(BeZero())
 
 			var pv corev1.PersistentVolume
 			err = k8sClient.Get(ctx, types.NamespacedName{Name: MakeZeroOutPVName(backup), Namespace: nsController}, &pv)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(pv.GetLabels()["app.kubernetes.io/name"]).To(Equal(labelAppNameValue))
 			Expect(pv.GetLabels()["app.kubernetes.io/component"]).To(Equal(labelComponentZeroOutVolume))
-			Expect(len(pv.Spec.AccessModes)).To(Equal(1))
+			Expect(pv.Spec.AccessModes).To(HaveLen(1))
 			Expect(pv.Spec.AccessModes[0]).To(Equal(corev1.ReadWriteOnce))
 			Expect(pv.Spec.Capacity).To(Equal(pvCapacity))
 			Expect(pv.Spec.CSI.Driver).To(Equal(pvDriver))
@@ -2287,7 +2287,7 @@ var _ = Describe("import", func() {
 			Expect(pvc.GetLabels()["app.kubernetes.io/name"]).To(Equal(labelAppNameValue))
 			Expect(pvc.GetLabels()["app.kubernetes.io/component"]).To(Equal(labelComponentZeroOutVolume))
 			Expect(*pvc.Spec.StorageClassName).To(Equal(""))
-			Expect(len(pvc.Spec.AccessModes)).To(Equal(1))
+			Expect(pvc.Spec.AccessModes).To(HaveLen(1))
 			Expect(pvc.Spec.AccessModes[0]).To(Equal(corev1.ReadWriteOnce))
 			Expect(pvc.Spec.Resources).To(Equal(pvcResources))
 			Expect(*pvc.Spec.VolumeMode).To(Equal(corev1.PersistentVolumeBlock))
@@ -2299,17 +2299,17 @@ var _ = Describe("import", func() {
 			Expect(job.GetLabels()["app.kubernetes.io/name"]).To(Equal(labelAppNameValue))
 			Expect(job.GetLabels()["app.kubernetes.io/component"]).To(Equal(labelComponentZeroOutJob))
 			Expect(*job.Spec.BackoffLimit).To(Equal(int32(65535)))
-			Expect(len(job.Spec.Template.Spec.Containers)).To(Equal(1))
+			Expect(job.Spec.Template.Spec.Containers).To(HaveLen(1))
 			Expect(job.Spec.Template.Spec.Containers[0].Name).To(Equal("zeroout"))
 			Expect(*job.Spec.Template.Spec.Containers[0].SecurityContext.Privileged).To(BeTrue())
 			Expect(*job.Spec.Template.Spec.Containers[0].SecurityContext.RunAsGroup).To(Equal(int64(0)))
 			Expect(*job.Spec.Template.Spec.Containers[0].SecurityContext.RunAsUser).To(Equal(int64(0)))
 			Expect(job.Spec.Template.Spec.Containers[0].Image).To(Equal(mbr.podImage))
-			Expect(len(job.Spec.Template.Spec.Containers[0].VolumeDevices)).To(Equal(1))
+			Expect(job.Spec.Template.Spec.Containers[0].VolumeDevices).To(HaveLen(1))
 			Expect(job.Spec.Template.Spec.Containers[0].VolumeDevices[0].Name).To(Equal("zeroout-rbd"))
 			Expect(job.Spec.Template.Spec.Containers[0].VolumeDevices[0].DevicePath).To(Equal("/dev/zeroout-rbd"))
 			Expect(job.Spec.Template.Spec.RestartPolicy).To(Equal(corev1.RestartPolicyOnFailure))
-			Expect(len(job.Spec.Template.Spec.Volumes)).To(Equal(1))
+			Expect(job.Spec.Template.Spec.Volumes).To(HaveLen(1))
 			Expect(job.Spec.Template.Spec.Volumes[0]).To(Equal(corev1.Volume{
 				Name: "zeroout-rbd",
 				VolumeSource: corev1.VolumeSource{
@@ -2325,7 +2325,7 @@ var _ = Describe("import", func() {
 			// A call to reconcileZeroOutJob should NOT requeue after the Job completed
 			result, err = mbr.reconcileZeroOutJob(ctx, backup, snapshotTarget)
 			Expect(err).NotTo(HaveOccurred())
-			Expect(result.Requeue).To(BeFalse())
+			Expect(result.RequeueAfter).To(BeZero())
 		})
 	})
 

internal/controller/replication.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cybozu-go/mantle/pkg/controller/proto"
 	"google.golang.org/grpc"
 	corev1 "k8s.io/api/core/v1"
+	aerrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -109,10 +110,10 @@ func (s *SecondaryServer) CreateOrUpdatePVC(
 	return &proto.CreateOrUpdatePVCResponse{Uid: string(pvc.GetUID())}, nil
 }
 
-func (s *SecondaryServer) CreateOrUpdateMantleBackup(
+func (s *SecondaryServer) CreateMantleBackup(
 	ctx context.Context,
-	req *proto.CreateOrUpdateMantleBackupRequest,
-) (*proto.CreateOrUpdateMantleBackupResponse, error) {
+	req *proto.CreateMantleBackupRequest,
+) (*proto.CreateMantleBackupResponse, error) {
 	var backupReceived mantlev1.MantleBackup
 	if err := json.Unmarshal(req.MantleBackup, &backupReceived); err != nil {
 		return nil, err
@@ -139,62 +140,59 @@ func (s *SecondaryServer) CreateOrUpdateMantleBackup(
 			annotRemoteUID, backupReceived.GetName(), backupReceived.GetNamespace())
 	}
 
-	var backup mantlev1.MantleBackup
-	backup.SetName(backupReceived.GetName())
-	backup.SetNamespace(backupReceived.GetNamespace())
-	if _, err := ctrl.CreateOrUpdate(ctx, s.client, &backup, func() error {
-		if !backup.CreationTimestamp.IsZero() {
-			errMsg := ""
-			if backup.Labels == nil {
-				errMsg = "labels field is nil in backup"
-			} else {
-				pvcUID, ok := backup.Labels[labelLocalBackupTargetPVCUID]
-				if !ok {
-					errMsg = "label not found"
-				} else if pvcUID != pvcUIDReceived {
-					errMsg = "label not matched"
-				}
-			}
-			if errMsg != "" {
-				return fmt.Errorf("%s: %s: %s: %s",
-					errMsg, labelLocalBackupTargetPVCUID, backupReceived.GetName(), backupReceived.GetNamespace())
-			}
-
-			if backup.Annotations == nil {
-				errMsg = "annotation field is nil in backup"
-			} else {
-				remoteUID, ok := backup.Annotations[annotRemoteUID]
-				if !ok {
-					errMsg = "annotation not found in backup"
-				} else if remoteUID != remoteUIDReceived {
-					errMsg = "annotation not matched"
-				}
-			}
-			if errMsg != "" {
-				return fmt.Errorf("%s: %s: %s: %s",
-					errMsg, annotRemoteUID, backupReceived.GetName(), backupReceived.GetNamespace())
-			}
+	var backupExists mantlev1.MantleBackup
+	if err := s.client.Get(ctx, types.NamespacedName{
+		Namespace: backupReceived.GetNamespace(),
+		Name:      backupReceived.GetName(),
+	}, &backupExists); err != nil {
+		if !aerrors.IsNotFound(err) {
+			return nil, fmt.Errorf("failed to get MantleBackup: %w", err)
+		}
+		// Not found, create a new one.
+		if err := s.client.Create(ctx, &backupReceived); err != nil {
+			return nil, fmt.Errorf("failed to create MantleBackup: %w", err)
 		}
+		if err := s.client.Status().Update(ctx, &backupReceived); err != nil {
+			return nil, fmt.Errorf("failed to update status of MantleBackup: %w", err)
+		}
+		return &proto.CreateMantleBackupResponse{}, nil
+	}
 
-		backup.Finalizers = backupReceived.Finalizers
-		backup.Annotations = backupReceived.Annotations
-		backup.Labels = backupReceived.Labels
-		backup.Spec = backupReceived.Spec
-		return nil
-	}); err != nil {
-		return nil, fmt.Errorf("CreateOrUpdate failed: %w", err)
+	// Found, check it.
+	if backupExists.Labels == nil || backupExists.Annotations == nil {
+		return nil, fmt.Errorf("both labels and annotations must not be nil in the existing MantleBackup: %s/%s",
+			backupReceived.GetNamespace(), backupReceived.GetName())
+	}
+
+	if pvcUID, ok := backupExists.Labels[labelLocalBackupTargetPVCUID]; !ok {
+		return nil, fmt.Errorf("label %s not found in the existing MantleBackup: %s/%s",
+			labelLocalBackupTargetPVCUID, backupReceived.GetNamespace(), backupReceived.GetName())
+	} else if pvcUID != pvcUIDReceived {
+		return nil, fmt.Errorf("label %s not matched in the existing MantleBackup: %s/%s",
+			labelLocalBackupTargetPVCUID, backupReceived.GetNamespace(), backupReceived.GetName())
 	}
 
-	// Use Patch here because updateStatus is likely to fail due to "the object has been modified" error.
-	newBackup := backup.DeepCopy()
-	newBackup.Status.CreatedAt = backupReceived.Status.CreatedAt
-	newBackup.Status.SnapSize = backupReceived.Status.SnapSize
-	newBackup.Status.TransferPartSize = backupReceived.Status.TransferPartSize
-	if err := s.client.Status().Patch(ctx, newBackup, client.MergeFrom(&backup)); err != nil {
-		return nil, fmt.Errorf("status patch failed: %w", err)
+	if remoteUID, ok := backupExists.Annotations[annotRemoteUID]; !ok {
+		return nil, fmt.Errorf("annotation %s not found in the existing MantleBackup: %s/%s",
+			annotRemoteUID, backupReceived.GetNamespace(), backupReceived.GetName())
+	} else if remoteUID != remoteUIDReceived {
+		return nil, fmt.Errorf("annotation %s not matched in the existing MantleBackup: %s/%s",
+			annotRemoteUID, backupReceived.GetNamespace(), backupReceived.GetName())
+	}
+
+	// Update status if not set.
+	if backupExists.Status.CreatedAt.IsZero() {
+		newBackup := backupExists.DeepCopy()
+		newBackup.Status.CreatedAt = backupReceived.Status.CreatedAt
+		newBackup.Status.SnapSize = backupReceived.Status.SnapSize
+		newBackup.Status.TransferPartSize = backupReceived.Status.TransferPartSize
+		if err := s.client.Status().Patch(ctx, newBackup, client.MergeFrom(&backupExists)); err != nil {
+			return nil, fmt.Errorf("status patch failed: %w", err)
+		}
 	}
 
-	return &proto.CreateOrUpdateMantleBackupResponse{}, nil
+	// Already exists with matching pvc-uid and remote-uid, do nothing.
+	return &proto.CreateMantleBackupResponse{}, nil
 }
 
 func (s *SecondaryServer) ListMantleBackup(

internal/controller/suite_test.go

@@ -71,6 +71,9 @@ var _ = BeforeSuite(func(ctx SpecContext) {
 	Expect(err).NotTo(HaveOccurred())
 	Expect(k8sClient).NotTo(BeNil())
 
+	err = os.Setenv("REQUEUE_RECONCILIATION_AFTER", "10s")
+	Expect(err).NotTo(HaveOccurred())
+
 	By("Setup common resources")
 	resMgr, err = testutil.NewResourceManager(k8sClient)
 	Expect(err).NotTo(HaveOccurred())

internal/controller/util.go

@@ -151,12 +151,12 @@ func IsJobConditionTrue(conditions []batchv1.JobCondition, conditionType batchv1
 
 func requeueReconciliation() ctrl.Result {
 	requeueAfter := os.Getenv("REQUEUE_RECONCILIATION_AFTER")
-	if requeueAfter != "" {
-		duration, err := time.ParseDuration(requeueAfter)
-		if err != nil {
-			panic(fmt.Sprintf("set REQUEUE_RECONCILIATION_AFTER properly: %v", err))
-		}
-		return ctrl.Result{RequeueAfter: duration}
+	if len(requeueAfter) == 0 {
+		panic("You should set REQUEUE_RECONCILIATION_AFTER env var.")
+	}
+	duration, err := time.ParseDuration(requeueAfter)
+	if err != nil {
+		panic(fmt.Sprintf("Set REQUEUE_RECONCILIATION_AFTER properly: %v", err))
 	}
-	return ctrl.Result{Requeue: true}
+	return ctrl.Result{RequeueAfter: duration}
 }

internal/webhook/v1/suite_test.go

@@ -73,6 +73,9 @@ var _ = BeforeSuite(func(ctx SpecContext) {
 	Expect(err).NotTo(HaveOccurred())
 	Expect(k8sClient).NotTo(BeNil())
 
+	err = os.Setenv("REQUEUE_RECONCILIATION_AFTER", "10s")
+	Expect(err).NotTo(HaveOccurred())
+
 	By("Setup common resources")
 	resMgr, err = testutil.NewResourceManager(k8sClient)
 	Expect(err).NotTo(HaveOccurred())