implement prepareForDataSynchronization func

docs/controller-protocol.md

@@ -8,6 +8,8 @@
     - [CreateOrUpdateMantleBackupResponse](#proto-CreateOrUpdateMantleBackupResponse)
     - [CreateOrUpdatePVCRequest](#proto-CreateOrUpdatePVCRequest)
     - [CreateOrUpdatePVCResponse](#proto-CreateOrUpdatePVCResponse)
+    - [ListMantleBackupRequest](#proto-ListMantleBackupRequest)
+    - [ListMantleBackupResponse](#proto-ListMantleBackupResponse)
 
     - [MantleService](#proto-MantleService)
 
@@ -78,6 +80,37 @@ CreateOrUpdatePVCResponse is a response message for CreateOrUpdatePVC RPC.
 
 
 
+
+<a name="proto-ListMantleBackupRequest"></a>
+
+### ListMantleBackupRequest
+ListMantleBackupRequest is a request message for ListMantleBackup RPC.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| pvcUID | [string](#string) |  |  |
+| namespace | [string](#string) |  |  |
+
+
+
+
+
+
+<a name="proto-ListMantleBackupResponse"></a>
+
+### ListMantleBackupResponse
+ListMantleBackupResponse is a response message for ListMantleBackup RPC.
+
+
+| Field | Type | Label | Description |
+| ----- | ---- | ----- | ----------- |
+| mantleBackupList | [bytes](#bytes) |  |  |
+
+
+
+
+
 
 
 
@@ -94,6 +127,7 @@ CreateOrUpdatePVCResponse is a response message for CreateOrUpdatePVC RPC.
 | ----------- | ------------ | ------------- | ------------|
 | CreateOrUpdatePVC | [CreateOrUpdatePVCRequest](#proto-CreateOrUpdatePVCRequest) | [CreateOrUpdatePVCResponse](#proto-CreateOrUpdatePVCResponse) |  |
 | CreateOrUpdateMantleBackup | [CreateOrUpdateMantleBackupRequest](#proto-CreateOrUpdateMantleBackupRequest) | [CreateOrUpdateMantleBackupResponse](#proto-CreateOrUpdateMantleBackupResponse) |  |
+| ListMantleBackup | [ListMantleBackupRequest](#proto-ListMantleBackupRequest) | [ListMantleBackupResponse](#proto-ListMantleBackupResponse) |  |
 
 
 

internal/controller/internal/testutil/resources.go

@@ -199,6 +199,15 @@ func (r *ResourceManager) WaitForBackupReady(ctx context.Context, backup *mantle
 	}).WithContext(ctx).Should(Succeed())
 }
 
+func (r *ResourceManager) WaitForBackupSyncedToRemote(ctx context.Context, backup *mantlev1.MantleBackup) {
+	EventuallyWithOffset(1, func(g Gomega, ctx context.Context) {
+		err := r.client.Get(ctx, types.NamespacedName{Name: backup.Name, Namespace: backup.Namespace}, backup)
+		g.Expect(err).NotTo(HaveOccurred())
+
+		g.Expect(meta.IsStatusConditionTrue(backup.Status.Conditions, mantlev1.BackupConditionSyncedToRemote)).Should(BeTrue())
+	}).WithContext(ctx).Should(Succeed())
+}
+
 // cf. https://go.googlesource.com/proposal/+/refs/heads/master/design/43651-type-parameters.md#pointer-method-example
 type ObjectConstraint[T any] interface {
 	client.Object

internal/controller/mantlebackup_controller.go

@@ -34,8 +34,13 @@ const (
 	labelLocalBackupTargetPVCUID  = "mantle.cybozu.io/local-backup-target-pvc-uid"
 	labelRemoteBackupTargetPVCUID = "mantle.cybozu.io/remote-backup-target-pvc-uid"
 	annotRemoteUID                = "mantle.cybozu.io/remote-uid"
+	annotDiffFrom                 = "mantle.cybozu.io/diff-from"
 	annotDiffTo                   = "mantle.cybozu.io/diff-to"
 	annotRetainIfExpired          = "mantle.cybozu.io/retain-if-expired"
+	annotSyncMode                 = "mantle.cybozu.io/sync-mode"
+
+	syncModeFull        = "full"
+	syncModeIncremental = "incremental"
 )
 
 // MantleBackupReconciler reconciles a MantleBackup object
@@ -397,10 +402,14 @@ func (r *MantleBackupReconciler) replicate(
 	if err != nil || result != (ctrl.Result{}) {
 		return result, err
 	}
-	prepareResult, result, err := r.prepareForDataSynchronization(ctx, backup, r.primarySettings.Client)
-	if err != nil || result != (ctrl.Result{}) {
-		return result, err
+	prepareResult, err := r.prepareForDataSynchronization(ctx, backup, r.primarySettings.Client)
+	if err != nil {
+		return ctrl.Result{}, err
 	}
+
+	// FIXME: Delete this code after implementing export().
+	prepareResult.isSecondaryMantleBackupReadyToUse = true
+
 	if prepareResult.isSecondaryMantleBackupReadyToUse {
 		return r.primaryCleanup(ctx, backup)
 	}
@@ -602,21 +611,146 @@ func (r *MantleBackupReconciler) finalize(
 }
 
 type dataSyncPrepareResult struct {
-	isIncremental                     bool
+	isIncremental                     bool // NOTE: The value is forcibly set to false if isSecondaryMantleBackupReadyToUse is true.
 	isSecondaryMantleBackupReadyToUse bool
 	diffFrom                          *mantlev1.MantleBackup // non-nil value iff isIncremental is true.
 }
 
 func (r *MantleBackupReconciler) prepareForDataSynchronization(
-	_ context.Context,
-	_ *mantlev1.MantleBackup,
-	_ proto.MantleServiceClient,
-) (*dataSyncPrepareResult, ctrl.Result, error) { //nolint:unparam
+	ctx context.Context,
+	backup *mantlev1.MantleBackup,
+	msc proto.MantleServiceClient,
+) (*dataSyncPrepareResult, error) {
+	exportTargetPVCUID, ok := backup.GetLabels()[labelLocalBackupTargetPVCUID]
+	if !ok {
+		return nil, fmt.Errorf(`"%s" label is missing`, labelLocalBackupTargetPVCUID)
+	}
+	resp, err := msc.ListMantleBackup(
+		ctx,
+		&proto.ListMantleBackupRequest{
+			PvcUID:    exportTargetPVCUID,
+			Namespace: backup.GetNamespace(),
+		},
+	)
+	if err != nil {
+		return nil, err
+	}
+	secondaryBackups := make([]mantlev1.MantleBackup, 0)
+	err = json.Unmarshal(resp.MantleBackupList, &secondaryBackups)
+	if err != nil {
+		return nil, err
+	}
+	secondaryBackupMap := convertToMap(secondaryBackups)
+
+	isSecondaryMantleBackupReadyToUse := false
+	secondaryBackup, ok := secondaryBackupMap[backup.GetName()]
+	if !ok {
+		return nil, fmt.Errorf("secondary MantleBackup not found: %s, %s",
+			backup.GetName(), backup.GetNamespace())
+	}
+	isSecondaryMantleBackupReadyToUse = meta.IsStatusConditionTrue(
+		secondaryBackup.Status.Conditions,
+		mantlev1.BackupConditionReadyToUse,
+	)
+
+	if isSecondaryMantleBackupReadyToUse {
+		return &dataSyncPrepareResult{
+			isIncremental:                     false,
+			isSecondaryMantleBackupReadyToUse: true,
+			diffFrom:                          nil,
+		}, nil
+	}
+
+	if syncMode, ok := backup.GetAnnotations()[annotSyncMode]; ok {
+		switch syncMode {
+		case syncModeFull:
+			return &dataSyncPrepareResult{
+				isIncremental:                     false,
+				isSecondaryMantleBackupReadyToUse: isSecondaryMantleBackupReadyToUse,
+				diffFrom:                          nil,
+			}, nil
+		case syncModeIncremental:
+			diffFromName, ok := backup.GetAnnotations()[annotDiffFrom]
+			if !ok {
+				return nil, fmt.Errorf(`"%s" annotation is missing`, annotDiffFrom)
+			}
+
+			var diffFrom mantlev1.MantleBackup
+			err = r.Client.Get(ctx, types.NamespacedName{
+				Name:      diffFromName,
+				Namespace: backup.GetNamespace(),
+			}, &diffFrom)
+			if err != nil {
+				return nil, err
+			}
+
+			return &dataSyncPrepareResult{
+				isIncremental:                     true,
+				isSecondaryMantleBackupReadyToUse: isSecondaryMantleBackupReadyToUse,
+				diffFrom:                          &diffFrom,
+			}, nil
+		default:
+			return nil, fmt.Errorf("unknown sync mode: %s", syncMode)
+		}
+	}
+
+	var primaryBackupList mantlev1.MantleBackupList
+	// TODO: Perhaps, we may have to use the client without cache.
+	err = r.Client.List(ctx, &primaryBackupList, &client.ListOptions{
+		LabelSelector: labels.SelectorFromSet(map[string]string{labelLocalBackupTargetPVCUID: exportTargetPVCUID}),
+		Namespace:     backup.GetNamespace(),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	diffFrom := searchForDiffOriginMantleBackup(backup, primaryBackupList.Items, secondaryBackupMap)
+	isIncremental := (diffFrom != nil)
+
 	return &dataSyncPrepareResult{
-		isIncremental:                     false,
-		isSecondaryMantleBackupReadyToUse: true,
-		diffFrom:                          nil,
-	}, ctrl.Result{}, nil
+		isIncremental:                     isIncremental,
+		isSecondaryMantleBackupReadyToUse: isSecondaryMantleBackupReadyToUse,
+		diffFrom:                          diffFrom,
+	}, nil
+}
+
+func convertToMap(mantleBackups []mantlev1.MantleBackup) map[string]*mantlev1.MantleBackup {
+	m := make(map[string]*mantlev1.MantleBackup)
+	for _, mantleBackup := range mantleBackups {
+		mantleBackup := mantleBackup
+		m[mantleBackup.GetName()] = &mantleBackup
+	}
+	return m
+}
+
+func searchForDiffOriginMantleBackup(
+	backup *mantlev1.MantleBackup,
+	primaryBackups []mantlev1.MantleBackup,
+	secondaryBackupMap map[string]*mantlev1.MantleBackup,
+) *mantlev1.MantleBackup {
+	var diffOrigin *mantlev1.MantleBackup
+	for _, primaryBackup := range primaryBackups {
+		primaryBackup := primaryBackup
+		secondaryBackup, ok := secondaryBackupMap[primaryBackup.Name]
+		if !ok {
+			continue
+		}
+		if !meta.IsStatusConditionTrue(primaryBackup.Status.Conditions, mantlev1.BackupConditionReadyToUse) ||
+			!meta.IsStatusConditionTrue(secondaryBackup.Status.Conditions, mantlev1.BackupConditionReadyToUse) {
+			continue
+		}
+		if !primaryBackup.DeletionTimestamp.IsZero() || !secondaryBackup.DeletionTimestamp.IsZero() {
+			continue
+		}
+		if *backup.Status.SnapID <= *primaryBackup.Status.SnapID {
+			continue
+		}
+		if diffOrigin == nil || *diffOrigin.Status.SnapID < *primaryBackup.Status.SnapID {
+			diffOrigin = &primaryBackup
+		}
+	}
+
+	return diffOrigin
 }
 
 func (r *MantleBackupReconciler) export(