refactor(dataprotection): clarify population completion and PV rebind flow

Author: leon-ape

controllers/dataprotection/volumepopulator_controller.go

@@ -60,7 +60,7 @@ import (
 	viper "github.com/apecloud/kubeblocks/pkg/viperx"
 )
 
-// VolumePopulatorReconciler reconciles Backup dataSource PVCs.
+// VolumePopulatorReconciler reconciles PVCs with Backup or Restore data sources.
 type VolumePopulatorReconciler struct {
 	client.Client
 	Scheme   *runtime.Scheme
@@ -389,8 +389,7 @@ func (r *VolumePopulatorReconciler) syncPVC(reqCtx intctrlutil.RequestCtx, pvc *
 // dispatchUnboundPVC routes an unbound PVC to either Populate or ProvisionOnly.
 // When mode is RestoreData but PrepareDataBackupSets is empty, it checks
 // PostReadyBackupSets: if postReady actions exist, fall back to ProvisionOnly
-// (data arrives via postReady); if neither stage exists, fail the invalid
-// restore contract rather than silently completing with an empty PVC.
+// (data arrives via postReady); if neither stage exists, fail the restore.
 func (r *VolumePopulatorReconciler) dispatchUnboundPVC(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim, restoreCtx *pvcRestoreContext) error {
 	if restoreCtx.mode == pvcRestoreModeRestoreData {
 		if len(restoreCtx.restoreMgr.PrepareDataBackupSets) == 0 {
@@ -1190,55 +1189,17 @@ func (r *VolumePopulatorReconciler) ProvisionOnly(reqCtx intctrlutil.RequestCtx,
 func (r *VolumePopulatorReconciler) completeBoundPVCIfNeeded(reqCtx intctrlutil.RequestCtx,
 	pvc *corev1.PersistentVolumeClaim,
 	restoreCtx *pvcRestoreContext) error {
-	populateReleased := pvcPopulateReleased(pvc)
-	for i := range pvc.Status.Conditions {
-		condition := pvc.Status.Conditions[i]
-		if string(condition.Type) != appsv1.ConditionTypeRestore {
-			continue
-		}
-		if condition.Status == corev1.ConditionFalse {
-			return nil
-		}
-		break
+	if condition := findPVCConditionByType(pvc, appsv1.ConditionTypeRestore); condition != nil && condition.Status == corev1.ConditionFalse {
+		return nil
 	}
-	if !populateReleased {
-		// A target PVC can be bound elsewhere while our PV-first handoff is in
-		// progress. Recover that handoff before validation makes the failure terminal.
-		populatePVC := &corev1.PersistentVolumeClaim{}
-		populateKey := types.NamespacedName{Namespace: pvc.Namespace, Name: getPopulatePVCName(pvc.UID)}
-		if err := r.Client.Get(reqCtx.Ctx, populateKey, populatePVC); err != nil {
-			if !apierrors.IsNotFound(err) {
-				return err
-			}
-		} else if populatePVC.Spec.VolumeName != "" && populatePVC.Spec.VolumeName != pvc.Spec.VolumeName {
-			rebound, err := r.rebindPVCAndPV(reqCtx, populatePVC, pvc)
-			if err != nil {
-				return err
-			}
-			if !rebound {
-				return intctrlutil.NewRequeueError(reconcileInterval, "waiting to recover interrupted PV handoff")
-			}
-		}
-		if err := r.validateBoundTargetPV(reqCtx, pvc); err != nil {
-			return err
-		}
-		if !pvcBindingCompleted(pvc) {
-			return intctrlutil.NewRequeueError(reconcileInterval, "waiting for Kubernetes to complete target PVC binding")
-		}
-		// Release the target PVC after prepareData and PV rebind. PostReady
-		// actions may need the workload pod to start, which cannot happen while
-		// the populate PVC still owns the restored PV or while the target PVC is
-		// still marked as being populated.
-		if err := r.releasePopulateResources(reqCtx, pvc); err != nil {
-			return err
-		}
-		reason := ReasonPopulatingSucceed
-		message := "Populator finished"
-		if restoreCtx.mode == pvcRestoreModeProvisionOnly {
-			reason = ReasonPopulatingProvisioned
-			message = "PVC provisioned without data restore"
-		}
-		if err := r.updatePVCPopulatingCondition(reqCtx, pvc, reason, message); err != nil {
+	reason := ReasonPopulatingSucceed
+	message := "Populator finished"
+	if restoreCtx.mode == pvcRestoreModeProvisionOnly {
+		reason = ReasonPopulatingProvisioned
+		message = "PVC provisioned without data restore"
+	}
+	if !pvcPopulateReleased(pvc) {
+		if err := r.completePVCPopulation(reqCtx, pvc, reason, message); err != nil {
 			return err
 		}
 	}
@@ -1249,17 +1210,51 @@ func (r *VolumePopulatorReconciler) completeBoundPVCIfNeeded(reqCtx intctrlutil.
 	if !postReadyCompleted {
 		return intctrlutil.NewRequeueError(reconcileInterval, "waiting for postReady restore")
 	}
-	reason := ReasonPopulatingSucceed
-	message := "Populator finished"
-	if restoreCtx.mode == pvcRestoreModeProvisionOnly {
-		reason = ReasonPopulatingProvisioned
-		message = "PVC provisioned without data restore"
-	}
 	return r.UpdatePVCConditions(reqCtx, pvc, reason, message)
 }
 
-// validateBoundTargetPV verifies that an already-bound target PVC points to
-// the PV rebound by this populator.
+// completePVCPopulation verifies the target binding, releases helper resources,
+// and records population completion. PostReady completion is checked separately.
+func (r *VolumePopulatorReconciler) completePVCPopulation(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim, reason, message string) error {
+	if err := r.recoverInterruptedPVRebind(reqCtx, pvc); err != nil {
+		return err
+	}
+	if err := r.validateBoundTargetPV(reqCtx, pvc); err != nil {
+		return err
+	}
+	if !pvcBindingCompleted(pvc) {
+		return intctrlutil.NewRequeueError(reconcileInterval, "waiting for Kubernetes to complete target PVC binding")
+	}
+	if err := r.releasePopulateResources(reqCtx, pvc); err != nil {
+		return err
+	}
+	return r.updatePVCPopulatingCondition(reqCtx, pvc, reason, message)
+}
+
+// recoverInterruptedPVRebind reconciles the helper PV when the target PVC
+// references a different PV, before target-binding validation can report failure.
+func (r *VolumePopulatorReconciler) recoverInterruptedPVRebind(reqCtx intctrlutil.RequestCtx,
+	pvc *corev1.PersistentVolumeClaim) error {
+	populatePVC := &corev1.PersistentVolumeClaim{}
+	populateKey := types.NamespacedName{Namespace: pvc.Namespace, Name: getPopulatePVCName(pvc.UID)}
+	if err := r.Client.Get(reqCtx.Ctx, populateKey, populatePVC); err != nil {
+		return client.IgnoreNotFound(err)
+	}
+	if populatePVC.Spec.VolumeName == "" || populatePVC.Spec.VolumeName == pvc.Spec.VolumeName {
+		return nil
+	}
+	rebound, err := r.rebindPVCAndPV(reqCtx, populatePVC, pvc)
+	if err != nil {
+		return err
+	}
+	if !rebound {
+		return intctrlutil.NewRequeueError(reconcileInterval, "waiting to recover interrupted PV handoff")
+	}
+	return nil
+}
+
+// validateBoundTargetPV checks the target PV's claimRef and population-source annotation.
 func (r *VolumePopulatorReconciler) validateBoundTargetPV(reqCtx intctrlutil.RequestCtx,
 	pvc *corev1.PersistentVolumeClaim) error {
 	if pvc.Spec.VolumeName == "" {
@@ -1862,8 +1857,8 @@ func postReadyRestoreName(componentUID types.UID) string {
 	return constant.ShortenKubeName(fmt.Sprintf("restore-%s-post-ready", componentUID), constant.KubeNameMaxLength)
 }
 
-// releasePopulateResources releases only the temporary PVC and the target PVC
-// finalizer after population has succeeded.
+// releasePopulateResources deletes the helper PVC and removes the target PVC's
+// population finalizer after target binding is verified.
 func (r *VolumePopulatorReconciler) releasePopulateResources(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error {
 	if err := r.deletePopulatePVC(reqCtx, pvc); err != nil {
 		return err
@@ -2042,7 +2037,7 @@ func pvClaimRefComplete(claimRef *corev1.ObjectReference) bool {
 
 func (r *VolumePopulatorReconciler) rebindPVCAndPV(reqCtx intctrlutil.RequestCtx, populatePVC, pvc *corev1.PersistentVolumeClaim) (bool, error) {
 	if populatePVC == nil {
-		return false, intctrlutil.NewFatalError(fmt.Sprintf("populate PVC is nil for target PVC %s/%s; restoreData path entered without prepareData backup set", pvc.Namespace, pvc.Name))
+		return false, intctrlutil.NewFatalError(fmt.Sprintf("populate PVC is nil for target PVC %s/%s", pvc.Namespace, pvc.Name))
 	}
 	if pvc.Spec.DataSourceRef == nil || pvc.Spec.DataSourceRef.Name == "" {
 		return false, intctrlutil.NewFatalError(fmt.Sprintf(
@@ -2062,48 +2057,61 @@ func (r *VolumePopulatorReconciler) rebindPVCAndPV(reqCtx intctrlutil.RequestCtx
 		if !apierrors.IsNotFound(err) {
 			return false, err
 		}
-		// We'll get called again later when the PV exists
+		// Wait until the helper PV is visible in the cache.
 		return false, nil
 	}
-	// Examine the claimref for the PV and see if it's bound to the correct PVC
-	claimRef := pv.Spec.ClaimRef
-	if pvClaimRefMatchesPVC(claimRef, pvc) {
-		if pv.Annotations[AnnPopulateFrom] != pvc.Spec.DataSourceRef.Name {
-			return false, intctrlutil.NewFatalError(fmt.Sprintf(
-				"PV %s already identifies target PVC %s/%s without expected population provenance %s=%q",
-				pv.Name, pvc.Namespace, pvc.Name, AnnPopulateFrom, pvc.Spec.DataSourceRef.Name))
-		}
-		if pvc.Spec.VolumeName != "" && pvc.Spec.VolumeName != pv.Name {
-			if err := r.restoreHelperPVBinding(reqCtx, pv, populatePVC); err != nil {
-				return false, err
-			}
-			if err := r.validateBoundTargetPV(reqCtx, pvc); err != nil {
-				return false, err
-			}
-			return true, nil
-		}
-		if err := r.bindTargetPVCToPV(reqCtx, pvc, pv.Name); err != nil {
-			return false, err
-		}
-		return true, nil
+	if pvClaimRefMatchesPVC(pv.Spec.ClaimRef, pvc) {
+		err := r.resumePVRebind(reqCtx, pv, populatePVC, pvc)
+		return err == nil, err
 	}
-	if !pvClaimRefMatchesPVC(claimRef, populatePVC) {
-		if !pvClaimRefComplete(claimRef) {
-			return false, intctrlutil.NewRequeueError(reconcileInterval, fmt.Sprintf(
-				"waiting for PV %s claimRef to identify helper PVC %s/%s",
-				pv.Name, populatePVC.Namespace, populatePVC.Name))
-		}
-		return false, intctrlutil.NewFatalError(fmt.Sprintf(
-			"refusing to rebind PV %s for target PVC %s/%s: claimRef does not identify helper PVC %s/%s UID %s",
-			pv.Name, pvc.Namespace, pvc.Name, populatePVC.Namespace, populatePVC.Name, populatePVC.UID))
+	if err := validateHelperPVClaimRef(pv, populatePVC, pvc); err != nil {
+		return false, err
 	}
 	if pvc.Spec.VolumeName != "" {
-		if err := r.validateBoundTargetPV(reqCtx, pvc); err != nil {
-			return false, err
+		err := r.validateBoundTargetPV(reqCtx, pvc)
+		return err == nil, err
+	}
+	if err := r.rebindPVToTarget(reqCtx, pv, pvc); err != nil {
+		return false, err
+	}
+	err := r.bindTargetPVCToPV(reqCtx, pvc, pv.Name)
+	return err == nil, err
+}
+
+// resumePVRebind completes a PV assignment to the target PVC, or returns the PV
+// to the helper if a concurrent binding assigned a different PV to the target.
+func (r *VolumePopulatorReconciler) resumePVRebind(reqCtx intctrlutil.RequestCtx,
+	pv *corev1.PersistentVolume, populatePVC, pvc *corev1.PersistentVolumeClaim) error {
+	if pv.Annotations[AnnPopulateFrom] != pvc.Spec.DataSourceRef.Name {
+		return intctrlutil.NewFatalError(fmt.Sprintf(
+			"PV %s already identifies target PVC %s/%s without expected population provenance %s=%q",
+			pv.Name, pvc.Namespace, pvc.Name, AnnPopulateFrom, pvc.Spec.DataSourceRef.Name))
+	}
+	if pvc.Spec.VolumeName != "" && pvc.Spec.VolumeName != pv.Name {
+		if err := r.restoreHelperPVBinding(reqCtx, pv, populatePVC); err != nil {
+			return err
 		}
-		return true, nil
+		return r.validateBoundTargetPV(reqCtx, pvc)
+	}
+	return r.bindTargetPVCToPV(reqCtx, pvc, pv.Name)
+}
+
+func validateHelperPVClaimRef(pv *corev1.PersistentVolume, populatePVC, pvc *corev1.PersistentVolumeClaim) error {
+	if pvClaimRefMatchesPVC(pv.Spec.ClaimRef, populatePVC) {
+		return nil
+	}
+	if !pvClaimRefComplete(pv.Spec.ClaimRef) {
+		return intctrlutil.NewRequeueError(reconcileInterval, fmt.Sprintf(
+			"waiting for PV %s claimRef to identify helper PVC %s/%s",
+			pv.Name, populatePVC.Namespace, populatePVC.Name))
 	}
-	// Make new PV with strategic patch values to perform the PV rebind
+	return intctrlutil.NewFatalError(fmt.Sprintf(
+		"refusing to rebind PV %s for target PVC %s/%s: claimRef does not identify helper PVC %s/%s UID %s",
+		pv.Name, pvc.Namespace, pvc.Name, populatePVC.Namespace, populatePVC.Name, populatePVC.UID))
+}
+
+func (r *VolumePopulatorReconciler) rebindPVToTarget(reqCtx intctrlutil.RequestCtx,
+	pv *corev1.PersistentVolume, pvc *corev1.PersistentVolumeClaim) error {
 	patchPV := client.MergeFromWithOptions(pv.DeepCopy(), client.MergeFromWithOptimisticLock{})
 	pv.Spec.ClaimRef = &corev1.ObjectReference{
 		Namespace:       pvc.Namespace,
@@ -2115,13 +2123,7 @@ func (r *VolumePopulatorReconciler) rebindPVCAndPV(reqCtx intctrlutil.RequestCtx
 		pv.Annotations = map[string]string{}
 	}
 	pv.Annotations[AnnPopulateFrom] = pvc.Spec.DataSourceRef.Name
-	if err := r.Client.Patch(reqCtx.Ctx, pv, patchPV); err != nil {
-		return false, err
-	}
-	if err := r.bindTargetPVCToPV(reqCtx, pvc, pv.Name); err != nil {
-		return false, err
-	}
-	return true, nil
+	return r.Client.Patch(reqCtx.Ctx, pv, patchPV)
 }
 
 func (r *VolumePopulatorReconciler) restoreHelperPVBinding(reqCtx intctrlutil.RequestCtx,
@@ -2195,7 +2197,7 @@ func (r *VolumePopulatorReconciler) UpdatePVCConditions(reqCtx intctrlutil.Reque
 			continue
 		}
 		if v.Reason == ReasonPopulatingSucceed {
-			// ignore succeed condition
+			// Preserve completed population while updating the Restore condition.
 			if pvcConditionMatches(pvc.Status.Conditions, restoreCondition) {
 				return nil
 			}