Merge pull request #202 from cybozu-go/followup-fix-for-createmantlebackup-rpc-call

Fix bugs introduced in #194

Author: llamerada-jp

internal/controller/mantlebackup_controller.go

@@ -800,6 +800,13 @@ func (r *MantleBackupReconciler) provisionRBDSnapshot(
 		backup.Status.CreatedAt = metav1.NewTime(snapshot.Timestamp.Time)
 		backup.Status.SnapSize = &snapshot.Size
 
+		if backup.Status.TransferPartSize == nil {
+			// .status.transferPartSize isn't necessary in the standalone mode,
+			// but its value must be set before CreateMantleBackup RPC, so we
+			// set it here.
+			backup.Status.TransferPartSize = &r.backupTransferPartSize
+		}
+
 		meta.SetStatusCondition(&backup.Status.Conditions, metav1.Condition{
 			Type: mantlev1.BackupConditionReadyToUse, Status: metav1.ConditionTrue, Reason: mantlev1.ConditionReasonReadyToUseNoProblem})
 		return nil
@@ -1062,7 +1069,7 @@ func (r *MantleBackupReconciler) startExportAndUpload(
 			DiffFrom:  sourceBackupName,
 		},
 	); err != nil {
-		return ctrl.Result{}, err
+		return ctrl.Result{}, fmt.Errorf("failed to call SetSynchronizing RPC: %w", err)
 	}
 
 	largestCompletedExportPartNum, err := r.startExport(ctx, targetBackup, sourceBackupName)
@@ -1146,10 +1153,6 @@ func (r *MantleBackupReconciler) startExport(
 		return largestCompletedPartNum, nil
 	}
 
-	if err := r.addStatusTransferPartSizeIfEmpty(ctx, targetBackup); err != nil {
-		return -1, fmt.Errorf("failed to patch .status.transferPartSize: %w", err)
-	}
-
 	if ok, err := r.haveAllExportJobsCompleted(targetBackup, largestCompletedPartNum); err != nil {
 		return -1, fmt.Errorf("failed to check if all export Jobs are completed: %w", err)
 	} else if ok {
@@ -1319,21 +1322,6 @@ func (r *MantleBackupReconciler) getPartNumRangeOfExpectedRunningUploadJobs(
 	return uploadedPartNum + 1, min(exportedPartNum, uploadedPartNum+throttle), nil
 }
 
-func (r *MantleBackupReconciler) addStatusTransferPartSizeIfEmpty(ctx context.Context, backup *mantlev1.MantleBackup) error {
-	if backup.Status.TransferPartSize != nil {
-		return nil
-	}
-
-	// Use PATCH here in order not to update backup with stale values.
-	oldBackup := backup.DeepCopy()
-	backup.Status.TransferPartSize = &r.backupTransferPartSize
-	if err := r.Client.Status().Patch(ctx, backup, client.MergeFrom(oldBackup)); err != nil {
-		return fmt.Errorf("failed to patch .status.transferPartSize: %s: %w", r.backupTransferPartSize.String(), err)
-	}
-
-	return nil
-}
-
 func (r *MantleBackupReconciler) getPoolAndImageFromStatusPVManifest(backup *mantlev1.MantleBackup) (string, string, error) {
 	var pv corev1.PersistentVolume
 	if err := json.Unmarshal([]byte(backup.Status.PVManifest), &pv); err != nil {
@@ -2447,7 +2435,7 @@ func (r *MantleBackupReconciler) reconcileImportJob(
 	// Check that all import Jobs are completed
 	finalPartNum, err := r.getNumberOfParts(backup)
 	if err != nil {
-		return ctrl.Result{}, fmt.Errorf("failed to calcuate num of export data parts: %w", err)
+		return ctrl.Result{}, fmt.Errorf("failed to calculate num of export data parts: %w", err)
 	}
 	if partNum == finalPartNum {
 		// Make sure the (final) RBD snapshot is created.

internal/controller/mantlebackup_controller_test.go

@@ -1318,7 +1318,12 @@ func createMantleBackupUsingDummyPVC(ctx context.Context, name, ns string) (*man
 	return target, nil
 }
 
-func createSnapshotForMantleBackupUsingDummyPVC(ctx context.Context, cephCmd ceph.CephCmd, backup *mantlev1.MantleBackup) error {
+func createSnapshotForMantleBackupUsingDummyPVC(
+	ctx context.Context,
+	cephCmd ceph.CephCmd,
+	backup *mantlev1.MantleBackup,
+	transferPartSize resource.Quantity,
+) error {
 	snapName := util.GetUniqueName("snap")
 	if err := cephCmd.RBDSnapCreate(dummyPoolName, dummyImageName, snapName); err != nil {
 		return err
@@ -1334,21 +1339,17 @@ func createSnapshotForMantleBackupUsingDummyPVC(ctx context.Context, cephCmd cep
 	if err := updateStatus(ctx, k8sClient, backup, func() error {
 		backup.Status.SnapID = &snaps[index].Id
 		backup.Status.SnapSize = &snaps[index].Size
+		backup.Status.TransferPartSize = &transferPartSize
 		return nil
 	}); err != nil {
 		return err
 	}
 	return nil
 }
 
-func setStatusTransferPartSize(ctx context.Context, backup *mantlev1.MantleBackup) error {
-	transferPartSize := resource.MustParse("1Gi")
-	backup.Status.TransferPartSize = &transferPartSize
-	if err := k8sClient.Status().Update(ctx, backup); err != nil {
-		return err
-	}
-	return nil
-}
+var (
+	defaultTransferPartSize = resource.MustParse("1Gi")
+)
 
 var _ = Describe("export and upload", func() {
 	var mockCtrl *gomock.Controller
@@ -1368,7 +1369,7 @@ var _ = Describe("export and upload", func() {
 		target, err := createMantleBackupUsingDummyPVC(ctx, name, ns)
 		Expect(err).NotTo(HaveOccurred())
 
-		err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, target)
+		err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, target, mbr.backupTransferPartSize)
 		Expect(err).NotTo(HaveOccurred())
 
 		grpcClient.EXPECT().SetSynchronizing(gomock.Any(), gomock.Any()).
@@ -1571,25 +1572,50 @@ var _ = Describe("export and upload", func() {
 		)
 
 		It("should use the original transfer part size if it's changed", func(ctx SpecContext) {
+			// drop all items in the expire queue to avoid getting stuck
+			go func() {
+				for {
+					select {
+					case <-mbr.expireQueueCh:
+					case <-ctx.Done():
+						return
+					}
+				}
+			}()
+			mbr.managedCephClusterID = resMgr.ClusterID
+			_, pvc, err := resMgr.CreateUniquePVAndPVC(ctx, ns)
+			Expect(err).NotTo(HaveOccurred())
+			backup, err := resMgr.CreateUniqueBackupFor(ctx, pvc)
+			Expect(err).NotTo(HaveOccurred())
+
 			origSize := *resource.NewQuantity(testutil.FakeRBDSnapshotSize-1, resource.BinarySI)
-			newSize := *resource.NewQuantity(testutil.FakeRBDSnapshotSize+1, resource.BinarySI)
 			mbr.backupTransferPartSize = origSize
-			backup := createAndExportMantleBackup(ctx, mbr, "backup", ns, false, false, nil)
-			mbr.backupTransferPartSize = newSize
 
-			// mimic as if the reconciler is called at least once after setting the new transfer part size.
-			runStartExportAndUpload(ctx, backup)
+			// We use reconcileAsStandalone to focus on the part size update logic.
+			Eventually(func(g Gomega) {
+				result, err := mbr.reconcileAsStandalone(ctx, backup)
+				g.Expect(result.IsZero()).To(BeTrue())
+				g.Expect(err).NotTo(HaveOccurred())
 
-			Eventually(ctx, func(g Gomega, ctx SpecContext) {
-				err := k8sClient.Get(ctx, types.NamespacedName{Name: backup.GetName(), Namespace: backup.GetNamespace()}, backup)
-				Expect(err).NotTo(HaveOccurred())
+				err = k8sClient.Get(ctx, types.NamespacedName{Name: backup.GetName(), Namespace: backup.GetNamespace()}, backup)
+				g.Expect(err).NotTo(HaveOccurred())
+				g.Expect(backup.Status.TransferPartSize).NotTo(BeNil())
+				g.Expect(backup.Status.TransferPartSize.Equal(origSize)).To(BeTrue())
 			}).Should(Succeed())
-			Expect(backup.Status.TransferPartSize.Equal(origSize)).To(BeTrue())
 
-			completeJob(ctx, MakeExportJobName(backup, 0))
-			runStartExportAndUpload(ctx, backup)
-			completeJob(ctx, MakeExportJobName(backup, 1))
-			runStartExportAndUpload(ctx, backup)
+			newSize := *resource.NewQuantity(testutil.FakeRBDSnapshotSize+1, resource.BinarySI)
+			mbr.backupTransferPartSize = newSize
+
+			Eventually(func(g Gomega) {
+				result, err := mbr.reconcileAsStandalone(ctx, backup)
+				g.Expect(result.IsZero()).To(BeTrue())
+				g.Expect(err).NotTo(HaveOccurred())
+			}, "10s").Should(Succeed())
+
+			err = k8sClient.Get(ctx, types.NamespacedName{Name: backup.GetName(), Namespace: backup.GetNamespace()}, backup)
+			Expect(err).NotTo(HaveOccurred())
+			Expect(backup.Status.TransferPartSize).NotTo(BeNil())
+			Expect(backup.Status.TransferPartSize.Equal(origSize)).To(BeTrue())
 		})
 	})
 
@@ -1703,13 +1729,12 @@ var _ = Describe("import", func() {
 		It("should work correctly", func(ctx SpecContext) {
 			backup, err := createMantleBackupUsingDummyPVC(ctx, "target", ns)
 			Expect(err).NotTo(HaveOccurred())
-			err = setStatusTransferPartSize(ctx, backup)
-			Expect(err).NotTo(HaveOccurred())
 
-			// set .status.snapSize
+			// set .status.snapSize and .status.transferPartSize
 			err = updateStatus(ctx, k8sClient, backup, func() error {
-				var i int64 = testutil.FakeRBDSnapshotSize
-				backup.Status.SnapSize = &i
+				backup.Status.SnapSize = ptr.To(int64(testutil.FakeRBDSnapshotSize))
+				transferPartSize := resource.MustParse("1Gi")
+				backup.Status.TransferPartSize = &transferPartSize
 				return nil
 			})
 			Expect(err).NotTo(HaveOccurred())
@@ -1793,9 +1818,7 @@ var _ = Describe("import", func() {
 		backup.SetAnnotations(m)
 		err = k8sClient.Update(ctx, backup)
 		Expect(err).NotTo(HaveOccurred())
-		err = setStatusTransferPartSize(ctx, backup)
-		Expect(err).NotTo(HaveOccurred())
-		err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, backup)
+		err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, backup, defaultTransferPartSize)
 		Expect(err).NotTo(HaveOccurred())
 		return backup
 	}
@@ -1888,7 +1911,7 @@ var _ = Describe("import", func() {
 			})
 			err = k8sClient.Update(ctx, source)
 			Expect(err).NotTo(HaveOccurred())
-			err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, source)
+			err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, source, defaultTransferPartSize)
 			Expect(err).NotTo(HaveOccurred())
 
 			// Create target MantleBackup
@@ -1932,9 +1955,7 @@ var _ = Describe("import", func() {
 		It("should work correctly if deletionTimestamp is set", func(ctx SpecContext) {
 			backup, err := createMantleBackupUsingDummyPVC(ctx, "target", ns)
 			Expect(err).NotTo(HaveOccurred())
-			err = setStatusTransferPartSize(ctx, backup)
-			Expect(err).NotTo(HaveOccurred())
-			err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, backup)
+			err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, backup, defaultTransferPartSize)
 			Expect(err).NotTo(HaveOccurred())
 
 			err = k8sClient.Delete(ctx, backup)
@@ -2039,9 +2060,7 @@ var _ = Describe("import", func() {
 			})
 			err = k8sClient.Update(ctx, source)
 			Expect(err).NotTo(HaveOccurred())
-			err = setStatusTransferPartSize(ctx, source)
-			Expect(err).NotTo(HaveOccurred())
-			err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, source)
+			err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, source, defaultTransferPartSize)
 			Expect(err).NotTo(HaveOccurred())
 
 			// Create target MantleBackup
@@ -2096,9 +2115,7 @@ var _ = Describe("import", func() {
 			})
 			err = k8sClient.Update(ctx, backup)
 			Expect(err).NotTo(HaveOccurred())
-			err = setStatusTransferPartSize(ctx, backup)
-			Expect(err).NotTo(HaveOccurred())
-			err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, backup)
+			err = createSnapshotForMantleBackupUsingDummyPVC(ctx, mbr.ceph, backup, defaultTransferPartSize)
 			Expect(err).NotTo(HaveOccurred())
 
 			err = k8sClient.Delete(ctx, backup)

test/e2e/testdata/values-mantle1.yaml

@@ -6,3 +6,6 @@ image:
 controller:
   overwriteMBCSchedule: "* * * * *"
   gcInterval: 1s
+  env:
+    - name: REQUEUE_RECONCILIATION_AFTER
+      value: "10s"

test/e2e/testdata/values-mantle2.yaml

@@ -5,3 +5,6 @@ image:
 
 controller:
   gcInterval: 1s
+  env:
+    - name: REQUEUE_RECONCILIATION_AFTER
+      value: "10s"