From ceedcff3b225845a18e0e75b2f6d9507a7f6777c Mon Sep 17 00:00:00 2001 From: Leon Date: Wed, 12 Aug 2026 13:33:06 +0800 Subject: [PATCH 1/7] fix: validate restore volume binding lifecycle --- .../volumepopulator_controller.go | 199 +++++++++++- .../volumepopulator_controller_test.go | 286 ++++++++++++++++++ 2 files changed, 480 insertions(+), 5 deletions(-) diff --git a/controllers/dataprotection/volumepopulator_controller.go b/controllers/dataprotection/volumepopulator_controller.go index e847695b52b..3764cd83227 100644 --- a/controllers/dataprotection/volumepopulator_controller.go +++ b/controllers/dataprotection/volumepopulator_controller.go @@ -173,7 +173,7 @@ func (r *VolumePopulatorReconciler) syncPVC(reqCtx intctrlutil.RequestCtx, pvc * return nil } if !pvc.DeletionTimestamp.IsZero() { - return r.Cleanup(reqCtx, pvc) + return r.cleanupDeletingPVC(reqCtx, pvc) } var restoreCtx *pvcRestoreContext if pvc.Spec.DataSourceRef.Kind == dptypes.RestoreKind { @@ -188,10 +188,173 @@ func (r *VolumePopulatorReconciler) syncPVC(reqCtx intctrlutil.RequestCtx, pvc * if pvc.Spec.VolumeName == "" { return r.dispatchUnboundPVC(reqCtx, pvc, restoreCtx) } - if err = r.completeBoundPVCIfNeeded(reqCtx, pvc, restoreCtx); err != nil { + if err = r.validateBoundPVCForCompletion(reqCtx, pvc, restoreCtx); err != nil { return err } - return r.Cleanup(reqCtx, pvc) + return r.completeBoundPVCIfNeeded(reqCtx, pvc, restoreCtx) +} + +type boundPVCArtifacts struct { + targetPV *corev1.PersistentVolume + populatePVC *corev1.PersistentVolumeClaim + populatePV *corev1.PersistentVolume + executionRestore *dpv1alpha1.Restore +} + +func (a *boundPVCArtifacts) describe(pvc *corev1.PersistentVolumeClaim) string { + claimRefDescription := func(claimRef *corev1.ObjectReference) string { + if claimRef == nil { + return "" + } + return fmt.Sprintf("%s/%s uid=%s", claimRef.Namespace, claimRef.Name, claimRef.UID) + } + targetPVDescription := fmt.Sprintf("target PV %q not found", pvc.Spec.VolumeName) + if a.targetPV != nil { + targetPVDescription = fmt.Sprintf("target PV %q claimRef=%s %s=%q", a.targetPV.Name, + claimRefDescription(a.targetPV.Spec.ClaimRef), AnnPopulateFrom, a.targetPV.Annotations[AnnPopulateFrom]) + } + populatePVCName := getPopulatePVCName(pvc.UID) + populatePVCDescription := fmt.Sprintf("helper PVC %s/%s not found", pvc.Namespace, populatePVCName) + if a.populatePVC != nil { + populatePVCDescription = fmt.Sprintf("helper PVC %s/%s volumeName=%q deleting=%t", a.populatePVC.Namespace, + a.populatePVC.Name, a.populatePVC.Spec.VolumeName, !a.populatePVC.DeletionTimestamp.IsZero()) + } + populatePVDescription := "helper PV not found" + if a.populatePV != nil { + populatePVDescription = fmt.Sprintf("helper PV %q claimRef=%s %s=%q", a.populatePV.Name, + claimRefDescription(a.populatePV.Spec.ClaimRef), AnnPopulateFrom, a.populatePV.Annotations[AnnPopulateFrom]) + } + restoreDescription := fmt.Sprintf("execution Restore %s/%s not found", pvc.Namespace, populatePVCName) + if a.executionRestore != nil { + restoreDescription = fmt.Sprintf("execution Restore %s/%s phase=%q", a.executionRestore.Namespace, + a.executionRestore.Name, a.executionRestore.Status.Phase) + } + return strings.Join([]string{ + fmt.Sprintf("target PVC %s/%s volumeName=%q", pvc.Namespace, pvc.Name, pvc.Spec.VolumeName), + targetPVDescription, + populatePVCDescription, + populatePVDescription, + restoreDescription, + }, "; ") +} + +func (r *VolumePopulatorReconciler) inspectBoundPVCArtifacts(reqCtx intctrlutil.RequestCtx, + pvc *corev1.PersistentVolumeClaim, + restoreCtx *pvcRestoreContext) (*boundPVCArtifacts, error) { + artifacts := &boundPVCArtifacts{} + populatePVCName := getPopulatePVCName(pvc.UID) + inspectionError := func(operation string, err error) error { + return intctrlutil.NewRequeueError(reconcileInterval, fmt.Sprintf( + "transient error while %s; will retry without releasing restore artifacts: %v; %s", + operation, err, artifacts.describe(pvc))) + } + if restoreCtx.mode == pvcRestoreModeRestoreData { + executionRestore := &dpv1alpha1.Restore{} + if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Namespace: pvc.Namespace, Name: populatePVCName}, executionRestore); err != nil { + if !apierrors.IsNotFound(err) { + return nil, inspectionError(fmt.Sprintf("getting execution Restore %s/%s", pvc.Namespace, populatePVCName), err) + } + } else { + artifacts.executionRestore = executionRestore + } + } + + populatePVC := &corev1.PersistentVolumeClaim{} + if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Namespace: pvc.Namespace, Name: populatePVCName}, populatePVC); err != nil { + if !apierrors.IsNotFound(err) { + return nil, inspectionError(fmt.Sprintf("getting helper PVC %s/%s", pvc.Namespace, populatePVCName), err) + } + } else { + artifacts.populatePVC = populatePVC + } + + targetPV := &corev1.PersistentVolume{} + if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Name: pvc.Spec.VolumeName}, targetPV); err != nil { + if !apierrors.IsNotFound(err) { + return nil, inspectionError(fmt.Sprintf("getting target PV %q for target PVC %s/%s", + pvc.Spec.VolumeName, pvc.Namespace, pvc.Name), err) + } + } else { + artifacts.targetPV = targetPV + } + + if artifacts.populatePVC != nil && artifacts.populatePVC.Spec.VolumeName != "" { + if artifacts.targetPV != nil && artifacts.populatePVC.Spec.VolumeName == artifacts.targetPV.Name { + artifacts.populatePV = artifacts.targetPV + } else { + populatePV := &corev1.PersistentVolume{} + if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Name: artifacts.populatePVC.Spec.VolumeName}, populatePV); err != nil { + if !apierrors.IsNotFound(err) { + return nil, inspectionError(fmt.Sprintf("getting helper PV %q for helper PVC %s/%s", + artifacts.populatePVC.Spec.VolumeName, artifacts.populatePVC.Namespace, artifacts.populatePVC.Name), err) + } + } else { + artifacts.populatePV = populatePV + } + } + } + return artifacts, nil +} + +// validateBoundPVCForCompletion proves that a bound target PVC was bound by this +// populator and that its prepareData Restore has finished. volumeName alone is +// insufficient: another provisioner may have bound an empty PV first. +func (r *VolumePopulatorReconciler) validateBoundPVCForCompletion(reqCtx intctrlutil.RequestCtx, + pvc *corev1.PersistentVolumeClaim, + restoreCtx *pvcRestoreContext) error { + artifacts, err := r.inspectBoundPVCArtifacts(reqCtx, pvc, restoreCtx) + if err != nil { + return err + } + details := artifacts.describe(pvc) + + if restoreCtx.mode == pvcRestoreModeRestoreData { + if artifacts.executionRestore == nil { + return intctrlutil.NewFatalError("bound target PVC has no execution Restore; refusing to treat restore as completed: " + details) + } + switch artifacts.executionRestore.Status.Phase { + case dpv1alpha1.RestorePhaseCompleted: + // Continue with binding provenance validation. + case dpv1alpha1.RestorePhaseFailed: + return intctrlutil.NewFatalError("execution Restore failed; preserving restore artifacts: " + details) + default: + message := "waiting for execution Restore before accepting target PVC binding; preserving restore artifacts: " + details + if err := r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, message); err != nil { + return err + } + return intctrlutil.NewRequeueError(reconcileInterval, message) + } + } + + if artifacts.targetPV == nil { + message := "waiting for target PV to become observable before validating restore binding; preserving restore artifacts: " + details + if err := r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, message); err != nil { + return err + } + return intctrlutil.NewRequeueError(reconcileInterval, message) + } + if !pvClaimRefMatchesPVC(artifacts.targetPV.Spec.ClaimRef, pvc) { + return intctrlutil.NewFatalError("target PV ClaimRef does not match target PVC; preserving restore artifacts: " + details) + } + expectedSource := pvc.Spec.DataSourceRef.Name + if artifacts.targetPV.Annotations[AnnPopulateFrom] != expectedSource { + return intctrlutil.NewFatalError(fmt.Sprintf( + "target PV %s annotation %s does not identify restore source %q; preserving restore artifacts: %s", + artifacts.targetPV.Name, AnnPopulateFrom, expectedSource, details)) + } + if artifacts.populatePVC != nil { + if artifacts.populatePVC.Spec.VolumeName == "" { + message := "waiting for helper PVC binding state to catch up before releasing restore artifacts: " + details + if err := r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, message); err != nil { + return err + } + return intctrlutil.NewRequeueError(reconcileInterval, message) + } + if artifacts.populatePVC.Spec.VolumeName != artifacts.targetPV.Name { + return intctrlutil.NewFatalError("helper PVC and target PVC refer to different PVs; preserving restore artifacts: " + details) + } + } + return nil } // dispatchUnboundPVC routes an unbound PVC to either Populate or ProvisionOnly. @@ -1014,7 +1177,7 @@ func (r *VolumePopulatorReconciler) completeBoundPVCIfNeeded(reqCtx intctrlutil. // 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.Cleanup(reqCtx, pvc); err != nil { + if err := r.releasePopulateResources(reqCtx, pvc); err != nil { return err } reason := ReasonPopulatingSucceed @@ -1613,7 +1776,33 @@ func postReadyRestoreName(componentUID types.UID) string { return constant.ShortenKubeName(fmt.Sprintf("restore-%s-post-ready", componentUID), constant.KubeNameMaxLength) } -func (r *VolumePopulatorReconciler) Cleanup(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error { +// cleanupDeletingPVC stops the per-PVC execution Restore before releasing its +// helper PVC and the target finalizer. This lets the Restore controller finish +// deleting its Jobs before the target PVC disappears. +func (r *VolumePopulatorReconciler) cleanupDeletingPVC(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error { + executionRestore := &dpv1alpha1.Restore{} + key := types.NamespacedName{Namespace: pvc.Namespace, Name: getPopulatePVCName(pvc.UID)} + if err := r.Client.Get(reqCtx.Ctx, key, executionRestore); err != nil { + if !apierrors.IsNotFound(err) { + return err + } + } else if metav1.IsControlledBy(executionRestore, pvc) { + if executionRestore.DeletionTimestamp.IsZero() { + if err := r.Client.Delete(reqCtx.Ctx, executionRestore); err != nil && !apierrors.IsNotFound(err) { + return err + } + } + return intctrlutil.NewRequeueError(reconcileInterval, + fmt.Sprintf("waiting for execution Restore %s/%s to be deleted before cleaning target PVC %s/%s", + executionRestore.Namespace, executionRestore.Name, pvc.Namespace, pvc.Name)) + } + return r.releasePopulateResources(reqCtx, pvc) +} + +// releasePopulateResources is used only after binding provenance and Restore +// completion have been verified, or after cleanupDeletingPVC has stopped the +// execution Restore. +func (r *VolumePopulatorReconciler) releasePopulateResources(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error { populatePVC := &corev1.PersistentVolumeClaim{} if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Name: getPopulatePVCName(pvc.UID), Namespace: pvc.Namespace}, populatePVC); err != nil { diff --git a/controllers/dataprotection/volumepopulator_controller_test.go b/controllers/dataprotection/volumepopulator_controller_test.go index 02fd8099895..19ce8d2beef 100644 --- a/controllers/dataprotection/volumepopulator_controller_test.go +++ b/controllers/dataprotection/volumepopulator_controller_test.go @@ -22,6 +22,7 @@ package dataprotection import ( "context" "encoding/json" + "errors" "fmt" "testing" @@ -40,6 +41,7 @@ import ( "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" kbappsv1 "github.com/apecloud/kubeblocks/apis/apps/v1" dpv1alpha1 "github.com/apecloud/kubeblocks/apis/dataprotection/v1alpha1" @@ -2159,6 +2161,290 @@ func TestCompleteBoundPVCMarksRestoreSucceededAfterPostReadyCompleted(t *testing require.Equal(t, ReasonPopulatingSucceed, restoreCondition.Reason) } +func TestValidateBoundPVCRunningRestorePreservesArtifacts(t *testing.T) { + reconciler, pvc, populatePVC, _, executionRestore, job, restoreCtx := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseRunning, false, true) + + err := reconciler.validateBoundPVCForCompletion( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx) + + require.Error(t, err) + require.True(t, intctrlutil.IsRequeueError(err), err.Error()) + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) + require.NotNil(t, restoreCondition) + require.Equal(t, corev1.ConditionUnknown, restoreCondition.Status) + require.Contains(t, restoreCondition.Message, pvc.Spec.VolumeName) + require.Contains(t, restoreCondition.Message, populatePVC.Name) + require.Contains(t, restoreCondition.Message, executionRestore.Name) + require.Contains(t, restoreCondition.Message, string(dpv1alpha1.RestorePhaseRunning)) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{})) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(job), &batchv1.Job{})) +} + +func TestValidateBoundPVCCompletedRestoreReleasesOnlyProvenBinding(t *testing.T) { + reconciler, pvc, populatePVC, _, _, _, restoreCtx := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseCompleted, true, false) + + err := reconciler.validateBoundPVCForCompletion( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx) + require.NoError(t, err) + require.NoError(t, reconciler.completeBoundPVCIfNeeded( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx)) + + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + require.NotContains(t, currentPVC.Finalizers, dptypes.DataProtectionFinalizerName) + restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) + require.NotNil(t, restoreCondition) + require.Equal(t, corev1.ConditionTrue, restoreCondition.Status) + err = reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{}) + require.True(t, apierrors.IsNotFound(err), "helper PVC should be released after verified completion, got: %v", err) +} + +func TestValidateBoundPVCCompletedRestoreRejectsForeignPV(t *testing.T) { + reconciler, pvc, populatePVC, pv, _, job, restoreCtx := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseCompleted, false, true) + + err := reconciler.validateBoundPVCForCompletion( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx) + + require.Error(t, err) + require.True(t, intctrlutil.IsTargetError(err, intctrlutil.ErrorTypeFatal), err.Error()) + require.Contains(t, err.Error(), AnnPopulateFrom) + require.Contains(t, err.Error(), populatePVC.Name) + _, reconcileErr := reconciler.handleSyncPVCError( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, err) + require.NoError(t, reconcileErr) + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) + require.NotNil(t, restoreCondition) + require.Equal(t, corev1.ConditionFalse, restoreCondition.Status) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{})) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pv), &corev1.PersistentVolume{})) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(job), &batchv1.Job{})) +} + +func TestValidateBoundPVCFailedRestorePreservesArtifacts(t *testing.T) { + reconciler, pvc, populatePVC, _, executionRestore, job, restoreCtx := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseFailed, true, true) + + err := reconciler.validateBoundPVCForCompletion( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx) + + require.Error(t, err) + require.True(t, intctrlutil.IsTargetError(err, intctrlutil.ErrorTypeFatal), err.Error()) + require.Contains(t, err.Error(), executionRestore.Name) + require.Contains(t, err.Error(), string(dpv1alpha1.RestorePhaseFailed)) + _, reconcileErr := reconciler.handleSyncPVCError( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, err) + require.NoError(t, reconcileErr) + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) + require.NotNil(t, restoreCondition) + require.Equal(t, corev1.ConditionFalse, restoreCondition.Status) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{})) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(job), &batchv1.Job{})) +} + +func TestValidateBoundPVCWaitsForTargetPVObservation(t *testing.T) { + reconciler, pvc, populatePVC, pv, _, _, restoreCtx := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseCompleted, true, false) + require.NoError(t, reconciler.Client.Delete(context.Background(), pv)) + + err := reconciler.validateBoundPVCForCompletion( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx) + + require.Error(t, err) + require.True(t, intctrlutil.IsRequeueError(err), err.Error()) + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) + require.NotNil(t, restoreCondition) + require.Equal(t, corev1.ConditionUnknown, restoreCondition.Status) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{})) +} + +func TestValidateBoundPVCRequeuesAfterTransientPVReadFailure(t *testing.T) { + reconciler, pvc, populatePVC, _, executionRestore, _, restoreCtx := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseCompleted, true, false) + baseClient := reconciler.Client.(client.WithWatch) + reconciler.Client = interceptor.NewClient(baseClient, interceptor.Funcs{ + Get: func(ctx context.Context, delegated client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*corev1.PersistentVolume); ok && key.Name == pvc.Spec.VolumeName { + return errors.New("temporary PV cache read failure") + } + return delegated.Get(ctx, key, obj, opts...) + }, + }) + + err := reconciler.validateBoundPVCForCompletion( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx) + + require.Error(t, err) + require.True(t, intctrlutil.IsRequeueError(err), err.Error()) + require.Contains(t, err.Error(), pvc.Spec.VolumeName) + require.Contains(t, err.Error(), populatePVC.Name) + require.Contains(t, err.Error(), executionRestore.Name) + require.Contains(t, err.Error(), string(dpv1alpha1.RestorePhaseCompleted)) + result, reconcileErr := reconciler.handleSyncPVCError( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, err) + require.NoError(t, reconcileErr) + require.Equal(t, reconcileInterval, result.RequeueAfter) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{})) +} + +func TestValidateBoundPVCWaitsForOutOfOrderHelperPVCObservation(t *testing.T) { + reconciler, pvc, populatePVC, _, _, _, restoreCtx := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseCompleted, true, false) + populatePVC.Spec.VolumeName = "" + require.NoError(t, reconciler.Client.Update(context.Background(), populatePVC)) + + err := reconciler.validateBoundPVCForCompletion( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx) + + require.Error(t, err) + require.True(t, intctrlutil.IsRequeueError(err), err.Error()) + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) + require.NotNil(t, restoreCondition) + require.Equal(t, corev1.ConditionUnknown, restoreCondition.Status) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{})) +} + +func TestValidateBoundPVCProvisionOnlyUsesBindingProvenance(t *testing.T) { + reconciler, pvc, populatePVC, _, executionRestore, _, restoreCtx := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseRunning, true, false) + restoreCtx.mode = pvcRestoreModeProvisionOnly + require.NoError(t, reconciler.Client.Delete(context.Background(), executionRestore)) + + err := reconciler.validateBoundPVCForCompletion( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx) + require.NoError(t, err) + require.NoError(t, reconciler.completeBoundPVCIfNeeded( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx)) + + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) + require.NotNil(t, restoreCondition) + require.Equal(t, corev1.ConditionTrue, restoreCondition.Status) + require.Equal(t, ReasonPopulatingProvisioned, restoreCondition.Reason) + err = reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{}) + require.True(t, apierrors.IsNotFound(err), "provision-only helper PVC should be released, got: %v", err) +} + +func TestCleanupDeletingPVCStopsExecutionRestoreBeforeReleasingHelper(t *testing.T) { + reconciler, pvc, populatePVC, _, _, _, _ := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseRunning, false, false) + now := metav1.Now() + pvc.DeletionTimestamp = &now + + err := reconciler.cleanupDeletingPVC(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc) + require.Error(t, err) + require.True(t, intctrlutil.IsRequeueError(err), err.Error()) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{})) + + require.NoError(t, reconciler.cleanupDeletingPVC(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc)) + err = reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{}) + require.True(t, apierrors.IsNotFound(err), "helper PVC should be deleted after execution Restore, got: %v", err) + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + require.NotContains(t, currentPVC.Finalizers, dptypes.DataProtectionFinalizerName) +} + +func newBoundPVCLifecycleTest(t *testing.T, + phase dpv1alpha1.RestorePhase, + validProvenance bool, + withJob bool) (*VolumePopulatorReconciler, + *corev1.PersistentVolumeClaim, + *corev1.PersistentVolumeClaim, + *corev1.PersistentVolume, + *dpv1alpha1.Restore, + *batchv1.Job, + *pvcRestoreContext) { + t.Helper() + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, batchv1.AddToScheme(scheme)) + require.NoError(t, dpv1alpha1.AddToScheme(scheme)) + apiGroup := dptypes.DataprotectionAPIGroup + pvc := &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "data-mysql-0", + UID: "target-pvc-uid", + Finalizers: []string{dptypes.DataProtectionFinalizerName}, + }, + Spec: corev1.PersistentVolumeClaimSpec{ + VolumeName: "target-pv", + DataSourceRef: &corev1.TypedObjectReference{ + APIGroup: &apiGroup, + Kind: dptypes.BackupKind, + Name: "backup", + }, + }, + } + populatePVC := &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{Namespace: pvc.Namespace, Name: getPopulatePVCName(pvc.UID)}, + Spec: corev1.PersistentVolumeClaimSpec{VolumeName: pvc.Spec.VolumeName}, + } + pv := &corev1.PersistentVolume{ + ObjectMeta: metav1.ObjectMeta{Name: pvc.Spec.VolumeName}, + Spec: corev1.PersistentVolumeSpec{ + Capacity: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1Gi")}, + AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce}, + ClaimRef: &corev1.ObjectReference{ + Namespace: pvc.Namespace, + Name: pvc.Name, + UID: pvc.UID, + }, + }, + } + if validProvenance { + pv.Annotations = map[string]string{AnnPopulateFrom: pvc.Spec.DataSourceRef.Name} + } + controller := true + executionRestore := &dpv1alpha1.Restore{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: pvc.Namespace, + Name: getPopulatePVCName(pvc.UID), + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: "v1", + Kind: "PersistentVolumeClaim", + Name: pvc.Name, + UID: pvc.UID, + Controller: &controller, + }}, + }, + Status: dpv1alpha1.RestoreStatus{Phase: phase}, + } + job := &batchv1.Job{ObjectMeta: metav1.ObjectMeta{Namespace: pvc.Namespace, Name: "restore-job"}} + objects := []client.Object{pvc, populatePVC, pv, executionRestore} + if withJob { + objects = append(objects, job) + } + reconciler := &VolumePopulatorReconciler{ + Client: fake.NewClientBuilder().WithScheme(scheme). + WithStatusSubresource(pvc, executionRestore). + WithObjects(objects...). + Build(), + Scheme: scheme, + Recorder: record.NewFakeRecorder(10), + } + restoreMgr := dprestore.NewRestoreManager(&dpv1alpha1.Restore{ + Spec: dpv1alpha1.RestoreSpec{Backup: dpv1alpha1.BackupRef{Name: "backup", Namespace: "default"}}, + }, nil, scheme, reconciler.Client) + return reconciler, pvc, populatePVC, pv, executionRestore, job, &pvcRestoreContext{ + mode: pvcRestoreModeRestoreData, + restoreMgr: restoreMgr, + } +} + func TestEnsurePostReadyRestoreCompletedRejectsMismatchedExistingRestore(t *testing.T) { scheme := runtime.NewScheme() require.NoError(t, corev1.AddToScheme(scheme)) From b2c8900815195752b75bc4aa2d16ad3ba164c92f Mon Sep 17 00:00:00 2001 From: Leon Date: Wed, 12 Aug 2026 16:06:40 +0800 Subject: [PATCH 2/7] fix: gate workloads on verified restore --- controllers/dataprotection/types.go | 8 +- .../volumepopulator_controller.go | 54 ++++++--- .../volumepopulator_controller_test.go | 110 +++++++++++++++++- .../reconciler_instance_alignment.go | 76 ++++++++++++ .../reconciler_instance_alignment_test.go | 48 ++++++++ pkg/dataprotection/types/types.go | 26 +++++ pkg/dataprotection/types/types_test.go | 48 ++++++++ 7 files changed, 346 insertions(+), 24 deletions(-) create mode 100644 pkg/dataprotection/types/types_test.go diff --git a/controllers/dataprotection/types.go b/controllers/dataprotection/types.go index b566dbce5f8..7b79d817ce2 100644 --- a/controllers/dataprotection/types.go +++ b/controllers/dataprotection/types.go @@ -22,7 +22,7 @@ package dataprotection import ( "time" - corev1 "k8s.io/api/core/v1" + dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types" ) const ( @@ -89,10 +89,10 @@ const ( // pvc condition type and reason ReasonPopulatingFailed = "Failed" ReasonPopulatingProcessing = "Processing" - ReasonPopulatingSucceed = "Succeed" - ReasonPopulatingProvisioned = "Provisioned" + ReasonPopulatingSucceed = dptypes.ReasonPopulatingSucceed + ReasonPopulatingProvisioned = dptypes.ReasonPopulatingProvisioned - PersistentVolumeClaimPopulating corev1.PersistentVolumeClaimConditionType = "Populating" + PersistentVolumeClaimPopulating = dptypes.PersistentVolumeClaimPopulating ) var reconcileInterval = time.Second diff --git a/controllers/dataprotection/volumepopulator_controller.go b/controllers/dataprotection/volumepopulator_controller.go index 3764cd83227..7eadf459f03 100644 --- a/controllers/dataprotection/volumepopulator_controller.go +++ b/controllers/dataprotection/volumepopulator_controller.go @@ -327,28 +327,30 @@ func (r *VolumePopulatorReconciler) validateBoundPVCForCompletion(reqCtx intctrl } if artifacts.targetPV == nil { - message := "waiting for target PV to become observable before validating restore binding; preserving restore artifacts: " + details - if err := r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, message); err != nil { - return err - } - return intctrlutil.NewRequeueError(reconcileInterval, message) + return r.waitForBoundPVCObservation(reqCtx, pvc, + "target PV to become observable before validating restore binding", details) } if !pvClaimRefMatchesPVC(artifacts.targetPV.Spec.ClaimRef, pvc) { + if helperPVCBindingMayBeOutOfOrder(artifacts) { + return r.waitForBoundPVCObservation(reqCtx, pvc, + "target PV ClaimRef cache to reflect the completed helper-to-target rebind", details) + } return intctrlutil.NewFatalError("target PV ClaimRef does not match target PVC; preserving restore artifacts: " + details) } expectedSource := pvc.Spec.DataSourceRef.Name if artifacts.targetPV.Annotations[AnnPopulateFrom] != expectedSource { + if helperPVCBindingMayBeOutOfOrder(artifacts) { + return r.waitForBoundPVCObservation(reqCtx, pvc, + fmt.Sprintf("target PV annotation %s cache to reflect the completed helper-to-target rebind", AnnPopulateFrom), details) + } return intctrlutil.NewFatalError(fmt.Sprintf( "target PV %s annotation %s does not identify restore source %q; preserving restore artifacts: %s", artifacts.targetPV.Name, AnnPopulateFrom, expectedSource, details)) } if artifacts.populatePVC != nil { if artifacts.populatePVC.Spec.VolumeName == "" { - message := "waiting for helper PVC binding state to catch up before releasing restore artifacts: " + details - if err := r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, message); err != nil { - return err - } - return intctrlutil.NewRequeueError(reconcileInterval, message) + return r.waitForBoundPVCObservation(reqCtx, pvc, + "helper PVC binding state to catch up before releasing restore artifacts", details) } if artifacts.populatePVC.Spec.VolumeName != artifacts.targetPV.Name { return intctrlutil.NewFatalError("helper PVC and target PVC refer to different PVs; preserving restore artifacts: " + details) @@ -357,6 +359,22 @@ func (r *VolumePopulatorReconciler) validateBoundPVCForCompletion(reqCtx intctrl return nil } +func helperPVCBindingMayBeOutOfOrder(artifacts *boundPVCArtifacts) bool { + return artifacts.populatePVC != nil && artifacts.targetPV != nil && + (artifacts.populatePVC.Spec.VolumeName == "" || artifacts.populatePVC.Spec.VolumeName == artifacts.targetPV.Name) +} + +func (r *VolumePopulatorReconciler) waitForBoundPVCObservation(reqCtx intctrlutil.RequestCtx, + pvc *corev1.PersistentVolumeClaim, + observation, + details string) error { + message := fmt.Sprintf("waiting for %s; preserving restore artifacts: %s", observation, details) + if err := r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, message); err != nil { + return err + } + return intctrlutil.NewRequeueError(reconcileInterval, message) +} + // 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 @@ -1245,7 +1263,7 @@ func (r *VolumePopulatorReconciler) waitForSerialPredecessors(reqCtx intctrlutil if cond != nil && cond.Status == corev1.ConditionFalse { return intctrlutil.NewFatalError(fmt.Sprintf("previous restore PVC %s/%s failed: %s", item.Namespace, item.Name, cond.Message)) } - if item.Spec.VolumeName != "" { + if pvcPopulateReleased(item) { continue } if err = r.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, @@ -1550,7 +1568,7 @@ func (r *VolumePopulatorReconciler) allRestorePVCsForComponentBound(reqCtx intct if cond != nil && cond.Status == corev1.ConditionFalse { return false, intctrlutil.NewFatalError(fmt.Sprintf("restore PVC %s/%s failed: %s", item.Namespace, item.Name, cond.Message)) } - if item.Spec.VolumeName == "" { + if !pvcPopulateReleased(item) { return false, nil } } @@ -1568,7 +1586,7 @@ func (r *VolumePopulatorReconciler) allRestorePVCsForClusterBound(reqCtx intctrl if cond != nil && cond.Status == corev1.ConditionFalse { return false, intctrlutil.NewFatalError(fmt.Sprintf("restore PVC %s/%s failed: %s", item.Namespace, item.Name, cond.Message)) } - if item.Spec.VolumeName == "" { + if !pvcPopulateReleased(item) { return false, nil } } @@ -1620,9 +1638,7 @@ func findPVCConditionByType(pvc *corev1.PersistentVolumeClaim, conditionType str } func pvcPopulateReleased(pvc *corev1.PersistentVolumeClaim) bool { - cond := findPVCConditionByType(pvc, string(PersistentVolumeClaimPopulating)) - return cond != nil && cond.Status == corev1.ConditionTrue && - (cond.Reason == ReasonPopulatingSucceed || cond.Reason == ReasonPopulatingProvisioned) + return dptypes.IsPVCPopulationCompleted(pvc) } func (r *VolumePopulatorReconciler) listRestorePVCsForComponent(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) ([]corev1.PersistentVolumeClaim, error) { @@ -1786,7 +1802,7 @@ func (r *VolumePopulatorReconciler) cleanupDeletingPVC(reqCtx intctrlutil.Reques if !apierrors.IsNotFound(err) { return err } - } else if metav1.IsControlledBy(executionRestore, pvc) { + } else if hasOwnerReference(executionRestore.OwnerReferences, pvc.UID) { if executionRestore.DeletionTimestamp.IsZero() { if err := r.Client.Delete(reqCtx.Ctx, executionRestore); err != nil && !apierrors.IsNotFound(err) { return err @@ -1795,6 +1811,10 @@ func (r *VolumePopulatorReconciler) cleanupDeletingPVC(reqCtx intctrlutil.Reques return intctrlutil.NewRequeueError(reconcileInterval, fmt.Sprintf("waiting for execution Restore %s/%s to be deleted before cleaning target PVC %s/%s", executionRestore.Namespace, executionRestore.Name, pvc.Namespace, pvc.Name)) + } else { + return intctrlutil.NewFatalError(fmt.Sprintf( + "execution Restore %s/%s is not owned by deleting target PVC %s/%s uid=%s; refusing to release helper resources", + executionRestore.Namespace, executionRestore.Name, pvc.Namespace, pvc.Name, pvc.UID)) } return r.releasePopulateResources(reqCtx, pvc) } diff --git a/controllers/dataprotection/volumepopulator_controller_test.go b/controllers/dataprotection/volumepopulator_controller_test.go index 19ce8d2beef..00d0dd98b2b 100644 --- a/controllers/dataprotection/volumepopulator_controller_test.go +++ b/controllers/dataprotection/volumepopulator_controller_test.go @@ -1810,6 +1810,7 @@ func TestEnsurePostReadyRestoreCompletedDoesNotReuseStaleRestore(t *testing.T) { } pvc.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind pvc.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(pvc, ReasonPopulatingSucceed) comp := &kbappsv1.Component{ ObjectMeta: metav1.ObjectMeta{ Namespace: "default", @@ -1876,12 +1877,14 @@ func TestEnsurePostReadyRestoreCompletedUsesOneRestorePerComponent(t *testing.T) } pvc1.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind pvc1.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(pvc1, ReasonPopulatingSucceed) pvc2 := newPVCForRestoreDecision("logs", "mysql", "") pvc2.UID = types.UID("logs-pvc") pvc2.Spec.VolumeName = "logs-pv" pvc2.Spec.DataSourceRef = pvc1.Spec.DataSourceRef.DeepCopy() pvc2.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind pvc2.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(pvc2, ReasonPopulatingSucceed) comp := &kbappsv1.Component{ ObjectMeta: metav1.ObjectMeta{ Namespace: "default", @@ -2227,6 +2230,30 @@ func TestValidateBoundPVCCompletedRestoreRejectsForeignPV(t *testing.T) { require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(job), &batchv1.Job{})) } +func TestValidateBoundPVCRequeuesForOutOfOrderPVClaimRefAndAnnotation(t *testing.T) { + reconciler, pvc, populatePVC, pv, _, _, restoreCtx := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseCompleted, true, false) + pv.Spec.ClaimRef = &corev1.ObjectReference{ + Namespace: populatePVC.Namespace, + Name: populatePVC.Name, + UID: populatePVC.UID, + } + delete(pv.Annotations, AnnPopulateFrom) + require.NoError(t, reconciler.Client.Update(context.Background(), pv)) + + err := reconciler.validateBoundPVCForCompletion( + intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, restoreCtx) + + require.Error(t, err) + require.True(t, intctrlutil.IsRequeueError(err), err.Error()) + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) + require.NotNil(t, restoreCondition) + require.Equal(t, corev1.ConditionUnknown, restoreCondition.Status) + require.Contains(t, restoreCondition.Message, "ClaimRef cache") +} + func TestValidateBoundPVCFailedRestorePreservesArtifacts(t *testing.T) { reconciler, pvc, populatePVC, _, executionRestore, job, restoreCtx := newBoundPVCLifecycleTest( t, dpv1alpha1.RestorePhaseFailed, true, true) @@ -2357,6 +2384,20 @@ func TestCleanupDeletingPVCStopsExecutionRestoreBeforeReleasingHelper(t *testing require.NotContains(t, currentPVC.Finalizers, dptypes.DataProtectionFinalizerName) } +func TestCleanupDeletingPVCRejectsExecutionRestoreOwnedByAnotherPVC(t *testing.T) { + reconciler, pvc, populatePVC, _, executionRestore, _, _ := newBoundPVCLifecycleTest( + t, dpv1alpha1.RestorePhaseRunning, false, false) + executionRestore.OwnerReferences[0].UID = "another-pvc-uid" + require.NoError(t, reconciler.Client.Update(context.Background(), executionRestore)) + + err := reconciler.cleanupDeletingPVC(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc) + + require.Error(t, err) + require.True(t, intctrlutil.IsTargetError(err, intctrlutil.ErrorTypeFatal), err.Error()) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(executionRestore), &dpv1alpha1.Restore{})) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(populatePVC), &corev1.PersistentVolumeClaim{})) +} + func newBoundPVCLifecycleTest(t *testing.T, phase dpv1alpha1.RestorePhase, validProvenance bool, @@ -2393,6 +2434,9 @@ func newBoundPVCLifecycleTest(t *testing.T, ObjectMeta: metav1.ObjectMeta{Namespace: pvc.Namespace, Name: getPopulatePVCName(pvc.UID)}, Spec: corev1.PersistentVolumeClaimSpec{VolumeName: pvc.Spec.VolumeName}, } + if !validProvenance { + populatePVC.Spec.VolumeName = "restored-helper-pv" + } pv := &corev1.PersistentVolume{ ObjectMeta: metav1.ObjectMeta{Name: pvc.Spec.VolumeName}, Spec: corev1.PersistentVolumeSpec{ @@ -2408,7 +2452,6 @@ func newBoundPVCLifecycleTest(t *testing.T, if validProvenance { pv.Annotations = map[string]string{AnnPopulateFrom: pvc.Spec.DataSourceRef.Name} } - controller := true executionRestore := &dpv1alpha1.Restore{ ObjectMeta: metav1.ObjectMeta{ Namespace: pvc.Namespace, @@ -2418,7 +2461,6 @@ func newBoundPVCLifecycleTest(t *testing.T, Kind: "PersistentVolumeClaim", Name: pvc.Name, UID: pvc.UID, - Controller: &controller, }}, }, Status: dpv1alpha1.RestoreStatus{Phase: phase}, @@ -2462,6 +2504,7 @@ func TestEnsurePostReadyRestoreCompletedRejectsMismatchedExistingRestore(t *test } pvc.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind pvc.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(pvc, ReasonPopulatingSucceed) comp := &kbappsv1.Component{ ObjectMeta: metav1.ObjectMeta{ Namespace: "default", @@ -2587,11 +2630,41 @@ func TestWaitForSerialPredecessorsWaitsForEarlierUnboundPVC(t *testing.T) { require.True(t, intctrlutil.IsRequeueError(err), err.Error()) } -func TestWaitForSerialPredecessorsAllowsAfterEarlierBoundPVC(t *testing.T) { +func TestWaitForSerialPredecessorsWaitsForEarlierBoundButUnverifiedPVC(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, dpv1alpha1.AddToScheme(scheme)) + previous := newRestorePVCForSerialTest("data-target-0", "pv-0") + current := newRestorePVCForSerialTest("data-target-1", "") + backup := newBackupForRestoreDecision([]string{"data"}, nil) + reconciler := &VolumePopulatorReconciler{ + Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(previous, current, backup).Build(), + Recorder: record.NewFakeRecorder(10), + } + restoreMgr := dprestore.NewRestoreManager(&dpv1alpha1.Restore{ + Spec: dpv1alpha1.RestoreSpec{ + PrepareDataConfig: &dpv1alpha1.PrepareDataConfig{ + VolumeClaimRestorePolicy: dpv1alpha1.VolumeClaimRestorePolicySerial, + }, + }, + }, nil, scheme, reconciler.Client) + + err := reconciler.waitForSerialPredecessors(intctrlutil.RequestCtx{Ctx: context.Background()}, current, restoreMgr) + + require.Error(t, err) + require.True(t, intctrlutil.IsRequeueError(err), err.Error()) +} + +func TestWaitForSerialPredecessorsAllowsAfterEarlierValidatedPVC(t *testing.T) { scheme := runtime.NewScheme() require.NoError(t, corev1.AddToScheme(scheme)) require.NoError(t, dpv1alpha1.AddToScheme(scheme)) previous := newRestorePVCForSerialTest("data-target-0", "pv-0") + previous.Status.Conditions = append(previous.Status.Conditions, corev1.PersistentVolumeClaimCondition{ + Type: PersistentVolumeClaimPopulating, + Status: corev1.ConditionTrue, + Reason: ReasonPopulatingSucceed, + }) current := newRestorePVCForSerialTest("data-target-1", "") backup := newBackupForRestoreDecision([]string{"data"}, nil) reconciler := &VolumePopulatorReconciler{Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(previous, current, backup).Build()} @@ -2631,6 +2704,23 @@ func TestWaitForSerialPredecessorsSkipsProvisionOnlyPVC(t *testing.T) { require.NoError(t, err) } +func TestPostReadyGatesIgnoreBoundButUnverifiedPVC(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + pvc := newRestorePVCForSerialTest("data-mysql-0", "premature-empty-pv") + reconciler := &VolumePopulatorReconciler{ + Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(pvc).Build(), + } + reqCtx := intctrlutil.RequestCtx{Ctx: context.Background()} + + componentReady, err := reconciler.allRestorePVCsForComponentBound(reqCtx, pvc) + require.NoError(t, err) + require.False(t, componentReady) + clusterReady, err := reconciler.allRestorePVCsForClusterBound(reqCtx, pvc) + require.NoError(t, err) + require.False(t, clusterReady) +} + func newRestorePVCForSerialTest(name, volumeName string) *corev1.PersistentVolumeClaim { apiGroup := dptypes.DataprotectionAPIGroup return &corev1.PersistentVolumeClaim{ @@ -2661,6 +2751,14 @@ func newRestorePVCForSerialTest(name, volumeName string) *corev1.PersistentVolum } } +func markPVCPopulationCompleted(pvc *corev1.PersistentVolumeClaim, reason string) { + upsertPVCCondition(&pvc.Status.Conditions, corev1.PersistentVolumeClaimCondition{ + Type: PersistentVolumeClaimPopulating, + Status: corev1.ConditionTrue, + Reason: reason, + }) +} + func newBackupForRestoreDecision(targetVolumes []string, targets []string) *dpv1alpha1.Backup { backup := &dpv1alpha1.Backup{ ObjectMeta: metav1.ObjectMeta{ @@ -3002,6 +3100,7 @@ func TestEnsurePostReadyRestore_MultiComponent_PostReadyOnly_ShouldNotSilentlySk } pdPVC.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind pdPVC.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(pdPVC, ReasonPopulatingProvisioned) // TiKV data PVC tikvPVC := newPVCForRestoreDecision("data", "tikv", "") @@ -3014,6 +3113,7 @@ func TestEnsurePostReadyRestore_MultiComponent_PostReadyOnly_ShouldNotSilentlySk } tikvPVC.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind tikvPVC.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(tikvPVC, ReasonPopulatingProvisioned) // All three components: pd, tikv, tidb (backup target) pdComp := &kbappsv1.Component{ @@ -3157,6 +3257,7 @@ func TestEnsurePostReadyRestore_MultiComponent_PrepareDataAndPostReady_Redirects } pdPVC.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind pdPVC.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(pdPVC, ReasonPopulatingSucceed) pdComp := &kbappsv1.Component{ ObjectMeta: metav1.ObjectMeta{ @@ -3378,6 +3479,7 @@ func TestEnsurePostReadyRestore_MultiComponent_PostReadyRedirectPreservesTargetE } pdPVC.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind pdPVC.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(pdPVC, ReasonPopulatingProvisioned) pdComp := &kbappsv1.Component{ ObjectMeta: metav1.ObjectMeta{ @@ -3506,6 +3608,7 @@ func TestEnsurePostReadyRestore_MultiComponent_PostReadyOnly_TargetComponentNotY } pdPVC.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind pdPVC.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(pdPVC, ReasonPopulatingProvisioned) // Only PD and TiKV components exist — tidb NOT yet created (sequential creation) pdComp := &kbappsv1.Component{ @@ -3710,6 +3813,7 @@ func TestEnsurePostReadyRestore_MultiComponent_PostReadyOnly_TargetsSlice(t *tes } pdPVC.Annotations[constant.RestoreSourceKindAnnotationKey] = dptypes.BackupKind pdPVC.Annotations[constant.RestoreSourceNamespaceAnnotationKey] = backup.Namespace + markPVCPopulationCompleted(pdPVC, ReasonPopulatingProvisioned) pdComp := &kbappsv1.Component{ ObjectMeta: metav1.ObjectMeta{ diff --git a/pkg/controller/instanceset/reconciler_instance_alignment.go b/pkg/controller/instanceset/reconciler_instance_alignment.go index c3ee7289ea0..9c525cd61ad 100644 --- a/pkg/controller/instanceset/reconciler_instance_alignment.go +++ b/pkg/controller/instanceset/reconciler_instance_alignment.go @@ -31,6 +31,7 @@ import ( "github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx" "github.com/apecloud/kubeblocks/pkg/controller/model" intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil" + dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types" ) // instanceAlignmentReconciler is responsible for aligning the actual instances(pods) with the desired replicas specified in the spec, @@ -89,6 +90,32 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( pod, _ := object.(*corev1.Pod) oldInstanceMap[object.GetName()] = pod } + // A Pod is needed to select a node for WaitForFirstConsumer storage. Keep it + // while the restore PVC is unbound, but remove it from the desired tree as + // soon as an unverified binding is observed. It will be recreated only after + // the populator has verified prepareData/provisioning completion. + for name, pod := range oldInstanceMap { + template, desired := nameToTemplateMap[name] + if !desired { + continue + } + restoreReady, err := instanceRestoreReadyForPod(tree, name, template, its) + if err != nil { + return kubebuilderx.Continue, err + } + if restoreReady { + continue + } + if err := tree.Delete(pod); err != nil { + return kubebuilderx.Continue, err + } + oldNameSet.Delete(name) + delete(oldInstanceMap, name) + if tree.EventRecorder != nil { + tree.EventRecorder.Eventf(its, corev1.EventTypeWarning, + "Waiting for restore PVC population to complete before starting Pod %s", name) + } + } createNameSet := newNameSet.Difference(oldNameSet) deleteNameSet := oldNameSet.Difference(newNameSet) @@ -138,6 +165,21 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( if isOrderedReady && predecessor != nil && !intctrlutil.IsPodAvailable(predecessor, its.Spec.MinReadySeconds) { break } + restoreReady, err := instanceRestoreReadyForPod(tree, name, nameToTemplateMap[name], its) + if err != nil { + return kubebuilderx.Continue, err + } + if !restoreReady { + // Include the instance in PVC alignment so its restore PVCs are + // created, but do not create the Pod until prepareData has been + // verified. This prevents a prematurely bound empty PV from being + // mounted and served as a healthy database. + currentAlignedNameList = append(currentAlignedNameList, name) + if isOrderedReady { + break + } + continue + } newPod, err := buildInstancePodByTemplate(name, nameToTemplateMap[name], its, "") if err != nil { return kubebuilderx.Continue, err @@ -227,4 +269,38 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( return kubebuilderx.Continue, nil } +func instanceRestoreReadyForPod(tree *kubebuilderx.ObjectTree, + instanceName string, + template *instancetemplate.InstanceTemplateExt, + its *workloads.InstanceSet) (bool, error) { + pvcs, err := buildInstancePVCByTemplate(instanceName, template, its) + if err != nil { + return false, err + } + for _, desiredPVC := range pvcs { + if desiredPVC.Spec.DataSourceRef == nil || desiredPVC.Annotations[constant.RestoreSourceKindAnnotationKey] == "" { + continue + } + current, err := tree.Get(desiredPVC) + if err != nil { + return false, err + } + if current == nil { + // The Pod may be required for WaitForFirstConsumer storage to select + // a node. It cannot mount the restore volume before the PVC exists. + continue + } + currentPVC := current.(*corev1.PersistentVolumeClaim) + if currentPVC.Spec.VolumeName == "" { + // Keep the unscheduled Pod as the first consumer. Once the PVC is + // bound, the next PVC event will either verify it or remove the Pod. + continue + } + if !dptypes.IsPVCPopulationCompleted(currentPVC) { + return false, nil + } + } + return true, nil +} + var _ kubebuilderx.Reconciler = &instanceAlignmentReconciler{} diff --git a/pkg/controller/instanceset/reconciler_instance_alignment_test.go b/pkg/controller/instanceset/reconciler_instance_alignment_test.go index d0118434df8..89de513a334 100644 --- a/pkg/controller/instanceset/reconciler_instance_alignment_test.go +++ b/pkg/controller/instanceset/reconciler_instance_alignment_test.go @@ -32,8 +32,10 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" workloads "github.com/apecloud/kubeblocks/apis/workloads/v1" + "github.com/apecloud/kubeblocks/pkg/constant" "github.com/apecloud/kubeblocks/pkg/controller/builder" "github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx" + dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types" ) var _ = Describe("replicas alignment reconciler test", func() { @@ -181,5 +183,51 @@ var _ = Describe("replicas alignment reconciler test", func() { } } }) + + It("removes Pods bound to restore PVCs until population is verified", func() { + replicas := int32(1) + its.Spec.Replicas = &replicas + its.Spec.VolumeClaimTemplates = []corev1.PersistentVolumeClaim{ + *its.Spec.VolumeClaimTemplates[0].DeepCopy(), + } + its.Spec.VolumeClaimTemplates[0].Annotations = map[string]string{ + constant.RestoreSourceKindAnnotationKey: "Backup", + } + apiGroup := "dataprotection.kubeblocks.io" + its.Spec.VolumeClaimTemplates[0].Spec.DataSourceRef = &corev1.TypedObjectReference{ + APIGroup: &apiGroup, + Kind: "Backup", + Name: "backup", + } + tree := kubebuilderx.NewObjectTree() + tree.SetRoot(its) + reconciler = NewReplicasAlignmentReconciler() + + By("creating the restore PVC and a Pod for WaitForFirstConsumer scheduling") + res, err := reconciler.Reconcile(tree) + Expect(err).Should(BeNil()) + Expect(res).Should(Equal(kubebuilderx.Continue)) + Expect(tree.List(&corev1.PersistentVolumeClaim{})).Should(HaveLen(1)) + Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) + + By("removing the Pod when its PVC is prematurely bound while prepareData is processing") + pvc := tree.List(&corev1.PersistentVolumeClaim{})[0].(*corev1.PersistentVolumeClaim) + pvc.Spec.VolumeName = "empty-pv" + pvc.Status.Conditions = []corev1.PersistentVolumeClaimCondition{{ + Type: dptypes.PersistentVolumeClaimPopulating, + Status: corev1.ConditionTrue, + Reason: "Processing", + }} + _, err = reconciler.Reconcile(tree) + Expect(err).Should(BeNil()) + Expect(tree.List(&corev1.Pod{})).Should(BeEmpty()) + + By("creating the Pod only after population completion is verified") + pvc = tree.List(&corev1.PersistentVolumeClaim{})[0].(*corev1.PersistentVolumeClaim) + pvc.Status.Conditions[0].Reason = dptypes.ReasonPopulatingSucceed + _, err = reconciler.Reconcile(tree) + Expect(err).Should(BeNil()) + Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) + }) }) }) diff --git a/pkg/dataprotection/types/types.go b/pkg/dataprotection/types/types.go index 3451155d990..5463f0fe5b5 100644 --- a/pkg/dataprotection/types/types.go +++ b/pkg/dataprotection/types/types.go @@ -19,7 +19,33 @@ along with this program. If not, see . package types +import corev1 "k8s.io/api/core/v1" + +const ( + PersistentVolumeClaimPopulating corev1.PersistentVolumeClaimConditionType = "Populating" + ReasonPopulatingSucceed = "Succeed" + ReasonPopulatingProvisioned = "Provisioned" +) + var ( // DefaultBackOffLimit is the default backoff limit for jobs. DefaultBackOffLimit = int32(2) ) + +// IsPVCPopulationCompleted reports whether the volume populator has verified +// the prepareData/provisioning result and released the helper PVC. A bound PVC +// alone is not sufficient because it may have been bound before restore data +// was prepared. +func IsPVCPopulationCompleted(pvc *corev1.PersistentVolumeClaim) bool { + if pvc == nil { + return false + } + for i := range pvc.Status.Conditions { + condition := &pvc.Status.Conditions[i] + if condition.Type != PersistentVolumeClaimPopulating || condition.Status != corev1.ConditionTrue { + continue + } + return condition.Reason == ReasonPopulatingSucceed || condition.Reason == ReasonPopulatingProvisioned + } + return false +} diff --git a/pkg/dataprotection/types/types_test.go b/pkg/dataprotection/types/types_test.go new file mode 100644 index 00000000000..777b92b2bb3 --- /dev/null +++ b/pkg/dataprotection/types/types_test.go @@ -0,0 +1,48 @@ +/* +Copyright (C) 2022-2026 ApeCloud Co., Ltd + +This file is part of KubeBlocks project + +This program is free software: you can redistribute it and/or modify +it under the terms of the GNU Affero General Public License as published by +the Free Software Foundation, either version 3 of the License, or +(at your option) any later version. + +This program is distributed in the hope that it will be useful, +but WITHOUT ANY WARRANTY; without even the implied warranty of +MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License +along with this program. If not, see . +*/ + +package types + +import ( + "testing" + + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" +) + +func TestIsPVCPopulationCompleted(t *testing.T) { + pvc := &corev1.PersistentVolumeClaim{} + require.False(t, IsPVCPopulationCompleted(nil)) + require.False(t, IsPVCPopulationCompleted(pvc)) + + pvc.Spec.VolumeName = "bound-before-restore" + pvc.Status.Conditions = []corev1.PersistentVolumeClaimCondition{{ + Type: PersistentVolumeClaimPopulating, + Status: corev1.ConditionTrue, + Reason: "Processing", + }} + require.False(t, IsPVCPopulationCompleted(pvc)) + + pvc.Status.Conditions[0].Reason = ReasonPopulatingSucceed + require.True(t, IsPVCPopulationCompleted(pvc)) + pvc.Status.Conditions[0].Reason = ReasonPopulatingProvisioned + require.True(t, IsPVCPopulationCompleted(pvc)) + pvc.Status.Conditions[0].Status = corev1.ConditionFalse + require.False(t, IsPVCPopulationCompleted(pvc)) +} From c001d007c64d580eb0fe789210d39a1fcdf66e3a Mon Sep 17 00:00:00 2001 From: Leon Date: Wed, 12 Aug 2026 16:13:45 +0800 Subject: [PATCH 3/7] fix: bootstrap restore pods safely --- .../reconciler_instance_alignment.go | 75 +++++++++++++------ .../reconciler_instance_alignment_test.go | 34 +++++++-- 2 files changed, 80 insertions(+), 29 deletions(-) diff --git a/pkg/controller/instanceset/reconciler_instance_alignment.go b/pkg/controller/instanceset/reconciler_instance_alignment.go index 9c525cd61ad..6c272d4e547 100644 --- a/pkg/controller/instanceset/reconciler_instance_alignment.go +++ b/pkg/controller/instanceset/reconciler_instance_alignment.go @@ -20,6 +20,7 @@ along with this program. If not, see . package instanceset import ( + "github.com/spf13/viper" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/util/sets" @@ -41,6 +42,12 @@ import ( // TODO(free6om): support membership reconfiguration type instanceAlignmentReconciler struct{} +const ( + restoreBootstrapAnnotationKey = "workloads.kubeblocks.io/restore-bootstrap" + restoreBootstrapContainerName = "restore-bootstrap" + restoreBootstrapReadinessGate = corev1.PodConditionType("workloads.kubeblocks.io/restore-ready") +) + func NewReplicasAlignmentReconciler() kubebuilderx.Reconciler { return &instanceAlignmentReconciler{} } @@ -90,9 +97,10 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( pod, _ := object.(*corev1.Pod) oldInstanceMap[object.GetName()] = pod } - // A Pod is needed to select a node for WaitForFirstConsumer storage. Keep it - // while the restore PVC is unbound, but remove it from the desired tree as - // soon as an unverified binding is observed. It will be recreated only after + replacingNameSet := sets.New[string]() + // A Pod is needed to select a node for WaitForFirstConsumer storage. Before + // restore completion, use a bootstrap Pod whose init container prevents the + // database containers from starting. Replace it with the real Pod only after // the populator has verified prepareData/provisioning completion. for name, pod := range oldInstanceMap { template, desired := nameToTemplateMap[name] @@ -103,7 +111,8 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( if err != nil { return kubebuilderx.Continue, err } - if restoreReady { + bootstrap := pod.Annotations[restoreBootstrapAnnotationKey] == "true" + if bootstrap == !restoreReady { continue } if err := tree.Delete(pod); err != nil { @@ -111,9 +120,10 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( } oldNameSet.Delete(name) delete(oldInstanceMap, name) + replacingNameSet.Insert(name) if tree.EventRecorder != nil { - tree.EventRecorder.Eventf(its, corev1.EventTypeWarning, - "Waiting for restore PVC population to complete before starting Pod %s", name) + tree.EventRecorder.Eventf(its, corev1.EventTypeNormal, + "Replacing restore bootstrap Pod %s after PVC population state changed", name) } } createNameSet := newNameSet.Difference(oldNameSet) @@ -161,6 +171,15 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( if !isOrderedReady && concurrency <= 0 { break } + if replacingNameSet.Has(name) { + // Wait until the old Pod has actually disappeared before creating a + // replacement whose immutable Pod spec differs. + currentAlignedNameList = append(currentAlignedNameList, name) + if isOrderedReady { + break + } + continue + } predecessor := getPredecessor(i) if isOrderedReady && predecessor != nil && !intctrlutil.IsPodAvailable(predecessor, its.Spec.MinReadySeconds) { break @@ -169,21 +188,13 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( if err != nil { return kubebuilderx.Continue, err } - if !restoreReady { - // Include the instance in PVC alignment so its restore PVCs are - // created, but do not create the Pod until prepareData has been - // verified. This prevents a prematurely bound empty PV from being - // mounted and served as a healthy database. - currentAlignedNameList = append(currentAlignedNameList, name) - if isOrderedReady { - break - } - continue - } newPod, err := buildInstancePodByTemplate(name, nameToTemplateMap[name], its, "") if err != nil { return kubebuilderx.Continue, err } + if !restoreReady { + makeRestoreBootstrapPod(newPod) + } if err := tree.Add(newPod); err != nil { return kubebuilderx.Continue, err @@ -286,15 +297,11 @@ func instanceRestoreReadyForPod(tree *kubebuilderx.ObjectTree, return false, err } if current == nil { - // The Pod may be required for WaitForFirstConsumer storage to select - // a node. It cannot mount the restore volume before the PVC exists. - continue + return false, nil } currentPVC := current.(*corev1.PersistentVolumeClaim) if currentPVC.Spec.VolumeName == "" { - // Keep the unscheduled Pod as the first consumer. Once the PVC is - // bound, the next PVC event will either verify it or remove the Pod. - continue + return false, nil } if !dptypes.IsPVCPopulationCompleted(currentPVC) { return false, nil @@ -303,4 +310,26 @@ func instanceRestoreReadyForPod(tree *kubebuilderx.ObjectTree, return true, nil } +func makeRestoreBootstrapPod(pod *corev1.Pod) { + if pod.Annotations == nil { + pod.Annotations = map[string]string{} + } + pod.Annotations[restoreBootstrapAnnotationKey] = "true" + image := viper.GetString(constant.KBToolsImage) + if image == "" && len(pod.Spec.Containers) > 0 { + // Keep the Pod valid if the tools image configuration is missing. The + // command is intentionally blocking (or repeatedly failing if the image + // lacks a shell), so application containers still cannot start. + image = pod.Spec.Containers[0].Image + } + pod.Spec.InitContainers = append(pod.Spec.InitContainers, corev1.Container{ + Name: restoreBootstrapContainerName, + Image: image, + Command: []string{"sh", "-c", "while true; do sleep 3600; done"}, + }) + pod.Spec.ReadinessGates = append(pod.Spec.ReadinessGates, corev1.PodReadinessGate{ + ConditionType: restoreBootstrapReadinessGate, + }) +} + var _ kubebuilderx.Reconciler = &instanceAlignmentReconciler{} diff --git a/pkg/controller/instanceset/reconciler_instance_alignment_test.go b/pkg/controller/instanceset/reconciler_instance_alignment_test.go index 89de513a334..fb6140fe3d2 100644 --- a/pkg/controller/instanceset/reconciler_instance_alignment_test.go +++ b/pkg/controller/instanceset/reconciler_instance_alignment_test.go @@ -25,6 +25,7 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + "github.com/spf13/viper" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" @@ -184,7 +185,10 @@ var _ = Describe("replicas alignment reconciler test", func() { } }) - It("removes Pods bound to restore PVCs until population is verified", func() { + It("uses a scheduling bootstrap Pod until restore population is verified", func() { + oldToolsImage := viper.GetString(constant.KBToolsImage) + DeferCleanup(func() { viper.Set(constant.KBToolsImage, oldToolsImage) }) + viper.Set(constant.KBToolsImage, "kubeblocks-tools:test") replicas := int32(1) its.Spec.Replicas = &replicas its.Spec.VolumeClaimTemplates = []corev1.PersistentVolumeClaim{ @@ -203,14 +207,22 @@ var _ = Describe("replicas alignment reconciler test", func() { tree.SetRoot(its) reconciler = NewReplicasAlignmentReconciler() - By("creating the restore PVC and a Pod for WaitForFirstConsumer scheduling") + By("creating the restore PVC and a blocked Pod for WaitForFirstConsumer scheduling") res, err := reconciler.Reconcile(tree) Expect(err).Should(BeNil()) Expect(res).Should(Equal(kubebuilderx.Continue)) Expect(tree.List(&corev1.PersistentVolumeClaim{})).Should(HaveLen(1)) Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) - - By("removing the Pod when its PVC is prematurely bound while prepareData is processing") + bootstrapPod := tree.List(&corev1.Pod{})[0].(*corev1.Pod) + Expect(bootstrapPod.Annotations).Should(HaveKeyWithValue(restoreBootstrapAnnotationKey, "true")) + Expect(bootstrapPod.Spec.InitContainers).Should(ContainElement(Satisfy(func(container corev1.Container) bool { + return container.Name == restoreBootstrapContainerName && container.Image == "kubeblocks-tools:test" + }))) + Expect(bootstrapPod.Spec.ReadinessGates).Should(ContainElement(corev1.PodReadinessGate{ + ConditionType: restoreBootstrapReadinessGate, + })) + + By("keeping the database blocked when its PVC binds while prepareData is processing") pvc := tree.List(&corev1.PersistentVolumeClaim{})[0].(*corev1.PersistentVolumeClaim) pvc.Spec.VolumeName = "empty-pv" pvc.Status.Conditions = []corev1.PersistentVolumeClaimCondition{{ @@ -220,14 +232,24 @@ var _ = Describe("replicas alignment reconciler test", func() { }} _, err = reconciler.Reconcile(tree) Expect(err).Should(BeNil()) - Expect(tree.List(&corev1.Pod{})).Should(BeEmpty()) + Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) - By("creating the Pod only after population completion is verified") + By("deleting the bootstrap Pod after population completion is verified") pvc = tree.List(&corev1.PersistentVolumeClaim{})[0].(*corev1.PersistentVolumeClaim) pvc.Status.Conditions[0].Reason = dptypes.ReasonPopulatingSucceed _, err = reconciler.Reconcile(tree) Expect(err).Should(BeNil()) + Expect(tree.List(&corev1.Pod{})).Should(BeEmpty()) + + By("creating the real Pod after the bootstrap Pod is gone") + _, err = reconciler.Reconcile(tree) + Expect(err).Should(BeNil()) Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) + realPod := tree.List(&corev1.Pod{})[0].(*corev1.Pod) + Expect(realPod.Annotations).ShouldNot(HaveKey(restoreBootstrapAnnotationKey)) + Expect(realPod.Spec.InitContainers).ShouldNot(ContainElement(Satisfy(func(container corev1.Container) bool { + return container.Name == restoreBootstrapContainerName + }))) }) }) }) From 3103eb614e64b14a53c38bad3b4d3dc02021c0a7 Mon Sep 17 00:00:00 2001 From: Leon Date: Wed, 12 Aug 2026 17:00:26 +0800 Subject: [PATCH 4/7] fix: decouple restore scheduling from workloads --- .../volumepopulator_controller.go | 161 +++++++++++++++++- .../volumepopulator_controller_test.go | 86 ++++++++++ pkg/constant/const.go | 5 + .../reconciler_instance_alignment.go | 74 +++----- .../reconciler_instance_alignment_test.go | 50 ++---- pkg/dataprotection/types/types.go | 7 +- 6 files changed, 294 insertions(+), 89 deletions(-) diff --git a/controllers/dataprotection/volumepopulator_controller.go b/controllers/dataprotection/volumepopulator_controller.go index 7eadf459f03..1ff17a7ec63 100644 --- a/controllers/dataprotection/volumepopulator_controller.go +++ b/controllers/dataprotection/volumepopulator_controller.go @@ -83,6 +83,7 @@ type pvcRestoreDecision struct { // +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims/status,verbs=get;update;patch // +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims/finalizers,verbs=update +// +kubebuilder:rbac:groups=core,resources=pods,verbs=get;create;delete // +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=components,verbs=get;list;watch // +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=componentdefinitions,verbs=get;list;watch @@ -1823,6 +1824,9 @@ func (r *VolumePopulatorReconciler) cleanupDeletingPVC(reqCtx intctrlutil.Reques // completion have been verified, or after cleanupDeletingPVC has stopped the // execution Restore. func (r *VolumePopulatorReconciler) releasePopulateResources(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error { + if err := r.deleteRestoreSchedulingPod(reqCtx, pvc); err != nil { + return err + } populatePVC := &corev1.PersistentVolumeClaim{} if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Name: getPopulatePVCName(pvc.UID), Namespace: pvc.Namespace}, populatePVC); err != nil { @@ -1874,14 +1878,139 @@ func (r *VolumePopulatorReconciler) waitForPVCSelectedNode(reqCtx intctrlutil.Re if storageClass.VolumeBindingMode != nil && storagev1.VolumeBindingWaitForFirstConsumer == *storageClass.VolumeBindingMode { nodeName = pvc.Annotations[AnnSelectedNode] if nodeName == "" { - // Wait for the PVC to get a node name before continuing + if err := r.ensureRestoreSchedulingPod(reqCtx, pvc); err != nil { + return false, nodeName, err + } + // The dedicated Pod is only a scheduler consumer. It has no + // InstanceSet/Component labels and cannot start database containers. return true, nodeName, nil } + if err := r.deleteRestoreSchedulingPod(reqCtx, pvc); err != nil { + return false, nodeName, err + } } } return false, nodeName, nil } +func (r *VolumePopulatorReconciler) ensureRestoreSchedulingPod(reqCtx intctrlutil.RequestCtx, + pvc *corev1.PersistentVolumeClaim) error { + pod := &corev1.Pod{} + key := types.NamespacedName{Namespace: pvc.Namespace, Name: getRestoreSchedulingPodName(pvc.UID)} + if err := r.Client.Get(reqCtx.Ctx, key, pod); err == nil { + return nil + } else if !apierrors.IsNotFound(err) { + return err + } + image := viper.GetString(constant.KBToolsImage) + if image == "" { + return fmt.Errorf("%s is empty; cannot create restore scheduling Pod for PVC %s/%s", + constant.KBToolsImage, pvc.Namespace, pvc.Name) + } + controller := true + schedulingSpec, err := r.restoreSchedulingPodSpec(reqCtx, pvc) + if err != nil { + return err + } + pod = &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: pvc.Namespace, + Name: key.Name, + OwnerReferences: []metav1.OwnerReference{{ + APIVersion: corev1.SchemeGroupVersion.String(), + Kind: constant.PersistentVolumeClaimKind, + Name: pvc.Name, + UID: pvc.UID, + Controller: &controller, + }}, + }, + Spec: corev1.PodSpec{ + NodeSelector: schedulingSpec.NodeSelector, + Affinity: schedulingSpec.Affinity, + Tolerations: schedulingSpec.Tolerations, + SchedulerName: schedulingSpec.SchedulerName, + PriorityClassName: schedulingSpec.PriorityClassName, + Priority: schedulingSpec.Priority, + PreemptionPolicy: schedulingSpec.PreemptionPolicy, + TopologySpreadConstraints: schedulingSpec.TopologySpreadConstraints, + RuntimeClassName: schedulingSpec.RuntimeClassName, + Overhead: schedulingSpec.Overhead, + RestartPolicy: corev1.RestartPolicyNever, + Containers: []corev1.Container{{ + Name: "scheduler", + Image: image, + ImagePullPolicy: corev1.PullPolicy(viper.GetString(constant.KBImagePullPolicy)), + Command: []string{"sh", "-c", "while true; do sleep 3600; done"}, + VolumeMounts: []corev1.VolumeMount{{ + Name: "target", + MountPath: "/var/lib/kubeblocks/restore-target", + }}, + }}, + Volumes: []corev1.Volume{{ + Name: "target", + VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ + ClaimName: pvc.Name, + }}, + }}, + }, + } + return client.IgnoreAlreadyExists(r.Client.Create(reqCtx.Ctx, pod)) +} + +func (r *VolumePopulatorReconciler) restoreSchedulingPodSpec(reqCtx intctrlutil.RequestCtx, + pvc *corev1.PersistentVolumeClaim) (*corev1.PodSpec, error) { + owner := metav1.GetControllerOf(pvc) + if owner == nil || owner.APIVersion != workloads.GroupVersion.String() { + return &corev1.PodSpec{}, nil + } + switch owner.Kind { + case workloads.InstanceSetKind: + its := &workloads.InstanceSet{} + if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Namespace: pvc.Namespace, Name: owner.Name}, its); err != nil { + return nil, err + } + itsExt, err := instancetemplate.BuildInstanceSetExt(its, nil) + if err != nil { + return nil, err + } + nameBuilder, err := instancetemplate.NewPodNameBuilder(itsExt, nil) + if err != nil { + return nil, err + } + nameTemplateMap, err := nameBuilder.BuildInstanceName2TemplateMap() + if err != nil { + return nil, err + } + template, ok := nameTemplateMap[pvc.Labels[constant.KBAppPodNameLabelKey]] + if !ok { + return nil, fmt.Errorf("PVC %s/%s does not identify an expected InstanceSet member", pvc.Namespace, pvc.Name) + } + return template.Spec.DeepCopy(), nil + case "Instance": + instance := &workloads.Instance{} + if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Namespace: pvc.Namespace, Name: owner.Name}, instance); err != nil { + return nil, err + } + return instance.Spec.Template.Spec.DeepCopy(), nil + default: + return &corev1.PodSpec{}, nil + } +} + +func (r *VolumePopulatorReconciler) deleteRestoreSchedulingPod(reqCtx intctrlutil.RequestCtx, + pvc *corev1.PersistentVolumeClaim) error { + pod := &corev1.Pod{} + key := types.NamespacedName{Namespace: pvc.Namespace, Name: getRestoreSchedulingPodName(pvc.UID)} + if err := r.Client.Get(reqCtx.Ctx, key, pod); err != nil { + return client.IgnoreNotFound(err) + } + return client.IgnoreNotFound(r.Client.Delete(reqCtx.Ctx, pod)) +} + +func getRestoreSchedulingPodName(pvcUID types.UID) string { + return fmt.Sprintf("kb-restore-scheduler-%s", pvcUID) +} + func (r *VolumePopulatorReconciler) getPopulatePVC(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim, backupSet dprestore.BackupActionSet, @@ -2085,11 +2214,27 @@ func (r *VolumePopulatorReconciler) UpdatePVCConditions(reqCtx intctrlutil.Reque Reason: reason, Message: message, } + dataReadyCondition := corev1.PersistentVolumeClaimCondition{ + Type: corev1.PersistentVolumeClaimConditionType(constant.RestoreDataReadyConditionType), + Status: corev1.ConditionUnknown, + LastTransitionTime: metav1.Now(), + Reason: reason, + Message: message, + } switch reason { case ReasonPopulatingSucceed, ReasonPopulatingProvisioned: restoreCondition.Status = corev1.ConditionTrue + dataReadyCondition.Status = corev1.ConditionTrue case ReasonPopulatingFailed: restoreCondition.Status = corev1.ConditionFalse + if existing := findPVCConditionByType(pvc, constant.RestoreDataReadyConditionType); existing != nil && + existing.Status == corev1.ConditionTrue { + // A postReady failure must not revoke the already verified + // prepareData contract or tear down a running workload. + dataReadyCondition = *existing + } else { + dataReadyCondition.Status = corev1.ConditionFalse + } } pvcPatch := client.MergeFrom(pvc.DeepCopy()) var existPopulating bool @@ -2098,7 +2243,8 @@ func (r *VolumePopulatorReconciler) UpdatePVCConditions(reqCtx intctrlutil.Reque continue } if reason == v.Reason { - if pvcConditionMatches(pvc.Status.Conditions, restoreCondition) { + if pvcConditionMatches(pvc.Status.Conditions, restoreCondition) && + pvcConditionMatches(pvc.Status.Conditions, dataReadyCondition) { return nil } existPopulating = true @@ -2107,7 +2253,8 @@ func (r *VolumePopulatorReconciler) UpdatePVCConditions(reqCtx intctrlutil.Reque } if v.Reason == ReasonPopulatingSucceed { // ignore succeed condition - if pvcConditionMatches(pvc.Status.Conditions, restoreCondition) { + if pvcConditionMatches(pvc.Status.Conditions, restoreCondition) && + pvcConditionMatches(pvc.Status.Conditions, dataReadyCondition) { return nil } existPopulating = true @@ -2120,6 +2267,7 @@ func (r *VolumePopulatorReconciler) UpdatePVCConditions(reqCtx intctrlutil.Reque pvc.Status.Conditions = append(pvc.Status.Conditions, progressCondition) } upsertPVCCondition(&pvc.Status.Conditions, restoreCondition) + upsertPVCCondition(&pvc.Status.Conditions, dataReadyCondition) switch reason { case ReasonPopulatingProcessing: r.Recorder.Event(pvc, corev1.EventTypeNormal, ReasonStartToVolumePopulate, message) @@ -2142,6 +2290,13 @@ func (r *VolumePopulatorReconciler) updatePVCPopulatingCondition(reqCtx intctrlu } pvcPatch := client.MergeFrom(pvc.DeepCopy()) upsertPVCCondition(&pvc.Status.Conditions, progressCondition) + upsertPVCCondition(&pvc.Status.Conditions, corev1.PersistentVolumeClaimCondition{ + Type: corev1.PersistentVolumeClaimConditionType(constant.RestoreDataReadyConditionType), + Status: corev1.ConditionTrue, + LastTransitionTime: metav1.Now(), + Reason: reason, + Message: message, + }) switch reason { case ReasonPopulatingSucceed, ReasonPopulatingProvisioned: r.Recorder.Event(pvc, corev1.EventTypeNormal, ReasonVolumePopulateSucceed, message) diff --git a/controllers/dataprotection/volumepopulator_controller_test.go b/controllers/dataprotection/volumepopulator_controller_test.go index 00d0dd98b2b..de6e27c3ae2 100644 --- a/controllers/dataprotection/volumepopulator_controller_test.go +++ b/controllers/dataprotection/volumepopulator_controller_test.go @@ -2721,6 +2721,92 @@ func TestPostReadyGatesIgnoreBoundButUnverifiedPVC(t *testing.T) { require.False(t, clusterReady) } +func TestWaitForPVCSelectedNodeUsesDedicatedSchedulingPod(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + require.NoError(t, storagev1.AddToScheme(scheme)) + mode := storagev1.VolumeBindingWaitForFirstConsumer + storageClassName := "wait-for-consumer" + storageClass := &storagev1.StorageClass{ + ObjectMeta: metav1.ObjectMeta{Name: storageClassName}, + Provisioner: "example.csi.k8s.io", + VolumeBindingMode: &mode, + } + pvc := newRestorePVCForSerialTest("data-mysql-0", "") + pvc.UID = types.UID("12345678-1234-1234-1234-123456789abc") + pvc.Spec.StorageClassName = &storageClassName + oldToolsImage := viper.GetString(constant.KBToolsImage) + t.Cleanup(func() { viper.Set(constant.KBToolsImage, oldToolsImage) }) + viper.Set(constant.KBToolsImage, "kubeblocks-tools:test") + reconciler := &VolumePopulatorReconciler{ + Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(storageClass, pvc).Build(), + Scheme: scheme, + } + reqCtx := intctrlutil.RequestCtx{Ctx: context.Background()} + + wait, nodeName, err := reconciler.waitForPVCSelectedNode(reqCtx, pvc) + require.NoError(t, err) + require.True(t, wait) + require.Empty(t, nodeName) + schedulingPod := &corev1.Pod{} + require.NoError(t, reconciler.Client.Get(context.Background(), types.NamespacedName{ + Namespace: pvc.Namespace, + Name: getRestoreSchedulingPodName(pvc.UID), + }, schedulingPod)) + require.Equal(t, "kubeblocks-tools:test", schedulingPod.Spec.Containers[0].Image) + require.Equal(t, pvc.Name, schedulingPod.Spec.Volumes[0].PersistentVolumeClaim.ClaimName) + require.Empty(t, schedulingPod.Labels[constant.AppInstanceLabelKey]) + require.Len(t, schedulingPod.OwnerReferences, 1) + require.Equal(t, pvc.UID, schedulingPod.OwnerReferences[0].UID) + + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + if currentPVC.Annotations == nil { + currentPVC.Annotations = map[string]string{} + } + currentPVC.Annotations[AnnSelectedNode] = "node-1" + require.NoError(t, reconciler.Client.Update(context.Background(), currentPVC)) + wait, nodeName, err = reconciler.waitForPVCSelectedNode(reqCtx, currentPVC) + require.NoError(t, err) + require.False(t, wait) + require.Equal(t, "node-1", nodeName) + err = reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(schedulingPod), &corev1.Pod{}) + require.True(t, apierrors.IsNotFound(err), err) +} + +func TestPVCDataReadyConditionContract(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + pvc := newRestorePVCForSerialTest("data-mysql-0", "") + reconciler := &VolumePopulatorReconciler{ + Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(pvc).Build(), + Recorder: record.NewFakeRecorder(10), + } + reqCtx := intctrlutil.RequestCtx{Ctx: context.Background()} + + require.NoError(t, reconciler.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, "preparing data")) + currentPVC := &corev1.PersistentVolumeClaim{} + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + dataReady := findPVCConditionByType(currentPVC, constant.RestoreDataReadyConditionType) + require.NotNil(t, dataReady) + require.Equal(t, corev1.ConditionUnknown, dataReady.Status) + + require.NoError(t, reconciler.updatePVCPopulatingCondition(reqCtx, currentPVC, ReasonPopulatingSucceed, "data ready")) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + dataReady = findPVCConditionByType(currentPVC, constant.RestoreDataReadyConditionType) + require.NotNil(t, dataReady) + require.Equal(t, corev1.ConditionTrue, dataReady.Status) + + require.NoError(t, reconciler.UpdatePVCConditions(reqCtx, currentPVC, ReasonPopulatingFailed, "postReady failed")) + require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) + dataReady = findPVCConditionByType(currentPVC, constant.RestoreDataReadyConditionType) + require.NotNil(t, dataReady) + require.Equal(t, corev1.ConditionTrue, dataReady.Status) + restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) + require.NotNil(t, restoreCondition) + require.Equal(t, corev1.ConditionFalse, restoreCondition.Status) +} + func newRestorePVCForSerialTest(name, volumeName string) *corev1.PersistentVolumeClaim { apiGroup := dptypes.DataprotectionAPIGroup return &corev1.PersistentVolumeClaim{ diff --git a/pkg/constant/const.go b/pkg/constant/const.go index 74ec22ad6e9..5062377f4ec 100644 --- a/pkg/constant/const.go +++ b/pkg/constant/const.go @@ -69,6 +69,11 @@ const InvalidContainerPort int32 = 0 const EmptyInsTemplateName = "" +// RestoreDataReadyConditionType is the PVC condition contract between the +// dataprotection and workload controllers. True means prepareData (or +// ProvisionOnly) is complete and workload containers may mount the target PVC. +const RestoreDataReadyConditionType = "RestoreDataReady" + type Key string // DryRunContextKey tells the KB Controllers to do dry-run reconciliations diff --git a/pkg/controller/instanceset/reconciler_instance_alignment.go b/pkg/controller/instanceset/reconciler_instance_alignment.go index 6c272d4e547..75a631efd35 100644 --- a/pkg/controller/instanceset/reconciler_instance_alignment.go +++ b/pkg/controller/instanceset/reconciler_instance_alignment.go @@ -20,7 +20,6 @@ along with this program. If not, see . package instanceset import ( - "github.com/spf13/viper" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/util/sets" @@ -32,7 +31,6 @@ import ( "github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx" "github.com/apecloud/kubeblocks/pkg/controller/model" intctrlutil "github.com/apecloud/kubeblocks/pkg/controllerutil" - dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types" ) // instanceAlignmentReconciler is responsible for aligning the actual instances(pods) with the desired replicas specified in the spec, @@ -42,12 +40,6 @@ import ( // TODO(free6om): support membership reconfiguration type instanceAlignmentReconciler struct{} -const ( - restoreBootstrapAnnotationKey = "workloads.kubeblocks.io/restore-bootstrap" - restoreBootstrapContainerName = "restore-bootstrap" - restoreBootstrapReadinessGate = corev1.PodConditionType("workloads.kubeblocks.io/restore-ready") -) - func NewReplicasAlignmentReconciler() kubebuilderx.Reconciler { return &instanceAlignmentReconciler{} } @@ -97,11 +89,9 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( pod, _ := object.(*corev1.Pod) oldInstanceMap[object.GetName()] = pod } - replacingNameSet := sets.New[string]() - // A Pod is needed to select a node for WaitForFirstConsumer storage. Before - // restore completion, use a bootstrap Pod whose init container prevents the - // database containers from starting. Replace it with the real Pod only after - // the populator has verified prepareData/provisioning completion. + // Remove a workload Pod if its restore PVC has not reached the explicit data + // readiness contract. The dataprotection controller uses a separate, + // non-workload scheduling Pod for WaitForFirstConsumer storage. for name, pod := range oldInstanceMap { template, desired := nameToTemplateMap[name] if !desired { @@ -111,8 +101,7 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( if err != nil { return kubebuilderx.Continue, err } - bootstrap := pod.Annotations[restoreBootstrapAnnotationKey] == "true" - if bootstrap == !restoreReady { + if restoreReady { continue } if err := tree.Delete(pod); err != nil { @@ -120,10 +109,9 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( } oldNameSet.Delete(name) delete(oldInstanceMap, name) - replacingNameSet.Insert(name) if tree.EventRecorder != nil { - tree.EventRecorder.Eventf(its, corev1.EventTypeNormal, - "Replacing restore bootstrap Pod %s after PVC population state changed", name) + tree.EventRecorder.Eventf(its, corev1.EventTypeWarning, + "Waiting for restore data readiness before starting Pod %s", name) } } createNameSet := newNameSet.Difference(oldNameSet) @@ -171,15 +159,6 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( if !isOrderedReady && concurrency <= 0 { break } - if replacingNameSet.Has(name) { - // Wait until the old Pod has actually disappeared before creating a - // replacement whose immutable Pod spec differs. - currentAlignedNameList = append(currentAlignedNameList, name) - if isOrderedReady { - break - } - continue - } predecessor := getPredecessor(i) if isOrderedReady && predecessor != nil && !intctrlutil.IsPodAvailable(predecessor, its.Spec.MinReadySeconds) { break @@ -188,14 +167,19 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( if err != nil { return kubebuilderx.Continue, err } + if !restoreReady { + // Align PVCs, but leave workload Pod creation gated. A dedicated + // dataprotection Pod handles WaitForFirstConsumer node selection. + currentAlignedNameList = append(currentAlignedNameList, name) + if isOrderedReady { + break + } + continue + } newPod, err := buildInstancePodByTemplate(name, nameToTemplateMap[name], its, "") if err != nil { return kubebuilderx.Continue, err } - if !restoreReady { - makeRestoreBootstrapPod(newPod) - } - if err := tree.Add(newPod); err != nil { return kubebuilderx.Continue, err } @@ -303,33 +287,21 @@ func instanceRestoreReadyForPod(tree *kubebuilderx.ObjectTree, if currentPVC.Spec.VolumeName == "" { return false, nil } - if !dptypes.IsPVCPopulationCompleted(currentPVC) { + if !pvcRestoreDataReady(currentPVC) { return false, nil } } return true, nil } -func makeRestoreBootstrapPod(pod *corev1.Pod) { - if pod.Annotations == nil { - pod.Annotations = map[string]string{} - } - pod.Annotations[restoreBootstrapAnnotationKey] = "true" - image := viper.GetString(constant.KBToolsImage) - if image == "" && len(pod.Spec.Containers) > 0 { - // Keep the Pod valid if the tools image configuration is missing. The - // command is intentionally blocking (or repeatedly failing if the image - // lacks a shell), so application containers still cannot start. - image = pod.Spec.Containers[0].Image +func pvcRestoreDataReady(pvc *corev1.PersistentVolumeClaim) bool { + for i := range pvc.Status.Conditions { + condition := &pvc.Status.Conditions[i] + if string(condition.Type) == constant.RestoreDataReadyConditionType { + return condition.Status == corev1.ConditionTrue + } } - pod.Spec.InitContainers = append(pod.Spec.InitContainers, corev1.Container{ - Name: restoreBootstrapContainerName, - Image: image, - Command: []string{"sh", "-c", "while true; do sleep 3600; done"}, - }) - pod.Spec.ReadinessGates = append(pod.Spec.ReadinessGates, corev1.PodReadinessGate{ - ConditionType: restoreBootstrapReadinessGate, - }) + return false } var _ kubebuilderx.Reconciler = &instanceAlignmentReconciler{} diff --git a/pkg/controller/instanceset/reconciler_instance_alignment_test.go b/pkg/controller/instanceset/reconciler_instance_alignment_test.go index fb6140fe3d2..e8305304ead 100644 --- a/pkg/controller/instanceset/reconciler_instance_alignment_test.go +++ b/pkg/controller/instanceset/reconciler_instance_alignment_test.go @@ -25,7 +25,6 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" - "github.com/spf13/viper" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" @@ -36,7 +35,6 @@ import ( "github.com/apecloud/kubeblocks/pkg/constant" "github.com/apecloud/kubeblocks/pkg/controller/builder" "github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx" - dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types" ) var _ = Describe("replicas alignment reconciler test", func() { @@ -185,10 +183,7 @@ var _ = Describe("replicas alignment reconciler test", func() { } }) - It("uses a scheduling bootstrap Pod until restore population is verified", func() { - oldToolsImage := viper.GetString(constant.KBToolsImage) - DeferCleanup(func() { viper.Set(constant.KBToolsImage, oldToolsImage) }) - viper.Set(constant.KBToolsImage, "kubeblocks-tools:test") + It("gates workload Pods on the restore data readiness condition", func() { replicas := int32(1) its.Spec.Replicas = &replicas its.Spec.VolumeClaimTemplates = []corev1.PersistentVolumeClaim{ @@ -207,49 +202,40 @@ var _ = Describe("replicas alignment reconciler test", func() { tree.SetRoot(its) reconciler = NewReplicasAlignmentReconciler() - By("creating the restore PVC and a blocked Pod for WaitForFirstConsumer scheduling") + By("creating the restore PVC without creating a workload Pod") res, err := reconciler.Reconcile(tree) Expect(err).Should(BeNil()) Expect(res).Should(Equal(kubebuilderx.Continue)) Expect(tree.List(&corev1.PersistentVolumeClaim{})).Should(HaveLen(1)) - Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) - bootstrapPod := tree.List(&corev1.Pod{})[0].(*corev1.Pod) - Expect(bootstrapPod.Annotations).Should(HaveKeyWithValue(restoreBootstrapAnnotationKey, "true")) - Expect(bootstrapPod.Spec.InitContainers).Should(ContainElement(Satisfy(func(container corev1.Container) bool { - return container.Name == restoreBootstrapContainerName && container.Image == "kubeblocks-tools:test" - }))) - Expect(bootstrapPod.Spec.ReadinessGates).Should(ContainElement(corev1.PodReadinessGate{ - ConditionType: restoreBootstrapReadinessGate, - })) - - By("keeping the database blocked when its PVC binds while prepareData is processing") + Expect(tree.List(&corev1.Pod{})).Should(BeEmpty()) + + By("removing an unsafe workload Pod while restore data is not ready") pvc := tree.List(&corev1.PersistentVolumeClaim{})[0].(*corev1.PersistentVolumeClaim) pvc.Spec.VolumeName = "empty-pv" pvc.Status.Conditions = []corev1.PersistentVolumeClaimCondition{{ - Type: dptypes.PersistentVolumeClaimPopulating, - Status: corev1.ConditionTrue, - Reason: "Processing", + Type: corev1.PersistentVolumeClaimConditionType(constant.RestoreDataReadyConditionType), + Status: corev1.ConditionUnknown, }} + instanceName := pvc.Labels[constant.KBAppPodNameLabelKey] + unsafePod := builder.NewPodBuilder(its.Namespace, instanceName). + AddContainer(corev1.Container{Name: "database", Image: "database:test"}). + GetObject() + Expect(tree.Add(unsafePod)).Should(Succeed()) _, err = reconciler.Reconcile(tree) Expect(err).Should(BeNil()) - Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) + Expect(tree.List(&corev1.Pod{})).Should(BeEmpty()) - By("deleting the bootstrap Pod after population completion is verified") + By("creating the real Pod only after the stable readiness condition is True") pvc = tree.List(&corev1.PersistentVolumeClaim{})[0].(*corev1.PersistentVolumeClaim) - pvc.Status.Conditions[0].Reason = dptypes.ReasonPopulatingSucceed + pvc.Status.Conditions[0].Status = corev1.ConditionTrue _, err = reconciler.Reconcile(tree) Expect(err).Should(BeNil()) - Expect(tree.List(&corev1.Pod{})).Should(BeEmpty()) + Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) - By("creating the real Pod after the bootstrap Pod is gone") - _, err = reconciler.Reconcile(tree) + By("surviving the update reconciler in the full reconcile chain") + _, err = NewUpdateReconciler().Reconcile(tree) Expect(err).Should(BeNil()) Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) - realPod := tree.List(&corev1.Pod{})[0].(*corev1.Pod) - Expect(realPod.Annotations).ShouldNot(HaveKey(restoreBootstrapAnnotationKey)) - Expect(realPod.Spec.InitContainers).ShouldNot(ContainElement(Satisfy(func(container corev1.Container) bool { - return container.Name == restoreBootstrapContainerName - }))) }) }) }) diff --git a/pkg/dataprotection/types/types.go b/pkg/dataprotection/types/types.go index 5463f0fe5b5..ea6ce414a91 100644 --- a/pkg/dataprotection/types/types.go +++ b/pkg/dataprotection/types/types.go @@ -21,10 +21,11 @@ package types import corev1 "k8s.io/api/core/v1" +const PersistentVolumeClaimPopulating corev1.PersistentVolumeClaimConditionType = "Populating" + const ( - PersistentVolumeClaimPopulating corev1.PersistentVolumeClaimConditionType = "Populating" - ReasonPopulatingSucceed = "Succeed" - ReasonPopulatingProvisioned = "Provisioned" + ReasonPopulatingSucceed = "Succeed" + ReasonPopulatingProvisioned = "Provisioned" ) var ( From bed333e409fe6c7a2420f794e9fa898b438f9574 Mon Sep 17 00:00:00 2001 From: Leon Date: Thu, 13 Aug 2026 15:42:22 +0800 Subject: [PATCH 5/7] refactor: centralize population condition constants --- controllers/dataprotection/types.go | 4 ++-- pkg/dataprotection/types/types.go | 9 +++++---- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/controllers/dataprotection/types.go b/controllers/dataprotection/types.go index 7b79d817ce2..6ee7359bd6f 100644 --- a/controllers/dataprotection/types.go +++ b/controllers/dataprotection/types.go @@ -87,8 +87,8 @@ const ( ReasonVolumePopulateFailed = "VolumePopulateFailed" // pvc condition type and reason - ReasonPopulatingFailed = "Failed" - ReasonPopulatingProcessing = "Processing" + ReasonPopulatingFailed = dptypes.ReasonPopulatingFailed + ReasonPopulatingProcessing = dptypes.ReasonPopulatingProcessing ReasonPopulatingSucceed = dptypes.ReasonPopulatingSucceed ReasonPopulatingProvisioned = dptypes.ReasonPopulatingProvisioned diff --git a/pkg/dataprotection/types/types.go b/pkg/dataprotection/types/types.go index ea6ce414a91..c2390877ac8 100644 --- a/pkg/dataprotection/types/types.go +++ b/pkg/dataprotection/types/types.go @@ -21,11 +21,12 @@ package types import corev1 "k8s.io/api/core/v1" -const PersistentVolumeClaimPopulating corev1.PersistentVolumeClaimConditionType = "Populating" - const ( - ReasonPopulatingSucceed = "Succeed" - ReasonPopulatingProvisioned = "Provisioned" + PersistentVolumeClaimPopulating corev1.PersistentVolumeClaimConditionType = "Populating" + ReasonPopulatingFailed string = "Failed" + ReasonPopulatingProcessing string = "Processing" + ReasonPopulatingSucceed string = "Succeed" + ReasonPopulatingProvisioned string = "Provisioned" ) var ( From f32b7aae539f9031ed80a721b3169fe882d5c3cb Mon Sep 17 00:00:00 2001 From: Leon Date: Thu, 13 Aug 2026 15:56:52 +0800 Subject: [PATCH 6/7] fix: preserve data source binding semantics --- .../volumepopulator_controller.go | 162 +----------------- .../volumepopulator_controller_test.go | 92 +--------- pkg/constant/const.go | 5 - .../reconciler_instance_alignment.go | 79 +-------- .../reconciler_instance_alignment_test.go | 56 ------ 5 files changed, 11 insertions(+), 383 deletions(-) diff --git a/controllers/dataprotection/volumepopulator_controller.go b/controllers/dataprotection/volumepopulator_controller.go index 1ff17a7ec63..260f8d10cad 100644 --- a/controllers/dataprotection/volumepopulator_controller.go +++ b/controllers/dataprotection/volumepopulator_controller.go @@ -83,7 +83,6 @@ type pvcRestoreDecision struct { // +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims/status,verbs=get;update;patch // +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims/finalizers,verbs=update -// +kubebuilder:rbac:groups=core,resources=pods,verbs=get;create;delete // +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=components,verbs=get;list;watch // +kubebuilder:rbac:groups=apps.kubeblocks.io,resources=componentdefinitions,verbs=get;list;watch @@ -1824,9 +1823,6 @@ func (r *VolumePopulatorReconciler) cleanupDeletingPVC(reqCtx intctrlutil.Reques // completion have been verified, or after cleanupDeletingPVC has stopped the // execution Restore. func (r *VolumePopulatorReconciler) releasePopulateResources(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) error { - if err := r.deleteRestoreSchedulingPod(reqCtx, pvc); err != nil { - return err - } populatePVC := &corev1.PersistentVolumeClaim{} if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Name: getPopulatePVCName(pvc.UID), Namespace: pvc.Namespace}, populatePVC); err != nil { @@ -1878,139 +1874,15 @@ func (r *VolumePopulatorReconciler) waitForPVCSelectedNode(reqCtx intctrlutil.Re if storageClass.VolumeBindingMode != nil && storagev1.VolumeBindingWaitForFirstConsumer == *storageClass.VolumeBindingMode { nodeName = pvc.Annotations[AnnSelectedNode] if nodeName == "" { - if err := r.ensureRestoreSchedulingPod(reqCtx, pvc); err != nil { - return false, nodeName, err - } - // The dedicated Pod is only a scheduler consumer. It has no - // InstanceSet/Component labels and cannot start database containers. + // Wait for the workload Pod to select a node. The target PVC remains + // unbound, so the Pod cannot start before population completes. return true, nodeName, nil } - if err := r.deleteRestoreSchedulingPod(reqCtx, pvc); err != nil { - return false, nodeName, err - } } } return false, nodeName, nil } -func (r *VolumePopulatorReconciler) ensureRestoreSchedulingPod(reqCtx intctrlutil.RequestCtx, - pvc *corev1.PersistentVolumeClaim) error { - pod := &corev1.Pod{} - key := types.NamespacedName{Namespace: pvc.Namespace, Name: getRestoreSchedulingPodName(pvc.UID)} - if err := r.Client.Get(reqCtx.Ctx, key, pod); err == nil { - return nil - } else if !apierrors.IsNotFound(err) { - return err - } - image := viper.GetString(constant.KBToolsImage) - if image == "" { - return fmt.Errorf("%s is empty; cannot create restore scheduling Pod for PVC %s/%s", - constant.KBToolsImage, pvc.Namespace, pvc.Name) - } - controller := true - schedulingSpec, err := r.restoreSchedulingPodSpec(reqCtx, pvc) - if err != nil { - return err - } - pod = &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{ - Namespace: pvc.Namespace, - Name: key.Name, - OwnerReferences: []metav1.OwnerReference{{ - APIVersion: corev1.SchemeGroupVersion.String(), - Kind: constant.PersistentVolumeClaimKind, - Name: pvc.Name, - UID: pvc.UID, - Controller: &controller, - }}, - }, - Spec: corev1.PodSpec{ - NodeSelector: schedulingSpec.NodeSelector, - Affinity: schedulingSpec.Affinity, - Tolerations: schedulingSpec.Tolerations, - SchedulerName: schedulingSpec.SchedulerName, - PriorityClassName: schedulingSpec.PriorityClassName, - Priority: schedulingSpec.Priority, - PreemptionPolicy: schedulingSpec.PreemptionPolicy, - TopologySpreadConstraints: schedulingSpec.TopologySpreadConstraints, - RuntimeClassName: schedulingSpec.RuntimeClassName, - Overhead: schedulingSpec.Overhead, - RestartPolicy: corev1.RestartPolicyNever, - Containers: []corev1.Container{{ - Name: "scheduler", - Image: image, - ImagePullPolicy: corev1.PullPolicy(viper.GetString(constant.KBImagePullPolicy)), - Command: []string{"sh", "-c", "while true; do sleep 3600; done"}, - VolumeMounts: []corev1.VolumeMount{{ - Name: "target", - MountPath: "/var/lib/kubeblocks/restore-target", - }}, - }}, - Volumes: []corev1.Volume{{ - Name: "target", - VolumeSource: corev1.VolumeSource{PersistentVolumeClaim: &corev1.PersistentVolumeClaimVolumeSource{ - ClaimName: pvc.Name, - }}, - }}, - }, - } - return client.IgnoreAlreadyExists(r.Client.Create(reqCtx.Ctx, pod)) -} - -func (r *VolumePopulatorReconciler) restoreSchedulingPodSpec(reqCtx intctrlutil.RequestCtx, - pvc *corev1.PersistentVolumeClaim) (*corev1.PodSpec, error) { - owner := metav1.GetControllerOf(pvc) - if owner == nil || owner.APIVersion != workloads.GroupVersion.String() { - return &corev1.PodSpec{}, nil - } - switch owner.Kind { - case workloads.InstanceSetKind: - its := &workloads.InstanceSet{} - if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Namespace: pvc.Namespace, Name: owner.Name}, its); err != nil { - return nil, err - } - itsExt, err := instancetemplate.BuildInstanceSetExt(its, nil) - if err != nil { - return nil, err - } - nameBuilder, err := instancetemplate.NewPodNameBuilder(itsExt, nil) - if err != nil { - return nil, err - } - nameTemplateMap, err := nameBuilder.BuildInstanceName2TemplateMap() - if err != nil { - return nil, err - } - template, ok := nameTemplateMap[pvc.Labels[constant.KBAppPodNameLabelKey]] - if !ok { - return nil, fmt.Errorf("PVC %s/%s does not identify an expected InstanceSet member", pvc.Namespace, pvc.Name) - } - return template.Spec.DeepCopy(), nil - case "Instance": - instance := &workloads.Instance{} - if err := r.Client.Get(reqCtx.Ctx, types.NamespacedName{Namespace: pvc.Namespace, Name: owner.Name}, instance); err != nil { - return nil, err - } - return instance.Spec.Template.Spec.DeepCopy(), nil - default: - return &corev1.PodSpec{}, nil - } -} - -func (r *VolumePopulatorReconciler) deleteRestoreSchedulingPod(reqCtx intctrlutil.RequestCtx, - pvc *corev1.PersistentVolumeClaim) error { - pod := &corev1.Pod{} - key := types.NamespacedName{Namespace: pvc.Namespace, Name: getRestoreSchedulingPodName(pvc.UID)} - if err := r.Client.Get(reqCtx.Ctx, key, pod); err != nil { - return client.IgnoreNotFound(err) - } - return client.IgnoreNotFound(r.Client.Delete(reqCtx.Ctx, pod)) -} - -func getRestoreSchedulingPodName(pvcUID types.UID) string { - return fmt.Sprintf("kb-restore-scheduler-%s", pvcUID) -} - func (r *VolumePopulatorReconciler) getPopulatePVC(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim, backupSet dprestore.BackupActionSet, @@ -2214,27 +2086,11 @@ func (r *VolumePopulatorReconciler) UpdatePVCConditions(reqCtx intctrlutil.Reque Reason: reason, Message: message, } - dataReadyCondition := corev1.PersistentVolumeClaimCondition{ - Type: corev1.PersistentVolumeClaimConditionType(constant.RestoreDataReadyConditionType), - Status: corev1.ConditionUnknown, - LastTransitionTime: metav1.Now(), - Reason: reason, - Message: message, - } switch reason { case ReasonPopulatingSucceed, ReasonPopulatingProvisioned: restoreCondition.Status = corev1.ConditionTrue - dataReadyCondition.Status = corev1.ConditionTrue case ReasonPopulatingFailed: restoreCondition.Status = corev1.ConditionFalse - if existing := findPVCConditionByType(pvc, constant.RestoreDataReadyConditionType); existing != nil && - existing.Status == corev1.ConditionTrue { - // A postReady failure must not revoke the already verified - // prepareData contract or tear down a running workload. - dataReadyCondition = *existing - } else { - dataReadyCondition.Status = corev1.ConditionFalse - } } pvcPatch := client.MergeFrom(pvc.DeepCopy()) var existPopulating bool @@ -2243,8 +2099,7 @@ func (r *VolumePopulatorReconciler) UpdatePVCConditions(reqCtx intctrlutil.Reque continue } if reason == v.Reason { - if pvcConditionMatches(pvc.Status.Conditions, restoreCondition) && - pvcConditionMatches(pvc.Status.Conditions, dataReadyCondition) { + if pvcConditionMatches(pvc.Status.Conditions, restoreCondition) { return nil } existPopulating = true @@ -2253,8 +2108,7 @@ func (r *VolumePopulatorReconciler) UpdatePVCConditions(reqCtx intctrlutil.Reque } if v.Reason == ReasonPopulatingSucceed { // ignore succeed condition - if pvcConditionMatches(pvc.Status.Conditions, restoreCondition) && - pvcConditionMatches(pvc.Status.Conditions, dataReadyCondition) { + if pvcConditionMatches(pvc.Status.Conditions, restoreCondition) { return nil } existPopulating = true @@ -2267,7 +2121,6 @@ func (r *VolumePopulatorReconciler) UpdatePVCConditions(reqCtx intctrlutil.Reque pvc.Status.Conditions = append(pvc.Status.Conditions, progressCondition) } upsertPVCCondition(&pvc.Status.Conditions, restoreCondition) - upsertPVCCondition(&pvc.Status.Conditions, dataReadyCondition) switch reason { case ReasonPopulatingProcessing: r.Recorder.Event(pvc, corev1.EventTypeNormal, ReasonStartToVolumePopulate, message) @@ -2290,13 +2143,6 @@ func (r *VolumePopulatorReconciler) updatePVCPopulatingCondition(reqCtx intctrlu } pvcPatch := client.MergeFrom(pvc.DeepCopy()) upsertPVCCondition(&pvc.Status.Conditions, progressCondition) - upsertPVCCondition(&pvc.Status.Conditions, corev1.PersistentVolumeClaimCondition{ - Type: corev1.PersistentVolumeClaimConditionType(constant.RestoreDataReadyConditionType), - Status: corev1.ConditionTrue, - LastTransitionTime: metav1.Now(), - Reason: reason, - Message: message, - }) switch reason { case ReasonPopulatingSucceed, ReasonPopulatingProvisioned: r.Recorder.Event(pvc, corev1.EventTypeNormal, ReasonVolumePopulateSucceed, message) diff --git a/controllers/dataprotection/volumepopulator_controller_test.go b/controllers/dataprotection/volumepopulator_controller_test.go index de6e27c3ae2..6782221bad8 100644 --- a/controllers/dataprotection/volumepopulator_controller_test.go +++ b/controllers/dataprotection/volumepopulator_controller_test.go @@ -25,6 +25,7 @@ import ( "errors" "fmt" "testing" + "time" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" @@ -474,6 +475,11 @@ var _ = Describe("Volume Populator Controller test", func() { g.Expect(restore.Status.Phase).Should(Equal(dpv1alpha1.RestorePhaseRunning)) })).Should(Succeed()) + By("keep the target PVC unbound until prepareData completes") + Consistently(testapps.CheckObj(&testCtx, pvcKey, func(g Gomega, targetPVC *corev1.PersistentVolumeClaim) { + g.Expect(targetPVC.Spec.VolumeName).Should(BeEmpty()) + }), time.Second).Should(Succeed()) + By("expect for job created") Eventually(testapps.List(&testCtx, generics.JobSignature, client.MatchingLabels{dprestore.DataProtectionRestoreLabelKey: populatePVCName}, @@ -2721,92 +2727,6 @@ func TestPostReadyGatesIgnoreBoundButUnverifiedPVC(t *testing.T) { require.False(t, clusterReady) } -func TestWaitForPVCSelectedNodeUsesDedicatedSchedulingPod(t *testing.T) { - scheme := runtime.NewScheme() - require.NoError(t, corev1.AddToScheme(scheme)) - require.NoError(t, storagev1.AddToScheme(scheme)) - mode := storagev1.VolumeBindingWaitForFirstConsumer - storageClassName := "wait-for-consumer" - storageClass := &storagev1.StorageClass{ - ObjectMeta: metav1.ObjectMeta{Name: storageClassName}, - Provisioner: "example.csi.k8s.io", - VolumeBindingMode: &mode, - } - pvc := newRestorePVCForSerialTest("data-mysql-0", "") - pvc.UID = types.UID("12345678-1234-1234-1234-123456789abc") - pvc.Spec.StorageClassName = &storageClassName - oldToolsImage := viper.GetString(constant.KBToolsImage) - t.Cleanup(func() { viper.Set(constant.KBToolsImage, oldToolsImage) }) - viper.Set(constant.KBToolsImage, "kubeblocks-tools:test") - reconciler := &VolumePopulatorReconciler{ - Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(storageClass, pvc).Build(), - Scheme: scheme, - } - reqCtx := intctrlutil.RequestCtx{Ctx: context.Background()} - - wait, nodeName, err := reconciler.waitForPVCSelectedNode(reqCtx, pvc) - require.NoError(t, err) - require.True(t, wait) - require.Empty(t, nodeName) - schedulingPod := &corev1.Pod{} - require.NoError(t, reconciler.Client.Get(context.Background(), types.NamespacedName{ - Namespace: pvc.Namespace, - Name: getRestoreSchedulingPodName(pvc.UID), - }, schedulingPod)) - require.Equal(t, "kubeblocks-tools:test", schedulingPod.Spec.Containers[0].Image) - require.Equal(t, pvc.Name, schedulingPod.Spec.Volumes[0].PersistentVolumeClaim.ClaimName) - require.Empty(t, schedulingPod.Labels[constant.AppInstanceLabelKey]) - require.Len(t, schedulingPod.OwnerReferences, 1) - require.Equal(t, pvc.UID, schedulingPod.OwnerReferences[0].UID) - - currentPVC := &corev1.PersistentVolumeClaim{} - require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) - if currentPVC.Annotations == nil { - currentPVC.Annotations = map[string]string{} - } - currentPVC.Annotations[AnnSelectedNode] = "node-1" - require.NoError(t, reconciler.Client.Update(context.Background(), currentPVC)) - wait, nodeName, err = reconciler.waitForPVCSelectedNode(reqCtx, currentPVC) - require.NoError(t, err) - require.False(t, wait) - require.Equal(t, "node-1", nodeName) - err = reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(schedulingPod), &corev1.Pod{}) - require.True(t, apierrors.IsNotFound(err), err) -} - -func TestPVCDataReadyConditionContract(t *testing.T) { - scheme := runtime.NewScheme() - require.NoError(t, corev1.AddToScheme(scheme)) - pvc := newRestorePVCForSerialTest("data-mysql-0", "") - reconciler := &VolumePopulatorReconciler{ - Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(pvc).Build(), - Recorder: record.NewFakeRecorder(10), - } - reqCtx := intctrlutil.RequestCtx{Ctx: context.Background()} - - require.NoError(t, reconciler.UpdatePVCConditions(reqCtx, pvc, ReasonPopulatingProcessing, "preparing data")) - currentPVC := &corev1.PersistentVolumeClaim{} - require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) - dataReady := findPVCConditionByType(currentPVC, constant.RestoreDataReadyConditionType) - require.NotNil(t, dataReady) - require.Equal(t, corev1.ConditionUnknown, dataReady.Status) - - require.NoError(t, reconciler.updatePVCPopulatingCondition(reqCtx, currentPVC, ReasonPopulatingSucceed, "data ready")) - require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) - dataReady = findPVCConditionByType(currentPVC, constant.RestoreDataReadyConditionType) - require.NotNil(t, dataReady) - require.Equal(t, corev1.ConditionTrue, dataReady.Status) - - require.NoError(t, reconciler.UpdatePVCConditions(reqCtx, currentPVC, ReasonPopulatingFailed, "postReady failed")) - require.NoError(t, reconciler.Client.Get(context.Background(), client.ObjectKeyFromObject(pvc), currentPVC)) - dataReady = findPVCConditionByType(currentPVC, constant.RestoreDataReadyConditionType) - require.NotNil(t, dataReady) - require.Equal(t, corev1.ConditionTrue, dataReady.Status) - restoreCondition := findPVCConditionByType(currentPVC, kbappsv1.ConditionTypeRestore) - require.NotNil(t, restoreCondition) - require.Equal(t, corev1.ConditionFalse, restoreCondition.Status) -} - func newRestorePVCForSerialTest(name, volumeName string) *corev1.PersistentVolumeClaim { apiGroup := dptypes.DataprotectionAPIGroup return &corev1.PersistentVolumeClaim{ diff --git a/pkg/constant/const.go b/pkg/constant/const.go index 5062377f4ec..74ec22ad6e9 100644 --- a/pkg/constant/const.go +++ b/pkg/constant/const.go @@ -69,11 +69,6 @@ const InvalidContainerPort int32 = 0 const EmptyInsTemplateName = "" -// RestoreDataReadyConditionType is the PVC condition contract between the -// dataprotection and workload controllers. True means prepareData (or -// ProvisionOnly) is complete and workload containers may mount the target PVC. -const RestoreDataReadyConditionType = "RestoreDataReady" - type Key string // DryRunContextKey tells the KB Controllers to do dry-run reconciliations diff --git a/pkg/controller/instanceset/reconciler_instance_alignment.go b/pkg/controller/instanceset/reconciler_instance_alignment.go index 75a631efd35..c3ee7289ea0 100644 --- a/pkg/controller/instanceset/reconciler_instance_alignment.go +++ b/pkg/controller/instanceset/reconciler_instance_alignment.go @@ -89,31 +89,6 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( pod, _ := object.(*corev1.Pod) oldInstanceMap[object.GetName()] = pod } - // Remove a workload Pod if its restore PVC has not reached the explicit data - // readiness contract. The dataprotection controller uses a separate, - // non-workload scheduling Pod for WaitForFirstConsumer storage. - for name, pod := range oldInstanceMap { - template, desired := nameToTemplateMap[name] - if !desired { - continue - } - restoreReady, err := instanceRestoreReadyForPod(tree, name, template, its) - if err != nil { - return kubebuilderx.Continue, err - } - if restoreReady { - continue - } - if err := tree.Delete(pod); err != nil { - return kubebuilderx.Continue, err - } - oldNameSet.Delete(name) - delete(oldInstanceMap, name) - if tree.EventRecorder != nil { - tree.EventRecorder.Eventf(its, corev1.EventTypeWarning, - "Waiting for restore data readiness before starting Pod %s", name) - } - } createNameSet := newNameSet.Difference(oldNameSet) deleteNameSet := oldNameSet.Difference(newNameSet) @@ -163,23 +138,11 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( if isOrderedReady && predecessor != nil && !intctrlutil.IsPodAvailable(predecessor, its.Spec.MinReadySeconds) { break } - restoreReady, err := instanceRestoreReadyForPod(tree, name, nameToTemplateMap[name], its) - if err != nil { - return kubebuilderx.Continue, err - } - if !restoreReady { - // Align PVCs, but leave workload Pod creation gated. A dedicated - // dataprotection Pod handles WaitForFirstConsumer node selection. - currentAlignedNameList = append(currentAlignedNameList, name) - if isOrderedReady { - break - } - continue - } newPod, err := buildInstancePodByTemplate(name, nameToTemplateMap[name], its, "") if err != nil { return kubebuilderx.Continue, err } + if err := tree.Add(newPod); err != nil { return kubebuilderx.Continue, err } @@ -264,44 +227,4 @@ func (r *instanceAlignmentReconciler) Reconcile(tree *kubebuilderx.ObjectTree) ( return kubebuilderx.Continue, nil } -func instanceRestoreReadyForPod(tree *kubebuilderx.ObjectTree, - instanceName string, - template *instancetemplate.InstanceTemplateExt, - its *workloads.InstanceSet) (bool, error) { - pvcs, err := buildInstancePVCByTemplate(instanceName, template, its) - if err != nil { - return false, err - } - for _, desiredPVC := range pvcs { - if desiredPVC.Spec.DataSourceRef == nil || desiredPVC.Annotations[constant.RestoreSourceKindAnnotationKey] == "" { - continue - } - current, err := tree.Get(desiredPVC) - if err != nil { - return false, err - } - if current == nil { - return false, nil - } - currentPVC := current.(*corev1.PersistentVolumeClaim) - if currentPVC.Spec.VolumeName == "" { - return false, nil - } - if !pvcRestoreDataReady(currentPVC) { - return false, nil - } - } - return true, nil -} - -func pvcRestoreDataReady(pvc *corev1.PersistentVolumeClaim) bool { - for i := range pvc.Status.Conditions { - condition := &pvc.Status.Conditions[i] - if string(condition.Type) == constant.RestoreDataReadyConditionType { - return condition.Status == corev1.ConditionTrue - } - } - return false -} - var _ kubebuilderx.Reconciler = &instanceAlignmentReconciler{} diff --git a/pkg/controller/instanceset/reconciler_instance_alignment_test.go b/pkg/controller/instanceset/reconciler_instance_alignment_test.go index e8305304ead..d0118434df8 100644 --- a/pkg/controller/instanceset/reconciler_instance_alignment_test.go +++ b/pkg/controller/instanceset/reconciler_instance_alignment_test.go @@ -32,7 +32,6 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" workloads "github.com/apecloud/kubeblocks/apis/workloads/v1" - "github.com/apecloud/kubeblocks/pkg/constant" "github.com/apecloud/kubeblocks/pkg/controller/builder" "github.com/apecloud/kubeblocks/pkg/controller/kubebuilderx" ) @@ -182,60 +181,5 @@ var _ = Describe("replicas alignment reconciler test", func() { } } }) - - It("gates workload Pods on the restore data readiness condition", func() { - replicas := int32(1) - its.Spec.Replicas = &replicas - its.Spec.VolumeClaimTemplates = []corev1.PersistentVolumeClaim{ - *its.Spec.VolumeClaimTemplates[0].DeepCopy(), - } - its.Spec.VolumeClaimTemplates[0].Annotations = map[string]string{ - constant.RestoreSourceKindAnnotationKey: "Backup", - } - apiGroup := "dataprotection.kubeblocks.io" - its.Spec.VolumeClaimTemplates[0].Spec.DataSourceRef = &corev1.TypedObjectReference{ - APIGroup: &apiGroup, - Kind: "Backup", - Name: "backup", - } - tree := kubebuilderx.NewObjectTree() - tree.SetRoot(its) - reconciler = NewReplicasAlignmentReconciler() - - By("creating the restore PVC without creating a workload Pod") - res, err := reconciler.Reconcile(tree) - Expect(err).Should(BeNil()) - Expect(res).Should(Equal(kubebuilderx.Continue)) - Expect(tree.List(&corev1.PersistentVolumeClaim{})).Should(HaveLen(1)) - Expect(tree.List(&corev1.Pod{})).Should(BeEmpty()) - - By("removing an unsafe workload Pod while restore data is not ready") - pvc := tree.List(&corev1.PersistentVolumeClaim{})[0].(*corev1.PersistentVolumeClaim) - pvc.Spec.VolumeName = "empty-pv" - pvc.Status.Conditions = []corev1.PersistentVolumeClaimCondition{{ - Type: corev1.PersistentVolumeClaimConditionType(constant.RestoreDataReadyConditionType), - Status: corev1.ConditionUnknown, - }} - instanceName := pvc.Labels[constant.KBAppPodNameLabelKey] - unsafePod := builder.NewPodBuilder(its.Namespace, instanceName). - AddContainer(corev1.Container{Name: "database", Image: "database:test"}). - GetObject() - Expect(tree.Add(unsafePod)).Should(Succeed()) - _, err = reconciler.Reconcile(tree) - Expect(err).Should(BeNil()) - Expect(tree.List(&corev1.Pod{})).Should(BeEmpty()) - - By("creating the real Pod only after the stable readiness condition is True") - pvc = tree.List(&corev1.PersistentVolumeClaim{})[0].(*corev1.PersistentVolumeClaim) - pvc.Status.Conditions[0].Status = corev1.ConditionTrue - _, err = reconciler.Reconcile(tree) - Expect(err).Should(BeNil()) - Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) - - By("surviving the update reconciler in the full reconcile chain") - _, err = NewUpdateReconciler().Reconcile(tree) - Expect(err).Should(BeNil()) - Expect(tree.List(&corev1.Pod{})).Should(HaveLen(1)) - }) }) }) From 31227d5acb2337590e492595cca37e7fda0e4c97 Mon Sep 17 00:00:00 2001 From: Leon Date: Thu, 13 Aug 2026 16:15:22 +0800 Subject: [PATCH 7/7] refactor: keep population state internal --- controllers/dataprotection/types.go | 12 ++--- .../volumepopulator_controller.go | 12 ++++- pkg/dataprotection/types/types.go | 28 ----------- pkg/dataprotection/types/types_test.go | 48 ------------------- 4 files changed, 17 insertions(+), 83 deletions(-) delete mode 100644 pkg/dataprotection/types/types_test.go diff --git a/controllers/dataprotection/types.go b/controllers/dataprotection/types.go index 6ee7359bd6f..b566dbce5f8 100644 --- a/controllers/dataprotection/types.go +++ b/controllers/dataprotection/types.go @@ -22,7 +22,7 @@ package dataprotection import ( "time" - dptypes "github.com/apecloud/kubeblocks/pkg/dataprotection/types" + corev1 "k8s.io/api/core/v1" ) const ( @@ -87,12 +87,12 @@ const ( ReasonVolumePopulateFailed = "VolumePopulateFailed" // pvc condition type and reason - ReasonPopulatingFailed = dptypes.ReasonPopulatingFailed - ReasonPopulatingProcessing = dptypes.ReasonPopulatingProcessing - ReasonPopulatingSucceed = dptypes.ReasonPopulatingSucceed - ReasonPopulatingProvisioned = dptypes.ReasonPopulatingProvisioned + ReasonPopulatingFailed = "Failed" + ReasonPopulatingProcessing = "Processing" + ReasonPopulatingSucceed = "Succeed" + ReasonPopulatingProvisioned = "Provisioned" - PersistentVolumeClaimPopulating = dptypes.PersistentVolumeClaimPopulating + PersistentVolumeClaimPopulating corev1.PersistentVolumeClaimConditionType = "Populating" ) var reconcileInterval = time.Second diff --git a/controllers/dataprotection/volumepopulator_controller.go b/controllers/dataprotection/volumepopulator_controller.go index 260f8d10cad..12c853ad2fd 100644 --- a/controllers/dataprotection/volumepopulator_controller.go +++ b/controllers/dataprotection/volumepopulator_controller.go @@ -1638,7 +1638,17 @@ func findPVCConditionByType(pvc *corev1.PersistentVolumeClaim, conditionType str } func pvcPopulateReleased(pvc *corev1.PersistentVolumeClaim) bool { - return dptypes.IsPVCPopulationCompleted(pvc) + if pvc == nil { + return false + } + for i := range pvc.Status.Conditions { + condition := &pvc.Status.Conditions[i] + if condition.Type != PersistentVolumeClaimPopulating || condition.Status != corev1.ConditionTrue { + continue + } + return condition.Reason == ReasonPopulatingSucceed || condition.Reason == ReasonPopulatingProvisioned + } + return false } func (r *VolumePopulatorReconciler) listRestorePVCsForComponent(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim) ([]corev1.PersistentVolumeClaim, error) { diff --git a/pkg/dataprotection/types/types.go b/pkg/dataprotection/types/types.go index c2390877ac8..3451155d990 100644 --- a/pkg/dataprotection/types/types.go +++ b/pkg/dataprotection/types/types.go @@ -19,35 +19,7 @@ along with this program. If not, see . package types -import corev1 "k8s.io/api/core/v1" - -const ( - PersistentVolumeClaimPopulating corev1.PersistentVolumeClaimConditionType = "Populating" - ReasonPopulatingFailed string = "Failed" - ReasonPopulatingProcessing string = "Processing" - ReasonPopulatingSucceed string = "Succeed" - ReasonPopulatingProvisioned string = "Provisioned" -) - var ( // DefaultBackOffLimit is the default backoff limit for jobs. DefaultBackOffLimit = int32(2) ) - -// IsPVCPopulationCompleted reports whether the volume populator has verified -// the prepareData/provisioning result and released the helper PVC. A bound PVC -// alone is not sufficient because it may have been bound before restore data -// was prepared. -func IsPVCPopulationCompleted(pvc *corev1.PersistentVolumeClaim) bool { - if pvc == nil { - return false - } - for i := range pvc.Status.Conditions { - condition := &pvc.Status.Conditions[i] - if condition.Type != PersistentVolumeClaimPopulating || condition.Status != corev1.ConditionTrue { - continue - } - return condition.Reason == ReasonPopulatingSucceed || condition.Reason == ReasonPopulatingProvisioned - } - return false -} diff --git a/pkg/dataprotection/types/types_test.go b/pkg/dataprotection/types/types_test.go deleted file mode 100644 index 777b92b2bb3..00000000000 --- a/pkg/dataprotection/types/types_test.go +++ /dev/null @@ -1,48 +0,0 @@ -/* -Copyright (C) 2022-2026 ApeCloud Co., Ltd - -This file is part of KubeBlocks project - -This program is free software: you can redistribute it and/or modify -it under the terms of the GNU Affero General Public License as published by -the Free Software Foundation, either version 3 of the License, or -(at your option) any later version. - -This program is distributed in the hope that it will be useful, -but WITHOUT ANY WARRANTY; without even the implied warranty of -MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -GNU Affero General Public License for more details. - -You should have received a copy of the GNU Affero General Public License -along with this program. If not, see . -*/ - -package types - -import ( - "testing" - - "github.com/stretchr/testify/require" - corev1 "k8s.io/api/core/v1" -) - -func TestIsPVCPopulationCompleted(t *testing.T) { - pvc := &corev1.PersistentVolumeClaim{} - require.False(t, IsPVCPopulationCompleted(nil)) - require.False(t, IsPVCPopulationCompleted(pvc)) - - pvc.Spec.VolumeName = "bound-before-restore" - pvc.Status.Conditions = []corev1.PersistentVolumeClaimCondition{{ - Type: PersistentVolumeClaimPopulating, - Status: corev1.ConditionTrue, - Reason: "Processing", - }} - require.False(t, IsPVCPopulationCompleted(pvc)) - - pvc.Status.Conditions[0].Reason = ReasonPopulatingSucceed - require.True(t, IsPVCPopulationCompleted(pvc)) - pvc.Status.Conditions[0].Reason = ReasonPopulatingProvisioned - require.True(t, IsPVCPopulationCompleted(pvc)) - pvc.Status.Conditions[0].Status = corev1.ConditionFalse - require.False(t, IsPVCPopulationCompleted(pvc)) -}