diff --git a/controllers/dataprotection/volumepopulator_controller.go b/controllers/dataprotection/volumepopulator_controller.go index 18877477e3b..9b1e4cfe04b 100644 --- a/controllers/dataprotection/volumepopulator_controller.go +++ b/controllers/dataprotection/volumepopulator_controller.go @@ -35,6 +35,7 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/tools/record" @@ -125,6 +126,9 @@ func (r *VolumePopulatorReconciler) handleSyncPVCError(reqCtx intctrlutil.Reques if requeueErr, ok := err.(intctrlutil.RequeueError); ok { return intctrlutil.RequeueAfter(requeueErr.RequeueAfter(), reqCtx.Log, requeueErr.Reason()) } + if apierrors.IsConflict(err) { + return intctrlutil.RequeueWithError(err, reqCtx.Log, "") + } if r.ContainPopulatingCondition(pvc) { // Ignore the error if an external controller handles this PVC. return intctrlutil.Reconciled() @@ -397,6 +401,17 @@ func (r *VolumePopulatorReconciler) decidePVCRestore(reqCtx intctrlutil.RequestC if restoreData && !skipPostReady { mode = pvcRestoreModeRestoreData } + // When no volume-level restore exists and the PVC belongs to the backup + // target component, postReady must not be blocked — it may be the only + // data delivery path (e.g. logical backup via dataLoad). Only clear + // skipPostReady for the target component; non-target component PVCs + // (e.g. pd/tikv in a TiDB backup) must stay skipped. + if !restoreData && skipPostReady { + if target := backupTargetForPVCComponent(backup, pvc); target != nil { + skipPostReady = false + sourceTarget = target + } + } return &pvcRestoreDecision{ mode: mode, sourceTarget: sourceTarget, @@ -468,6 +483,62 @@ func resolveSingleSourceTargetForPVC(target *dpv1alpha1.BackupStatusTarget, pvc return target, false, nil } +// backupTargetForPVCComponent returns the backup target whose component label +// matches the PVC's component. This is a fallback for when backupTargetMatchesPVC +// rejected the PVC due to pod-only labels (role, pod-name, etc.) that PVCs +// never carry. Returns nil if no target matches the PVC's component. +func backupTargetForPVCComponent(backup *dpv1alpha1.Backup, pvc *corev1.PersistentVolumeClaim) *dpv1alpha1.BackupStatusTarget { + pvcComp := pvc.Labels[constant.KBAppComponentLabelKey] + if pvcComp == "" { + return nil + } + if backup.Status.Target != nil { + if targetMatchesComponent(backup.Status.Target, pvcComp) { + return backup.Status.Target + } + return nil + } + for i := range backup.Status.Targets { + if targetMatchesComponent(&backup.Status.Targets[i], pvcComp) { + return &backup.Status.Targets[i] + } + } + return nil +} + +func targetMatchesComponent(target *dpv1alpha1.BackupStatusTarget, componentName string) bool { + if target.PodSelector == nil || target.PodSelector.LabelSelector == nil { + return true + } + componentSelector := componentLabelSelector(target.PodSelector.LabelSelector) + if componentSelector == nil { + return false + } + selector, err := metav1.LabelSelectorAsSelector(componentSelector) + if err != nil { + return false + } + return selector.Matches(labels.Set{constant.KBAppComponentLabelKey: componentName}) +} + +func componentLabelSelector(selector *metav1.LabelSelector) *metav1.LabelSelector { + componentSelector := &metav1.LabelSelector{} + if component, ok := selector.MatchLabels[constant.KBAppComponentLabelKey]; ok { + componentSelector.MatchLabels = map[string]string{ + constant.KBAppComponentLabelKey: component, + } + } + for _, expression := range selector.MatchExpressions { + if expression.Key == constant.KBAppComponentLabelKey { + componentSelector.MatchExpressions = append(componentSelector.MatchExpressions, expression) + } + } + if len(componentSelector.MatchLabels) == 0 && len(componentSelector.MatchExpressions) == 0 { + return nil + } + return componentSelector +} + func (r *VolumePopulatorReconciler) resolveShardingSourceTarget(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim, backup *dpv1alpha1.Backup) (*dpv1alpha1.BackupStatusTarget, bool, error) { @@ -905,7 +976,7 @@ func (r *VolumePopulatorReconciler) Populate(reqCtx intctrlutil.RequestCtx, pvc if err = r.patchInternalRestoreStatus(reqCtx, restoreMgr); err != nil { return err } - return nil + return intctrlutil.NewRequeueError(reconcileInterval, "waiting for prepareData job") } if actionFailed { dprestore.SetRestoreStageCondition(restoreMgr.Restore, dpv1alpha1.PrepareData, dprestore.ReasonFailed, "restore prepareData job failed") @@ -969,7 +1040,17 @@ func (r *VolumePopulatorReconciler) patchInternalRestoreStatus(reqCtx intctrluti if reflect.DeepEqual(restoreMgr.OriginalRestore.Status, restoreMgr.Restore.Status) { return nil } - return r.Client.Status().Patch(reqCtx.Ctx, restoreMgr.Restore, client.MergeFrom(restoreMgr.OriginalRestore)) + latest := &dpv1alpha1.Restore{} + if err := r.Client.Get(reqCtx.Ctx, client.ObjectKeyFromObject(restoreMgr.Restore), latest); err != nil { + return err + } + mergedStatus := mergeRestoreStatusChanges(restoreMgr.OriginalRestore.Status, restoreMgr.Restore.Status, latest.Status) + if reflect.DeepEqual(latest.Status, mergedStatus) { + return nil + } + statusPatch := client.MergeFrom(latest.DeepCopy()) + latest.Status = mergedStatus + return r.Client.Status().Patch(reqCtx.Ctx, latest, statusPatch) } func (r *VolumePopulatorReconciler) patchInternalRestoreMetaAndStatus(reqCtx intctrlutil.RequestCtx, @@ -987,10 +1068,33 @@ func (r *VolumePopulatorReconciler) patchInternalRestoreMetaAndStatus(reqCtx int return err } statusPatch := client.MergeFrom(latest.DeepCopy()) - latest.Status = restore.Status + latest.Status = mergeRestoreStatusChanges(original.Status, restore.Status, latest.Status) return r.Client.Status().Patch(reqCtx.Ctx, latest, statusPatch) } +func mergeRestoreStatusChanges(original, desired, latest dpv1alpha1.RestoreStatus) dpv1alpha1.RestoreStatus { + merged := latest + if !reflect.DeepEqual(original.Phase, desired.Phase) { + merged.Phase = desired.Phase + } + if !reflect.DeepEqual(original.StartTimestamp, desired.StartTimestamp) { + merged.StartTimestamp = desired.StartTimestamp + } + if !reflect.DeepEqual(original.CompletionTimestamp, desired.CompletionTimestamp) { + merged.CompletionTimestamp = desired.CompletionTimestamp + } + if !reflect.DeepEqual(original.Duration, desired.Duration) { + merged.Duration = desired.Duration + } + if !reflect.DeepEqual(original.Actions, desired.Actions) { + merged.Actions = desired.Actions + } + if !reflect.DeepEqual(original.Conditions, desired.Conditions) { + merged.Conditions = desired.Conditions + } + return merged +} + func (r *VolumePopulatorReconciler) completeBoundPVCIfNeeded(reqCtx intctrlutil.RequestCtx, pvc *corev1.PersistentVolumeClaim, restoreCtx *pvcRestoreContext) error { diff --git a/controllers/dataprotection/volumepopulator_controller_test.go b/controllers/dataprotection/volumepopulator_controller_test.go index 9853931410b..41f2cf7aae3 100644 --- a/controllers/dataprotection/volumepopulator_controller_test.go +++ b/controllers/dataprotection/volumepopulator_controller_test.go @@ -31,14 +31,17 @@ import ( batchv1 "k8s.io/api/batch/v1" corev1 "k8s.io/api/core/v1" storagev1 "k8s.io/api/storage/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/tools/record" "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" @@ -859,6 +862,125 @@ func TestDecidePVCRestoreTreatsNilTargetVolumesAsProvisionOnly(t *testing.T) { require.False(t, decision.skipPostReady) } +func TestDecidePVCRestoreNoTargetVolumesTargetComponent(t *testing.T) { + reconciler := &VolumePopulatorReconciler{} + backup := newBackupForRestoreDecision(nil, nil) + backup.Status.BackupMethod.TargetVolumes = nil + backup.Status.Target.PodSelector = &dpv1alpha1.PodSelector{ + LabelSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + constant.AppInstanceLabelKey: "source-cluster", + constant.KBAppComponentLabelKey: "tidb", + constant.RoleLabelKey: "leader", + }, + }, + Strategy: dpv1alpha1.PodSelectionStrategyAny, + } + + // PVC from the backup target component: postReady must not be skipped + pvc := newPVCForRestoreDecision("data", "tidb", "") + decision, err := reconciler.decidePVCRestore(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, backup, nil) + require.NoError(t, err) + require.Equal(t, pvcRestoreModeProvisionOnly, decision.mode) + require.False(t, decision.skipPostReady, + "target component PVC must not skip postReady when no volume-level restore exists") + require.NotNil(t, decision.sourceTarget, + "sourceTarget must be recovered for the target component") + + // PVC from a different component: postReady must be skipped + pvc = newPVCForRestoreDecision("data", "pd", "") + decision, err = reconciler.decidePVCRestore(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, backup, nil) + require.NoError(t, err) + require.Equal(t, pvcRestoreModeProvisionOnly, decision.mode) + require.True(t, decision.skipPostReady, + "non-target component PVC must skip postReady to avoid running restore on wrong component") + require.Nil(t, decision.sourceTarget) +} + +func TestDecidePVCRestoreNoTargetVolumesTargetComponentMatchExpression(t *testing.T) { + reconciler := &VolumePopulatorReconciler{} + backup := newBackupForRestoreDecision(nil, nil) + backup.Status.BackupMethod.TargetVolumes = nil + backup.Status.Target.PodSelector = &dpv1alpha1.PodSelector{ + LabelSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + constant.AppInstanceLabelKey: "source-cluster", + constant.RoleLabelKey: "leader", + }, + MatchExpressions: []metav1.LabelSelectorRequirement{{ + Key: constant.KBAppComponentLabelKey, + Operator: metav1.LabelSelectorOpIn, + Values: []string{"tidb"}, + }}, + }, + Strategy: dpv1alpha1.PodSelectionStrategyAny, + } + + pvc := newPVCForRestoreDecision("data", "tidb", "") + decision, err := reconciler.decidePVCRestore(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, backup, nil) + require.NoError(t, err) + require.Equal(t, pvcRestoreModeProvisionOnly, decision.mode) + require.False(t, decision.skipPostReady, + "target component selected by MatchExpressions must not skip postReady") + require.NotNil(t, decision.sourceTarget) + + pvc = newPVCForRestoreDecision("data", "pd", "") + decision, err = reconciler.decidePVCRestore(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, backup, nil) + require.NoError(t, err) + require.Equal(t, pvcRestoreModeProvisionOnly, decision.mode) + require.True(t, decision.skipPostReady, + "non-target component must still skip postReady when component expression does not match") + require.Nil(t, decision.sourceTarget) +} + +func TestDecidePVCRestoreNoTargetVolumesMultiComponentTargets(t *testing.T) { + reconciler := &VolumePopulatorReconciler{} + backup := newBackupForRestoreDecision(nil, []string{"etcd"}) + backup.Status.BackupMethod.TargetVolumes = nil + backup.Status.Targets[0].PodSelector = &dpv1alpha1.PodSelector{ + LabelSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + constant.AppInstanceLabelKey: "source-cluster", + constant.KBAppComponentLabelKey: "etcd", + constant.RoleLabelKey: "leader", + }, + }, + Strategy: dpv1alpha1.PodSelectionStrategyAny, + } + + // Etcd PVC matches target component — postReady enabled + pvc := newPVCForRestoreDecision("data", "etcd", "") + decision, err := reconciler.decidePVCRestore(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, backup, nil) + require.NoError(t, err) + require.Equal(t, pvcRestoreModeProvisionOnly, decision.mode) + require.False(t, decision.skipPostReady) + require.NotNil(t, decision.sourceTarget) + require.Equal(t, "etcd", decision.sourceTarget.Name) +} + +func TestDecidePVCRestoreNoTargetVolumesNoComponentLabel(t *testing.T) { + reconciler := &VolumePopulatorReconciler{} + backup := newBackupForRestoreDecision(nil, nil) + backup.Status.BackupMethod.TargetVolumes = nil + backup.Status.Target.PodSelector = &dpv1alpha1.PodSelector{ + LabelSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{ + constant.AppInstanceLabelKey: "source-cluster", + constant.RoleLabelKey: "leader", + }, + }, + Strategy: dpv1alpha1.PodSelectionStrategyAny, + } + + pvc := newPVCForRestoreDecision("data", "mysql", "") + decision, err := reconciler.decidePVCRestore(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, backup, nil) + require.NoError(t, err) + require.Equal(t, pvcRestoreModeProvisionOnly, decision.mode) + require.True(t, decision.skipPostReady, + "when target selector has no component label, cannot confirm PVC belongs to target — keep skipPostReady") + require.Nil(t, decision.sourceTarget) +} + func TestMatchToPopulateSupportsRestoreAndBackupDataSources(t *testing.T) { apiGroup := dptypes.DataprotectionAPIGroup reconciler := &VolumePopulatorReconciler{} @@ -1142,6 +1264,70 @@ func TestHandleSyncPVCErrorKeepsInternalRequeueWhenPVCIsPopulating(t *testing.T) require.Equal(t, reconcileInterval, result.RequeueAfter) } +func TestHandleSyncPVCErrorRequeuesConflictWhenPVCIsPopulating(t *testing.T) { + pvc := &corev1.PersistentVolumeClaim{ + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "target"}, + Status: corev1.PersistentVolumeClaimStatus{ + Conditions: []corev1.PersistentVolumeClaimCondition{{ + Type: PersistentVolumeClaimPopulating, + Status: corev1.ConditionTrue, + }}, + }, + } + reconciler := &VolumePopulatorReconciler{Recorder: record.NewFakeRecorder(10)} + conflict := apierrors.NewConflict( + schema.GroupResource{Resource: "persistentvolumeclaims"}, + pvc.Name, + fmt.Errorf("stale resource version"), + ) + + _, err := reconciler.handleSyncPVCError(intctrlutil.RequestCtx{Ctx: context.Background()}, pvc, conflict) + + require.Error(t, err) + require.True(t, apierrors.IsConflict(err), err.Error()) +} + +func TestPatchInternalRestoreStatusUsesLatestRestoreResourceVersion(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, dpv1alpha1.AddToScheme(scheme)) + latest := &dpv1alpha1.Restore{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "restore", + ResourceVersion: "42", + }, + Status: dpv1alpha1.RestoreStatus{Phase: dpv1alpha1.RestorePhaseAsDataSource}, + } + original := latest.DeepCopy() + original.ResourceVersion = "1" + original.Status = dpv1alpha1.RestoreStatus{} + var patchedResourceVersion string + cli := fake.NewClientBuilder(). + WithScheme(scheme). + WithStatusSubresource(latest). + WithObjects(latest). + WithInterceptorFuncs(interceptor.Funcs{ + SubResourcePatch: func(ctx context.Context, c client.Client, subResourceName string, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + require.Equal(t, "status", subResourceName) + patchedResourceVersion = obj.GetResourceVersion() + return c.SubResource(subResourceName).Patch(ctx, obj, patch, opts...) + }, + }). + Build() + reconciler := &VolumePopulatorReconciler{Client: cli} + restoreMgr := dprestore.NewRestoreManager(original.DeepCopy(), nil, scheme, cli) + dprestore.SetRestoreStageCondition(restoreMgr.Restore, dpv1alpha1.PrepareData, dprestore.ReasonSucceed, "prepare data successfully") + + err := reconciler.patchInternalRestoreStatus(intctrlutil.RequestCtx{Ctx: context.Background()}, restoreMgr) + + require.NoError(t, err) + require.Equal(t, "42", patchedResourceVersion) + got := &dpv1alpha1.Restore{} + require.NoError(t, cli.Get(context.Background(), types.NamespacedName{Namespace: "default", Name: "restore"}, got)) + require.Equal(t, dpv1alpha1.RestorePhaseAsDataSource, got.Status.Phase) + require.NotEmpty(t, got.Status.Conditions) +} + func TestDecidePVCRestoreAssignsShardingTargetsByStableComponentOrder(t *testing.T) { scheme := runtime.NewScheme() require.NoError(t, corev1.AddToScheme(scheme)) @@ -1256,6 +1442,126 @@ func TestProvisionOnlyCreatesPopulatePVCWithoutJob(t *testing.T) { require.Empty(t, jobs.Items) } +func TestPopulateRequeuesWhenRestoreContainerFinishedBeforeJobComplete(t *testing.T) { + 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-target-0", + UID: "target-pvc-uid", + ResourceVersion: "1", + }, + Spec: corev1.PersistentVolumeClaimSpec{ + AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce}, + Resources: corev1.VolumeResourceRequirements{ + Requests: corev1.ResourceList{corev1.ResourceStorage: resource.MustParse("1Gi")}, + }, + DataSourceRef: &corev1.TypedObjectReference{ + APIGroup: &apiGroup, + Kind: dptypes.BackupKind, + Name: "backup", + }, + }, + } + backup := newBackupForRestoreDecision([]string{"data"}, nil) + backup.Status.Target.PodSelector = &dpv1alpha1.PodSelector{Strategy: dpv1alpha1.PodSelectionStrategyAny} + actionSet := &dpv1alpha1.ActionSet{ + ObjectMeta: metav1.ObjectMeta{Name: "prepare-action"}, + Spec: dpv1alpha1.ActionSetSpec{ + BackupType: dpv1alpha1.BackupTypeFull, + Restore: &dpv1alpha1.RestoreActionSpec{ + PrepareData: &dpv1alpha1.JobActionSpec{ + BaseJobActionSpec: dpv1alpha1.BaseJobActionSpec{ + Image: "busybox", + Command: []string{"restore"}, + }, + }, + }, + }, + } + restore := &dpv1alpha1.Restore{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "internal-restore", + }, + Spec: dpv1alpha1.RestoreSpec{ + Backup: dpv1alpha1.BackupRef{Name: backup.Name, Namespace: backup.Namespace}, + PrepareDataConfig: &dpv1alpha1.PrepareDataConfig{ + DataSourceRef: &dpv1alpha1.VolumeConfig{ + VolumeSource: "data", + MountPath: "/data", + }, + }, + }, + } + populatePVCName := getPopulatePVCName(pvc.UID) + jobName := fmt.Sprintf("%s-%d", populatePVCName, 0) + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: jobName, + Labels: map[string]string{ + dprestore.DataProtectionRestoreLabelKey: restore.Name, + dprestore.DataProtectionPopulatePVCLabelKey: populatePVCName, + }, + }, + } + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: jobName + "-pod", + Labels: map[string]string{"job-name": jobName}, + }, + Status: corev1.PodStatus{ + ContainerStatuses: []corev1.ContainerStatus{ + { + Name: dprestore.Restore, + State: corev1.ContainerState{Terminated: &corev1.ContainerStateTerminated{ExitCode: 0}}, + }, + { + Name: "restore-manager", + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{}}, + }, + }, + }, + } + cli := fake.NewClientBuilder(). + WithScheme(scheme). + WithStatusSubresource(pvc, restore). + WithObjects(pvc, backup, actionSet, restore, job, pod). + Build() + reconciler := &VolumePopulatorReconciler{ + Client: cli, + Scheme: scheme, + Recorder: record.NewFakeRecorder(10), + } + restoreMgr := dprestore.NewRestoreManager(restore.DeepCopy(), nil, scheme, cli) + restoreMgr.PrepareDataBackupSets = []dprestore.BackupActionSet{{ + Backup: backup, + ActionSet: actionSet, + }} + + err := reconciler.Populate( + intctrlutil.RequestCtx{Ctx: context.Background()}, + pvc, + &pvcRestoreContext{restoreMgr: restoreMgr, mode: pvcRestoreModeRestoreData}, + ) + + require.Error(t, err) + require.True(t, intctrlutil.IsRequeueError(err), err.Error()) + currentPod := &corev1.Pod{} + require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(pod), currentPod)) + require.Equal(t, "true", currentPod.Annotations[dprestore.DataProtectionStopRestoreManagerAnnotationKey]) + currentRestore := &dpv1alpha1.Restore{} + require.NoError(t, cli.Get(context.Background(), client.ObjectKeyFromObject(restore), currentRestore)) + require.Len(t, currentRestore.Status.Actions.PrepareData, 1) + require.Equal(t, dpv1alpha1.RestoreActionProcessing, currentRestore.Status.Actions.PrepareData[0].Status) +} + func TestBuildPostReadyRestoreSelectsHighestPriorityRole(t *testing.T) { scheme := runtime.NewScheme() require.NoError(t, corev1.AddToScheme(scheme))